symfonic.memory.janitor_dedup

janitor_dedup

The graph-facing half of SemanticMerge: dedupe one node, or a whole layer.

Split out of :mod:symfonic.memory.janitor (330 lines against the 300-line budget). What stayed behind is the similarity decision -- find_duplicates and merge, which are pure functions of two nodes and a threshold. What moved here is the pass that talks to the graph store: reading candidates, writing merges back, and walking a whole layer.

SemanticMerge mixes this in, so both entry points are called as before.

SemanticDeduplicationMixin

Store-backed deduplication passes for :class:SemanticMerge.

deduplicate async

deduplicate(scope: TenantScope, graph_store: GraphMemoryStore, new_node: MemoryNode) -> MemoryNode

Check for duplicates and merge when found; otherwise return new_node.

If one or more duplicate nodes exist this method: 1. Merges new_node into the first duplicate (sequential merge for multiple duplicates would require an additional loop; a single merge covers the most common case). 2. Persists the merged node back to the graph via update_node. 3. Returns the canonical (merged) node.

If no duplicates are found the caller should persist new_node as a fresh graph record.

Parameters:

Name	Type	Description	Default
scope	TenantScope	Tenant context for graph queries.	required
graph_store	GraphMemoryStore	Live graph store to query and update.	required
new_node	MemoryNode	Incoming node that may duplicate an existing record.	required

Returns:

Type	Description
MemoryNode	The canonical MemoryNode (either the merged existing node or the
MemoryNode	unchanged new_node when no duplicate was found).

Source code in src/symfonic/memory/janitor_dedup.py

async def deduplicate(
    self,
    scope: TenantScope,
    graph_store: GraphMemoryStore,
    new_node: MemoryNode,
) -> MemoryNode:
    """Check for duplicates and merge when found; otherwise return *new_node*.

    If one or more duplicate nodes exist this method:
    1. Merges *new_node* into the first duplicate (sequential merge for
       multiple duplicates would require an additional loop; a single merge
       covers the most common case).
    2. Persists the merged node back to the graph via ``update_node``.
    3. Returns the canonical (merged) node.

    If no duplicates are found the caller should persist *new_node* as a
    fresh graph record.

    Args:
        scope: Tenant context for graph queries.
        graph_store: Live graph store to query and update.
        new_node: Incoming node that may duplicate an existing record.

    Returns:
        The canonical MemoryNode (either the merged existing node or the
        unchanged *new_node* when no duplicate was found).
    """
    existing_nodes = await graph_store.query_nodes(
        scope, layer=MemoryLayer.PROCEDURAL
    )
    duplicates = await self.find_duplicates(scope, existing_nodes, new_node)

    if not duplicates:
        return new_node

    existing = duplicates[0]
    if self._reviewed_canonical_blocks_auto_draft(existing, new_node):
        logger.info(
            "Kept reviewed procedural canonical %r unchanged (id=%s)",
            existing.label,
            existing.id,
        )
        return existing

    # Merge into the first (highest-priority) duplicate.
    canonical = await self.merge(existing, new_node)

    # Persist the merged state back to the graph.
    await graph_store.update_node(
        scope,
        canonical.id,
        {
            "label": canonical.label,
            "importance": canonical.importance,
            "properties": canonical.properties,
        },
    )
    logger.info(
        "Merged procedural node %r into canonical %r (id=%s)",
        new_node.label,
        canonical.label,
        canonical.id,
    )
    return canonical

deduplicate_all async

deduplicate_all(scope: TenantScope, graph_store: GraphMemoryStore) -> tuple[int, list[MemoryNode]]

Scan all procedural nodes for the tenant and merge duplicates.

This is the bulk operation exposed by the /procedures/deduplicate endpoint. It iterates over all nodes and greedily merges pairs whose similarity exceeds the threshold, then deletes the absorbed nodes.

Returns:

Type	Description
tuple[int, list[MemoryNode]]	A tuple of (merged_count, remaining_nodes).

Source code in src/symfonic/memory/janitor_dedup.py

