symfonic.kernel.prompt_assembly

prompt_assembly

Running the prompt-assembly phase — the first phase that dispatches.

Split from :mod:symfonic.kernel.invoker because it is a different kind of thing. The invoker holds the kernel's phase operations — bind, assemble, finish — each a small operation on a plan. This module is the composition of one of those with the dispatcher: which handlers are in the table, how a contribution is applied, what the caller gets back. Keeping them together took invoker.py over its 300-line budget, and the budget was right: the two are read for different reasons.

Prompt-assembly went first for a structural reason, not an arbitrary one. Memory hands its hydration to the prompt compiler (capabilities/memory/contribution.py), so any capability that wants to reach a turn's prompt needs this phase to dispatch before it can. It is the phase the rest of the migration is gated on.

StageContext dataclass

StageContext(plan: Any, request: Any, stage: Any, assembly: Any = None, resolved: ResolvedInputs = ResolvedInputs())

What a prompt-assembly stage is handed.

Deliberately no RequestContext: decision 2 says a stage returns rather than mutates, so handing it the mutable run context would be handing it precisely the thing it must not touch. A stage that needs to record something returns it; the dispatcher decides what happens next.

assembly is the exception that makes composition possible, and it is read-only in the sense that matters: the stage cannot write to it, only return a successor. Without it a second prompt-assembly stage could not see what the first produced, so the only assembly it could return was one built from scratch -- and the dispatcher would adopt it, discarding the first stage's work with no trace. That is what invoker.assemble_prompt meant by "something to receive and something to hand back"; dispatch landed with only the handing back.

kernel_prompt_handler

kernel_prompt_handler(assemble: Any, request: TurnRequest) -> StageHandler

The handler for the compiler's synthesized kernel.prompt stage.

W2/1b decision 4. The compiler synthesizes that stage and gave it no handler, so a rule of "every stage resolves a handler" would have failed every compilation — and the obvious escape, exempting kernel-owned stages, creates a class of stage whose inertness no rule covers. The distinction that holds is executable stage vs structural marker: if it runs, it has a handler, no matter who declared it.

No recursive dispatch. assemble stays the pure function it was; this is the thin adapter that brings it to the stage protocol, so correct logic is reused rather than rewritten one layer up.

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

def kernel_prompt_handler(assemble: Any, request: TurnRequest) -> StageHandler:
    """The handler for the compiler's synthesized ``kernel.prompt`` stage.

    W2/1b decision 4. The compiler synthesizes that stage and gave it no
    handler, so a rule of "every stage resolves a handler" would have failed
    every compilation — and the obvious escape, exempting kernel-owned stages,
    creates a class of stage whose inertness no rule covers. The distinction
    that holds is **executable stage vs structural marker**: if it runs, it has
    a handler, no matter who declared it.

    No recursive dispatch. ``assemble`` stays the pure function it was; this is
    the thin adapter that brings it to the stage protocol, so correct logic is
    reused rather than rewritten one layer up.
    """

    async def handle(context: Any) -> StageResult[PromptAssembly]:
        plan = getattr(context, "plan", None)
        if plan is None:
            return no_change("no plan on the stage context; nothing to assemble from")
        return applied(assemble(plan, request))

    return handle

run_prompt_assembly async

run_prompt_assembly(kernel: Any, plan: Any, ctx: Any, request: TurnRequest, *, handlers: Mapping[str, StageHandler] | None = None) -> tuple[Any, tuple[StageTrace, ...]]

bind, then run prompt-assembly through the dispatcher.

This is what makes W2/1b not-inert: the stage program the compiler has always produced is finally executed. The kernel's own kernel.prompt stage is registered here rather than special-cased — it resolves a handler like anyone else.

Handlers come from plan.bindings.stage_handlers — G6, where the plan already keeps its live callables. The handlers argument overlays them and exists for tests and for a composition root that has not yet moved: it is additive, never a replacement, so a plan's own handlers cannot be silently swapped out by a caller.

Returns the opened turn and the traces, so a caller can see what ran, what changed nothing and why — the record that makes an inert stage visible at runtime the way RCH-1 makes an unreached package visible statically.

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

