Skip to content

symfonic.kernel

kernel

symfonic.kernel — compiled invocation and shared event projections.

Two things live here and nothing else: the function that turns a normalized configuration into a frozen :class:InvocationPlan, and the loop that executes one. Both are written against symfonic.kernel.contracts and the standard library alone — no provider SDK, no message library, no transport — which is the property the architecture gate enforces on every commit.

The W1 bootstrap invocation core (symfonic.agent._bootstrap) was absorbed into this package by T2.3.2, retiring the transitional second path its charter sanctioned. The public facade's own behavior is unchanged, and the T2.1.3 contract suite reruns against this kernel to prove it.

AdapterMetrics dataclass 

AdapterMetrics(high_watermark: int = 0, byte_high_watermark: int = 0, events_shed: dict[str, int] = dict(), blocked_seconds: float = 0.0, terminal_delivery_failed: bool = False, abandoned: bool = False)

Observable pressure for one adapter on one run (BP-12).

BackgroundRegistry 

BackgroundRegistry(run_id: str, *, run_deadline_seconds: float | None = None)

Every task one invocation spawned, owned from creation to teardown.

Source code in src/symfonic/kernel/background.py 
54
55
56
57
58
59
def __init__(self, run_id: str, *, run_deadline_seconds: float | None = None) -> None:
    self.run_id = run_id
    self._run_deadline = run_deadline_seconds
    self._tasks: list[asyncio.Task[Any]] = []
    self._entries: list[BackgroundEntry] = []
    self._closed = False

entries property 

entries: tuple[BackgroundEntry, ...]

The owner/purpose/deadline of every task still held.

drain async 

drain(*, grace_seconds: float) -> DrainReport

Await what finishes inside the grace window, cancel and await the rest.

Closing first is what makes the window a bound: a task that spawned another task on its way out would otherwise refill the registry behind the drain, and the loop would be as long as the work chose to make it.

Source code in src/symfonic/kernel/background.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
async def drain(self, *, grace_seconds: float) -> DrainReport:
    """Await what finishes inside the grace window, cancel and await the rest.

    Closing *first* is what makes the window a bound: a task that spawned
    another task on its way out would otherwise refill the registry behind
    the drain, and the loop would be as long as the work chose to make it.
    """
    self._closed = True
    owned = tuple(zip(self._tasks, self._entries, strict=True))
    self._tasks.clear()
    self._entries.clear()
    if not owned:
        return DrainReport()

    remaining = [task for task, _ in owned if not task.done()]
    if remaining and grace_seconds > 0:
        _, unfinished = await asyncio.wait(remaining, timeout=grace_seconds)
        remaining = list(unfinished)

    for task in remaining:
        task.cancel()
    if remaining:
        await asyncio.gather(*remaining, return_exceptions=True)
    return DrainReport(
        awaited=len(owned) - len(remaining),
        cancelled=len(remaining),
        failures=_failures(owned),
    )

spawn 

spawn(work: Coroutine[Any, Any, Any], *, owner: str, purpose: str, deadline_seconds: float | None = None) -> asyncio.Task[Any]

Register and start one unit of run-owned work.

Every rejection closes work first. A coroutine that is refused and then left unawaited would surface as a RuntimeWarning from whatever code happened to run next, attributing this module's refusal to an innocent bystander.

Source code in src/symfonic/kernel/background.py 
 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
def spawn(
    self,
    work: Coroutine[Any, Any, Any],
    *,
    owner: str,
    purpose: str,
    deadline_seconds: float | None = None,
) -> asyncio.Task[Any]:
    """Register and start one unit of run-owned work.

    Every rejection closes ``work`` first. A coroutine that is refused and
    then left unawaited would surface as a ``RuntimeWarning`` from whatever
    code happened to run next, attributing this module's refusal to an
    innocent bystander.
    """
    deadline = deadline_seconds if deadline_seconds is not None else self._run_deadline
    problem = self._reject(owner, purpose, deadline_seconds)
    if problem is not None:
        _discard(work)
        raise ContractViolationError(problem)

    entry = BackgroundEntry(owner=owner, purpose=purpose, deadline_seconds=deadline)
    task = asyncio.create_task(
        _bounded(work, deadline, owner, purpose),
        name=f"{owner}:{purpose}:{self.run_id}",
    )
    self._tasks.append(task)
    self._entries.append(entry)
    return task

BoundedEventBuffer 

BoundedEventBuffer(policy: EventAdapter, *, deadline_seconds: float | None = None)

A per-run queue bounded by both event count and approximate bytes.

One slot is held back from non-terminal events under reserve. Under preempt, a terminal event evicts only the oldest non-terminal entries. No module-global registry or storage is involved (BP-8/BP-13).

Source code in src/symfonic/kernel/backpressure.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
def __init__(
    self, policy: EventAdapter, *, deadline_seconds: float | None = None
) -> None:
    if policy.buffer != "bounded":
        raise ContractViolationError(
            f"{policy.name!r} is not declared as a bounded adapter (BP-1)."
        )
    self.policy = policy
    self.metrics = AdapterMetrics()
    self._items: deque[tuple[KernelEvent, int]] = deque()
    self._bytes = 0
    self._closed = False
    self._drops: dict[tuple[str, int], tuple[int, int, int, str]] = {}
    self._condition = asyncio.Condition()
    self._deadline = (
        time.monotonic() + deadline_seconds
        if deadline_seconds is not None
        else None
    )

close async 

close(*, abandoned: bool = False) -> None

Detach this run's consumer and release all buffered references.

Source code in src/symfonic/kernel/backpressure.py 
127
128
129
130
131
132
133
134
135
async def close(self, *, abandoned: bool = False) -> None:
    """Detach this run's consumer and release all buffered references."""
    async with self._condition:
        self.metrics.abandoned = abandoned
        self._closed = True
        self._items.clear()
        self._drops.clear()
        self._bytes = 0
        self._condition.notify_all()

get async 

get() -> KernelEvent

Take the oldest surviving event, preserving emission order.

Source code in src/symfonic/kernel/backpressure.py 
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
async def get(self) -> KernelEvent:
    """Take the oldest surviving event, preserving emission order."""
    async with self._condition:
        while not self._items and not self._drops:
            self._require_open()
            await self._condition.wait()
        if self._drops:
            first_drop = min(item[1] for item in self._drops.values())
            first_item = self._items[0][0].index if self._items else None
            if first_item is None or first_drop < first_item:
                return self._drop_notice()
        event, size = self._items.popleft()
        self._bytes -= size
        self._condition.notify_all()
        return event

put async 

put(event: KernelEvent) -> bool

Put an event, returning False only for a declared shed.

Source code in src/symfonic/kernel/backpressure.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
async def put(self, event: KernelEvent) -> bool:
    """Put an event, returning ``False`` only for a declared shed."""
    size = _event_size(event)
    terminal = event.kind in _TERMINAL
    async with self._condition:
        self._require_open()
        if terminal and self.policy.terminal_policy == "preempt":
            self._preempt_until_fits(size)
        if self._fits(size, terminal):
            self._append(event, size)
            return True
        if not terminal and event.kind in self.policy.sheddable:
            self.metrics.record_shed(event.kind)
            self._record_drop(event)
            return False
        await self._wait_for_space(size, terminal)
        self._require_open()
        self._append(event, size)
        return True

