symfonic.core.learning

learning

symfonic.core.learning -- self-reflection, insight extraction, and consolidation.

ConsolidationReport dataclass

ConsolidationReport(
    tenant_id: str,
    started_at: datetime = (lambda: datetime.now(UTC))(),
    finished_at: datetime | None = None,
    nodes_strengthened: int = 0,
    nodes_pruned: int = 0,
    risk_nodes_tagged: int = 0,
    meta_nodes_created: int = 0,
    soul_updates: int = 0,
    edges_created: int = 0,
    working_ttl_pruned: int = 0,
    nodes_decayed: int = 0,
    retracted_nodes_pruned: int = 0,
    episodes_summarized: int = 0,
    synthetic_edges_created: int = 0,
    draft_skills_promoted: int = 0,
    entity_nodes_created: int = 0,
    mentions_edges_created: int = 0,
    transient_rejected: int = 0,
    semantic_merges: int = 0,
    errors: list[str] = list(),
    phases_run: list[str] = list(),
)

Summary of a single consolidation run.

phases_run is an observability-only list populated by each phase implementation in phases.py / phases_maintenance.py / phases_episodic.py etc. when the phase actually executes (even if its mutation count is zero). It exists to disambiguate "the phase ran and found nothing" from "the phase was short-circuited and never ran" -- both produce a 0 counter, but only the former appears in phases_run. Purely additive; existing consumers reading the counter fields continue to work unchanged.

InsightExtractor

InsightExtractor(
    llm: Any = None,
    custom_prompt: str | None = None,
    min_confidence: float = 0.5,
    *,
    observability_hook: ObservabilityHook | None = None,
)

Extracts LearnedInsight objects from conversation text using an LLM.

The extraction prompt can be customized with domain-specific instructions via custom_prompt. The default prompt targets general business insights.

Graceful degradation: when no LLM is configured, extract() returns an empty list without raising.

Source code in src/symfonic/core/learning/refiner.py

def __init__(
    self,
    llm: Any = None,
    custom_prompt: str | None = None,
    min_confidence: float = 0.5,
    *,
    observability_hook: ObservabilityHook | None = None,
) -> None:
    self._llm = llm
    self._prompt_template = custom_prompt or _DEFAULT_PROMPT
    self._min_confidence = min_confidence
    # v8.3.2 observability-bug fix (G2): the insight-extractor LLM call
    # emitted ONLY on the merge-dependent callback manager and made ZERO
    # ``ObservabilityHook`` calls -- the same bug class as the v8.3.1
    # critic.  The engine now threads in the always-on hook
    # (``deps.require(ObservabilityHook)``); ``None`` resolves to a
    # ``NoOpObservabilityHook`` so the default is byte-identical
    # (additive only).  Attribution is MODEL-BASED (no node_name on the
    # hook signature); node-level disambiguation is deferred to Build 2.
    if observability_hook is None:
        from symfonic.core.observability.hooks import NoOpObservabilityHook

        observability_hook = NoOpObservabilityHook()
    self._observability_hook = observability_hook

extract async

extract(
    conversation: str,
    tenant_id: str = "",
    *,
    callback_manager: CallbackManager | None = None,
    run_id: str = "",
    model_name: str = "",
) -> list[LearnedInsight]

Extract insights from conversation text.

Parameters:

Name	Type	Description	Default
conversation	str	The raw conversation text to analyze.	required
tenant_id	str	Tenant identifier to stamp on each insight.	''
callback_manager	CallbackManager | None	v7.4.3 -- when present, fires on_llm_end(node_name="insight_extractor") after the LLM call (mirrors react.py:303-311). None preserves pre-7.4.3 behaviour.	None
run_id	str	Engine run identifier for callback correlation.	''
model_name	str	Optional model identifier to stamp on the LLMEndEvent. When empty, falls back to "insight_extractor".	''

Returns:

Type	Description
list[LearnedInsight]	List of LearnedInsight objects above the min_confidence threshold.
list[LearnedInsight]	Returns empty list when no LLM is configured or the conversation
list[LearnedInsight]	is blank (never raises).

Source code in src/symfonic/core/learning/refiner.py

async def extract(
    self,
    conversation: str,
    tenant_id: str = "",
    *,
    callback_manager: CallbackManager | None = None,
    run_id: str = "",
    model_name: str = "",
) -> list[LearnedInsight]:
    """Extract insights from conversation text.

    Args:
        conversation: The raw conversation text to analyze.
        tenant_id: Tenant identifier to stamp on each insight.
        callback_manager: v7.4.3 -- when present, fires
            ``on_llm_end(node_name="insight_extractor")`` after the
            LLM call (mirrors react.py:303-311).  ``None`` preserves
            pre-7.4.3 behaviour.
        run_id: Engine run identifier for callback correlation.
        model_name: Optional model identifier to stamp on the
            LLMEndEvent.  When empty, falls back to ``"insight_extractor"``.

    Returns:
        List of LearnedInsight objects above the min_confidence threshold.
        Returns empty list when no LLM is configured or the conversation
        is blank (never raises).
    """
    if not self._llm or not conversation.strip():
        return []

    prompt = self._prompt_template.format(conversation=conversation)

    try:
        raw_text = await self._invoke_llm(
            prompt,
            callback_manager=callback_manager,
            run_id=run_id,
            model_name=model_name or "insight_extractor",
        )
        return self._parse_insights(raw_text, tenant_id)
    except Exception:
        logger.warning("Insight extraction failed", exc_info=True)
        return []

