symfonic.memory.janitor

janitor

SemanticMerge -- deduplicates procedural memory nodes by semantic similarity.

Uses difflib.SequenceMatcher for label/content comparison so no additional dependencies are required. The default threshold of 0.65 captures cross-language similarities (e.g. "Ventas" vs "Sales") while still avoiding false positives on unrelated content.

SemanticMerge

SemanticMerge(threshold: float = 0.65)

Deduplicates procedural nodes by semantic (textual) similarity.

Parameters:

Name	Type	Description	Default
threshold	float	Minimum similarity score (0.0–1.0) required for two nodes to be considered duplicates. Defaults to 0.75.	0.65

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

def __init__(self, threshold: float = 0.65) -> None:
    if not 0.0 <= threshold <= 1.0:
        raise ValueError(f"threshold must be between 0.0 and 1.0, got {threshold}")
    self.threshold = threshold

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

    # Merge into the first (highest-priority) duplicate.
    canonical = await self.merge(duplicates[0], 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.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()
    merged_count = 0

    for i, primary in enumerate(nodes):
        if str(primary.id) in absorbed_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)
        for dup in duplicates:
            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
            )

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

find_duplicates async

find_duplicates(
    scope: TenantScope,
    existing_nodes: list[MemoryNode],
    new_node: MemoryNode,
) -> list[MemoryNode]

Return nodes from existing_nodes that overlap with new_node.

A node is considered a duplicate when the similarity between its representative text and that of new_node meets or exceeds self.threshold. Nodes whose representative text is shorter than _MIN_TEXT_LENGTH characters are skipped to avoid false positives caused by incidental substring overlap in short identifiers.

v7.5 authored-tier guard (Jarvio Ask 7 Option B): * When new_node itself is authored, return [] -- the caller's store_skill call must produce a fresh graph row even if a near-identical auto-promoted entry already exists. Domain plugins own the authored skill's lifecycle. * Authored entries in existing_nodes are skipped as merge candidates so an incoming auto-promoted draft cannot absorb the authored canonical (which would mutate the hand-edited fields).

Parameters:

Name	Type	Description	Default
scope	TenantScope	Tenant context (used for logging only at this level).	required
existing_nodes	list[MemoryNode]	Candidate nodes already stored in the graph.	required
new_node	MemoryNode	The incoming node to compare against.	required

Returns:

Type	Description
list[MemoryNode]	Subset of existing_nodes whose similarity score >= threshold.

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

async def find_duplicates(
    self,
    scope: TenantScope,
    existing_nodes: list[MemoryNode],
    new_node: MemoryNode,
) -> list[MemoryNode]:
    """Return nodes from *existing_nodes* that overlap with *new_node*.

    A node is considered a duplicate when the similarity between its
    representative text and that of *new_node* meets or exceeds
    ``self.threshold``.  Nodes whose representative text is shorter than
    ``_MIN_TEXT_LENGTH`` characters are skipped to avoid false positives
    caused by incidental substring overlap in short identifiers.

    v7.5 authored-tier guard (Jarvio Ask 7 Option B):
        * When *new_node* itself is authored, return ``[]`` -- the
          caller's ``store_skill`` call must produce a fresh graph
          row even if a near-identical auto-promoted entry already
          exists.  Domain plugins own the authored skill's
          lifecycle.
        * Authored entries in *existing_nodes* are skipped as merge
          candidates so an incoming auto-promoted draft cannot
          absorb the authored canonical (which would mutate the
          hand-edited fields).

    Args:
        scope: Tenant context (used for logging only at this level).
        existing_nodes: Candidate nodes already stored in the graph.
        new_node: The incoming node to compare against.

    Returns:
        Subset of *existing_nodes* whose similarity score >= threshold.
    """
    if self._is_authored(new_node):
        return []
    new_text = _representative_text(new_node)
    if len(new_text) < self._MIN_TEXT_LENGTH:
        return []
    # Isolation: never merge a node from a DIFFERENT scope level. Under v8.0
    # prefix isolation ``query_nodes`` also returns ancestor-visible rows,
    # so without this guard a child-scope dedup could absorb (and delete) an
    # ancestor node -- a cross-scope data leak. v8.0 rows always carry a
    # scope_path, so exclude any candidate whose scope_path DIFFERS from the
    # current scope. Candidates without a scope_path (legacy/un-backfilled
    # rows) are left as-is for backward compatibility. Round-2 fix.
    own_path = materialise_scope_path(scope)
    duplicates: list[MemoryNode] = []
    for node in existing_nodes:
        if self._is_authored(node):
            # Authored entries are immutable from the janitor's POV.
            continue
        cand_path = node.properties.get(SCOPE_PATH_KEY)
        if cand_path is not None and cand_path != own_path:
            continue
        candidate_text = _representative_text(node)
        if len(candidate_text) < self._MIN_TEXT_LENGTH:
            continue
        score = _similarity(new_text, candidate_text)
        if score >= self.threshold:
            logger.debug(
                "Duplicate candidate (score=%.3f): %r vs %r",
                score,
                new_node.label,
                node.label,
            )
            duplicates.append(node)
    return duplicates

