symfonic.capabilities.memory.phases.merge

merge

Phase 13: LLM-judged semantic-duplicate merge (true supersede).

Moved here from symfonic.capabilities.memory.phases.merge, which now imports it back: phase 13 is on the quick roster -- v8.17 put it there because quick is the only cadence the engine schedules in-process -- and a roster the kernel composes cannot reach into the legacy package.

Closes the write-time dedup's documented misses: restatements that reword a fact's OPENING ("Uses pytest and ruff" vs "The user uses pytest and ruff"), front rewording, and clause reordering all escape the prefix-gated string match in symfonic.memory.dedup -- see tests/memory/test_restatement_corpus.py for the full matrix. Those classes need semantic-equivalence judgment, which is deliberately kept OFF the per-turn write path and run here, offline, during consolidation.

Safety model (why an LLM judge and not an embedding threshold): cosine similarity cannot distinguish "deploys Tuesdays, never Fridays" from "deploys Fridays, never Tuesdays", and a false merge OVERWRITES a real memory. Embeddings (or a lexical ratio when no embeddings are available) only PREFILTER candidate pairs; every merge requires the judge to confirm the two phrasings state the same fact and to produce the merged phrasing.

Supersede mechanics: the canonical node (higher importance, else the older row) absorbs the pair -- merged phrasing, max importance, union of tags -- and the absorbed node is SOFT-RETRACTED with the namespaced marker (symfonic.memory.retraction), so it vanishes from every read path immediately, remains auditable, and is hard-deleted by the existing prune_retracted phase after the grace window.

Exclusions: only nodes materialised at exactly the run scope are eligible (visibility != ownership); AGENT_IDENTITY, SOUL and entity nodes are never merged (they have their own identity rules). SOUL: is excluded for the same reason as AGENT_IDENTITY: once a block layer renders these nodes into a cached prompt prefix, a false merge here would silently corrupt the user's profile for the whole session rather than just producing one wrong MEMORY_CONTEXT line (see .claude/docs/2026-08-04-prompt-part-blocks-architecture.md §6).

merge_semantic_duplicates async

merge_semantic_duplicates(graph: Any, scope: TenantScope, *, chat_model: Any, embedding_provider: Any | None = None, embedding_threshold: float = DEFAULT_EMBEDDING_THRESHOLD, lexical_threshold: float = DEFAULT_LEXICAL_THRESHOLD, max_pairs_per_run: int = 10, on_candidates: Callable[[Iterable[Any]], None] | None = None, on_supersede: Callable[[Any, Any], None] | None = None) -> int

Merge LLM-confirmed duplicate semantic facts; return merge count.

Per-pair failures are logged and skipped -- one bad judge response must never abort the phase.

on_candidates and on_supersede report node ids, and nothing else, to a caller keeping a cycle ledger. The supersede hook fires where the retraction is written rather than being derived from the return value, so the count stays true if one merge ever absorbs more than one row. Both are optional and the legacy consolidator passes neither.

Source code in src/symfonic/capabilities/memory/phases/merge.py

