symfonic.kernel.run_observation

run_observation

What a runner opens around one turn so the turn can be observed.

Two things belong to the run rather than to any step of it: the identity every observer correlates on, and the stage traces the run publishes as its capabilities execute. They were two modules of about twenty lines each, and the runner had to remember to open one, thread run_id and the trace sink through five call sites, and close the other in a finally.

One object owns both now. The runner opens it, hands it the traces each phase produced, and closes it -- and the correlation identifier is never passed again, because the thing publishing already knows which run it is.

Metadata only, and that is a contract not a convenience. A stage trace carries the stage's name, phase, capability and outcome. It never carries the subject a stage acted on or the payload it produced: this stream reaches durable storage and an operator's screen, and a memory's text on either would be a leak that no consumer could opt out of.

Emitter

Emitter(run_id: str, attempt_id: str = '')

Assign event identity, ordering, and terminal cardinality once (RCX-6).

Source code in src/symfonic/kernel/run_observation.py

def __init__(self, run_id: str, attempt_id: str = "") -> None:
    self._index = 0
    self._run_id = run_id
    self._attempt_id = attempt_id
    self._closed = False

RunObservation

RunObservation(context: Any, request: Any, callbacks: Any, emitter: Any = None, plan: Any = None)

The identity a run carries, and the traces it publishes.

Parameters:

Name	Type	Description	Default
context	Any	the invocation context, which owns run_id.	required
request	Any	the turn request, which may name a root and a parent when this run is a delegated child.	required
callbacks	Any	unused now that stage events ride the ordinary channel; kept so the signature still says what a run is given.	required
emitter	Any	the run's event factory. Stage events take their index from it, so they sit in the same dense sequence as the model and tool events and a reader can reconstruct the turn in order. They were emitted at index=-1 before this, which is not a position -- two of them could not be told apart, and neither could be placed against the round it belonged to.	None
plan	Any	the compiled plan; its effect grants and stage effects stamp the run's monotone memory marker (CORE-194).	None

Source code in src/symfonic/kernel/run_observation.py

def __init__(
    self, context: Any, request: Any, callbacks: Any, emitter: Any = None,
    plan: Any = None,
) -> None:
    """
    Args:
        context: the invocation context, which owns ``run_id``.
        request: the turn request, which may name a root and a parent when
            this run is a delegated child.
        callbacks: unused now that stage events ride the ordinary
            channel; kept so the signature still says what a run is given.
        emitter: the run's event factory. Stage events take their index
            from it, so they sit in the same dense sequence as the model
            and tool events and a reader can reconstruct the turn in
            order. They were emitted at ``index=-1`` before this, which
            is not a position -- two of them could not be told apart, and
            neither could be placed against the round it belonged to.
        plan: the compiled plan; its effect grants and stage effects stamp
            the run's monotone memory marker (CORE-194).
    """
    self._run_id = context.run_id
    self._emitter = emitter
    self._identity = RunIdentity(
        run_id=context.run_id,
        root_run_id=request.root_run_id or context.run_id,
        parent_run_id=request.parent_run_id,
    )
    self._closed = False
    self._token: Any = _issue_run_identity(self._identity)
    self._node = self._token.node
    # CORE-194: stamp from the compiled authority; no plan reads as memory.
    from symfonic.kernel.contracts.run_admission import memory_in

    self._node.note_memory(
        plan is None
        or memory_in(plan.effect_grants, plan.stage_program.stages)
    )

bind

bind() -> None

Make this run current for the generator step about to execute (CORE-194).

The runner is a generator its consumer may step from a fresh task, or alternately with another run. Bound per step and released before the step yields, the run's own code always sees its own lineage, and the consumer keeps none of it between steps.

Source code in src/symfonic/kernel/run_observation.py

def bind(self) -> None:
    """Make this run current for the generator step about to execute (CORE-194).

    The runner is a generator its consumer may step from a fresh task, or
    alternately with another run. Bound per step and released before the
    step yields, the run's own code always sees its own lineage, and the
    consumer keeps none of it between steps.
    """
    if not self._closed and self._token is None:
        self._token = _bind_issued(self._identity, self._node)

close

close() -> None

Release the identity. Idempotent, because teardown has two paths.

Source code in src/symfonic/kernel/run_observation.py

def close(self) -> None:
    """Release the identity. Idempotent, because teardown has two paths."""
    self._closed = True
    self.unbind()

