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()