Skip to content

symfonic.capabilities.memory.phases.deep_native_graph_cooccurrence

deep_native_graph_cooccurrence

Bounded native DEEP cooccurrence, with atomic endpoint and degree guards.

NativeCooccurrencePhase 

NativeCooccurrencePhase(graph: JournalledGraph, owner: DeepTenantOwner, port: NativeGraphMutationPort)

Link same-owner durable nodes updated within fifteen minutes.

Source code in src/symfonic/capabilities/memory/phases/deep_native_graph_cooccurrence.py 
33
34
35
36
37
38
def __init__(self, graph: JournalledGraph, owner: DeepTenantOwner,
             port: NativeGraphMutationPort) -> None:
    require_graph(graph, owner)
    if type(port) is NativeGraphMutationPort or not isinstance(port, NativeGraphMutationPort):
        raise MemoryContractError("native DEEP requires an implemented guarded edge port")
    self.graph, self.owner, self.port = graph, owner, port

run async 

run(context: PhaseContext) -> int

Return staged writes; receipt replay supplies the committed count.

A concurrent degree change can skip a staged edge at replay. Final reporting must use the journal's actual written count, not this value.

Source code in src/symfonic/capabilities/memory/phases/deep_native_graph_cooccurrence.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
async def run(self, context: PhaseContext) -> int:
    """Return staged writes; receipt replay supplies the committed count.

    A concurrent degree change can skip a staged edge at replay. Final
    reporting must use the journal's actual written count, not this value.
    """
    self.port.require_active(self.graph)
    selected = await candidates(self.graph, self.owner, context)
    nodes = []
    for candidate, scope in selected:
        result = await self.graph.durable.read_owned_node(scope, candidate.id)
        if result is None:
            raise MemoryContractError("native DEEP candidate changed before owned read")
        node, version = result
        if type(version) is not OwnedNodeVersion or exact_owner(node, self.owner) != scope:
            raise MemoryContractError("native DEEP candidate owner changed before edge")
        if (node.updated_at is None
                or node.layer in {MemoryLayer.WORKING, MemoryLayer.EPISODIC}
                or node.label.startswith(f"{ENTITY_LABEL_PREFIX}:")
                or is_pending(node.properties) or is_retracted(node.properties)):
            continue
        nodes.append((node, scope, version))
    changed = 0
    for index, (left, scope, left_version) in enumerate(nodes):
        for right, right_scope, right_version in nodes[index + 1:]:
            if scope != right_scope or left.id == right.id:
                continue
            if left.label.strip().casefold() == right.label.strip().casefold():
                continue
            if abs(left.updated_at - right.updated_at) > WINDOW:
                continue
            if await self.port.guarded_cooccurrence(
                scope, left.id, right.id, left_version, right_version,
                weight=0.3, max_degree=MAX_DEGREE,
            ):
                if changed >= MAX_WRITES:
                    raise MemoryContractError(
                        "native DEEP cooccurrence incomplete: eligible write budget exceeded"
                    )
                changed += 1
    return changed