Skip to content

symfonic.capabilities.memory.nap_stage

nap_stage

The handler behind the nap stage: fire the due cycle, do not hold the answer.

Three things happen here that used to be spread across the engine, and each of them was a defect where it was:

  • The trigger is the turn, not the agent. _maybe_schedule_quick_nap counted turns on the agent, so two tenants served by one process shared a cadence and either consolidated on the other's turns. The coordinator counts per scope, and this stage tells it about the scope this turn actually ran for.
  • The nap runs after the flush. Legacy fired it once the turn was over; this sits after memory.lifecycle in the same phase, so the roster reads a store that already contains what the turn produced rather than one that is about to.
  • The task has an owner, and it is not the run. asyncio.create_task put the cycle in a process-wide set nobody waited for and nobody reported; the coordinator holds it and drains it at shutdown. Not the run's background registry, which was the other candidate: that bounds the cycle to the run's deadline and cancels it at that run's teardown, so a consumer hanging up would abandon maintenance for memories already published -- and because draining a run's events runs its teardown, Agent.run would wait for the very nap the deployment asked to run in the background.

A nap never fails the turn. Every stage result here is applied or no_change: the answer is delivered, the memories are already published, and a maintenance pass that could not run is a line in a log, not a broken turn.

nap_handler 

nap_handler(capability: Any) -> Callable[[Any], Any]

Advance this scope's cadence, and run the cycle it makes due.

Source code in src/symfonic/capabilities/memory/nap_stage.py 
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
def nap_handler(capability: Any) -> Callable[[Any], Any]:
    """Advance this scope's cadence, and run the cycle it makes due."""

    async def handle_nap(context: Any) -> StageResult[Any]:
        turn_request = getattr(context, "request", None)
        if turn_request is None:  # pragma: no cover - defensive
            return no_change(NO_SCOPE)

        coordinator = capability.consolidation
        # The turn's scope wins over the configured one, for the reason the
        # write half states: a deployment folded once and serving many tenants
        # would otherwise nap the same default scope on every tenant's turns.
        scope = getattr(turn_request, "scope", None) or capability.scope
        cycle = coordinator.after_turn(scope)
        if cycle is None:
            return no_change(NOT_DUE)

        run_id = str(getattr(turn_request, "run_id", "") or "")
        root_run_id = str(getattr(turn_request, "root_run_id", "") or run_id)

        if coordinator.background:
            try:
                coordinator.schedule_cycle(
                    scope, cycle, run_id=run_id, root_run_id=root_run_id
                )
            except Exception as exc:  # noqa: BLE001 - a nap never fails the turn
                # Reported, not raised: this stage runs after the answer
                # exists, and a crashed finalize stage ends the turn (EVT-7).
                logger.warning("consolidation could not be scheduled: %s", exc)
                return no_change(f"the {cycle.value} cycle could not be scheduled")
            # The cycle name, not its state: the state does not exist yet,
            # and reporting one here would be reporting a cycle that has not
            # run. The counts are deliberately absent rather than zero -- this
            # stage counted nothing, which is a different sentence from a nap
            # that ran and found nothing.
            return applied(
                ResolvedInput(
                    capability=HMS_CAPABILITY,
                    value=SCHEDULED.format(cycle=cycle.value),
                    provenance=scope.path,
                )
            )

        # Inline, because the deployment asked the turn to wait for its own
        # consolidation.
        state = await coordinator.consolidate(
            scope, cycle, run_id=run_id, root_run_id=root_run_id
        )
        if state is None:
            return no_change(f"the {cycle.value} cycle failed; nothing was published")
        # The ledger rides on ``counts`` because that is the field whose
        # contract already says integers only, for the reason a consolidation
        # makes acute: these numbers reach an execution log and durable
        # storage, and the thing counted is a tenant's memories. The
        # contribution is the cycle's status, projected into a string -- a
        # resolved input must be immutable, and ``ConsolidationState`` is the
        # runtime's own mutable record.
        return applied(
            ResolvedInput(
                capability=HMS_CAPABILITY,
                value=f"{cycle.value}:{state.status}",
                provenance=scope.path,
            ),
            counts=state.ledger,
        )

    return handle_nap