symfonic.kernel.dispatch

dispatch

Running the stages the plan compiled — W2/1b.

Before this, StageProgram was compiled, totally ordered, carried in the plan, and executed by nothing. invoker.finish said so:

The kernel compiles a total StageProgram but dispatches no stage callable
yet ... a registered stage affects ordering and diagnostics and nothing
else.

This is the "and nothing else" ending. One phase's stages run here, in compiled order, each as a transactional unit: invoke, validate, apply, publish.

Four decisions are frozen into this module, and each is written where it is enforced rather than in a document that drifts:

• A stage returns; it does not mutate (decision 2). The dispatcher owns application, so "did this stage change anything?" has one answer and one place that knows it.
• Events ride with the result (decision 3). One execution publishes one contiguous block, after its contribution is applied. A stage that fails publishes nothing — no event announcing a change that never landed.
• emits is a checked upper bound. A stage may emit fewer events than it declared; an undeclared event is a contract violation. This is what stops CompiledStage.emits from becoming the next declared-and-never-read field.
• Every executable stage resolves exactly one handler. The distinction is executable stage vs structural marker, not kernel-owned vs capability-owned — a kernel stage that runs is as obliged as anyone's.

Cardinality (decision 1) is fixed per phase and lives at the call sites in runner, not here: this module runs one phase's stages once, and the caller decides how often a phase happens.

StageDispatcher

StageDispatcher(handlers: Mapping[str, StageHandler])

Runs one phase's compiled stages, in order, against their handlers.

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

def __init__(self, handlers: Mapping[str, StageHandler]) -> None:
    self._handlers = dict(handlers)

run_phase async

run_phase(program: StageProgram, phase: Phase, *, context_for: Callable[[CompiledStage], Any], apply: Callable[[CompiledStage, Any], Awaitable[None]] | None = None, publish: Callable[[Any], Awaitable[None]] | None = None, grants: frozenset[str] | None = None, kind: StageKind | None = None, stop_on_failure: bool = False) -> tuple[StageTrace, ...]

Execute the stages compiled into phase, in order, until one's contribution cannot be applied.

By default a handler that fails stops nothing: it applied no returned contribution, so the stages after it run on untouched state. An apply that fails failed partway through adopting a contribution, so the phase ends there rather than handing the next stage its remains. Commit-capable phases opt into stop_on_failure because their handlers can perform effects without an apply hook.

Parameters:

Name	Type	Description	Default
program	StageProgram	the plan's compiled stage program.	required
phase	Phase	which phase to run. Stages outside it are untouched.	required
context_for	Callable[[CompiledStage], Any]	builds the per-stage context. A callable rather than a value because a phase that runs per tool call needs a different context per stage invocation.	required
apply	Callable[[CompiledStage, Any], Awaitable[None]] | None	applies an APPLIED result's contribution. None means the caller only wants the stages run and traced — used by the phases that have no applicable state yet.	None
publish	Callable[[Any], Awaitable[None]] | None	publishes one event. None drops them, and the trace records the gap rather than hiding it.	None
kind	StageKind | None	run only stages of this kind, in compiled order. None runs every stage in the phase. The caller passes a kind when a phase is split -- prompt-assembly is, into resolution then compilation (STG-7) -- because the split is an absolute ordering like the ladder itself, not something a constraint or a priority may reorder.	None
grants	frozenset[str] | None	the invocation's effect grants (STG-8). A stage declaring an effect outside them is not invoked — the check is before the call, because a handler that ran already performed whatever it was going to perform. None skips the check, for callers that have no plan to read grants from.	None
stop_on_failure	bool	stop before any later stage after a FAILED trace, including an absent handler or denied grant. FINALIZE uses this because handlers can commit effects directly. Other phases retain their existing continuation policy by default.	False

Returns:

Name	Type	Description
One	StageTrace	class:StageTrace per stage that ran, in execution order
	...	— a prefix of the phase when an apply failed, not one trace per
	tuple[StageTrace, ...]	compiled stage. Nothing is fabricated for the stages that never
started	tuple[StageTrace, ...]	:class:StageTrace means "one stage execution", and none
	tuple[StageTrace, ...]	of the four outcomes truthfully says "never reached". The terminal
	tuple[StageTrace, ...]	FAILED trace is what explains why the prefix is short, and exposing
	tuple[StageTrace, ...]	skips deliberately would need a status this model does not have.

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