CallbackEventAdapter 

CallbackEventAdapter(policy: EventAdapter, callbacks: Iterable[Callable[[KernelEvent], Any]], *, deadline_seconds: float | None, context: RequestContext | None = None)

Awaited, ordered, error-isolated callback event delivery.

Source code in src/symfonic/kernel/fanout.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
def __init__(
    self,
    policy: EventAdapter,
    callbacks: Iterable[Callable[[KernelEvent], Any]],
    *,
    deadline_seconds: float | None,
    context: RequestContext | None = None,
) -> None:
    self._policy = policy
    self._callbacks = tuple(callbacks)
    self._deadline_seconds = deadline_seconds
    self._context = context
    self._buffer = (
        BoundedEventBuffer(policy, deadline_seconds=deadline_seconds)
        if policy.buffer == "bounded"
        else None
    )
    self._worker: asyncio.Task[Any] | None = None
    self._terminal_owed = False
    self.failures = 0

CompileRequest dataclass 

CompileRequest(config_digest: str, scope: RequestScope = RequestScope(), model: ModelResolution = ModelResolution(), instructions: str | None = None, tools: Sequence[ToolDescriptor] = (), stages: Sequence[StageDescriptor] = (), bindings: ServiceBindings = ServiceBindings(), effect_grants: frozenset[str] = frozenset(), limits: PlanLimits = PlanLimits(), event_program: EventProgram = EventProgram(), capabilities: Sequence[str] = ())

The compiler's only accepted input (IPL-1 step 1).

It is a normalized value, never raw adopter configuration: re-parsing legacy input inside the compiler is how a second, subtly different interpretation of a config file gets born.

DrainReport dataclass 

DrainReport(awaited: int = 0, cancelled: int = 0, failures: tuple[BackgroundFailure, ...] = ())

What draining the registry cost: work that finished, work cut, work that failed.

failures is not a subset of cancelled: a task can finish well inside the grace window and still have raised. Reporting the two separately is what lets teardown say "nothing was forced, but something owned by this run broke" — a sentence the old awaited/cancelled pair could not form.

InvocationKernel

Bases: CapabilityRungs

Executes a compiled plan. Holds no state of its own, ever.

The three per-round capability rungs -- post-model, pre-tool and post-tool -- are inherited from :class:~symfonic.kernel.rungs.CapabilityRungs, which is a split for the 300-line module budget and nothing more: they are still methods on this object, because the runner reaches them through the kernel it was handed.

One instance may serve unlimited concurrent invocations: everything a run accumulates lives in the :class:RequestContext created for that run, so two runs of the same plan share nothing but the frozen plan itself.

Every entry point accepts that context as an optional argument. It is a diagnostics seam, not a sharing mechanism — a context that has already served an invocation is refused (RCX-3) — and it is what makes RCX-11's teardown record readable from all four projections rather than from none of them. A lifecycle guarantee no caller can observe is one no caller can hold us to.

assemble_prompt 

assemble_prompt(plan: InvocationPlan, request: TurnRequest) -> PromptAssembly

The prompt-assembly phase, as a value.

A pure function of (plan, request) (STG-7): it produces a value and performs nothing, which is what keeps the assembled prompt reproducible from the plan alone. Returning the assembly rather than opening the turn with it is what gives a future PROMPT_ASSEMBLY stage something to receive and something to hand back — the seam W2 needs, without any dispatch yet.

Synchronous on purpose. A deadline reaches work through lifecycle.bounded, which bounds an awaitable; there is no await point here for one to fire at, and adding one to a pure function would buy nothing. When stage dispatch lands it is the dispatcher that must establish the bounded await, not this method.

Source code in src/symfonic/kernel/invoker.py 
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
def assemble_prompt(self, plan: InvocationPlan, request: TurnRequest) -> PromptAssembly:
    """The ``prompt-assembly`` phase, as a value.

    A pure function of ``(plan, request)`` (STG-7): it produces a value and
    performs nothing, which is what keeps the assembled prompt reproducible
    from the plan alone. Returning the assembly rather than opening the turn
    with it is what gives a future ``PROMPT_ASSEMBLY`` stage something to
    receive and something to hand back — the seam W2 needs, without any
    dispatch yet.

    Synchronous on purpose. A deadline reaches work through
    ``lifecycle.bounded``, which bounds an *awaitable*; there is no await
    point here for one to fire at, and adding one to a pure function would
    buy nothing. When stage dispatch lands it is the dispatcher that must
    establish the bounded await, not this method."""
    return PromptAssembly(
        instructions=plan.instructions,
        prompt=request.prompt,
        attachments=tuple(request.attachments),
        history=tuple(request.history),
    )

bind 

bind(plan: InvocationPlan, ctx: RequestContext) -> None

The kernel-owned bind phase: capture the generation vector.

Separated from :meth:assemble_prompt so the two phases the ladder names as distinct are distinct in the code as well. They were one method, which meant prompt-assembly had no seam a stage could ever run at — a capability could declare a stage there and the kernel had nowhere to call it, which is one of the ways "declared but inert" happened.

bind stays kernel-only (KERNEL_OWNED_PHASES): a capability that could inject here would observe or outlive a run it does not own.

Source code in src/symfonic/kernel/invoker.py 
138
139
140
141
142
143
144
145
146
147
148
149
150
151
def bind(self, plan: InvocationPlan, ctx: RequestContext) -> None:
    """The kernel-owned ``bind`` phase: capture the generation vector.

    Separated from :meth:`assemble_prompt` so the two phases the ladder
    names as distinct are distinct in the code as well. They were one
    method, which meant ``prompt-assembly`` had no seam a stage could ever
    run at — a capability could declare a stage there and the kernel had
    nowhere to call it, which is one of the ways "declared but inert"
    happened.

    ``bind`` stays kernel-only (``KERNEL_OWNED_PHASES``): a capability that
    could inject here would observe or outlive a run it does not own.
    """
    ctx.bind_generation(plan.identity.plan_id)

finish async 

finish(plan: InvocationPlan, ctx: RequestContext, transcript: Any) -> InvocationOutcome

Prepare the outcome before the runner dispatches commit finalizers.

Extraction and conversation projection can fail. The runner calls this before FINALIZE and refreshes duration after it, without repeating either potentially failing operation after a successful commit.

Source code in src/symfonic/kernel/invoker.py 
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
async def finish(
    self, plan: InvocationPlan, ctx: RequestContext, transcript: Any
) -> InvocationOutcome:
    """Prepare the outcome before the runner dispatches commit finalizers.

    Extraction and conversation projection can fail. The runner calls this
    before FINALIZE and refreshes duration after it, without repeating either
    potentially failing operation after a successful commit.
    """
    response = plan.bindings.response
    output = (
        await StructuredOutputAdapter.attach(plan).extract(transcript)
        if response.structured
        else None
    )
    return InvocationOutcome(
        text=ctx.text,
        output=output,
        messages=plan.bindings.conversation.messages(transcript),
        tool_outcomes=tuple(ctx.tool_outcomes),
        usage=ctx.usage,
        run_id=ctx.run_id,
        duration_ms=ctx.duration_ms,
        stop_reason=ctx.stop_reason,
    )