async def run_prompt_assembly(
    kernel: Any,
    plan: Any,
    ctx: Any,
    request: TurnRequest,
    *,
    handlers: Mapping[str, StageHandler] | None = None,
) -> tuple[Any, tuple[StageTrace, ...]]:
    """``bind``, then run ``prompt-assembly`` **through the dispatcher**.

    This is what makes W2/1b not-inert: the stage program the compiler has
    always produced is finally executed. The kernel's own ``kernel.prompt``
    stage is registered here rather than special-cased — it resolves a handler
    like anyone else.

    Handlers come from ``plan.bindings.stage_handlers`` — G6, where the plan
    already keeps its live callables. The ``handlers`` argument overlays them and
    exists for tests and for a composition root that has not yet moved: it is
    additive, never a replacement, so a plan's own handlers cannot be silently
    swapped out by a caller.

    Returns the opened turn and the traces, so a caller can see what ran, what
    changed nothing and why — the record that makes an inert stage visible at
    runtime the way RCH-1 makes an unreached package visible statically.
    """
    kernel.bind(plan, ctx)

    bound = getattr(plan.bindings, "stage_handlers", None) or {}
    table: dict[str, StageHandler] = {**bound, **(handlers or {})}
    reserved = sorted(table.keys() & KERNEL_RESERVED_STAGES)
    if reserved:
        # Found by adversarial review of this function's first version, which
        # built the kernel's entry then ``update``d the caller's over it. A
        # capability passing ``{"kernel.prompt": ...}`` replaced the kernel's
        # own prompt assembly outright — the same capture that
        # ``KERNEL_OWNED_PHASES`` refuses at the *phase* level, walked in one
        # layer up through the handler table. Stage ids are the other half of
        # that boundary, so they are refused here too.
        raise ConfigurationError(
            f"handler(s) supplied for kernel-reserved stage id(s) {reserved}. "
            "These stages are the kernel's own; a capability that could replace "
            "one would control the turn it was invited into. Register under your "
            "own stage id and order against the kernel stage instead."
        )
    table[KERNEL_PROMPT_STAGE] = kernel_prompt_handler(kernel.assemble_prompt, request)

    assembly: PromptAssembly = kernel.assemble_prompt(plan, request)

    malformed: list[str] = []
    resolved = ResolvedInputs()

    async def take_resolved(stage: Any, contribution: Any) -> None:
        """Adopt one resolution stage's entry into the snapshot.

        A resolution stage contributes to the snapshot and *only* to the
        snapshot. Returning a ``PromptAssembly`` here would be an effectful
        stage writing the prompt directly, which is the thing STG-7's split
        exists to prevent -- so it is recorded as malformed rather than quietly
        ignored or quietly applied.
        """
        nonlocal resolved
        if isinstance(contribution, ResolvedInput):
            # ``extend`` rebuilds through the exact class, so a subclass that
            # overrode ``__post_init__`` cannot skip the payload contract on the
            # way in. Admission stays ``isinstance`` -- subclassing is allowed,
            # just powerless.
            resolved = resolved.extend(contribution)
            return
        malformed.append(
            f"{getattr(stage, 'capability', '?')}:{getattr(stage, 'stage_id', '?')} "
            f"is a resolution stage and returned {type(contribution).__name__}; "
            "resolution contributes a ResolvedInput"
        )

    async def apply(stage: Any, contribution: Any) -> None:
        """Adopt an applied contribution, or record that it could not be.

        The first version was ``if isinstance(...): assembly = contribution``
        with no else, so a handler returning ``applied(<wrong type>)`` was
        skipped in silence: the trace said APPLIED, nothing failed, and the
        model was called on a prompt missing that capability's contribution.
        That is the same shape as the raising-handler defect this module
        already fixed once -- a contribution that did not land, with nobody
        told -- surviving in the other branch of the ``if``.

        Recorded rather than raised here because this runs inside the
        dispatcher's apply hook, where a raise would be caught and turned into
        a FAILED result attributed to the *stage's* execution. It failed after
        the stage returned, and the distinction matters to whoever reads the
        trace.
        """
        nonlocal assembly
        if isinstance(contribution, PromptAssembly):
            assembly = contribution
            return
        malformed.append(
            f"{getattr(stage, 'capability', '?')}:{getattr(stage, 'stage_id', '?')} "
            f"returned {type(contribution).__name__}"
        )

    dispatcher = StageDispatcher(table)
    grants = frozenset(getattr(plan, "effect_grants", frozenset()))

    # Resolution first, then compilation, always (STG-7). This is a partition of
    # the phase, not a priority: it is absolute the way the ladder is, so no
    # constraint and no priority can interleave the two. The compiled order is
    # preserved *within* each half, so STG-3's tie-break still decides
    # everything it decided before.
    resolution_traces = await dispatcher.run_phase(
        plan.stage_program,
        Phase.PROMPT_ASSEMBLY,
        kind=StageKind.RESOLUTION,
        context_for=lambda stage: StageContext(
            plan=plan, request=request, stage=stage, assembly=assembly
        ),
        apply=take_resolved,
        grants=grants,
    )
    require_no_failed_stage(resolution_traces)
    # Before compilation, not after. A resolution stage returning the wrong type
    # was recorded here and not consulted until the end, so compilation ran
    # against a snapshot the kernel already knew was incomplete -- and when a
    # compilation stage then failed for want of the missing entry,
    # require_no_failed_stage named *it*. Memory broke the contract and
    # prompting got blamed, which is the same mis-attribution the contribution
    # guard in prompting/capability.py exists to avoid, one function away and
    # committed by the kernel itself.
    require_no_malformed(malformed)

    snapshot = resolved
    compilation_traces = await dispatcher.run_phase(
        plan.stage_program,
        Phase.PROMPT_ASSEMBLY,
        kind=StageKind.COMPILATION,
        # ``assembly`` is read at call time on purpose: ``apply`` rebinds it
        # between stages, so each stage receives what its predecessors left
        # rather than the value at phase entry. ``snapshot`` is bound once and
        # never rebound -- every compilation stage sees the same frozen value,
        # which is what makes the compile a pure function of it.
        context_for=lambda stage: StageContext(
            plan=plan,
            request=request,
            stage=stage,
            assembly=assembly,
            resolved=snapshot,
        ),
        apply=apply,
        grants=grants,
    )
    traces = resolution_traces + compilation_traces
    require_no_failed_stage(traces)
    require_no_malformed(malformed)
    # #24: the snapshot survives this function. It was a local -- built by the
    # resolution pass, read by the compilation pass, unreachable the moment this
    # returned -- so a value a resolution stage published could not be read by a
    # later phase at all. Bound write-once, after both passes, so what crosses
    # is what the compile actually saw.
    ctx.bind_resolved(snapshot)
    return plan.bindings.conversation.open_turn(assembly), traces