stage_events

stage_events(traces: tuple[Any, ...]) -> tuple[Any, ...]

One phase's stage records, as events on the run's own sequence.

Returned rather than published, because the caller is an async generator and these have to be yielded: the public stream and the durable execution log must be two readers of one event, and an event delivered only to callbacks is a second channel that can disagree with the first about what a turn did.

Indices come from the run's emitter, so a stage sits in order among the model and tool events and a reader can place a retrieval against the round it fed. They were emitted at index=-1 before this, which is not a position: two of them could not be told apart.

Source code in src/symfonic/kernel/run_observation.py

def stage_events(self, traces: tuple[Any, ...]) -> tuple[Any, ...]:
    """One phase's stage records, as events on the run's own sequence.

    Returned rather than published, because the caller is an async
    generator and these have to be *yielded*: the public stream and the
    durable execution log must be two readers of one event, and an event
    delivered only to callbacks is a second channel that can disagree
    with the first about what a turn did.

    Indices come from the run's emitter, so a stage sits in order among
    the model and tool events and a reader can place a retrieval against
    the round it fed. They were emitted at ``index=-1`` before this, which
    is not a position: two of them could not be told apart.
    """
    if self._emitter is None:
        return ()
    events = []
    for trace in traces:
        try:
            events.append(
                self._emitter.make(
                    "stage",
                    stage_id=trace.stage_id,
                    phase=str(getattr(trace.phase, "value", trace.phase)),
                    capability=trace.capability,
                    stage_outcome=str(
                        getattr(trace.outcome, "value", trace.outcome)
                    ),
                    stage_reason=trace.reason,
                    counts=trace.counts,
                )
            )
        except ContractViolationError:
            # The stream went terminal first. A stage record is worth
            # having and never worth reopening a closed stream for.
            break
    return tuple(events)

stepped async

stepped(source: AsyncGenerator[Any, None], holder: list[Any]) -> AsyncIterator[Any]

Relay source, binding its run around each step and its teardown.

holder receives the run's :class:RunObservation once the source opens it; until then the source's first step binds the run itself.

Source code in src/symfonic/kernel/run_observation.py

async def stepped(source: AsyncGenerator[Any, None], holder: list[Any]) -> AsyncIterator[Any]:
    """Relay ``source``, binding its run around each step and its teardown.

    ``holder`` receives the run's :class:`RunObservation` once the source
    opens it; until then the source's first step binds the run itself.
    """
    try:
        while True:
            if holder:
                holder[0].bind()
            try:
                event = await anext(source)
            except StopAsyncIteration:
                return
            finally:
                if holder:
                    holder[0].unbind()
            yield event
    finally:
        if holder:
            holder[0].bind()
        try:
            await source.aclose()
        finally:
            if holder:
                holder[0].unbind()

terminal_on_exit async

terminal_on_exit(lifecycle: Any, emitter: Emitter, publish: Any, reason: str) -> None

Record — and where owed, deliver — the cancelled terminal (CXL-2).

Cancellation is not an error, so it never emits an error event. If the stream already went terminal the emitter refuses a second one (EVT-1), and there is nothing left to say.

reason is the run's own teardown word — disconnected when the consumer walked away, cancelled when this task was cancelled — and it rides on the terminal so an observer can tell the two apart. Both are kind="cancelled": they end the stream the same way, and they mean different things about who stopped it (TA8.29).

Source code in src/symfonic/kernel/run_observation.py

async def terminal_on_exit(
lifecycle: Any, emitter: Emitter, publish: Any, reason: str
) -> None:
    """Record — and where owed, deliver — the cancelled terminal (CXL-2).

    Cancellation is not an error, so it never emits an ``error`` event. If
    the stream already went terminal the emitter refuses a second one
    (EVT-1), and there is nothing left to say.

    ``reason`` is the run's own teardown word — ``disconnected`` when the
    consumer walked away, ``cancelled`` when this task was cancelled — and
    it rides on the terminal so an observer can tell the two apart. Both
    are ``kind="cancelled"``: they end the stream the same way, and they
    mean different things about who stopped it (TA8.29).
    """
    try:
        cancelled = emitter.make("cancelled", termination=reason)
    except ContractViolationError:
        return
    await lifecycle.deliver_terminal(cancelled, publish)