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)

Bases: SemanticDeduplicationMixin

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

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 (adopter 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 (adopter 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: - Reviewed canonical: An automatic Phase 12 draft cannot modify an existing approved or rejected procedure. - 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:
    - **Reviewed canonical**: An automatic Phase 12 draft cannot modify an
      existing approved or rejected procedure.
    - **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.
    """
    # Human review owns the canonical procedure after approval or
    # rejection. Phase 12 may rediscover the same episode pattern on every
    # later DEEP cycle; its regenerated draft is evidence of a duplicate,
    # not authority to rewrite the reviewed status, steps or guard.
    if self._reviewed_canonical_blocks_auto_draft(existing, new):
        return existing

    # --- 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(
        cast(list[str], existing.properties.get("steps") or [])
    )
    new_steps = list(cast(list[str], 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,
        }
    )