symfonic.core.learning.phases

phases

The consolidation phases, now owned by the memory capability.

Every phase this module used to define lives in symfonic.capabilities.memory.phases and is imported back here for the legacy consolidator to call. A phase transcribed into two packages is two behaviours under one name, drifting until the copy an adopter runs is not the copy the tests exercise -- and the roster tables in capabilities.memory.rosters carry three findings that were exactly that mistake made in three places.

This module is the name the shipped consolidator and its tests already import.

consolidate_pending_edges async

consolidate_pending_edges(graph: GraphMemoryStore, scope: TenantScope, all_nodes: list[MemoryNode], pending_connections: list[dict[str, Any]]) -> int

Write inferred pending_connections as real graph edges.

Source code in symfonic/capabilities/memory/phases/pruning.py

async def consolidate_pending_edges(
    graph: GraphMemoryStore,
    scope: TenantScope,
    all_nodes: list[MemoryNode],
    pending_connections: list[dict[str, Any]],
) -> int:
    """Write inferred pending_connections as real graph edges."""
    label_to_id: dict[str, NodeId] = {n.label: n.id for n in all_nodes}
    created = 0

    for conn in pending_connections:
        src_label = conn.get("source", "")
        tgt_label = conn.get("target", "")
        rel = conn.get("relationship", "INFERRED")

        if not src_label or not tgt_label:
            logger.info(
                "Pending edge skip (missing labels): %r -> %r [%s]",
                src_label,
                tgt_label,
                rel,
            )
            continue

        src_id = label_to_id.get(src_label)
        tgt_id = label_to_id.get(tgt_label)
        if callable(getattr(type(graph), "related_candidates", None)):
            for label in (src_label, tgt_label):
                if label not in label_to_id:
                    matches = await graph.query_nodes(scope, label=label)
                    if matches:
                        label_to_id[label] = matches[0].id
            src_id, tgt_id = label_to_id.get(src_label), label_to_id.get(tgt_label)

        if src_id is None or tgt_id is None:
            logger.info(
                "Pending edge skip (node not found): %r (found=%s) -> %r (found=%s) [%s]",
                src_label,
                src_id is not None,
                tgt_label,
                tgt_id is not None,
                rel,
            )
            continue

        try:
            from symfonic.capabilities.memory.phases.adjacency import neighbor_probe
            if await neighbor_probe(graph, scope, src_id, target=str(tgt_id)):
                logger.info(
                    "Pending edge skip (edge exists): %s -> %s [%s]",
                    src_label,
                    tgt_label,
                    rel,
                )
                continue

            edge = MemoryEdge(
                source=src_id,
                target=tgt_id,
                relationship=rel,
                tenant_id=scope.tenant_id,
            )
            from symfonic.capabilities.memory.growth import edge_or_defer
            if await edge_or_defer(graph, scope, edge) is None:
                return created
            created += 1
            logger.info("Pending edge created: %s -> %s [%s]", src_label, tgt_label, rel)
        except Exception:
            logger.debug("Failed pending edge %s -> %s", src_label, tgt_label, exc_info=True)
    return created

create_cooccurrence_edges async

create_cooccurrence_edges(graph: GraphMemoryStore, scope: TenantScope, recent_nodes: list[MemoryNode]) -> int

Create CO_OCCURRED edges between nodes updated within 1 hour.

Weight reinforcement: repeated co-occurrence increments the edge weight rather than inserting a duplicate.

Degree cap: a node that already has >= MAX_CO_OCCUR_DEGREE CO_OCCURRED neighbours will not receive new CO_OCCURRED edges (existing edges still get reinforced).

Source code in symfonic/capabilities/memory/phases/structural.py

