symfonic.capabilities.memory.phases.pruning

pruning

Consolidation phases 2.5-4: reconcile pending edges, prune, roll up.

Split out of :mod:symfonic.core.learning.phases (439 lines against the 300-line budget), following the convention that module already established -- phases_profile, phases_maintenance and phases_episodic each hold a band of phases and say so in their docstring.

These three are the reconciling phases. The ones before them read the turn and add edges and attributes; these settle what that left behind -- pending edges become real ones or are dropped, stale low-confidence nodes go, and the survivors roll up into cluster meta-nodes. phases re-exports all three, plus the constants their tests import, so from symfonic.core.learning.phases import prune_orphans and its siblings are unchanged.

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/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

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 src/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 src/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