open_turn 

open_turn(plan: InvocationPlan, ctx: RequestContext, request: TurnRequest) -> Any

Run the kernel-owned bind phase, then assemble the prompt.

Composition only — the two phases are :meth:bind and :meth:assemble_prompt. Kept as one entry point so runner and every existing caller are unchanged by the split.

The synchronous path, and it dispatches nothing: bind and assemble_prompt, no capability stage. A capability that contributed a prompt-assembly stage sees it run on the other path and not on this one.

It has no caller in src/. All four public projections — run, stream, stream_text, stream_typed — go through InvocationRunner.events, which calls :meth:open_turn_dispatched. The docstring here used to say this was "what runner calls today", which stopped being true when the runner moved.

Kept public and kept dispatch-free deliberately: a caller that wants the two phases without adopter code running in them has one entry point that says so, and a contract test pins that it stays silent (#23 slice 3).

Source code in src/symfonic/kernel/invoker.py 
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
def open_turn(self, plan: InvocationPlan, ctx: RequestContext, request: TurnRequest) -> Any:
    """Run the kernel-owned ``bind`` phase, then assemble the prompt.

    Composition only — the two phases are :meth:`bind` and
    :meth:`assemble_prompt`. Kept as one entry point so ``runner`` and every
    existing caller are unchanged by the split.

    The synchronous path, and it **dispatches nothing**: ``bind`` and
    ``assemble_prompt``, no capability stage. A capability that contributed
    a ``prompt-assembly`` stage sees it run on the other path and not on
    this one.

    It has no caller in ``src/``. All four public projections — ``run``,
    ``stream``, ``stream_text``, ``stream_typed`` — go through
    ``InvocationRunner.events``, which calls :meth:`open_turn_dispatched`.
    The docstring here used to say this was "what ``runner`` calls today",
    which stopped being true when the runner moved.

    Kept public and kept dispatch-free deliberately: a caller that wants the
    two phases without adopter code running in them has one entry point that
    says so, and a contract test pins that it stays silent (#23 slice 3).
    """
    self.bind(plan, ctx)
    assembly = self.assemble_prompt(plan, request)
    return plan.bindings.conversation.open_turn(assembly)

open_turn_dispatched async 

open_turn_dispatched(plan: InvocationPlan, ctx: RequestContext, request: TurnRequest, *, handlers: Mapping[str, Any] | None = None) -> tuple[Any, tuple[Any, ...]]

bind + prompt-assembly with the stages actually dispatched.

The composition lives in :mod:symfonic.kernel.prompt_assembly; this is the kernel-facing name for it, so a caller reaches one object not two.

Source code in src/symfonic/kernel/invoker.py 
201
202
203
204
205
206
207
208
209
210
211
212
213
214
async def open_turn_dispatched(
    self,
    plan: InvocationPlan,
    ctx: RequestContext,
    request: TurnRequest,
    *,
    handlers: Mapping[str, Any] | None = None,
) -> tuple[Any, tuple[Any, ...]]:
    """``bind`` + ``prompt-assembly`` with the stages actually dispatched.

    The composition lives in :mod:`symfonic.kernel.prompt_assembly`; this is
    the kernel-facing name for it, so a caller reaches one object not two.
    """
    return await run_prompt_assembly(self, plan, ctx, request, handlers=handlers)

reserve_requests staticmethod 

reserve_requests(ctx: RequestContext, requests: Sequence[ToolRequest]) -> tuple[ToolRequest, ...]

Give every requested call a run-unique join key (RES-3).

Tool authorization is not re-derived here: the bound tool port is built from plan group G4 and is the single manifest source, so a second allowlist check in the loop would be a second answer to the same question (IPL-5).

Source code in src/symfonic/kernel/invoker.py 
216
217
218
219
220
221
222
223
224
225
226
227
@staticmethod
def reserve_requests(
    ctx: RequestContext, requests: Sequence[ToolRequest]
) -> tuple[ToolRequest, ...]:
    """Give every requested call a run-unique join key (RES-3).

    Tool *authorization* is not re-derived here: the bound tool port is
    built from plan group G4 and is the single manifest source, so a second
    allowlist check in the loop would be a second answer to the same
    question (IPL-5).
    """
    return tuple(ctx.reserve(request) for request in requests)

run async 

run(plan: InvocationPlan, request: TurnRequest, *, context: RequestContext | None = None) -> Any

Collect the blocking projection of the runner's event stream.

The run's deadline is not re-applied here. It is billed once, by the runner, on every step of forward progress (RunLifecycle.bounded); a second enforcer would be a second answer to "did this run run out of time", the two would race, and the teardown record would name whichever timer won (IPL-5). Everything outside the runner is the collector's own iteration and one synchronous build_result; neither can hang.

Source code in src/symfonic/kernel/invoker.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
async def run(
    self,
    plan: InvocationPlan,
    request: TurnRequest,
    *,
    context: RequestContext | None = None,
) -> Any:
    """Collect the blocking projection of the runner's event stream.

    The run's deadline is *not* re-applied here. It is billed once, by the
    runner, on every step of forward progress (``RunLifecycle.bounded``); a
    second enforcer would be a second answer to "did this run run out of
    time", the two would race, and the teardown record would name whichever
    timer won (IPL-5). Everything outside the runner is the collector's own
    iteration and one synchronous ``build_result``; neither can hang."""
    from symfonic.kernel.runner import InvocationRunner

    ctx = context if context is not None else RequestContext(
        plan, run_id=request.run_id or None
    )
    source = InvocationRunner().events(self, plan, request, stream_model=False, context=ctx)
    return await ResultCollector.attach(plan).collect(source)

stream 

stream(plan: InvocationPlan, request: TurnRequest, *, context: RequestContext | None = None) -> AsyncIterator[KernelEvent]

Return the event stream for one invocation (EVT-1…EVT-10).

Source code in src/symfonic/kernel/invoker.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def stream(
    self,
    plan: InvocationPlan,
    request: TurnRequest,
    *,
    context: RequestContext | None = None,
) -> AsyncIterator[KernelEvent]:
    """Return the event stream for one invocation (EVT-1…EVT-10)."""
    from symfonic.kernel.runner import InvocationRunner

    source = InvocationRunner().events(self, plan, request, stream_model=True, context=context)
    return TypedStreamAdapter.attach(plan).project(source)

stream_text 

stream_text(plan: InvocationPlan, request: TurnRequest, *, context: RequestContext | None = None) -> AsyncIterator[str]

Return the string-delta projection without widening the facade API.

Source code in src/symfonic/kernel/invoker.py 
103
104
105
106
107
108
109
110
111
112
113
114
def stream_text(
    self,
    plan: InvocationPlan,
    request: TurnRequest,
    *,
    context: RequestContext | None = None,
) -> AsyncIterator[str]:
    """Return the string-delta projection without widening the facade API."""
    from symfonic.kernel.runner import InvocationRunner

    source = InvocationRunner().events(self, plan, request, stream_model=True, context=context)
    return TextStreamAdapter.attach(plan).project(source)

stream_typed 

stream_typed(plan: InvocationPlan, request: TurnRequest, *, context: RequestContext | None = None) -> AsyncIterator[KernelEvent]

The typed/structured projection, attached in its own right.

It used to be return self.stream(...). ST2 (TA8.29) builds the facade's typed route on this method, and the alias became a hazard: a later concession to stream would silently retarget the typed route, and not inheriting the StreamChunk projection's losses is that route's whole reason to exist. Same runner, adapter and events, separately attached, so either can change alone.

Source code in src/symfonic/kernel/invoker.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
def stream_typed(
    self,
    plan: InvocationPlan,
    request: TurnRequest,
    *,
    context: RequestContext | None = None,
) -> AsyncIterator[KernelEvent]:
    """The typed/structured projection, attached in its own right.

    It used to be ``return self.stream(...)``. ST2 (TA8.29) builds the
    facade's typed route on *this* method, and the alias became a hazard: a
    later concession to ``stream`` would silently retarget the typed route,
    and not inheriting the ``StreamChunk`` projection's losses is that
    route's whole reason to exist. Same runner, adapter and events,
    separately attached, so either can change alone."""
    from symfonic.kernel.runner import InvocationRunner

    source = InvocationRunner().events(self, plan, request, stream_model=True, context=context)
    return TypedStreamAdapter.attach(plan).project(source)

InvocationRunner

Drive a run and emit its sole internal stream.

stream_model selects only the provider transport. It does not select a loop, accumulator, result builder, lifecycle, or event pipeline: those are all this method, once.

events async 

events(kernel: Any, plan: InvocationPlan, request: TurnRequest, *, stream_model: bool, context: RequestContext | None = None) -> AsyncIterator[KernelEvent]

Yield one run's events, owning its lifecycle on every exit path.

context is a seam, not a sharing mechanism: a caller that needs to read the run's teardown record supplies the context it constructed for this run. A context that has already run is refused, which is RCX-3 enforced rather than merely documented.

Source code in src/symfonic/kernel/runner.py 
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 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
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
async def events(
    self,
    kernel: Any,
    plan: InvocationPlan,
    request: TurnRequest,
    *,
    stream_model: bool,
    context: RequestContext | None = None,
) -> AsyncIterator[KernelEvent]:
    """Yield one run's events, owning its lifecycle on every exit path.

    ``context`` is a seam, not a sharing mechanism: a caller that needs to
    read the run's teardown record supplies the context it constructed for
    *this* run. A context that has already run is refused, which is RCX-3
    enforced rather than merely documented.
    """
    ctx = context if context is not None else RequestContext(
        plan, run_id=request.run_id or None
    )
    lifecycle = ctx.lifecycle
    if lifecycle.closed or ctx.generation is not None:
        raise ContractViolationError(
            f"request context {ctx.run_id} has already served an invocation; "
            "a context belongs to exactly one invocation (RCX-3)."
        )
    # Deferred: a turn needs these, resolving the facade does not, and
    # the import-footprint gate counts the latter.
    from symfonic.kernel.run_observation import (
        Emitter,
        RunObservation,
        terminal_on_exit,
    )

    emitter = Emitter(ctx.run_id)
    from symfonic.kernel.observability_binding import bind_observability
    bind_observability(plan.bindings.event_sink, request)
    callbacks = CallbackEventAdapter.attach(plan, ctx) if plan.bindings.event_sink else None
    reason = "completed"
    async def publish(event: KernelEvent) -> KernelEvent:
        if callbacks is not None:
            await callbacks.deliver(event)
        return event
    observed = RunObservation(ctx, request, callbacks, emitter)
    try:
        try:
            # W2/1b: the turn opens through the dispatcher, so the
            # ``prompt-assembly`` stages a capability contributed actually
            # run. Bounded like every other step of forward progress —
            # ``open_turn`` was synchronous and therefore outside the
            # deadline, and a stage is adopter code that can block.
            transcript, stage_traces = await lifecycle.bounded(
                kernel.open_turn_dispatched(plan, ctx, request),
                doing="assembling the prompt",
            )
            for staged in observed.stage_events(stage_traces):
                yield await publish(staged)
            model = model_for(plan, ctx)  # TA8.63: see kernel/palette.py
            # Egress governance must accept a completed draft before it is
            # exposed; ungoverned streams keep immediate delta delivery.
            buffer_egress_text = any(
                stage.stage_id == "governance.egress"
                for stage in plan.stage_program.stages
            )
            for _ in range(plan.limits.max_model_rounds):
                # #147: the round reaches pre-model before it is sent.
                # Inside the loop, not beside prompt-assembly: on round two
                # and later this transcript already carries the previous
                # round's tool results, and seeing that is the only thing
                # this rung offers that prompt-assembly cannot.
                #
                # Charged to the same deadline as the other rungs, for
                # open_turn_dispatched's reason: a stage is adopter code
                # and it can block.
                stage_traces = await lifecycle.bounded(
                    kernel.dispatch_pre_model(
                        plan, request, resolved=ctx.resolved,
                        transcript=transcript,
                    ),
                    doing="dispatching pre-model stages",
                )
                for staged in observed.stage_events(stage_traces):
                    yield await publish(staged)
                # Do not allocate a public event while egress may still
                # refuse it.  ``Emitter.make`` consumes a sequence index,
                # so withholding an already-made event would leave an
                # observable hole when the accepted text is later flushed.
                withheld_text: list[tuple[str, str]] = []
                if stream_model:
                    round_ = model.stream(transcript)
                    deltas = round_.deltas()
                    try:
                        while True:
                            try:
                                delta = await lifecycle.bounded(
                                    deltas.__anext__(),
                                    doing="streaming a model round",
                                )
                            except StopAsyncIteration:
                                break
                            if buffer_egress_text and delta.kind == "text_delta":
                                withheld_text.append((delta.kind, delta.text))
                            else:
                                yield await publish(
                                    emitter.make(delta.kind, text=delta.text)
                                )
                    finally:
                        await closing(deltas)
                    turn = round_.result()
                else:
                    turn = await lifecycle.bounded(
                        model.invoke(transcript),
                        doing="calling the model",
                    )
                    if turn.text:
                        if buffer_egress_text:
                            withheld_text.append(("text_delta", turn.text))
                        else:
                            yield await publish(emitter.make("text_delta", text=turn.text))

                ctx.absorb(turn.text, turn.usage, turn.stop_reason)

                # #23 slice 1: the completed round reaches post-model.
                # A stage is adopter code that can block, and it is charged to
                # the same deadline for the reason open_turn_dispatched is.
                stage_traces = await lifecycle.bounded(
                    kernel.dispatch_post_model(
                        plan, request, turn, resolved=ctx.resolved,
                        transcript=transcript,
                    ),
                    doing="running post-model stages",
                )
                for staged in observed.stage_events(stage_traces):
                    yield await publish(staged)
                for kind, text in withheld_text:
                    yield await publish(emitter.make(kind, text=text))
                if not turn.tool_requests:
                    plan.bindings.conversation.close_round(transcript, turn, (), ())
                    break

                requests = kernel.reserve_requests(ctx, turn.tool_requests)
                # #23 slice 2: the calls are reserved, so a stage sees
                # the ids the transcript will join on. Only reached on a
                # round that has calls -- the zero case is the break above,
                # and it has its own test because nothing else guards it.
                # Rebound: a stage may amend a call. See ``run_pre_tool``.
                stage_traces, requests = await lifecycle.bounded(
                    kernel.dispatch_pre_tool(
                        plan,
                        request,
                        turn,
                        requests,
                        resolved=ctx.resolved,
                        # HK2: a stage that pauses here is the only thing
                        # that can record a state the run is continuable
                        # from, and it needs the messages as well as the
                        # ids. Carried, never read by the loop.
                        transcript=transcript,
                    ),
                    doing="running pre-tool stages",
                )
                for staged in observed.stage_events(stage_traces):
                    yield await publish(staged)

                outcomes: list[Any] = []
                async for event in run_tool_round(
                    plan, requests, emitter=emitter, publish=publish,
                    lifecycle=lifecycle, outcomes=outcomes,
                ):
                    yield event
                kernel.record_tools(plan, ctx, transcript, turn, requests, outcomes)

                # TA8.50, #23's sixth rung: the calls have run AND the conversation
                # port has closed the round into the transcript, so a
                # post-tool stage is handed the object the next
                # ``model.invoke`` is called with -- which is what makes
                # "before the next round sees the result" a position and
                # not a slogan. One line earlier the results would not be
                # in the transcript yet, and a stage rewriting it would be
                # rewriting a round close_round was about to overwrite.
                #
                # Below the tool execution on purpose: a run cancelled or
                # timed out while a tool is still running never reaches
                # this line, so there is no half-dispatched rung to unwind.
                stage_traces = await lifecycle.bounded(
                    kernel.dispatch_post_tool(
                        plan,
                        request,
                        turn,
                        requests,
                        outcomes,
                        resolved=ctx.resolved,
                        transcript=transcript,
                    ),
                    doing="running post-tool stages",
                )
                for staged in observed.stage_events(stage_traces):
                    yield await publish(staged)
            else:
                ctx.stop_reason = TOOL_LIMIT

            # Validate extraction and project messages before commit-capable
            # finalizers. A failed response must not publish its memories.
            outcome = await lifecycle.bounded(
                kernel.finish(plan, ctx, transcript), doing="preparing the outcome"
            )
            stage_traces = await lifecycle.bounded(  # TA8.61: see kernel/finalize.py
                kernel.dispatch_finalize(
                    plan, request, turn,
                    resolved=ctx.resolved, transcript=transcript),
                doing="running finalize stages",
            )
            for staged in observed.stage_events(stage_traces):
                yield await publish(staged)
            outcome = replace(outcome, duration_ms=ctx.duration_ms)
        except InvocationPaused as paused:
            # HK1: the run stopped to ask a person. It is terminal and it
            # is NOT an error -- no ``error`` event, nothing re-raised at
            # the consumer, and no ``done`` after it. A pause reported as
            # a failure would tell an adopter's error handling that the
            # turn broke, and a pause reported as ``done`` would hand a
            # consumer a result the run never produced.
            #
            # The signal is a ``BaseException`` precisely so it arrives
            # here: the tool executor and the stage dispatcher between the
            # pause point and this line both catch ``Exception`` and would
            # otherwise have converted it into an observation or a FAILED
            # trace, and the run would have continued without the answer
            # it stopped for.
            reason = "paused"
            pause = emitter.make(
                paused.interrupt.kind, interrupt=paused.interrupt
            )
            await lifecycle.deliver_terminal(pause, publish)
            yield pause
            return
        except Exception as exc:  # noqa: BLE001 - EVT-7: event and raise
            # CXL-6: an elapsed budget is a `deadline`, and only this run's
            # own budget is. An adopter tool's ``ServiceTimeoutError`` is a
            # plain error, which is why the lifecycle is asked by identity
            # rather than the exception by type.
            reason = "deadline" if lifecycle.owns_timeout(exc) else "error"
            error = emitter.make("error", error=f"{type(exc).__name__}: {exc}")
            await lifecycle.deliver_terminal(error, publish)
            yield error
            raise
        done = emitter.make("done", outcome=outcome)
        await lifecycle.deliver_terminal(done, publish)
        yield done
    except GeneratorExit:
        # CXL-5: a consumer that walked away is a cancellation, and BP-10
        # case 2 still owes the *other* attached adapters their terminal
        # event and the diagnostics a terminal event proves.
        reason = "disconnected"
        ctx.cancelled = True
        await terminal_on_exit(lifecycle, emitter, publish, reason)
        raise
    except asyncio.CancelledError:
        reason = "cancelled"
        ctx.cancelled = True
        await terminal_on_exit(lifecycle, emitter, publish, reason)
        raise  # CXL-3: never swallowed
    finally:
        # Through the context, not around it: the context is the seam every
        # other collaborator reaches teardown by, and a second door into the
        # same stack is how two exit paths start disagreeing about it.
        await ctx.teardown(reason)
        observed.close()

RequestContext 

RequestContext(plan: InvocationPlan, *, run_id: str | None = None)

Request-local mutable state, with one writer per slot (RCX-2).

Source code in src/symfonic/kernel/context.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
def __init__(self, plan: InvocationPlan, *, run_id: str | None = None) -> None:
    self.plan = plan
    self.run_id = run_id or uuid.uuid4().hex
    self.started = time.perf_counter()
    self.usage = UsageDelta()
    self.answer: list[str] = []
    self.messages: tuple[Any, ...] = ()
    self.tool_outcomes: list[Any] = []
    self.stop_reason: str | None = None
    self.cancelled = False
    self._used_call_ids: set[str] = set()
    self._generation: str | None = None
    self._scratch: dict[str, dict[str, Any]] = {}
    self._resolved: Any = None
    self._lifecycle = RunLifecycle(
        self.run_id,
        deadline_seconds=plan.limits.deadline_seconds,
        grace_seconds=plan.limits.teardown_grace_seconds,
    )

lifecycle property 

lifecycle: RunLifecycle

The run's lifecycle coordinator — resources, work, and teardown.

resolved property 

resolved: Any

The turn's resolved inputs, or None before prompt assembly ends.

Read-only by construction: ResolvedInputs has no mutating method and freezes its payloads, so handing it to a later phase hands a fact rather than a channel.

bind_generation 

bind_generation(generation: str) -> None

Record the generation this invocation runs against, exactly once.

Write-once is what keeps a long-lived agent from mixing generations mid-invocation when a cutover flips underneath it (T2.3.6): the plan holds the binding stage, and the context holds the binding result.

Source code in src/symfonic/kernel/context.py 
73
74
75
76
77
78
79
80
81
82
83
84
85
def bind_generation(self, generation: str) -> None:
    """Record the generation this invocation runs against, exactly once.

    Write-once is what keeps a long-lived agent from mixing generations
    mid-invocation when a cutover flips underneath it (T2.3.6): the plan
    holds the binding *stage*, and the context holds the binding *result*.
    """
    if self._generation is not None:
        raise ContractViolationError(
            f"the generation vector is already bound to {self._generation!r}; "
            "a run observes exactly one generation (RCX-5)."
        )
    self._generation = generation

bind_resolved 

bind_resolved(snapshot: Any) -> None

Record the turn's resolved-input snapshot, exactly once (#24).

Before this, the snapshot was a local in run_prompt_assembly: built by the resolution pass, handed to the compilation pass, and unreachable the moment that function returned. A value published by a prompt-assembly stage could not be read by a later phase at all — not because the envelope forbade it, but because nothing carried it.

Write-once for the same reason bind_generation is, and the reason matters more here. scratch(namespace) could also transport a value, and it is the wrong carrier: it hands back the live dict, so any holder can rewrite what a resolution stage decided. A governance verdict read three phases later must have one answer per turn, not a last-writer-wins one. ResolvedInputs is already deeply immutable and refuses payloads it cannot freeze, so the only mutability left to remove was the binding itself.

Source code in src/symfonic/kernel/context.py 
 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
def bind_resolved(self, snapshot: Any) -> None:
    """Record the turn's resolved-input snapshot, exactly once (#24).

    Before this, the snapshot was a local in ``run_prompt_assembly``: built
    by the resolution pass, handed to the compilation pass, and unreachable
    the moment that function returned. A value published by a
    prompt-assembly stage could not be read by a later phase at all — not
    because the envelope forbade it, but because nothing carried it.

    Write-once for the same reason ``bind_generation`` is, and the reason
    matters more here. ``scratch(namespace)`` could also transport a value,
    and it is the wrong carrier: it hands back the live dict, so any holder
    can rewrite what a resolution stage decided. A governance verdict read
    three phases later must have one answer per turn, not a
    last-writer-wins one. ``ResolvedInputs`` is already deeply immutable and
    refuses payloads it cannot freeze, so the only mutability left to remove
    was the binding itself.
    """
    if self._resolved is not None:
        raise ContractViolationError(
            "the resolved-input snapshot is already bound; a turn resolves "
            "its inputs once, and a second binding would make 'what did the "
            "resolution pass produce?' depend on who asked last."
        )
    self._resolved = snapshot

reserve 

reserve(request: ToolRequest) -> ToolRequest

Return request with a run-unique call_id.

Source code in src/symfonic/kernel/context.py 
137
138
139
140
141
142
def reserve(self, request: ToolRequest) -> ToolRequest:
    """Return ``request`` with a run-unique ``call_id``."""
    call_id = self.reserve_call_id(request.call_id)
    if call_id == request.call_id:
        return request
    return ToolRequest(call_id=call_id, name=request.name, arguments=request.arguments)

reserve_call_id 

reserve_call_id(proposed: str | None) -> str

Return a run-unique tool-call id, keeping the provider's when usable.

Source code in src/symfonic/kernel/context.py 
129
130
131
132
133
134
135
def reserve_call_id(self, proposed: str | None) -> str:
    """Return a run-unique tool-call id, keeping the provider's when usable."""
    call_id = proposed
    while not call_id or call_id in self._used_call_ids:
        call_id = uuid.uuid4().hex
    self._used_call_ids.add(call_id)
    return call_id

scratch 

scratch(namespace: str) -> dict[str, Any]

Per-run scratch for one capability. Namespaced so nobody collides.

Source code in src/symfonic/kernel/context.py 
146
147
148
def scratch(self, namespace: str) -> dict[str, Any]:
    """Per-run scratch for one capability. Namespaced so nobody collides."""
    return self._scratch.setdefault(namespace, {})

spawn 

spawn(work: Any, *, owner: str, purpose: str, deadline_seconds: float | None = None) -> asyncio.Task[Any]

Create run-owned background work (RCX-8/BP-8); never fire-and-forget.

Source code in src/symfonic/kernel/context.py 
171
172
173
174
175
176
177
178
179
180
181
182
def spawn(
    self,
    work: Any,
    *,
    owner: str,
    purpose: str,
    deadline_seconds: float | None = None,
) -> asyncio.Task[Any]:
    """Create run-owned background work (RCX-8/BP-8); never fire-and-forget."""
    return self._lifecycle.spawn(
        work, owner=owner, purpose=purpose, deadline_seconds=deadline_seconds
    )

teardown async 

teardown(reason: TeardownReason = 'completed') -> TeardownReport

Unwind this run's lifecycle exactly once and return its record.

Source code in src/symfonic/kernel/context.py 
184
185
186
async def teardown(self, reason: TeardownReason = "completed") -> TeardownReport:
    """Unwind this run's lifecycle exactly once and return its record."""
    return await self._lifecycle.teardown(reason)

ResultCollector 

ResultCollector(policy: EventAdapter, response: Any)

The single assembler for the public blocking result.

Source code in src/symfonic/kernel/adapters.py 
37
38
39
def __init__(self, policy: EventAdapter, response: Any) -> None:
    self._policy = policy
    self._response = response

RunLifecycle 

RunLifecycle(run_id: str, *, deadline_seconds: float | None = None, grace_seconds: float = DEFAULT_TEARDOWN_GRACE_SECONDS)

Owns one run's releasable state and unwinds it exactly once.

Source code in src/symfonic/kernel/lifecycle.py 
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
def __init__(
    self,
    run_id: str,
    *,
    deadline_seconds: float | None = None,
    grace_seconds: float = DEFAULT_TEARDOWN_GRACE_SECONDS,
) -> None:
    self.run_id = run_id
    self._deadline = RunDeadline(run_id, deadline_seconds)
    self._grace = grace_seconds
    self._registry = BackgroundRegistry(run_id, run_deadline_seconds=deadline_seconds)
    self._finalizers = FinalizerStack(grace_seconds=grace_seconds)
    self._pressure: list[PressureRecord] = []
    self._ready = False
    self._closed = False
    self._report: TeardownReport | None = None
    self._terminal_kind: str | None = None
    self._terminal_delivered = False

deadline property 

deadline: RunDeadline

The run's clock: how much time is left, and what running out means.

report property 

report: TeardownReport | None

The teardown record, or None while the run is still open.

acquire 

acquire(resource: ResourcePort) -> ResourcePort

Take a resource and register its release in the same expression.

Returning the resource is what makes the acquisition and the release impossible to separate in a diff: there is no way to write the first without the second. The release lands in post_drain: background work this run spawned may still be writing through the resource, and giving it back first is a use-after-release the drain would then hide.

Source code in src/symfonic/kernel/lifecycle.py 
144
145
146
147
148
149
150
151
152
153
154
155
def acquire(self, resource: ResourcePort) -> ResourcePort:
    """Take a resource and register its release in the same expression.

    Returning the resource is what makes the acquisition and the release
    impossible to separate in a diff: there is no way to write the first
    without the second. The release lands in ``post_drain``: background work
    this run spawned may still be writing through the resource, and giving
    it back first is a use-after-release the drain would then hide.
    """
    self.require_open("acquiring a resource")
    self._finalizers.push(resource.release, name=f"resource:{resource.name}")
    return resource

bounded async 

bounded(step: Awaitable[Any], *, doing: str) -> Any

Await one step of forward progress inside the run's remaining budget.

This is how a deadline reaches a consumer-paced entry point, and it is the reason the budget is not left to :meth:scope alone: a scope can only bound a call one task both enters and leaves, so before this existed deadline_seconds was enforced on run() and silently ignored on every streaming projection — one plan value with two meanings, which is the entry-point drift the kernel exists to remove.

Source code in src/symfonic/kernel/lifecycle.py 
205
206
207
208
209
210
211
212
213
214
215
async def bounded(self, step: Awaitable[Any], *, doing: str) -> Any:
    """Await one step of forward progress inside the run's remaining budget.

    This is how a deadline reaches a *consumer-paced* entry point, and it
    is the reason the budget is not left to :meth:`scope` alone: a scope
    can only bound a call one task both enters and leaves, so before this
    existed ``deadline_seconds`` was enforced on ``run()`` and silently
    ignored on every streaming projection — one plan value with two
    meanings, which is the entry-point drift the kernel exists to remove.
    """
    return await self._deadline.bounded(step, doing=doing)

deliver_terminal async 

deliver_terminal(event: KernelEvent, deliver: Callable[[KernelEvent], Awaitable[Any]] | None = None) -> bool

Record the run's terminal event and attempt its delivery (BP-10).

The attempt is bounded and unshielded, and that is deliberate. On the cancellation path the delivering task is already cancelled, so a shielded attempt would keep a dead consumer's write alive past the run that owed it; failing fast and recording BP-10 case 3 is the honest outcome. Buffer pressure never reaches here — capacity for a terminal event is reserved or preempted upstream (BP-4).

A grace of 0 means "cancel owned work immediately", not "skip teardown"; the delivery it bounds is left unbounded rather than cut before it can write a byte.

Source code in src/symfonic/kernel/lifecycle.py 
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
async def deliver_terminal(
    self,
    event: KernelEvent,
    deliver: Callable[[KernelEvent], Awaitable[Any]] | None = None,
) -> bool:
    """Record the run's terminal event and attempt its delivery (BP-10).

    The attempt is bounded and unshielded, and that is deliberate. On the
    cancellation path the delivering task is already cancelled, so a
    shielded attempt would keep a dead consumer's write alive past the run
    that owed it; failing fast and recording BP-10 case 3 is the honest
    outcome. Buffer pressure never reaches here — capacity for a terminal
    event is reserved or preempted upstream (BP-4).

    A grace of ``0`` means "cancel owned work immediately", not "skip
    teardown"; the delivery it bounds is left unbounded rather than cut
    before it can write a byte.
    """
    self._terminal_kind = event.kind
    if deliver is None:
        self._terminal_delivered = True
        return True
    try:
        async with teardown_budget(self._grace):
            await deliver(event)
    except asyncio.CancelledError:
        self._terminal_delivered = False
        raise  # CXL-3: cancellation is recorded, never swallowed
    except Exception:  # noqa: BLE001 — a failed delivery is counted, not raised
        self._terminal_delivered = False
        return False
    self._terminal_delivered = True
    return True

ensure_ready async 

ensure_ready(checkpointer: CheckpointerPort | None) -> None

Make durable state ready once, registering its teardown on success.

Registration happens here rather than at construction because a checkpointer that never became ready has nothing to flush, and a checkpointer that did must be flushed no matter which of the eight call sites happened to trigger readiness first.

Both land in post_drain: a spawned checkpoint writer flushed before it was drained would have its writes silently dropped.

Source code in src/symfonic/kernel/lifecycle.py 
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
async def ensure_ready(self, checkpointer: CheckpointerPort | None) -> None:
    """Make durable state ready once, registering its teardown on success.

    Registration happens *here* rather than at construction because a
    checkpointer that never became ready has nothing to flush, and a
    checkpointer that did must be flushed no matter which of the eight call
    sites happened to trigger readiness first.

    Both land in ``post_drain``: a spawned checkpoint writer flushed before
    it was drained would have its writes silently dropped.
    """
    if checkpointer is None or self._ready:
        return
    self.require_open("preparing the checkpointer")
    await checkpointer.ensure_ready()
    self._ready = True
    self._finalizers.push(checkpointer.close, name="checkpointer.close")
    self._finalizers.push(checkpointer.flush, name="checkpointer.flush")

owns_timeout 

owns_timeout(error: BaseException) -> bool

True only for the timeout this run's own deadline raised (CXL-6).

Source code in src/symfonic/kernel/lifecycle.py 
108
109
110
def owns_timeout(self, error: BaseException) -> bool:
    """True only for the timeout this run's own deadline raised (CXL-6)."""
    return self._deadline.owns(error)

push_finalizer 

push_finalizer(finalizer: Callable[[], Any], *, name: str | None = None, phase: FinalizerPhase = 'post_drain') -> None

Register cleanup to run in reverse registration order, within its phase.

The default phase runs after owned work is drained, which is what keeps whatever the finalizer releases alive while the run's own tasks may still be using it. Pass phase="pre_drain" only for the inverse dependency: a finalizer the drain itself is waiting on.

Source code in src/symfonic/kernel/lifecycle.py 
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
def push_finalizer(
    self,
    finalizer: Callable[[], Any],
    *,
    name: str | None = None,
    phase: FinalizerPhase = "post_drain",
) -> None:
    """Register cleanup to run in reverse registration order, within its phase.

    The default phase runs *after* owned work is drained, which is what
    keeps whatever the finalizer releases alive while the run's own tasks
    may still be using it. Pass ``phase="pre_drain"`` only for the inverse
    dependency: a finalizer the drain itself is waiting on.
    """
    self.require_open("registering a finalizer")
    label = name or getattr(finalizer, "__name__", "finalizer")
    self._finalizers.push(finalizer, name=label, phase=phase)

record_pressure 

record_pressure(adapter: str, metrics: AdapterPressure) -> None

Freeze one adapter's BP-12 numbers into the run's record.

Source code in src/symfonic/kernel/lifecycle.py 
189
190
191
192
193
194
195
196
197
198
199
200
201
def record_pressure(self, adapter: str, metrics: AdapterPressure) -> None:
    """Freeze one adapter's BP-12 numbers into the run's record."""
    self._pressure.append(
        PressureRecord(
            adapter=adapter,
            high_watermark=metrics.high_watermark,
            byte_high_watermark=getattr(metrics, "byte_high_watermark", 0),
            blocked_seconds=metrics.blocked_seconds,
            events_shed=MappingProxyType(dict(metrics.events_shed)),
            terminal_delivery_failed=metrics.terminal_delivery_failed,
            abandoned=metrics.abandoned,
        )
    )

require_open 

require_open(action: str = 'write') -> None

Refuse post-close mutation and emission (RCX-10, BP-9).

Reads stay legal — diagnostics about a finished run are the reason the object survives its teardown at all.

Source code in src/symfonic/kernel/lifecycle.py 
112
113
114
115
116
117
118
119
120
121
122
def require_open(self, action: str = "write") -> None:
    """Refuse post-close mutation and emission (RCX-10, BP-9).

    Reads stay legal — diagnostics about a finished run are the reason the
    object survives its teardown at all.
    """
    if self._closed:
        raise ContractViolationError(
            f"run {self.run_id} is closed; {action} after teardown is a "
            "defect, not a late arrival to accommodate (BP-9)."
        )

scope async 

scope() -> AsyncIterator[None]

The run's cancellation scope: a deadline is an error, not a cancel.

asyncio.timeout implements an elapsed deadline by cancelling the body, so without this translation the two situations CXL-6 insists on separating would reach the caller as the same exception. External cancellation passes straight through, unswallowed (CXL-2/CXL-3).

Enter it only in a task that both enters and leaves it. The kernel's own entry points do not: they bill the same clock through :meth:bounded, one step at a time, which is legal from a generator that may be resumed by a different task than the one that suspended it.

Source code in src/symfonic/kernel/lifecycle.py 
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
@asynccontextmanager
async def scope(self) -> AsyncIterator[None]:
    """The run's cancellation scope: a deadline is an error, not a cancel.

    ``asyncio.timeout`` implements an elapsed deadline by cancelling the
    body, so without this translation the two situations CXL-6 insists on
    separating would reach the caller as the same exception. External
    cancellation passes straight through, unswallowed (CXL-2/CXL-3).

    Enter it only in a task that both enters and leaves it. The kernel's own
    entry points do not: they bill the same clock through :meth:`bounded`,
    one step at a time, which is legal from a generator that may be resumed
    by a different task than the one that suspended it.
    """
    async with self._deadline.scope():
        yield

spawn 

spawn(work: Any, *, owner: str, purpose: str, deadline_seconds: float | None = None) -> asyncio.Task[Any]

Create run-owned background work (RCX-8/BP-8); never fire-and-forget.

Source code in src/symfonic/kernel/lifecycle.py 
176
177
178
179
180
181
182
183
184
185
186
187
def spawn(
    self,
    work: Any,
    *,
    owner: str,
    purpose: str,
    deadline_seconds: float | None = None,
) -> asyncio.Task[Any]:
    """Create run-owned background work (RCX-8/BP-8); never fire-and-forget."""
    return self._registry.spawn(
        work, owner=owner, purpose=purpose, deadline_seconds=deadline_seconds
    )

teardown async 

teardown(reason: TeardownReason = 'completed') -> TeardownReport

Unwind everything, once, and return the record of having done it.

Idempotent because it runs from a finally that several exit paths can reach: a second call returns the first call's report rather than re-running finalizers against state they already released.

Source code in src/symfonic/kernel/lifecycle.py 
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
async def teardown(self, reason: TeardownReason = "completed") -> TeardownReport:
    """Unwind everything, once, and return the record of having done it.

    Idempotent because it runs from a ``finally`` that several exit paths
    can reach: a second call returns the first call's report rather than
    re-running finalizers against state they already released.
    """
    if self._report is not None:
        return self._report
    started = time.perf_counter()
    ran, failures = await self._finalizers.unwind("pre_drain")
    drained = await self._registry.drain(grace_seconds=self._grace)
    after_ran, after_failures = await self._finalizers.unwind("post_drain")
    self._closed = True
    self._report = TeardownReport(
        run_id=self.run_id,
        reason=reason,
        elapsed_ms=(time.perf_counter() - started) * 1000.0,
        finalizers_run=ran + after_ran,
        finalizer_failures=failures + after_failures,
        tasks_awaited=drained.awaited,
        tasks_cancelled=drained.cancelled,
        background_failures=drained.failures,
        terminal_kind=self._terminal_kind,
        terminal_delivered=self._terminal_delivered,
        pressure=tuple(self._pressure),
    )
    return self._report

StructuredOutputAdapter 

StructuredOutputAdapter(policy: EventAdapter, response: Any)

Run terminal extraction through the response port under its G9 row.

Source code in src/symfonic/kernel/adapters.py 
132
133
134
def __init__(self, policy: EventAdapter, response: Any) -> None:
    self._policy = policy
    self._response = response

TextStreamAdapter 

TextStreamAdapter(policy: EventAdapter)

Project answer deltas to strings while consuming the terminal event.

Source code in src/symfonic/kernel/adapters.py 
104
105
def __init__(self, policy: EventAdapter) -> None:
    self._policy = policy

TypedStreamAdapter 

TypedStreamAdapter(policy: EventAdapter, response: Any)

Project kernel events through ResponsePort.build_event in order.

Source code in src/symfonic/kernel/adapters.py 
83
84
85
def __init__(self, policy: EventAdapter, response: Any) -> None:
    self._policy = policy
    self._response = response

compile_invocation_plan 

compile_invocation_plan(request: CompileRequest) -> InvocationPlan

Compile one immutable, stateless plan. Performs no effect of any kind.

Source code in src/symfonic/kernel/compiler.py 
 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
def compile_invocation_plan(request: CompileRequest) -> InvocationPlan:
    """Compile one immutable, stateless plan. Performs no effect of any kind."""
    records: list[DiagnosticRecord] = list(_input_records(request))

    # The third gate. A stage's declaration and a capability's grant are each
    # checked at construction; this is where the *invocation's* grants land --
    # today the facade's baseline, and the seam an adopter-supplied grant will
    # arrive through. Checked here so an unknown family cannot enter a plan at
    # all, rather than being caught only if some stage happens to want it.
    require_known_families(request.effect_grants, subject="the invocation")
    validate_event_program(request.event_program)
    validate_limits(request.limits)
    validate_bindings(request.bindings)
    model = _resolve_model(request.model)
    tool_manifest = _resolve_tools(request.tools, records)
    stage_program, stage_records = _resolve_stages(request, records)
    records.extend(stage_records)
    records.extend(_binding_records(request.bindings))
    records.extend(_grant_records(request))

    return InvocationPlan(
        identity=PlanIdentity(
            plan_id=uuid.uuid4().hex,
            config_digest=request.config_digest,
            compiled_at=time.time(),
            schema_version=SCHEMA_VERSION,
        ),
        scope=request.scope,
        model=model,
        tool_manifest=tool_manifest,
        stage_program=stage_program,
        bindings=request.bindings,
        effect_grants=frozenset(request.effect_grants),
        limits=request.limits,
        event_program=request.event_program,
        diagnostics=PlanDiagnostics(records=tuple(records)),
    )

derive_child_plan 

derive_child_plan(parent: InvocationPlan, request: CompileRequest) -> InvocationPlan

Compile a child plan by narrowing the parent (IPL-6).

Every widening attempt is rejected here rather than at the point of use. A sub-agent that could add a tool, a grant, a tenant or a token budget its parent did not have would make the parent's plan a suggestion.

Source code in src/symfonic/kernel/compiler.py 
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
def derive_child_plan(parent: InvocationPlan, request: CompileRequest) -> InvocationPlan:
    """Compile a child plan by narrowing the parent (IPL-6).

    Every widening attempt is rejected here rather than at the point of use.
    A sub-agent that could add a tool, a grant, a tenant or a token budget its
    parent did not have would make the parent's plan a suggestion.
    """
    child = compile_invocation_plan(request)
    _require_narrowing(parent, child)
    return InvocationPlan(
        identity=PlanIdentity(
            plan_id=child.identity.plan_id,
            config_digest=child.identity.config_digest,
            compiled_at=child.identity.compiled_at,
            schema_version=child.identity.schema_version,
            parent_plan_id=parent.identity.plan_id,
        ),
        scope=child.scope,
        model=child.model,
        tool_manifest=child.tool_manifest,
        stage_program=child.stage_program,
        bindings=child.bindings,
        effect_grants=child.effect_grants,
        limits=child.limits,
        event_program=child.event_program,
        diagnostics=child.diagnostics,
    )