LearnedInsight dataclass

LearnedInsight(
    fact: str,
    confidence: float,
    category: str,
    tags: tuple[str, ...] = (),
    tenant_id: str = "",
    timestamp: datetime = (lambda: datetime.now(UTC))(),
    metadata: dict[str, Any] = dict(),
    scope: TenantScope | None = None,
)

A business fact extracted from conversation by the InsightExtractor.

Immutable by design — once extracted, an insight is a fixed record that can be safely passed across async boundaries without copying.

Attributes:

Name	Type	Description
fact	str	The specific business observation (1-2 sentences).
confidence	float	Certainty score between 0.0 and 1.0.
category	str	Domain bucket (e.g. "customer_behavior", "inventory").
tags	tuple[str, ...]	Optional keyword tuple for filtering and retrieval.
tenant_id	str	Owning tenant; empty string means unscoped.
timestamp	datetime	UTC creation time; auto-set on construction.
metadata	dict[str, Any]	Arbitrary extra data from the extraction process.
scope	TenantScope | None	T-7.21.7 -- optional full TenantScope carrying sub_tenant_id / namespace (and any future field). Preferred by :class:CommitToMemoryTool over the bare tenant_id reconstruction path because the latter silently dropped namespace. Backward compatible: when None (the v7.20 default), the tool falls back to the bare- tenant_id construction.

SleepConsolidator

SleepConsolidator(
    graph_store: GraphMemoryStore,
    *,
    lookback_hours: int = 24,
    llm_summariser: Any | None = None,
    episodic_layer: Any | None = None,
    procedural_layer: Any | None = None,
    promotion_min_pattern_count: int = 3,
    promotion_recency_days: int = 30,
    promotion_max_drafts_per_run: int = 5,
    promotion_use_tool_calls_fallback: bool = False,
    promotion_promote_assistant_content: bool = False,
    phase_12_use_llm_extractor: bool = False,
    phase_12_llm_max_episodes_per_run: int = 100,
    phase_12_llm_max_drafts_per_run: int = 5,
    phase_12_llm_model: str = "claude-haiku-4-5",
    chat_model: Any | None = None,
    episodic_summarization_max_entries: int = 100,
    episodic_summarization_batch_size: int = 50,
    phase1_spreading_weight: float = 0.5,
    synthetic_link_min_co_count: int = 2,
    enable_entity_linker: bool = False,
    entity_linker_extractor_kind: Literal[
        "regex", "spacy", "llm"
    ] = "regex",
    entity_linker_min_mention_count: int = 2,
    entity_linker_max_episodics_per_run: int = 200,
    entity_linker_confidence_threshold: float = 0.5,
    entity_linker_extractor: Any | None = None,
    enable_semantic_merge: bool = False,
    semantic_merge_embedding_provider: Any | None = None,
    semantic_merge_embedding_threshold: float = 0.9,
    semantic_merge_lexical_threshold: float = 0.6,
    semantic_merge_max_pairs_per_run: int = 10,
)

Offline memory consolidation engine.

Phases: 1. Strengthen -- boost importance for recently recurring nodes. 1.5 Co-occurrence edges -- create CO_OCCURRED edges between nodes updated within the same 1-hour window that have no direct edge. 2. Tag risk -- mark nodes with negative-sentiment context. 2.5 Pending edges -- write inferred pending_connections as real edges. 3. Prune orphans -- delete stale low-confidence orphan nodes. 4. Generate meta-nodes -- cluster nodes by label prefix. 5. SOUL update -- apply user corrections to the domain SOUL schema. 8. Working TTL cleanup -- prune WORKING-layer nodes with expired TTL. 9. Semantic importance decay -- reduce importance of stale nodes. 10. Episodic summarization -- compress old episodic entries into a semantic meta-node.

Source code in src/symfonic/core/learning/consolidation.py