merge async

merge(existing: MemoryNode, new: MemoryNode) -> MemoryNode

Merge new into existing and return the canonical node.

Merge rules: - Label: Keep the longer label (assumed more specific / descriptive). - Steps: Union of both step lists, preserving order and removing exact duplicates. - Importance: Take the maximum of the two scores. - Active flag: True if either node was active. - Properties: Shallow-merge; existing values are overridden by new values, except for the fields governed by the rules above. - Identity: The existing node's ID is preserved so callers can update the graph record in-place rather than creating a new one.

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

async def merge(self, existing: MemoryNode, new: MemoryNode) -> MemoryNode:
    """Merge *new* into *existing* and return the canonical node.

    Merge rules:
    - **Label**: Keep the longer label (assumed more specific / descriptive).
    - **Steps**: Union of both step lists, preserving order and removing
      exact duplicates.
    - **Importance**: Take the maximum of the two scores.
    - **Active flag**: True if either node was active.
    - **Properties**: Shallow-merge; *existing* values are overridden by
      *new* values, except for the fields governed by the rules above.
    - **Identity**: The *existing* node's ID is preserved so callers can
      update the graph record in-place rather than creating a new one.
    """
    # --- label: prefer the more descriptive (longer) one ---
    merged_label = (
        new.label if len(new.label) > len(existing.label) else existing.label
    )

    # --- steps: ordered union without duplicates ---
    existing_steps: list[str] = list(existing.properties.get("steps") or [])
    new_steps: list[str] = list(new.properties.get("steps") or [])
    seen: set[str] = set()
    merged_steps: list[str] = []
    for step in existing_steps + new_steps:
        normalised = step.strip()
        if normalised and normalised not in seen:
            seen.add(normalised)
            merged_steps.append(normalised)

    # --- importance: take the max ---
    merged_importance = max(existing.importance, new.importance)

    # --- active: True if either was active ---
    existing_active = existing.properties.get("active", True)
    new_active = new.properties.get("active", True)
    merged_active = bool(existing_active) or bool(new_active)

    # --- properties: shallow merge, structured fields overridden below ---
    merged_props: dict[str, object] = {**existing.properties, **new.properties}
    merged_props["steps"] = merged_steps
    merged_props["active"] = merged_active

    # Prefer the longer / richer content description
    existing_content = str(existing.properties.get("content", existing.label))
    new_content = str(new.properties.get("content", new.label))
    merged_props["content"] = (
        new_content if len(new_content) > len(existing_content) else existing_content
    )

    # Return a copy of the existing node with updated fields.
    # model_copy keeps the original id and created_at intact.
    return existing.model_copy(
        update={
            "label": merged_label,
            "properties": merged_props,
            "importance": merged_importance,
        }
    )