symfonic.core.learning.phases

phases

Graph-based consolidation phase implementations.

Extracted from consolidation.py to keep each module under ~300 lines. Each function is a standalone async phase that operates on a GraphMemoryStore and returns a count of mutations performed.

Phases 1-5 live here (structural graph operations). Phases 8-9 (maintenance) live in phases_maintenance.py. Phase 10 (episodic summarization) lives in phases_episodic.py.

apply_soul_corrections

apply_soul_corrections(
    soul_schema: dict[str, Any],
    recent_nodes: list[MemoryNode],
) -> int

Apply user corrections from recent nodes to the SOUL schema.

Source code in src/symfonic/core/learning/phases.py

def apply_soul_corrections(
    soul_schema: dict[str, Any],
    recent_nodes: list[MemoryNode],
) -> int:
    """Apply user corrections from recent nodes to the SOUL schema."""
    updates = 0
    for node in recent_nodes:
        props = node.properties or {}
        if props.get("_last_edited_by") != "user_manual_edit":
            continue
        for key in ("name", "role", "personality", "language"):
            if key in props and key in soul_schema and soul_schema[key] != props[key]:
                soul_schema[key] = props[key]
                updates += 1
    return updates

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 src/symfonic/core/learning/phases.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 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:
            neighbors = await graph.get_neighbors(scope, src_id)
            neighbor_ids = {str(n.id) for n in neighbors}
            if str(tgt_id) in neighbor_ids:
                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,
            )
            await graph.add_edge(scope, edge)
            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 src/symfonic/core/learning/phases.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
    nodes_with_ts = [n for n in recent_nodes if n.updated_at is not None]

    # 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:
            neighbours = await graph.get_neighbors(scope, node_id, "CO_OCCURRED")
            co_degree[key] = len(neighbours)
        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 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,
                )
                result = await graph.upsert_edge(scope, edge)
                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: nodes with the META: prefix are excluded from the clustering input. Without this 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 (first-run meta nodes trigger a second-run meta of metas). Phase 4 is clustering RAW domain nodes; the existing META: nodes are outputs of this phase, not inputs to it.

Source code in src/symfonic/core/learning/phases.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: nodes with the ``META:`` prefix are excluded from the
    clustering input.  Without this 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 (first-run meta nodes trigger a second-run meta
    of metas).  Phase 4 is clustering RAW domain nodes; the existing
    ``META:`` nodes are outputs of this phase, not inputs to it.
    """
    clusters: dict[str, list[MemoryNode]] = defaultdict(list)
    for node in all_nodes:
        # Skip existing meta-nodes so we don't cluster outputs as inputs.
        if node.label and node.label.startswith("META:"):
            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 src/symfonic/core/learning/phases.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:
            neighbors = await graph.get_neighbors(scope, node.id)
            if len(neighbors) > ORPHAN_MAX_EDGES:
                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 src/symfonic/core/learning/phases.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 src/symfonic/core/learning/phases.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