symfonic.capabilities.memory.phases

phases

The consolidation phases, as implementations the capability owns.

A roster names phases; this package is where they are. Until now the names on PHASE_ROSTER had no implementations behind them at all, so ConsolidationRuntime() composed with nothing ran nothing, reported an empty phases_run, and answered clean -- a cycle that did nothing and called it a success.

Each module here holds one phase's logic and nothing else, moved from symfonic.core.learning rather than rewritten. The legacy consolidator imports these same functions during the transition: two transcriptions of one phase is two behaviours, and the second is found by an adopter rather than by a test.

:mod:.quick holds the four phase objects the QUICK cadence runs and the factory that builds the complete set.

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

promote_profile_corrections async

promote_profile_corrections(graph: GraphMemoryStore, scope: TenantScope, recent_nodes: list[MemoryNode], profile_fields: frozenset[str], now: datetime | None = None) -> int

Promote user-corrected profile facts onto the tenant's SOUL node.

Trigger condition (kept verbatim from the pre-redesign apply_soul_corrections, phases.py:446): only nodes carrying properties["_last_edited_by"] == "user_manual_edit" are treated as corrections. This keeps agent self-edits (extraction writes, this phase's own promotion writes, etc.) out of the promotion loop -- load-bearing, do not relax without also revisiting the idempotency guarantee below.

Parameters:

Name	Type	Description	Default
graph	GraphMemoryStore	Tenant-scoped graph store. Both the read (locating the canonical SOUL node) and the write (applying corrections) go through this store with scope, so a promotion is structurally confined to one tenant.	required
scope	TenantScope	Tenant isolation scope for every graph operation.	required
recent_nodes	list[MemoryNode]	Nodes considered for this consolidation pass (typically updated_at >= lookback cutoff). Any node satisfying the guard above is a candidate correction source, whether or not it is itself the canonical SOUL node.	required
profile_fields	frozenset[str]	The set of field names that constitute "profile" for this domain. Callers derive this from domain.soul_schema.keys() -- the schema is READ to learn which fields matter, but this function never writes to it. An empty set is a safe no-op (nothing is eligible to promote).	required
now	datetime | None	Instant recorded as each promoted field's provenance. Defaults to wall-clock UTC. Injectable because a caller that pins its own clock -- a test, a replay, a walkthrough -- would otherwise write a stamp it cannot predict, and the rendered provenance would read as being from the future.	None

Returns:

Type	Description
int	Count of profile fields whose value actually changed on the
int	canonical SOUL node (mirrors the legacy apply_soul_corrections
int	"count of key assignments made" semantics). A guard-marker-only
int	rewrite (see idempotency note below) does not increment this
int	count even though a graph write occurs.

Idempotency: when the correction source IS the canonical SOUL node (a human edited it directly rather than via a separate correction record), its own _last_edited_by guard is flipped from "user_manual_edit" to PROMOTION_MARKER even if the "corrected" values already match (i.e. even when the differential count is zero). Without this, the node would satisfy the guard again on the very next pass -- reading this promotion's own prior output back as a fresh user correction and re-promoting forever.

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

async def promote_profile_corrections(
    graph: GraphMemoryStore,
    scope: TenantScope,
    recent_nodes: list[MemoryNode],
    profile_fields: frozenset[str],
    now: datetime | None = None,
) -> int:
    """Promote user-corrected profile facts onto the tenant's SOUL node.

    Trigger condition (kept verbatim from the pre-redesign
    ``apply_soul_corrections``, ``phases.py:446``): only nodes carrying
    ``properties["_last_edited_by"] == "user_manual_edit"`` are treated as
    corrections. This keeps agent self-edits (extraction writes, this
    phase's own promotion writes, etc.) out of the promotion loop --
    load-bearing, do not relax without also revisiting the idempotency
    guarantee below.

    Args:
        graph: Tenant-scoped graph store. Both the read (locating the
            canonical SOUL node) and the write (applying corrections) go
            through this store with ``scope``, so a promotion is
            structurally confined to one tenant.
        scope: Tenant isolation scope for every graph operation.
        recent_nodes: Nodes considered for this consolidation pass
            (typically ``updated_at >= lookback cutoff``). Any node
            satisfying the guard above is a candidate correction source,
            whether or not it is itself the canonical SOUL node.
        profile_fields: The set of field names that constitute "profile"
            for this domain. Callers derive this from
            ``domain.soul_schema.keys()`` -- the schema is READ to learn
            which fields matter, but this function never writes to it.
            An empty set is a safe no-op (nothing is eligible to promote).

        now: Instant recorded as each promoted field's provenance.
            Defaults to wall-clock UTC. Injectable because a caller that
            pins its own clock -- a test, a replay, a walkthrough -- would
            otherwise write a stamp it cannot predict, and the rendered
            provenance would read as being from the future.

    Returns:
        Count of profile fields whose value actually changed on the
        canonical SOUL node (mirrors the legacy ``apply_soul_corrections``
        "count of key assignments made" semantics). A guard-marker-only
        rewrite (see idempotency note below) does not increment this
        count even though a graph write occurs.

    Idempotency: when the correction source IS the canonical SOUL node
    (a human edited it directly rather than via a separate correction
    record), its own ``_last_edited_by`` guard is flipped from
    ``"user_manual_edit"`` to ``PROMOTION_MARKER`` even if the "corrected"
    values already match (i.e. even when the differential count is zero).
    Without this, the node would satisfy the guard again on the very next
    pass -- reading this promotion's own prior output back as a fresh
    user correction and re-promoting forever.
    """
    correction_nodes = [
        n for n in recent_nodes
        if (n.properties or {}).get(EDITED_BY_KEY) == "user_manual_edit"
    ]
    if not correction_nodes or not profile_fields:
        return 0

    # Last-writer-wins across every correction node touched this pass,
    # restricted to the caller-declared profile field set. Mirrors the
    # per-node accumulation order of the legacy function.
    corrected: dict[str, Any] = {}
    for node in correction_nodes:
        for key, value in _field_values(node).items():
            if key in profile_fields:
                corrected[key] = value

    if not corrected:
        return 0

    try:
        soul_nodes = await graph.query_nodes(
            scope, layer=MemoryLayer.SEMANTIC, label_prefix="SOUL",
        )
    except Exception:
        logger.debug(
            "promote_profile_corrections: query_nodes failed for tenant %s",
            scope.tenant_id, exc_info=True,
        )
        return 0

    target = _freshest(soul_nodes)
    if target is None:
        # No canonical profile node yet. The correction(s) stay flagged
        # user_manual_edit and are re-evaluated next pass -- harmless
        # (costs a query, not a mutation) and self-resolving once a SOUL
        # node is created (onboarding form / extractor).
        return 0

    existing = target.properties or {}
    merged = dict(existing)
    raw_provenance = existing.get("_field_provenance")
    provenance = dict(raw_provenance) if isinstance(raw_provenance, dict) else {}
    recorded_at = (now or datetime.now(UTC)).isoformat()
    updates = 0
    for key, value in corrected.items():
        if existing.get(key) != value:
            merged[key] = value
            # Per-field, because the node-level stamp answers "when was
            # this node written", which stops being the same question
            # once one node carries facts recorded at different times.
            provenance[key] = {
                "source": PROMOTED_FIELD_SOURCE,
                "recorded_at": recorded_at,
            }
            updates += 1
    if updates:
        merged["_field_provenance"] = provenance

    stale_guard = existing.get(EDITED_BY_KEY) == "user_manual_edit"
    if updates == 0 and not stale_guard:
        return 0

    merged[EDITED_BY_KEY] = PROMOTION_MARKER
    try:
        await graph.update_node(scope, target.id, {"properties": merged})
    except Exception:
        logger.debug(
            "promote_profile_corrections: update_node failed for %s (tenant %s)",
            target.id, scope.tenant_id, exc_info=True,
        )
        return 0

    return updates

summarize_episodic async

summarize_episodic(episodic_layer: Any, scope: Any, *, llm_summarise: Any | None = None, max_entries: int = 100, summarize_batch: int = 50, gate: DurabilityGate | None = None) -> tuple[str, int]

Summarize oldest episodic entries into a single text, delete originals.

When the total event count exceeds max_entries, the oldest summarize_batch entries are retrieved, concatenated (or LLM- summarised when llm_summarise is provided), and the originals are deleted from the vector store.

Parameters:

Name	Type	Description	Default
episodic_layer	Any	An EpisodicLayer instance.	required
scope	Any	Tenant scope for isolation.	required
llm_summarise	Any | None	Optional async callable (text) -> summary.	None
max_entries	int	Threshold below which summarization is skipped.	100
summarize_batch	int	Number of oldest entries to summarize per run.	50

Returns:

Type	Description
str	Tuple of (summary_text, count_deleted). When no
int	summarization was needed, returns ("", 0).

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

async def summarize_episodic(
    episodic_layer: Any,
    scope: Any,
    *,
    llm_summarise: Any | None = None,
    max_entries: int = 100,
    summarize_batch: int = 50,
    gate: DurabilityGate | None = None,
) -> tuple[str, int]:
    """Summarize oldest episodic entries into a single text, delete originals.

    When the total event count exceeds ``max_entries``, the oldest
    ``summarize_batch`` entries are retrieved, concatenated (or LLM-
    summarised when ``llm_summarise`` is provided), and the originals
    are deleted from the vector store.

    Args:
        episodic_layer: An ``EpisodicLayer`` instance.
        scope: Tenant scope for isolation.
        llm_summarise: Optional async callable ``(text) -> summary``.
        max_entries: Threshold below which summarization is skipped.
        summarize_batch: Number of oldest entries to summarize per run.

    Returns:
        Tuple of ``(summary_text, count_deleted)``.  When no
        summarization was needed, returns ``("", 0)``.
    """
    count = await episodic_layer.count_events(scope)
    if count <= max_entries:
        return ("", 0)

    # Retrieve oldest entries.  list_events() returns results in
    # cosine-descending order (arbitrary for zero-vectors), NOT time order.
    # Phase 10 requires the *oldest* ``summarize_batch`` entries, so we
    # fetch all entries (up to ``count``), sort by created_at ascending,
    # and slice the front.  created_at is stored in metadata by store_event().
    all_entries = await episodic_layer.list_events(scope, limit=count)
    all_entries = sorted(all_entries, key=lambda e: e.metadata.get("created_at", ""))
    entries = all_entries[:summarize_batch]
    if not entries:
        return ("", 0)

    # Build raw text from entry contents.  v7.26.2: transient/expired
    # entries are gated out of the summary so they never become a
    # semantic meta-node.  They are NOT deleted here -- Phase 8 (cleanup)
    # owns their pruning; Phase 10 only refuses to promote them.
    _gate = gate if gate is not None else DurabilityGate()
    raw_parts: list[str] = []
    summarised_entries = []
    for entry in entries:
        if _gate.is_gated(entry, phase="10"):
            continue
        raw_parts.append(entry.content or "")
        summarised_entries.append(entry)
    raw_text = "\n".join(part for part in raw_parts if part)

    # Summarize
    if llm_summarise is not None:
        try:
            summary_text: str = await llm_summarise(raw_text)
        except Exception:
            logger.warning("LLM episodic summarization failed, using raw concat")
            summary_text = raw_text
    else:
        summary_text = raw_text

    # Delete originals from vector store -- only the entries actually
    # folded into the summary.  Gated (transient/expired) entries are left
    # in place for Phase 8 to prune.
    entry_ids = [entry.id for entry in summarised_entries if entry.id]
    if entry_ids:
        await episodic_layer._vector.delete(scope, entry_ids)

    deleted = len(entry_ids)
    logger.info(
        "Episodic summarization: summarized %d entries for tenant=%s",
        deleted,
        scope.tenant_id,
    )

    return (summary_text, deleted)