symfonic.core.learning.phases_pruning

phases_pruning

Moved to :mod:symfonic.capabilities.memory.phases.pruning.

These phases are on a roster the kernel composes, so they live in the capability now and are imported back here for the legacy consolidator to call. One phase, one implementation, for as long as both routes ship.

The private helpers travel with them: the shipped per-phase tests import them by name, and a shim that re-exported only the public surface would have moved the code away from the tests that watch it.

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