def __init__(
    self,
    graph_store: GraphMemoryStore,
    *,
    lookback_hours: int = 24,
    llm_summariser: Any | None = None,
    episodic_layer: Any | None = None,
    procedural_layer: Any | None = None,
    promotion_min_pattern_count: int = 3,
    promotion_recency_days: int = 30,
    promotion_max_drafts_per_run: int = 5,
    # v7.4.7 Jarvio Ask 9 PR-A: opt-in tool-call fallback for the
    # regex-based pattern extractor. Off by default so existing
    # consumers see no behaviour change.
    promotion_use_tool_calls_fallback: bool = False,
    # v7.7.2 Jarvio fabrication-promotion ask: default-off
    # promotion of assistant-narrated content.  When ``False``
    # (default), the regex AND LLM extractor paths filter out
    # episodic rows whose ``metadata['speaker'] == 'assistant'``
    # and legacy rows without a speaker discriminator.
    promotion_promote_assistant_content: bool = False,
    # v7.6 Jarvio Ask 9 Option A: opt-in LLM-backed procedural
    # extractor.  ``phase_12_use_llm_extractor=False`` (default) runs
    # the regex extractor unchanged.  When True AND ``chat_model`` is
    # non-None, ``_build_procedural_extractor`` instantiates
    # :class:`LlmProceduralExtractor` and the regex path is skipped
    # (either-or dispatch -- the eval rejected merged streams).
    phase_12_use_llm_extractor: bool = False,
    phase_12_llm_max_episodes_per_run: int = 100,
    phase_12_llm_max_drafts_per_run: int = 5,
    phase_12_llm_model: str = "claude-haiku-4-5",
    chat_model: Any | None = None,
    episodic_summarization_max_entries: int = 100,
    episodic_summarization_batch_size: int = 50,
    phase1_spreading_weight: float = 0.5,
    synthetic_link_min_co_count: int = 2,
    # v7.2 Roadmap Item 12 -- EntityLinker phase (default-off).
    enable_entity_linker: bool = False,
    entity_linker_extractor_kind: Literal["regex", "spacy", "llm"] = "regex",
    entity_linker_min_mention_count: int = 2,
    entity_linker_max_episodics_per_run: int = 200,
    entity_linker_confidence_threshold: float = 0.5,
    entity_linker_extractor: Any | None = None,
    # v8.16 Phase 13 -- LLM-judged semantic-duplicate merge
    # (default-off).  Catches the restatement classes the write-time
    # dedup documents as misses (front reword, clause reorder; see
    # tests/memory/test_restatement_corpus.py).  Requires
    # ``chat_model``; ``semantic_merge_embedding_provider`` upgrades
    # the candidate prefilter from lexical ratio to cosine
    # similarity.  The judge is the correctness gate either way.
    enable_semantic_merge: bool = False,
    semantic_merge_embedding_provider: Any | None = None,
    semantic_merge_embedding_threshold: float = 0.90,
    semantic_merge_lexical_threshold: float = 0.60,
    semantic_merge_max_pairs_per_run: int = 10,
) -> None:
    self._graph = graph_store
    self._lookback_hours = lookback_hours
    self._llm_summarise = llm_summariser
    self._episodic = episodic_layer
    self._procedural = procedural_layer
    self._promotion_min_pattern_count = promotion_min_pattern_count
    self._promotion_recency_days = promotion_recency_days
    self._promotion_max_drafts_per_run = promotion_max_drafts_per_run
    self._promotion_use_tool_calls_fallback = (
        promotion_use_tool_calls_fallback
    )
    self._promotion_promote_assistant_content = (
        promotion_promote_assistant_content
    )
    # v7.6 LLM extractor knobs -- stored for the factory in
    # ``_build_procedural_extractor`` to consume.
    self._phase_12_use_llm_extractor = phase_12_use_llm_extractor
    self._phase_12_llm_max_episodes_per_run = (
        phase_12_llm_max_episodes_per_run
    )
    self._phase_12_llm_max_drafts_per_run = (
        phase_12_llm_max_drafts_per_run
    )
    self._phase_12_llm_model = phase_12_llm_model
    self._chat_model = chat_model
    self._episodic_summarization_max_entries = episodic_summarization_max_entries
    self._episodic_summarization_batch_size = episodic_summarization_batch_size
    # v6.2 T02: weight applied to spreading_access_count inside
    # Phase 1 strengthen. 0.5 preserves v6.1.x observable behaviour.
    self._phase1_spreading_weight = phase1_spreading_weight
    # v6.2 T03: minimum co-occurrence count for Phase 11 synthetic
    # edges. 2 matches the hardcoded v6.1 floor.
    self._synthetic_link_min_co_count = synthetic_link_min_co_count
    # v7.2 Roadmap Item 12 -- EntityLinker phase configuration.
    # When ``enable_entity_linker`` is False the phase is a strict
    # no-op (not appended to ``report.phases_run``).
    self._enable_entity_linker = enable_entity_linker
    self._entity_linker_extractor_kind = entity_linker_extractor_kind
    self._entity_linker_min_mention_count = entity_linker_min_mention_count
    self._entity_linker_max_episodics_per_run = (
        entity_linker_max_episodics_per_run
    )
    self._entity_linker_confidence_threshold = (
        entity_linker_confidence_threshold
    )
    # Test escape hatch: pre-built extractor instance trumps the
    # ``_kind`` knob so tests can drop in mocks without spinning up
    # spaCy or the LLM provider.
    self._entity_linker_extractor = entity_linker_extractor
    # v8.16 Phase 13 -- semantic-merge configuration.  When
    # ``enable_semantic_merge`` is False the phase is a strict no-op
    # (not appended to ``report.phases_run``).
    self._enable_semantic_merge = enable_semantic_merge
    self._semantic_merge_embedding_provider = (
        semantic_merge_embedding_provider
    )
    self._semantic_merge_embedding_threshold = (
        semantic_merge_embedding_threshold
    )
    self._semantic_merge_lexical_threshold = (
        semantic_merge_lexical_threshold
    )
    self._semantic_merge_max_pairs_per_run = (
        semantic_merge_max_pairs_per_run
    )

bind_layers

bind_layers(
    *,
    episodic_layer: Any | None = None,
    procedural_layer: Any | None = None,
    overwrite: bool = False,
) -> None

Late-bind the episodic and/or procedural layers after construction.

The constructor accepts episodic_layer and procedural_layer as kwargs (see :meth:__init__). Callers that build the :class:SymfonicAgent first and only obtain layer references via agent._orchestrator.get_layer(...) afterwards can use this method to wire the layers without resorting to direct attribute assignment.