async def run_phase(
    self,
    program: StageProgram,
    phase: Phase,
    *,
    context_for: Callable[[CompiledStage], Any],
    apply: Callable[[CompiledStage, Any], Awaitable[None]] | None = None,
    publish: Callable[[Any], Awaitable[None]] | None = None,
    grants: frozenset[str] | None = None,
    kind: StageKind | None = None,
    stop_on_failure: bool = False,
) -> tuple[StageTrace, ...]:
    """Execute the stages compiled into ``phase``, in order, until one's
    contribution cannot be applied.

    By default a handler that fails stops nothing: it applied no returned
    contribution, so the stages after it run on
    untouched state. An *apply* that fails failed partway through adopting a
    contribution, so the phase ends there rather than handing the next stage
    its remains. Commit-capable phases opt into ``stop_on_failure`` because
    their handlers can perform effects without an apply hook.

    Args:
        program: the plan's compiled stage program.
        phase: which phase to run. Stages outside it are untouched.
        context_for: builds the per-stage context. A callable rather than a
            value because a phase that runs per tool call needs a different
            context per stage invocation.
        apply: applies an APPLIED result's contribution. ``None`` means the
            caller only wants the stages run and traced — used by the phases
            that have no applicable state yet.
        publish: publishes one event. ``None`` drops them, and the trace
            records the gap rather than hiding it.
        kind: run only stages of this kind, in compiled order. ``None``
            runs every stage in the phase. The caller passes a kind when a
            phase is split -- ``prompt-assembly`` is, into resolution then
            compilation (STG-7) -- because the split is an absolute
            ordering like the ladder itself, not something a constraint or
            a priority may reorder.
        grants: the invocation's effect grants (STG-8). A stage declaring an
            effect outside them is **not invoked** — the check is before the
            call, because a handler that ran already performed whatever it
            was going to perform. ``None`` skips the check, for callers that
            have no plan to read grants from.
        stop_on_failure: stop before any later stage after a FAILED trace,
            including an absent handler or denied grant. FINALIZE uses
            this because handlers can commit effects directly. Other
            phases retain their existing continuation policy by default.

    Returns:
        One :class:`StageTrace` per stage that **ran**, in execution order
        — a prefix of the phase when an apply failed, not one trace per
        compiled stage. Nothing is fabricated for the stages that never
        started: :class:`StageTrace` means "one stage execution", and none
        of the four outcomes truthfully says "never reached". The terminal
        FAILED trace is what explains why the prefix is short, and exposing
        skips deliberately would need a status this model does not have.
    """
    traces: list[StageTrace] = []
    for stage in program.in_phase(phase):
        if stop_on_failure and traces and traces[-1].outcome is StageOutcome.FAILED:
            break
        if kind is not None and stage.kind is not kind:
            continue
        handler = self.resolve(stage)
        if handler is None:
            # Not silently skipped: the compiler is supposed to have
            # rejected this, so reaching here means the plan and the
            # handler table disagree, and that is worth a loud trace.
            traces.append(
                StageTrace(
                    stage_id=stage.stage_id,
                    phase=phase,
                    capability=stage.capability,
                    outcome=StageOutcome.FAILED,
                    reason="no handler resolved for an executable stage",
                    declared_events=tuple(stage.emits),
                )
            )
            continue

        ungranted = self._ungranted(stage, grants)
        if ungranted:
            # Before the call, not after. STG-8 says an ungranted effect
            # raises *before* the effect, and a handler that has already
            # run has already done whatever it does -- checking its return
            # value would be checking the receipt of a purchase the plan
            # refused. The handler is never entered.
            traces.append(
                StageTrace(
                    stage_id=stage.stage_id,
                    phase=phase,
                    capability=stage.capability,
                    outcome=StageOutcome.FAILED,
                    reason=(
                        f"stage declares effect(s) {ungranted} the invocation "
                        f"did not grant (STG-8); the handler was not invoked"
                    ),
                    declared_events=tuple(stage.emits),
                )
            )
            continue

        result = await self._invoke(handler, context_for(stage))
        published = 0
        if result.outcome is StageOutcome.APPLIED:
            self._check_declared(stage, result)
            published, failure = await apply_and_publish(
                stage, result, apply, publish
            )
            if failure is not None:
                # The phase stops here, and this is the one place in the
                # loop where that is right. A *handler* that fails applied
                # nothing, so the next stage runs on untouched state and
                # continuing is safe. An *apply* that fails failed partway
                # through adopting a contribution, so whatever it did before
                # raising is already visible -- and continuing hands the next
                # stage its wreckage. Reproduced: an apply hook that mutates
                # and then raises had its mutation observed by the following
                # stage, which then applied on top of it.
                #
                # ``run_phase`` is public and ``apply`` is the caller's own
                # callback, so this cannot be left to the caller being
                # careful: today's two hooks happen to rebind immutable
                # values, and the next one will not. "Transactional unit" in
                # this module's docstring has to mean the unit does not get
                # followed by more units acting on its remains.
                traces.append(apply_failure(stage, phase, result, failure))
                break

        traces.append(
            StageTrace(
                stage_id=stage.stage_id,
                phase=phase,
                capability=stage.capability,
                outcome=result.outcome,
                reason=result.reason,
                declared_events=tuple(stage.emits),
                returned_events=len(result.events),
                published_events=published,
                error=result.error,
                counts=result.counts,
            )
        )
    return tuple(traces)

StageHandler

Bases: Protocol

What a stage actually is: a callable returning a :class:StageResult.

Async by contract even when the work is synchronous. A handler that could be either forces every call site to branch, and the one place that forgets is the one that blocks the loop.

StageTrace dataclass

StageTrace(stage_id: str, phase: Phase, capability: str, outcome: StageOutcome, reason: str = '', declared_events: tuple[str, ...] = (), returned_events: int = 0, published_events: int = 0, counts: Mapping[str, int] = (lambda: EMPTY_COUNTS)(), error: BaseException | None = None)

One stage execution, as the record a reader needs to explain a turn.

declared/returned/published are separate on purpose. Equal numbers are the boring case; the interesting bugs are a stage that declared events and returned none, or returned some the dispatcher refused to publish.

inert property

inert: bool

Ran, changed nothing, emitted nothing.

Not an error — a stage can legitimately have nothing to do. But it is the shape worth counting, because a capability whose stages are always inert is a capability that is wired and doing nothing, which is exactly the state RCH-1 catches statically and this catches at runtime.

UndeclaredEventError

Bases: Exception

A stage returned an event kind it never declared in emits.