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

deduplicate_all async

deduplicate_all(scope: TenantScope, graph_store: Any, *, audit: Any = None, context: Any = None) -> tuple[int, list[MemoryNode]]

The one audited bulk boundary; see _deduplicate_all_audited.

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

async def deduplicate_all(
    self, scope: TenantScope, graph_store: Any, *, audit: Any = None, context: Any = None,
) -> tuple[int, list[MemoryNode]]:
    """The one audited bulk boundary; see ``_deduplicate_all_audited``."""
    return await self._deduplicate_all_audited(scope, graph_store, audit, context)

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	Effective exact owner permitted to merge both nodes.	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: Effective exact owner permitted to merge both nodes.
        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 not self._is_exact_owner(scope, new_node):
        return []
    if self._is_authored(new_node) or _lifecycle().procedure_authority_class(
        new_node.properties
    ) == "historical_terminal":
        return []
    new_text = _representative_text(new_node)
    if len(new_text) < self._MIN_TEXT_LENGTH:
        return []
    # Ancestor visibility is read authority, never merge authority.
    duplicates: list[MemoryNode] = []
    for node in existing_nodes:
        if self._is_authored(node):
            # Authored entries are immutable from the janitor's POV.
            continue
        if _lifecycle().procedure_authority_class(node.properties) == "historical_terminal":
            # Historic terminal rows have no native review seal. They are
            # independent evidence, not rows a cleanup can rewrite.
            continue
        if not self._is_exact_owner(scope, node):
            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. - Lifecycle: takes the incoming explicit lifecycle, or retains the canonical lifecycle when the incoming row has none. Missing evidence is a draft/inactive row; drafts never become active by merging. - 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.
    - **Lifecycle**: takes the incoming explicit lifecycle, or retains the
      canonical lifecycle when the incoming row has none. Missing evidence
      is a draft/inactive row; drafts never become active by merging.
    - **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
    if self._is_reviewed(existing) and not self._is_automatic(new):
        # ``merge`` is also a public pure helper.  It must never let a
        # manual draft rewrite a sealed terminal, even though the store
        # pass elects to retain that draft as a separate row.
        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)

    # --- properties: shallow merge, structured fields overridden below ---
    merged_props: dict[str, object] = {**existing.properties, **new.properties}
    merged_props["steps"] = merged_steps
    lifecycle_source = (
        new.properties
        if _lifecycle().procedure_lifecycle_declared(new.properties)
        else existing.properties
    )
    status, active = _lifecycle().procedure_lifecycle(lifecycle_source)
    merged_props["status"] = status
    merged_props["active"] = 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,
        }
    )