Direct consolidator._episodic = ... mutation silently creates new attributes on a typo (_episodic_layer vs _episodic) because the class is not slotted. This method validates the kwarg names and the binding state, so a typo raises TypeError at the call site (Python's keyword-only enforcement) instead of silently no-op'ing.

Parameters:

Name	Type	Description	Default
episodic_layer	Any | None	Optional EpisodicLayer to bind. When None, leaves the existing binding unchanged.	None
procedural_layer	Any | None	Optional ProceduralLayer to bind. When None, leaves the existing binding unchanged.	None
overwrite	bool	When False (default), refuses to overwrite an already-bound layer (raises RuntimeError). Set True to replace an existing binding (e.g. test rebinding).	False

Raises:

Type	Description
TypeError	When an unknown kwarg is passed (typo-safety). The keyword-only parameters above mean Python raises TypeError naturally at the call site for unknown kwargs.
RuntimeError	When overwrite=False (default) and an attempt is made to rebind a layer that's already set.

Example::

agent = SymfonicAgent(model_provider=provider, config=config)
consolidator = SleepConsolidator(graph_store=store)
consolidator.bind_layers(
    episodic_layer=agent._orchestrator.get_layer(
        MemoryLayer.EPISODIC,
    ),
    procedural_layer=agent._orchestrator.get_layer(
        MemoryLayer.PROCEDURAL,
    ),
)

Source code in src/symfonic/core/learning/consolidation.py

def bind_layers(
    self,
    *,
    episodic_layer: Any | None = None,
    procedural_layer: Any | None = None,
    overwrite: bool = False,
) -> None:
    """Late-bind the episodic and/or procedural layers after construction.

    The constructor accepts ``episodic_layer`` and ``procedural_layer``
    as kwargs (see :meth:`__init__`).  Callers that build the
    :class:`SymfonicAgent` first and only obtain layer references via
    ``agent._orchestrator.get_layer(...)`` afterwards can use this
    method to wire the layers without resorting to direct attribute
    assignment.

    Direct ``consolidator._episodic = ...`` mutation silently creates
    new attributes on a typo (``_episodic_layer`` vs ``_episodic``)
    because the class is not slotted.  This method validates the
    kwarg names and the binding state, so a typo raises ``TypeError``
    at the call site (Python's keyword-only enforcement) instead of
    silently no-op'ing.

    Args:
        episodic_layer: Optional EpisodicLayer to bind.  When None,
            leaves the existing binding unchanged.
        procedural_layer: Optional ProceduralLayer to bind.  When
            None, leaves the existing binding unchanged.
        overwrite: When False (default), refuses to overwrite an
            already-bound layer (raises ``RuntimeError``).  Set
            True to replace an existing binding (e.g. test
            rebinding).

    Raises:
        TypeError: When an unknown kwarg is passed (typo-safety).
            The keyword-only parameters above mean Python raises
            ``TypeError`` naturally at the call site for unknown
            kwargs.
        RuntimeError: When ``overwrite=False`` (default) and an
            attempt is made to rebind a layer that's already set.

    Example::

        agent = SymfonicAgent(model_provider=provider, config=config)
        consolidator = SleepConsolidator(graph_store=store)
        consolidator.bind_layers(
            episodic_layer=agent._orchestrator.get_layer(
                MemoryLayer.EPISODIC,
            ),
            procedural_layer=agent._orchestrator.get_layer(
                MemoryLayer.PROCEDURAL,
            ),
        )
    """
    if episodic_layer is not None:
        if self._episodic is not None and not overwrite:
            raise RuntimeError(
                "SleepConsolidator.episodic_layer is already bound. "
                "Pass overwrite=True to replace it.",
            )
        self._episodic = episodic_layer
    if procedural_layer is not None:
        if self._procedural is not None and not overwrite:
            raise RuntimeError(
                "SleepConsolidator.procedural_layer is already bound. "
                "Pass overwrite=True to replace it.",
            )
        self._procedural = procedural_layer

nightly_nap async

nightly_nap(
    scope: TenantScope,
    *,
    soul_schema: dict[str, Any] | None = None,
    pending_connections: list[dict[str, Any]] | None = None,
    max_entries: int | None = None,
    summarize_batch: int | None = None,
) -> ConsolidationReport

Run the full 11-phase consolidation roster (v7.0 T12).

quick_nap runs the 3-phase subset on every-N-turns cadence; nightly_nap runs the full roster (Phase 1, 1.5, 2, 2.5, 3, 4, 5, 8, 9, 10, 11, 12) and is intended for a cron / celery- beat / APScheduler job firing once per quiet window per tenant.

Accepts per-call overrides for the Phase 10 knobs so the scheduler can tune each tenant independently without rebuilding the consolidator.

Returns:

Type	Description
ConsolidationReport	ConsolidationReport with every phase that actually fired
ConsolidationReport	populated in phases_run.

Source code in src/symfonic/core/learning/consolidation.py

async def nightly_nap(
    self,
    scope: TenantScope,
    *,
    soul_schema: dict[str, Any] | None = None,
    pending_connections: list[dict[str, Any]] | None = None,
    max_entries: int | None = None,
    summarize_batch: int | None = None,
) -> ConsolidationReport:
    """Run the full 11-phase consolidation roster (v7.0 T12).

    ``quick_nap`` runs the 3-phase subset on every-N-turns cadence;
    ``nightly_nap`` runs the full roster (Phase 1, 1.5, 2, 2.5, 3,
    4, 5, 8, 9, 10, 11, 12) and is intended for a cron / celery-
    beat / APScheduler job firing once per quiet window per
    tenant.

    Accepts per-call overrides for the Phase 10 knobs so the
    scheduler can tune each tenant independently without rebuilding
    the consolidator.

    Returns:
        ConsolidationReport with every phase that actually fired
        populated in ``phases_run``.
    """
    # Apply per-call overrides by temporarily swapping instance
    # defaults so the existing ``run`` path picks them up without
    # duplicating phase orchestration here.
    original_max = self._episodic_summarization_max_entries
    original_batch = self._episodic_summarization_batch_size
    try:
        if max_entries is not None:
            self._episodic_summarization_max_entries = max_entries
        if summarize_batch is not None:
            self._episodic_summarization_batch_size = summarize_batch
        logger.info(
            "nightly_nap starting: tenant=%s",
            scope.tenant_id,
        )
        report = await self.run(
            scope,
            soul_schema=soul_schema,
            pending_connections=pending_connections,
        )
        logger.info(
            "nightly_nap complete: tenant=%s phases_run=%d errors=%d",
            scope.tenant_id,
            len(report.phases_run),
            len(report.errors),
        )
        return report
    finally:
        self._episodic_summarization_max_entries = original_max
        self._episodic_summarization_batch_size = original_batch

quick_nap async

quick_nap(
    scope: TenantScope,
    *,
    soul_schema: dict[str, Any] | None = None,
    max_entries: int | None = None,
    summarize_batch: int | None = None,
) -> ConsolidationReport

Lightweight consolidation: Phase 1 (strengthen) + Phase 10 (episodic summary).

Runs in <1s vs full run(). Intended for every-N-turns auto-consolidation. Only operates on recently updated semantic nodes within the lookback window.

Parameters:

Name	Type	Description	Default
scope	TenantScope	Tenant scope for multi-tenant isolation.	required
soul_schema	dict[str, Any] | None	Optional SOUL schema. When provided, Phase 5 runs so user corrections recorded in recent nodes are applied to the live schema. Auto-forwarded by SymfonicAgent from FrameworkConfig.domain.soul_schema so callers get Phase 5 without touching the auto-consolidation plumbing.	None
max_entries	int | None	Optional Phase 10 threshold override (per-call). When None, the instance default set at construction is used. Forwarded by the engine from FrameworkConfig.episodic_summarization_max_entries.	None
summarize_batch	int | None	Optional Phase 10 batch size override (per-call). When None, the instance default set at construction is used. Forwarded by the engine from FrameworkConfig.episodic_summarization_batch_size.	None

Returns:

Type	Description
ConsolidationReport	ConsolidationReport with nodes_strengthened, soul_updates,
ConsolidationReport	and episodes_summarized populated.

Source code in src/symfonic/core/learning/consolidation.py

async def quick_nap(
    self,
    scope: TenantScope,
    *,
    soul_schema: dict[str, Any] | None = None,
    max_entries: int | None = None,
    summarize_batch: int | None = None,
) -> ConsolidationReport:
    """Lightweight consolidation: Phase 1 (strengthen) + Phase 10 (episodic summary).

    Runs in <1s vs full run(). Intended for every-N-turns auto-consolidation.
    Only operates on recently updated semantic nodes within the lookback window.

    Args:
        scope: Tenant scope for multi-tenant isolation.
        soul_schema: Optional SOUL schema. When provided, Phase 5 runs so
            user corrections recorded in recent nodes are applied to the
            live schema. Auto-forwarded by ``SymfonicAgent`` from
            ``FrameworkConfig.domain.soul_schema`` so callers get Phase 5
            without touching the auto-consolidation plumbing.
        max_entries: Optional Phase 10 threshold override (per-call). When
            ``None``, the instance default set at construction is used.
            Forwarded by the engine from
            ``FrameworkConfig.episodic_summarization_max_entries``.
        summarize_batch: Optional Phase 10 batch size override (per-call).
            When ``None``, the instance default set at construction is
            used. Forwarded by the engine from
            ``FrameworkConfig.episodic_summarization_batch_size``.

    Returns:
        ConsolidationReport with nodes_strengthened, soul_updates,
        and episodes_summarized populated.
    """
    report = ConsolidationReport(tenant_id=scope.tenant_id)
    logger.debug("quick_nap starting: tenant=%s", scope.tenant_id)

    try:
        all_nodes = await self._graph.query_nodes(scope, layer=MemoryLayer.SEMANTIC)
    except Exception as exc:
        report.errors.append(f"quick_nap: failed to query nodes: {exc}")
        report.finished_at = datetime.now(UTC)
        return report

    cutoff = datetime.now(UTC) - timedelta(hours=self._lookback_hours)
    recent_nodes = [n for n in all_nodes if n.updated_at and n.updated_at >= cutoff]

    # Phase 1: Strengthen recurring nodes
    report.phases_run.append("strengthen")
    try:
        report.nodes_strengthened = await phases.strengthen(
            self._graph, scope, recent_nodes,
            spreading_weight=self._phase1_spreading_weight,
        )
    except Exception as exc:
        logger.warning("quick_nap phase 1 failed: %s", exc)
        report.errors.append(f"quick_nap phase 1 (strengthen): {exc}")

    # Phase 5: SOUL corrections (auto-forwarded by engine from domain).
    # apply_soul_corrections mutates soul_schema in place but only via
    # explicit SOUL_CORRECTION nodes -- safe to re-run because corrections
    # are consumed/cleared as they're applied. Idempotent on repeat calls.
    if soul_schema is not None:
        report.phases_run.append("soul_corrections")
        try:
            report.soul_updates = phases.apply_soul_corrections(
                soul_schema, recent_nodes,
            )
        except Exception as exc:
            logger.warning("quick_nap phase 5 failed: %s", exc)
            report.errors.append(f"quick_nap phase 5 (soul): {exc}")

    # Phase 10: Episodic summarization (only if episodic layer provided).
    # Per-call overrides fall back to instance defaults so call-sites that
    # do not thread FrameworkConfig still respect the ctor-time config.
    if self._episodic is not None:
        report.phases_run.append("episodic_summary")
        effective_max = (
            max_entries
            if max_entries is not None
            else self._episodic_summarization_max_entries
        )
        effective_batch = (
            summarize_batch
            if summarize_batch is not None
            else self._episodic_summarization_batch_size
        )
        try:
            _summary_text, deleted = await summarize_episodic(
                self._episodic,
                scope,
                llm_summarise=self._llm_summarise,
                max_entries=effective_max,
                summarize_batch=effective_batch,
            )
            report.episodes_summarized = deleted
        except Exception as exc:
            logger.warning("quick_nap phase 10 failed: %s", exc)
            report.errors.append(f"quick_nap phase 10 (episodic): {exc}")

    # Phase 13: LLM-judged semantic-duplicate merge (v8.17). Added to
    # the quick_nap roster because quick_nap is the ONLY cadence the
    # engine schedules in-process -- Phase 13 living solely in run()
    # made ``enable_semantic_merge`` dead config for every adopter who
    # did not own a scheduler (adopter-verified against 9.6.0).
    # Default-off gate keeps the pre-8.17 roster byte-identical; the
    # per-run pair budget bounds the LLM cost on this hot-ish path.
    if self._enable_semantic_merge and self._chat_model is not None:
        report.phases_run.append("semantic_merge")
        try:
            from symfonic.core.learning.phases_semantic_merge import (
                merge_semantic_duplicates,
            )

            report.semantic_merges = await merge_semantic_duplicates(
                self._graph,
                scope,
                chat_model=self._chat_model,
                embedding_provider=(
                    self._semantic_merge_embedding_provider
                ),
                embedding_threshold=(
                    self._semantic_merge_embedding_threshold
                ),
                lexical_threshold=(
                    self._semantic_merge_lexical_threshold
                ),
                max_pairs_per_run=(
                    self._semantic_merge_max_pairs_per_run
                ),
            )
        except Exception as exc:
            logger.warning(
                "quick_nap phase 13 (semantic_merge) failed: %s", exc
            )
            report.errors.append(
                f"quick_nap phase 13 (semantic_merge): {exc}"
            )

    report.finished_at = datetime.now(UTC)
    logger.debug(
        "quick_nap complete: tenant=%s strengthened=%d episodes_summarized=%d errors=%d",
        scope.tenant_id,
        report.nodes_strengthened,
        report.episodes_summarized,
        len(report.errors),
    )
    return report

run async

run(
    scope: TenantScope,
    *,
    soul_schema: dict[str, Any] | None = None,
    pending_connections: list[dict[str, Any]] | None = None,
) -> ConsolidationReport

Execute the full consolidation pipeline.

Source code in src/symfonic/core/learning/consolidation.py

async def run(
    self,
    scope: TenantScope,
    *,
    soul_schema: dict[str, Any] | None = None,
    pending_connections: list[dict[str, Any]] | None = None,
) -> ConsolidationReport:
    """Execute the full consolidation pipeline."""
    from symfonic.core.learning.durability_gate import DurabilityGate

    report = ConsolidationReport(tenant_id=scope.tenant_id)
    # v7.26.2: one gate instance shared across Phases 10/11/12 so the
    # per-run transient/expired rejection tally is aggregated.
    gate = DurabilityGate()
    cutoff = datetime.now(UTC) - timedelta(hours=self._lookback_hours)

    logger.info(
        "Consolidation starting: tenant=%s lookback_hours=%d pending_connections=%d",
        scope.tenant_id,
        self._lookback_hours,
        len(pending_connections) if pending_connections else 0,
    )

    try:
        all_nodes = await self._graph.query_nodes(
            scope, layer=MemoryLayer.SEMANTIC,
        )
    except Exception as exc:
        report.errors.append(f"Failed to query nodes: {exc}")
        report.finished_at = datetime.now(UTC)
        return report

    recent_nodes = [
        n for n in all_nodes
        if n.updated_at and n.updated_at >= cutoff
    ]
    logger.info(
        "Consolidation nodes: total=%d recent=%d",
        len(all_nodes), len(recent_nodes),
    )

    # Each phase appends its canonical name to ``report.phases_run``
    # BEFORE invocation so a half-way failure still records the fact
    # the phase started. Conditionally-skipped phases (2.5 / 5 / 10 /
    # 11 / 12) only append inside the branch that actually runs --
    # omission from the list proves short-circuit, presence with a
    # zero counter proves "ran and found nothing". Names match the
    # `phase_<n>_*` test file suffixes for easy correlation.

    # Phase 1: Strengthen recurring nodes
    report.phases_run.append("strengthen")
    try:
        report.nodes_strengthened = await phases.strengthen(
            self._graph, scope, recent_nodes,
            spreading_weight=self._phase1_spreading_weight,
        )
    except Exception as exc:
        logger.warning("phase 1 (strengthen) failed: %s", exc)
        report.errors.append(f"phase 1 (strengthen): {exc}")

    # Phase 1.5: Co-occurrence edges
    report.phases_run.append("cooccurrence")
    try:
        report.edges_created += await phases.create_cooccurrence_edges(
            self._graph, scope, recent_nodes,
        )
    except Exception as exc:
        logger.warning("phase 1.5 (cooccurrence) failed: %s", exc)
        report.errors.append(f"phase 1.5 (cooccurrence): {exc}")

    # Phase 2: Tag risk nodes
    report.phases_run.append("tag_risk")
    try:
        report.risk_nodes_tagged = await phases.tag_risk_nodes(
            self._graph, scope, recent_nodes,
        )
    except Exception as exc:
        logger.warning("phase 2 (tag_risk) failed: %s", exc)
        report.errors.append(f"phase 2 (tag_risk): {exc}")

    # Phase 2.5: Consolidate pending (inferred) edges
    if pending_connections:
        report.phases_run.append("pending_edges")
        try:
            report.edges_created += await phases.consolidate_pending_edges(
                self._graph, scope, all_nodes, pending_connections,
            )
        except Exception as exc:
            logger.warning("phase 2.5 (pending_edges) failed: %s", exc)
            report.errors.append(f"phase 2.5 (pending_edges): {exc}")

    # Phase 3: Prune orphans
    report.phases_run.append("prune_orphans")
    try:
        report.nodes_pruned = await phases.prune_orphans(
            self._graph, scope, all_nodes,
        )
    except Exception as exc:
        logger.warning("phase 3 (prune_orphans) failed: %s", exc)
        report.errors.append(f"phase 3 (prune_orphans): {exc}")

    # Phase 4: Generate meta-nodes from clusters
    report.phases_run.append("meta_nodes")
    try:
        report.meta_nodes_created = await phases.generate_meta_nodes(
            self._graph, scope, all_nodes, self._llm_summarise,
        )
    except Exception as exc:
        logger.warning("phase 4 (meta_nodes) failed: %s", exc)
        report.errors.append(f"phase 4 (meta_nodes): {exc}")

    # Phase 5: SOUL schema update from user corrections
    if soul_schema is not None:
        report.phases_run.append("soul_corrections")
        try:
            report.soul_updates = phases.apply_soul_corrections(
                soul_schema, recent_nodes,
            )
        except Exception as exc:
            logger.warning("phase 5 (soul_corrections) failed: %s", exc)
            report.errors.append(f"phase 5 (soul_corrections): {exc}")

    # Phase 8: Working TTL cleanup
    report.phases_run.append("working_ttl")
    try:
        report.working_ttl_pruned = await cleanup_working_ttl(
            self._graph, scope,
        )
    except Exception as exc:
        logger.warning("phase 8 (working_ttl) failed: %s", exc)
        report.errors.append(f"phase 8 (working_ttl): {exc}")

    # Phase 9: Semantic importance decay
    report.phases_run.append("decay_importance")
    try:
        report.nodes_decayed = await decay_importance(
            self._graph, scope, all_nodes,
        )
    except Exception as exc:
        logger.warning("phase 9 (decay_importance) failed: %s", exc)
        report.errors.append(f"phase 9 (decay_importance): {exc}")

    # Phase 9.5: Prune soft-retracted nodes past their grace window.
    # Soft-retracted (false-positive/corrected) nodes are already invisible
    # to reads via the query_nodes exclusion; this reclaims them once the
    # reversibility window has elapsed.
    report.phases_run.append("prune_retracted")
    try:
        report.retracted_nodes_pruned = await prune_retracted(
            self._graph, scope,
        )
    except Exception as exc:
        logger.warning("phase 9.5 (prune_retracted) failed: %s", exc)
        report.errors.append(f"phase 9.5 (prune_retracted): {exc}")

    # Phase 10: Episodic summarization
    if self._episodic is not None:
        report.phases_run.append("episodic_summary")
        try:
            summary_text, deleted = await summarize_episodic(
                self._episodic,
                scope,
                llm_summarise=self._llm_summarise,
                max_entries=self._episodic_summarization_max_entries,
                summarize_batch=self._episodic_summarization_batch_size,
                gate=gate,
            )
            report.episodes_summarized = deleted
            if summary_text:
                # Late import to avoid circular import via core.__init__
                from symfonic.memory.models.node import MemoryNode

                # Write summary as a semantic meta-node
                meta = MemoryNode(
                    layer=MemoryLayer.SEMANTIC,
                    tenant_id=scope.tenant_id,
                    label="META:episodic_summary",
                    properties={
                        "type": "episodic_summary",
                        "summary": summary_text,
                        "entries_compressed": deleted,
                    },
                    importance=6.0,
                )
                try:
                    await self._graph.add_node(scope, meta)
                    logger.info(
                        "Episodic summary meta-node created: %d entries compressed",
                        deleted,
                    )
                except Exception:
                    logger.debug(
                        "Failed to create episodic summary meta-node",
                        exc_info=True,
                    )
        except Exception as exc:
            logger.warning("phase 10 (episodic_summary) failed: %s", exc)
            report.errors.append({"phase": "episodic_summary", "error": str(exc)})

    # Phase 11: Synthetic linking (episodic -> semantic cross-pollination)
    if self._episodic is not None:
        report.phases_run.append("synthetic_links")
        try:
            from symfonic.core.learning.phases_synthetic import (
                build_synthetic_links,
            )

            report.synthetic_edges_created = await build_synthetic_links(
                self._graph, self._episodic, scope,
                min_co_count=self._synthetic_link_min_co_count,
                gate=gate,
            )
        except Exception as exc:
            logger.warning("phase 11 failed: %s", exc)
            report.errors.append(f"phase 11 (synthetic): {exc}")

    # Phase 12.5: EntityLinker (v7.2 Roadmap Item 12). Default-off via
    # ``enable_entity_linker``. Skipping the flag short-circuits the
    # whole block so ``"entity_links"`` does NOT appear in
    # ``report.phases_run`` (Decision 7.5 / acceptance #1).
    if self._enable_entity_linker and self._episodic is not None:
        report.phases_run.append("entity_links")
        try:
            from symfonic.core.learning.phases_entity import build_entity_links

            extractor = self._build_entity_extractor()
            nodes_added, edges_added = await build_entity_links(
                self._graph, self._episodic, scope,
                extractor=extractor,
                min_mention_count=self._entity_linker_min_mention_count,
                max_episodics_per_run=self._entity_linker_max_episodics_per_run,
                confidence_threshold=self._entity_linker_confidence_threshold,
            )
            report.entity_nodes_created = nodes_added
            report.mentions_edges_created = edges_added
        except Exception as exc:
            logger.warning("phase 12.5 (entity_links) failed: %s", exc)
            report.errors.append(f"phase_12.5 (entity_links): {exc}")

    # Phase 12: Episodic -> procedural promotion (draft skills)
    if self._episodic is not None and self._procedural is not None:
        report.phases_run.append("procedural_promotion")
        try:
            from symfonic.core.learning.phases_procedural_promotion import (
                promote_episodic_to_procedural,
            )

            report.draft_skills_promoted = await promote_episodic_to_procedural(
                self._episodic,
                self._procedural,
                scope,
                min_pattern_count=self._promotion_min_pattern_count,
                recency_days=self._promotion_recency_days,
                max_drafts_per_run=self._promotion_max_drafts_per_run,
                use_tool_calls_fallback=(
                    self._promotion_use_tool_calls_fallback
                ),
                # v7.6: pass the LLM extractor instance when opt-in;
                # ``None`` keeps the regex path running unchanged.
                llm_extractor=self._build_procedural_extractor(),
                llm_model_name=self._phase_12_llm_model,
                # v7.7.2: speaker filter for both extractor paths.
                promote_assistant_content=(
                    self._promotion_promote_assistant_content
                ),
                gate=gate,
            )
        except Exception as exc:
            logger.warning("phase 12 failed: %s", exc)
            report.errors.append(f"phase_12: {exc}")

    # Phase 13: LLM-judged semantic-duplicate merge (v8.16,
    # default-off via ``enable_semantic_merge``; also requires a
    # ``chat_model`` -- the judge IS the correctness gate).  Runs
    # last so it sees the nodes the earlier phases produced.  Newly
    # stamped supersede-retractions are reclaimed by phase 9.5
    # (prune_retracted) in a later run, after the grace window.
    if self._enable_semantic_merge and self._chat_model is not None:
        report.phases_run.append("semantic_merge")
        try:
            from symfonic.core.learning.phases_semantic_merge import (
                merge_semantic_duplicates,
            )

            report.semantic_merges = await merge_semantic_duplicates(
                self._graph,
                scope,
                chat_model=self._chat_model,
                embedding_provider=(
                    self._semantic_merge_embedding_provider
                ),
                embedding_threshold=(
                    self._semantic_merge_embedding_threshold
                ),
                lexical_threshold=(
                    self._semantic_merge_lexical_threshold
                ),
                max_pairs_per_run=(
                    self._semantic_merge_max_pairs_per_run
                ),
            )
        except Exception as exc:
            logger.warning("phase 13 (semantic_merge) failed: %s", exc)
            report.errors.append(f"phase_13 (semantic_merge): {exc}")

    # v7.26.2: surface the aggregate durability-gate rejection tally.
    report.transient_rejected = gate.rejected

    report.finished_at = datetime.now(UTC)
    logger.info(
        "Consolidation complete: tenant=%s strengthened=%d pruned=%d "
        "edges_created=%d meta_nodes=%d soul_updates=%d "
        "working_ttl_pruned=%d nodes_decayed=%d episodes_summarized=%d "
        "errors=%d",
        scope.tenant_id,
        report.nodes_strengthened,
        report.nodes_pruned,
        report.edges_created,
        report.meta_nodes_created,
        report.soul_updates,
        report.working_ttl_pruned,
        report.nodes_decayed,
        report.episodes_summarized,
        len(report.errors),
    )
    return report