async def create_cooccurrence_edges(
    graph: GraphMemoryStore,
    scope: TenantScope,
    recent_nodes: list[MemoryNode],
) -> int:
    """Create CO_OCCURRED edges between nodes updated within 1 hour.

    Weight reinforcement: repeated co-occurrence increments the edge weight
    rather than inserting a duplicate.

    Degree cap: a node that already has >= MAX_CO_OCCUR_DEGREE CO_OCCURRED
    neighbours will not receive new CO_OCCURRED edges (existing edges still
    get reinforced).
    """
    window = timedelta(hours=CO_OCCUR_WINDOW_HOURS)
    created = 0
    # Conversation transcripts are evidence from which durable memories are
    # derived, not concepts in the knowledge graph. Linking both the working
    # and episodic copy creates edges between whole paragraphs (and between
    # two copies of the same sentence), producing a dense graph that cannot
    # improve recall. Entity linking separately reads episodics and emits
    # canonical entity relations; co-occurrence operates on durable nodes.
    nodes_with_ts = [
        n
        for n in recent_nodes
        if n.updated_at is not None
        and n.layer not in {MemoryLayer.WORKING, MemoryLayer.EPISODIC}
        # EntityLinker already gives canonical entities explicit MENTIONS
        # relationships. Treating its outputs as raw co-occurrence inputs on
        # the next cycle creates a redundant clique one cycle late, so a graph
        # that was stable after DEEP mutates on an unchanged replay.
        and not n.label.startswith(f"{ENTITY_LABEL_PREFIX}:")
    ]

    # Cache degree counts to avoid redundant neighbour fetches inside the loop.
    co_degree: dict[str, int] = {}

    async def _co_degree(node_id: NodeId) -> int:
        key = str(node_id)
        if key not in co_degree:
            from symfonic.capabilities.memory.phases.adjacency import neighbor_probe
            saturated = await neighbor_probe(
                graph, scope, node_id, minimum=MAX_CO_OCCUR_DEGREE, relationship="CO_OCCURRED"
            )
            co_degree[key] = MAX_CO_OCCUR_DEGREE if saturated else 0
        return co_degree[key]

    for i, a in enumerate(nodes_with_ts):
        for b in nodes_with_ts[i + 1 :]:
            assert a.updated_at is not None and b.updated_at is not None
            if a.label.strip().casefold() == b.label.strip().casefold():
                continue
            if abs((a.updated_at - b.updated_at).total_seconds()) > window.total_seconds():
                continue
            try:
                # Check degree cap for both endpoints before creating a new edge.
                deg_a = await _co_degree(a.id)
                if deg_a >= MAX_CO_OCCUR_DEGREE:
                    logger.info(
                        "CO_OCCURRED degree cap reached for %s (%d edges), skipping new edge to %s",
                        a.label,
                        deg_a,
                        b.label,
                    )
                    continue
                deg_b = await _co_degree(b.id)
                if deg_b >= MAX_CO_OCCUR_DEGREE:
                    logger.info(
                        "CO_OCCURRED degree cap reached for %s (%d edges), skipping new edge to %s",
                        b.label,
                        deg_b,
                        a.label,
                    )
                    continue

                edge = MemoryEdge(
                    source=a.id,
                    target=b.id,
                    relationship="CO_OCCURRED",
                    tenant_id=scope.tenant_id,
                    weight=0.3,
                )
                from symfonic.capabilities.memory.growth import edge_or_defer
                result = await edge_or_defer(graph, scope, edge, upsert=True)
                if result is None:
                    return created
                created += 1
                # Invalidate cached degree so subsequent pairs see the updated count.
                co_degree.pop(str(a.id), None)
                co_degree.pop(str(b.id), None)
                logger.info(
                    "CO_OCCURRED upsert: %s <-> %s (weight=%.1f)",
                    a.label,
                    b.label,
                    result.weight,
                )
            except Exception:
                logger.debug("Failed CO_OCCURRED %s <-> %s", a.id, b.id, exc_info=True)
    return created

generate_meta_nodes async

generate_meta_nodes(graph: GraphMemoryStore, scope: TenantScope, all_nodes: list[MemoryNode], llm_summarise: Any | None = None) -> int

Create meta-nodes from clusters of related nodes.

