symfonic.agent.cutover.memory_fold

memory_fold

When a folded bundle gets a memory capability, and which ports it gets.

TA8.71. One call site held two defects, and each produced a store that stayed empty while :meth:RetrievalBundle.serves_consolidation answered True:

• the capability was built only if hydrator is not None, so a deployment that consolidates without hydrating got none at all -- no write stage was declared, the producer was never called, and the turn recorded nothing. Measured as zero records produced, which reads like "nothing worth keeping" rather than like a wiring gap;
• lifecycle was never passed, and the capability declares its finalize stage only when it has one. So even a hydrating deployment staged its records and never published them -- write once, flush never -- and pending rows are not retrievable by contract.

The second is the sharper lesson. serves_consolidation() asks whether the ports can flush, and they could. What nobody asked is whether anything calls them. Capability is not occurrence, and that predicate's licence now rests on tests/agent/test_turn_publishes_to_the_store.py, which takes a turn through a real agent and reads the store back, rather than on the ports being present.

CORE-227 found the same shape one stage later: the producer filed the exchange and nothing filed the reply's extraction operations, because no extractor was folded. A consolidating bundle now folds :class:~symfonic.agent.cutover.compat_extraction.CompatibilityExtractor, and :func:observed_handlers reports the retrieval stage's admitted count to the facade's turn observation so memory_entries_used is measured, not zero, and logs extracted memories the write port did not accept.

It also found the hydrating half losing one legacy effect: legacy hydration re-reads each recalled node through GraphMemoryStore.get_node, which advances its access_count and updated_at. Over the engine-built :class:~symfonic.capabilities.memory.legacy_port.LegacyRetrievalPort the hydrator is wrapped in :class:~symfonic.capabilities.memory.legacy_access.LegacyAccessHydrator, which restores that re-read and nothing else.

memory_capability

memory_capability(hydrator: Any, *, scope: Any, limit: int, writes: Any, records_from: Any, lifecycle: Any, consolidating: bool, retrieval: Any = None) -> MemoryCapability | None

The capability this bundle should fold, or None for neither half.

Folded when the deployment hydrates or consolidates: the two halves are independent, and a deployment asking for one must not be silently given neither. Which stages it then declares is the capability's own decision, made per port -- it declares retrieval only with a hydrator, the write stage only with a writer and a producer, and finalize only with a lifecycle.

Source code in src/symfonic/agent/cutover/memory_fold.py

def memory_capability(
    hydrator: Any,
    *,
    scope: Any,
    limit: int,
    writes: Any,
    records_from: Any,
    lifecycle: Any,
    consolidating: bool,
    retrieval: Any = None,
) -> MemoryCapability | None:
    """The capability this bundle should fold, or ``None`` for neither half.

    Folded when the deployment hydrates **or** consolidates: the two halves are
    independent, and a deployment asking for one must not be silently given
    neither. Which stages it then declares is the capability's own decision,
    made per port -- it declares retrieval only with a hydrator, the write
    stage only with a writer and a producer, and finalize only with a
    lifecycle.
    """
    if hydrator is None and not consolidating:
        return None
    extractor = None
    if consolidating:
        from symfonic.agent.cutover.compat_extraction import CompatibilityExtractor

        extractor = CompatibilityExtractor()
    return MemoryCapability(
        _with_legacy_access(hydrator, retrieval),
        scope=scope,
        limit=limit,
        writer=writes,
        records_from=records_from,
        lifecycle=lifecycle,
        extractor=extractor,
    )

observed_handlers

observed_handlers(handlers: Mapping[str, Any]) -> dict[str, Any]

handlers, with retrieval and consolidation reporting what they did.

Both wrapped handlers return their result unchanged. Retrieval reports its admitted count, a no-op unless the facade opened a turn observation. Consolidation logs when fewer memories were written than extracted: the engine-built write port refuses a layer it has no store for (prospective, procedural) by name on its receipt, which the stage reports only as a count, and a fact a legacy turn would have filed must not vanish silently.

Source code in src/symfonic/agent/cutover/memory_fold.py

def observed_handlers(handlers: Mapping[str, Any]) -> dict[str, Any]:
    """``handlers``, with retrieval and consolidation reporting what they did.

    Both wrapped handlers return their result unchanged. Retrieval reports its
    admitted count, a no-op unless the facade opened a turn observation.
    Consolidation logs when fewer memories were written than extracted: the
    engine-built write port refuses a layer it has no store for (prospective,
    procedural) by name on its receipt, which the stage reports only as a
    count, and a fact a legacy turn would have filed must not vanish silently.
    """
    wrapped = dict(handlers)
    recall_inner = wrapped.get(RETRIEVAL_STAGE)
    consolidate_inner = wrapped.get(CONSOLIDATION_STAGE)

    async def recall(context: Any) -> Any:
        result = await recall_inner(context)
        observe_recall(getattr(result, "counts", None))
        return result

    async def consolidate(context: Any) -> Any:
        result = await consolidate_inner(context)
        counts = getattr(result, "counts", None) or {}
        extracted, written = counts.get("extracted", 0), counts.get("written", 0)
        if extracted > written:
            reason = getattr(result, "reason", "") or "refused by the write port"
            logger.warning(
                "%d of %d extracted memories were not written on the kernel route: %s",
                extracted - written,
                extracted,
                reason,
            )
        return result

    if recall_inner is not None:
        wrapped[RETRIEVAL_STAGE] = recall
    if consolidate_inner is not None:
        wrapped[CONSOLIDATION_STAGE] = consolidate
    return wrapped