async def merge_semantic_duplicates(
    graph: Any,
    scope: TenantScope,
    *,
    chat_model: Any,
    embedding_provider: Any | None = None,
    embedding_threshold: float = DEFAULT_EMBEDDING_THRESHOLD,
    lexical_threshold: float = DEFAULT_LEXICAL_THRESHOLD,
    max_pairs_per_run: int = 10,
    on_candidates: Callable[[Iterable[Any]], None] | None = None,
    on_supersede: Callable[[Any, Any], None] | None = None,
) -> int:
    """Merge LLM-confirmed duplicate semantic facts; return merge count.

    Per-pair failures are logged and skipped -- one bad judge response
    must never abort the phase.

    ``on_candidates`` and ``on_supersede`` report node ids, and nothing else,
    to a caller keeping a cycle ledger. The supersede hook fires where the
    retraction is written rather than being derived from the return value, so
    the count stays true if one merge ever absorbs more than one row. Both are
    optional and the legacy consolidator passes neither.
    """
    if chat_model is None:
        return 0

    nodes = await graph.query_nodes(scope, layer=MemoryLayer.SEMANTIC)
    run_path = materialise_scope_path(scope)
    eligible = []
    for n in nodes:
        label = str(getattr(n, "label", "") or "")
        if label.startswith(_EXCLUDED_LABEL_PREFIXES):
            continue
        props = getattr(n, "properties", None) or {}
        if props.get("label_prefix") in _EXCLUDED_LABEL_PREFIXES:
            continue
        if stored_scope_path(props, getattr(n, "tenant_id", "")) != run_path:
            # Visible but ancestor-owned: not this scope's to merge.
            continue
        eligible.append(n)

    if on_candidates is not None:
        on_candidates([node.id for node in eligible])

    merges = 0
    judged_pairs = 0
    absorbed_ids: set[str] = set()
    for i, a in enumerate(eligible):
        if str(a.id) in absorbed_ids:
            continue
        peers = {str(node.id): node for node in eligible[i + 1 :]}
        if callable(getattr(type(graph), "related_candidates", None)):
            for node in await graph.related_candidates(scope, _node_text(a)):
                props = node.properties or {}
                if (str(node.id) != str(a.id)
                    and not node.label.startswith(_EXCLUDED_LABEL_PREFIXES)
                    and props.get("label_prefix") not in _EXCLUDED_LABEL_PREFIXES
                    and stored_scope_path(props, node.tenant_id) == run_path):
                    peers.setdefault(str(node.id), node)
        for b in peers.values():
            if judged_pairs >= max_pairs_per_run:
                logger.info(
                    "semantic_merge: pair budget (%d) reached; remaining "
                    "candidates deferred to the next run",
                    max_pairs_per_run,
                )
                return merges
            if str(b.id) in absorbed_ids or str(a.id) in absorbed_ids:
                continue
            if not carry_classification(a, b, {}):
                continue
            try:
                similarity, kind = await _pair_similarity(a, b, embedding_provider)
                threshold = embedding_threshold if kind == "embedding" else lexical_threshold
                if similarity < threshold:
                    continue
                judged_pairs += 1
                same, merged_fact = await _judge(chat_model, _node_text(a), _node_text(b))
                if not same:
                    continue
                # Canonical = higher importance, else the earlier row.
                canonical, absorbed = (
                    (a, b) if float(a.importance or 0) >= float(b.importance or 0) else (b, a)
                )
                phrasing = merged_fact or _node_text(canonical)
                merged_props = dict(canonical.properties or {})
                if not carry_classification(canonical, absorbed, merged_props):
                    continue
                merged_props["content"] = phrasing
                tags_a = list((canonical.properties or {}).get("tags") or [])
                tags_b = list((absorbed.properties or {}).get("tags") or [])
                if tags_a or tags_b:
                    merged_props["tags"] = sorted(set(tags_a) | set(tags_b))
                await graph.update_node(
                    scope,
                    canonical.id,
                    {
                        "label": phrasing[:100],
                        "importance": max(
                            float(canonical.importance or 1.0),
                            float(absorbed.importance or 1.0),
                        ),
                        "properties": merged_props,
                    },
                )
                await graph.update_node(
                    scope,
                    absorbed.id,
                    {
                        "properties": {
                            **(absorbed.properties or {}),
                            RETRACTED_KEY: True,
                            RETRACTED_AT_KEY: datetime.now(UTC).isoformat(),
                            RETRACTION_REASON_KEY: (
                                f"superseded by semantic merge into {canonical.id}"
                            )[:200],
                        },
                    },
                )
                absorbed_ids.add(str(absorbed.id))
                merges += 1
                if on_supersede is not None:
                    on_supersede(absorbed.id, canonical.id)
                logger.info(
                    "semantic_merge: %r absorbed into %s (%s sim=%.2f)",
                    _node_text(absorbed)[:60],
                    canonical.id,
                    kind,
                    similarity,
                )
            except Exception:
                logger.warning(
                    "semantic_merge: pair (%s, %s) failed; skipping",
                    a.id,
                    b.id,
                    exc_info=True,
                )
    return merges