Idempotence: generated META: and Entity: nodes are excluded from the clustering input. Without the first filter, a second consolidation run groups every META:foo/META:bar node under the shared prefix META and emits a META:META cascade node -- correct but cosmetically noisy. Without the second, every entity minted late in DEEP's roster groups under META:Entity on the next unchanged cycle. Phase 4 clusters raw domain nodes; both kinds are phase outputs, not inputs.

Source code in symfonic/capabilities/memory/phases/pruning.py

async def generate_meta_nodes(
    graph: GraphMemoryStore,
    scope: TenantScope,
    all_nodes: list[MemoryNode],
    llm_summarise: Any | None = None,
) -> int:
    """Create meta-nodes from clusters of related nodes.

    Idempotence: generated ``META:`` and ``Entity:`` nodes are excluded from
    the clustering input. Without the first filter, a second consolidation run
    groups every ``META:foo``/``META:bar`` node under the shared prefix
    ``META`` and emits a ``META:META`` cascade node -- correct but
    cosmetically noisy. Without the second, every entity minted late in DEEP's
    roster groups under ``META:Entity`` on the next unchanged cycle. Phase 4
    clusters raw domain nodes; both kinds are phase outputs, not inputs.
    """
    clusters: dict[str, list[MemoryNode]] = defaultdict(list)
    for node in all_nodes:
        # Working and episodic rows are transcript evidence. A meta-node over
        # messages that merely begin with the same word ("The", "La", "Lo")
        # creates a fluent-looking but meaningless cluster and makes one
        # conversation's prose look like durable knowledge.
        if node.layer in {MemoryLayer.WORKING, MemoryLayer.EPISODIC}:
            continue
        # Skip existing meta-nodes so we don't cluster outputs as inputs.
        if node.label and node.label.startswith("META:"):
            continue
        if node.label and node.label.startswith(f"{ENTITY_LABEL_PREFIX}:"):
            continue
        if not node.label:
            prefix = ""
        elif ":" in node.label:
            prefix = node.label.split(":")[0].strip()
        else:
            prefix = node.label.split()[0]
        if prefix:
            clusters[prefix].append(node)

    created = 0
    for prefix, members in clusters.items():
        if len(members) < MIN_CLUSTER_SIZE:
            continue
        existing = await graph.query_nodes(scope, label=f"META:{prefix}")
        if existing:
            continue

        member_labels = [m.label for m in members[:10]]
        if llm_summarise is not None:
            try:
                summary = await llm_summarise(member_labels)
            except Exception:
                summary = f"Cluster of {len(members)} nodes: {', '.join(member_labels[:5])}"
        else:
            summary = f"Cluster of {len(members)} nodes: {', '.join(member_labels[:5])}"

        meta_node = MemoryNode(
            layer=MemoryLayer.SEMANTIC,
            tenant_id=scope.tenant_id,
            label=f"META:{prefix}",
            properties={
                "type": "meta_node",
                "member_count": len(members),
                "summary": summary,
                "source_labels": member_labels[:10],
            },
            importance=7.0,
        )
        try:
            added = await graph.add_node(scope, meta_node)
            logger.info("Meta-node created: META:%s covering %d members", prefix, len(members))
            for member in members[:10]:
                edge = MemoryEdge(
                    source=added.id,
                    target=member.id,
                    relationship="CLUSTERS",
                    tenant_id=scope.tenant_id,
                )
                await graph.add_edge(scope, edge)
            created += 1
        except Exception:
            logger.debug("Failed to create meta-node for %s", prefix, exc_info=True)
    return created

prune_orphans async

prune_orphans(graph: GraphMemoryStore, scope: TenantScope, all_nodes: list[MemoryNode]) -> int

Delete orphan nodes: few edges, stale, low confidence, or TTL-expired.

Source code in symfonic/capabilities/memory/phases/pruning.py