async def deduplicate_all(
    self,
    scope: TenantScope,
    graph_store: GraphMemoryStore,
) -> tuple[int, list[MemoryNode]]:
    """Scan all procedural nodes for the tenant and merge duplicates.

    This is the bulk operation exposed by the ``/procedures/deduplicate``
    endpoint.  It iterates over all nodes and greedily merges pairs whose
    similarity exceeds the threshold, then deletes the absorbed nodes.

    Returns:
        A tuple of (merged_count, remaining_nodes).
    """
    nodes = await graph_store.query_nodes(scope, layer=MemoryLayer.PROCEDURAL)

    # Track which node IDs have already been absorbed so we skip them.
    absorbed_ids: set[str] = set()
    processed_ids: set[str] = set()
    merged_count = 0

    for i, primary in enumerate(nodes):
        if str(primary.id) in absorbed_ids or str(primary.id) in processed_ids:
            continue
        # v7.5 authored-tier guard: never use an authored node as the
        # primary that absorbs other nodes; ``find_duplicates`` already
        # filters authored candidates so the merge would only return
        # auto-promoted entries -- but absorbing them into a
        # hand-edited primary would mutate the authored properties.
        if self._is_authored(primary):
            continue
        candidates = [
            n for j, n in enumerate(nodes)
            if j > i and str(n.id) not in absorbed_ids
        ]
        duplicates = await self.find_duplicates(scope, candidates, primary)

        # Resolve review authority across the whole duplicate cluster
        # before performing any pairwise merge. Otherwise an automatic
        # draft that happens to precede the reviewed row can update another
        # draft first, churning durable state before review wins later.
        cluster = [primary, *duplicates]
        reviewed = next(
            (
                candidate
                for candidate in cluster
                if any(
                    self._reviewed_canonical_blocks_auto_draft(candidate, other)
                    for other in cluster
                    if str(other.id) != str(candidate.id)
                )
            ),
            None,
        )
        if reviewed is not None:
            automatic = [
                candidate
                for candidate in cluster
                if str(candidate.id) != str(reviewed.id)
                and self._reviewed_canonical_blocks_auto_draft(
                    reviewed, candidate
                )
            ]
            for draft in automatic:
                await graph_store.delete_node(scope, draft.id)
                absorbed_ids.add(str(draft.id))
                merged_count += 1
            primary = reviewed
            duplicates = [
                candidate
                for candidate in cluster
                if str(candidate.id) != str(reviewed.id)
                and str(candidate.id) not in absorbed_ids
            ]
        for dup in duplicates:
            # Human review chooses the canonical identity. Phase 12 emits
            # the same automatic draft again on every qualifying cycle,
            # and backend query order is not part of the review contract:
            # whether the reviewed row is ``primary`` or ``dup``, only
            # the automatic draft is removed and the reviewed row is not
            # updated (which would churn its review timestamp).
            if self._reviewed_canonical_blocks_auto_draft(primary, dup):
                await graph_store.delete_node(scope, dup.id)
                absorbed_ids.add(str(dup.id))
                merged_count += 1
                continue
            if self._reviewed_canonical_blocks_auto_draft(dup, primary):
                await graph_store.delete_node(scope, primary.id)
                absorbed_ids.add(str(primary.id))
                merged_count += 1
                primary = dup
                continue
            canonical = await self.merge(primary, dup)
            await graph_store.update_node(
                scope,
                canonical.id,
                {
                    "label": canonical.label,
                    "importance": canonical.importance,
                    "properties": canonical.properties,
                },
            )
            await graph_store.delete_node(scope, dup.id)
            absorbed_ids.add(str(dup.id))
            merged_count += 1
            # Advance the primary to the accumulated canonical: without this
            # each subsequent merge rebuilt from the ORIGINAL primary and the
            # final update_node overwrote earlier merges (dropped merged
            # steps/properties on 2+-way dedup). Round-2 fix.
            primary = canonical
            logger.info(
                "Bulk dedup: absorbed %r into %r", dup.label, canonical.label
            )
        processed_ids.add(str(primary.id))

    remaining = await graph_store.query_nodes(scope, layer=MemoryLayer.PROCEDURAL)
    return merged_count, remaining