symfonic.memory.janitor_dedup

janitor_dedup

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

This mixin reads candidates, writes merges, and walks a whole layer. Similarity and pure merging remain in :mod:symfonic.memory.janitor.

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

    # Backend query order is not review authority.  Scan the whole
    # duplicate set before selecting a canonical: a sealed rejection must
    # discard a Phase-12 replay even when an ordinary duplicate happens to
    # be returned first.
    reviewed = next(
        (
            candidate for candidate in duplicates
            if self._reviewed_canonical_blocks_auto_draft(candidate, new_node)
        ),
        None,
    )
    if reviewed is not None:
        logger.info(
            "Kept reviewed procedural canonical %r unchanged (id=%s)",
            reviewed.label,
            reviewed.id,
        )
        return reviewed
    if (
        any(self._is_reviewed(candidate) for candidate in duplicates)
        and not self._is_automatic(new_node)
    ):
        # A manual/imported draft is independent evidence.  Similarity is
        # not authority to delete it beneath a reviewed terminal row.
        return new_node
    # Independent rows do not veto compaction of another compatible
    # Phase-12 replay in the same duplicate set.  For example, a backend
    # may return [manual draft, automatic draft] in either order: retain
    # the manual evidence but still merge a new automatic replay into the
    # automatic draft.  The same rule preserves legacy automatic
    # terminals beside drafts.  Only merge candidates survive this filter.
    duplicates = [
        candidate for candidate in duplicates
        if not self._automatic_terminal_and_draft_are_independent(candidate, new_node)
        and not self._automatic_and_manual_drafts_are_independent(candidate, new_node)
    ]
    if not duplicates:
        return new_node

    # Merge into the first (highest-priority) duplicate.
    existing = duplicates[0]
    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