async def prune_orphans(
    graph: GraphMemoryStore,
    scope: TenantScope,
    all_nodes: list[MemoryNode],
) -> int:
    """Delete orphan nodes: few edges, stale, low confidence, or TTL-expired."""
    stale_cutoff = datetime.now(UTC) - timedelta(days=ORPHAN_STALE_DAYS)
    pruned = 0
    for node in all_nodes:
        # TTL-expired nodes are pruned unconditionally regardless of importance.
        if _is_ttl_expired(node):
            try:
                await graph.delete_node(scope, node.id)
                pruned += 1
                logger.info("TTL-expired node pruned: %s", node.id)
            except Exception:
                logger.debug("Failed to prune TTL-expired node %s", node.id, exc_info=True)
            continue

        if node.importance >= (ORPHAN_MIN_CONFIDENCE * 10):
            continue
        if node.updated_at and node.updated_at > stale_cutoff:
            continue
        try:
            from symfonic.capabilities.memory.phases.adjacency import neighbor_probe
            if await neighbor_probe(graph, scope, node.id, minimum=ORPHAN_MAX_EDGES + 1):
                continue
            await graph.delete_node(scope, node.id)
            pruned += 1
        except Exception:
            logger.debug("Failed to prune orphan %s", node.id, exc_info=True)
    return pruned

strengthen async

strengthen(graph: GraphMemoryStore, scope: TenantScope, recent_nodes: list[MemoryNode], *, spreading_weight: float = 0.5) -> int

Increment importance on recently accessed (recurring) nodes.

A node qualifies as "recurring" when EITHER:

1. It appears >= 2 times in recent_nodes (duplicate-in-list signal). This covers access-log replay / neighbourhood traversal callers that intentionally feed per-hit records into the consolidator.

2. Its combined recurrence score access_count + spreading_weight * spreading_access_count >= 2.

3. access_count is bumped by GraphMemoryStore.get_node on every direct fetch (one-shot reads).

4. spreading_access_count is bumped by GraphMemoryStore.bump_spreading on every BFS-induced visit (spreading activation through GraphTraversal.bfs).

spreading_weight (default 0.5) comes from FrameworkConfig.phase1_spreading_weight. At 0.5 the combined score matches v6.1.x observable behaviour: two BFS visits count like one direct fetch. Setting the weight to 0.0 yields the direct-only semantic from the user task description.

Before the v6.0.x fix, Phase 1 silently never fired in production: GraphMemoryStore.query_nodes returns each node exactly once, so the duplicate-in-list count was always 1 and the < 2 guard always skipped. Falling back to the per-node access_count surfaces the access-recurrence signal already captured at retrieval time without requiring a new DB schema or access log.

The count semantics of the list-duplicate path are preserved: a node that appears N>=2 times in the list still gets N successful update attempts (each subsequent read loads the freshly-boosted value).

ADR (v6.1 T02/T03, revised v6.2 T02): the canonical recurrence signal is the combined access_count + weighted spreading_access_count. v6.1 aligned spreading-activation reads with the single access_count counter by routing GraphTraversal.bfs through store.get_node. v6.2 decomposed the counter so callers can distinguish direct fetches from one-shot spreading visits. BFS now bumps spreading_access_count via GraphMemoryStore.bump_spreading. Intentionally excluded from bumping: query_nodes (returns candidate sets for scoring -- one query bumping N nodes would distort the frequency signal that scoring.py also reads), and shortest_path / graph_proximity (run inside the scoring loop itself -- same contamination risk).

Source code in symfonic/capabilities/memory/phases/strengthen.py

