symfonic.kernel.apply

apply

Adopting a stage's contribution, and publishing the events that ride with it.

Split from :mod:.dispatch when that module crossed its 300-line budget for the third time in one change. The seam is honest rather than convenient: dispatch decides which stages run and in what order; this decides what happens to what one of them returned. The asymmetry between applying and publishing — the whole point of the module — is easier to see on its own than buried inside a loop.

apply_and_publish async

apply_and_publish(stage: CompiledStage, result: StageResult[Any], apply: Callable[[CompiledStage, Any], Awaitable[None]] | None, publish: Callable[[Any], Awaitable[None]] | None) -> tuple[int, BaseException | None]

Adopt the contribution, then publish. Only the first is captured.

Applying is inside the failure handling. The dispatcher's docstring calls a stage "invoke, validate, apply, publish ... as a transactional unit", and for a while only the first of the four was covered: an exception raised while applying escaped the dispatcher entirely, so it became neither a FAILED trace nor a turn-ending ContractViolationError — it came out of Agent.run raw. Reachable rather than theoretical: the resolved-input snapshot reads a contribution's attributes in order to rebuild it, so a hostile __getattribute__ fires here, after the handler returned successfully.

Publishing is deliberately not, and the first version of this fix got that wrong by spanning both. Measured: a publish that threw halfway left a partially published block, the trace said the stage failed, and the loop ran the next stage, which published more — breaking decision 3 twice over ("a stage that fails publishes nothing", and one execution publishing one contiguous block). Propagating stops the phase, which is what the code did before this fix and what a broken block calls for.

There is a second reason, and it is the one that generalises: publish comes from the caller, not from the capability. A raise there is the host's own machinery failing — the UndeclaredEventError category — so charging it to the stage would name the wrong party, which is a mistake this phase has now made in three different places.

Returns the number of events published and the apply failure, if any.

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

async def apply_and_publish(
    stage: CompiledStage,
    result: StageResult[Any],
    apply: Callable[[CompiledStage, Any], Awaitable[None]] | None,
    publish: Callable[[Any], Awaitable[None]] | None,
) -> tuple[int, BaseException | None]:
    """Adopt the contribution, then publish. Only the first is captured.

    **Applying** is inside the failure handling. The dispatcher's docstring
    calls a stage "invoke, validate, apply, publish ... as a transactional
    unit", and for a while only the first of the four was covered: an exception
    raised while applying escaped the dispatcher entirely, so it became neither
    a FAILED trace nor a turn-ending ``ContractViolationError`` — it came out of
    ``Agent.run`` raw. Reachable rather than theoretical: the resolved-input
    snapshot reads a contribution's attributes in order to rebuild it, so a
    hostile ``__getattribute__`` fires here, *after* the handler returned
    successfully.

    **Publishing is deliberately not**, and the first version of this fix got
    that wrong by spanning both. Measured: a publish that threw halfway left a
    partially published block, the trace said the stage failed, and the loop ran
    the next stage, which published more — breaking decision 3 twice over ("a
    stage that fails publishes nothing", and one execution publishing one
    contiguous block). Propagating stops the phase, which is what the code did
    before this fix and what a broken block calls for.

    There is a second reason, and it is the one that generalises: ``publish``
    comes from the caller, not from the capability. A raise there is the host's
    own machinery failing — the ``UndeclaredEventError`` category — so charging
    it to the stage would name the wrong party, which is a mistake this phase
    has now made in three different places.

    Returns the number of events published and the apply failure, if any.
    """
    try:
        if apply is not None:
            await apply(stage, result.contribution)
    except Exception as exc:  # noqa: BLE001 - becomes a trace, not swallowed
        # Nothing is published: the contribution did not land, and decision 3
        # says a stage that fails publishes nothing — no event announcing a
        # change that never happened.
        return 0, exc

    published = 0
    if publish is not None:
        for event in result.events:
            await publish(event)
            published += 1
    return published, None

apply_failure

apply_failure(stage: CompiledStage, phase: Phase, result: StageResult[Any], failure: BaseException) -> StageTrace

The trace for a contribution that could not be adopted.

Its own function because the reason has to be exact: the stage returned successfully, so blaming the handler would send a reader to the wrong place.

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

def apply_failure(
    stage: CompiledStage,
    phase: Phase,
    result: StageResult[Any],
    failure: BaseException,
) -> StageTrace:
    """The trace for a contribution that could not be adopted.

    Its own function because the reason has to be exact: the stage *returned*
    successfully, so blaming the handler would send a reader to the wrong place.
    """
    return StageTrace(
        stage_id=stage.stage_id,
        phase=phase,
        capability=stage.capability,
        outcome=StageOutcome.FAILED,
        reason=(
            "the stage returned successfully and its contribution "
            f"could not be applied: {failure!r}"
        ),
        declared_events=tuple(stage.emits),
        returned_events=len(result.events),
        published_events=0,
        counts=result.counts,
        error=failure,
    )