async def strengthen(
    graph: GraphMemoryStore,
    scope: TenantScope,
    recent_nodes: list[MemoryNode],
    *,
    spreading_weight: float = 0.5,
) -> int:
    """Increment importance on recently accessed (recurring) nodes.

    A node qualifies as "recurring" when EITHER:

    1. It appears >= 2 times in ``recent_nodes`` (duplicate-in-list signal).
       This covers access-log replay / neighbourhood traversal callers that
       intentionally feed per-hit records into the consolidator.

    2. Its combined recurrence score
       ``access_count + spreading_weight * spreading_access_count >= 2``.

       - ``access_count`` is bumped by ``GraphMemoryStore.get_node`` on
         every direct fetch (one-shot reads).
       - ``spreading_access_count`` is bumped by
         ``GraphMemoryStore.bump_spreading`` on every BFS-induced visit
         (spreading activation through ``GraphTraversal.bfs``).

       ``spreading_weight`` (default 0.5) comes from
       ``FrameworkConfig.phase1_spreading_weight``. At 0.5 the combined
       score matches v6.1.x observable behaviour: two BFS visits count
       like one direct fetch. Setting the weight to 0.0 yields the
       direct-only semantic from the user task description.

    Before the v6.0.x fix, Phase 1 silently never fired in production:
    ``GraphMemoryStore.query_nodes`` returns each node exactly once, so
    the duplicate-in-list count was always 1 and the ``< 2`` guard always
    skipped. Falling back to the per-node ``access_count`` surfaces the
    access-recurrence signal already captured at retrieval time without
    requiring a new DB schema or access log.

    The count semantics of the list-duplicate path are preserved: a node
    that appears N>=2 times in the list still gets N successful update
    attempts (each subsequent read loads the freshly-boosted value).

    ADR (v6.1 T02/T03, revised v6.2 T02): the canonical recurrence signal
    is the combined ``access_count`` + weighted ``spreading_access_count``.
    v6.1 aligned spreading-activation reads with the single ``access_count``
    counter by routing ``GraphTraversal.bfs`` through ``store.get_node``.
    v6.2 decomposed the counter so callers can distinguish direct fetches
    from one-shot spreading visits. BFS now bumps
    ``spreading_access_count`` via ``GraphMemoryStore.bump_spreading``.
    Intentionally excluded from bumping: ``query_nodes`` (returns
    candidate sets for scoring -- one query bumping N nodes would distort
    the frequency signal that scoring.py also reads), and
    ``shortest_path`` / ``graph_proximity`` (run inside the scoring
    loop itself -- same contamination risk).
    """
    list_counts: Counter[str] = Counter()
    for node in recent_nodes:
        list_counts[str(node.id)] += 1

    strengthened = 0
    # Deduplicate while preserving iteration count for list-duplicate nodes:
    # nodes passed multiple times in the list are strengthened per-iteration
    # (legacy behaviour). Nodes passed ONCE are strengthened when their
    # combined access score signals recurrence.
    for node in recent_nodes:
        nid = str(node.id)
        recurs_in_list = list_counts[nid] >= 2
        spread = getattr(node, "spreading_access_count", 0) or 0
        combined = node.access_count + spreading_weight * spread
        recurs_via_access = combined >= 2
        if not (recurs_in_list or recurs_via_access):
            continue
        new_importance = min(node.importance + IMPORTANCE_BOOST, MAX_IMPORTANCE)
        if new_importance == node.importance:
            continue
        try:
            await graph.update_node(scope, node.id, {"importance": new_importance})
            strengthened += 1
        except Exception:
            logger.debug("Failed to strengthen node %s", nid, exc_info=True)
    return strengthened

tag_risk_nodes async

tag_risk_nodes(graph: GraphMemoryStore, scope: TenantScope, recent_nodes: list[MemoryNode]) -> int

Mark nodes with negative context as RISK_NODE.

Source code in symfonic/capabilities/memory/phases/structural.py

async def tag_risk_nodes(
    graph: GraphMemoryStore,
    scope: TenantScope,
    recent_nodes: list[MemoryNode],
) -> int:
    """Mark nodes with negative context as RISK_NODE."""
    tagged = 0
    for node in recent_nodes:
        props = node.properties or {}
        text = " ".join(str(v).lower() for v in [node.label, *props.values()] if isinstance(v, str))
        if not any(marker in text for marker in NEGATIVE_MARKERS):
            continue
        if props.get("risk_node"):
            continue
        try:
            new_props = {
                **props,
                "risk_node": True,
                "risk_tagged_at": datetime.now(UTC).isoformat(),
            }
            await graph.update_node(scope, node.id, {"properties": new_props})
            tagged += 1
        except Exception:
            logger.debug("Failed to tag risk node %s", node.id, exc_info=True)
    return tagged