symfonic.capabilities.tools.selection

selection

Running the selection pipeline for one turn, and refusing what cannot run.

ToolSelectionPipeline and its seven stages have been public since T3.1.2 and had no kernel consumer: ToolsCapability called resolve_palette and nothing else, so ForcedChoiceStage was a class an adopter could construct, compose into a pipeline, and never reach the model with. This is the half that was missing -- what a composed pipeline means for one turn.

One writer for the force. SelectionContext.forced_choice is written by whichever stage sets it, last writer winning, which is invisible and order-dependent. Two stages that both answer "which tool is required" is a composition mistake, so it is refused when the pipeline is composed rather than resolved silently at every turn.

candidates_for

candidates_for(names: Sequence[str]) -> tuple[ToolDescriptor, ...]

The registered tools as the stages read them.

Descriptors rather than the live tool objects, and the reason is that the capability is handed names: every shipped stage is name-keyed -- an allowlist, a role palette, a policy ceiling, a forced choice -- so a descriptor carries everything they consult. The kernel binds the live objects afterwards, from the names this produces.

Source code in src/symfonic/capabilities/tools/selection.py

def candidates_for(names: Sequence[str]) -> tuple[ToolDescriptor, ...]:
    """The registered tools as the stages read them.

    Descriptors rather than the live tool objects, and the reason is that the
    capability is handed names: every shipped stage is name-keyed -- an
    allowlist, a role palette, a policy ceiling, a forced choice -- so a
    descriptor carries everything they consult. The kernel binds the live
    objects afterwards, from the names this produces.
    """
    return tuple(ToolDescriptor(name=str(name)) for name in names)

check_one_force_writer

check_one_force_writer(stages: Sequence[Any]) -> None

Refuse a pipeline where two stages answer the same question.

Detected by declaration -- a stage that resolves a force is a ForcedChoiceStage or something that names itself one -- rather than by running it and seeing what it writes, because a stage that forces only under some conditions would compose cleanly and collide in production on the turn where both fire.

Source code in src/symfonic/capabilities/tools/selection.py

def check_one_force_writer(stages: Sequence[Any]) -> None:
    """Refuse a pipeline where two stages answer the same question.

    Detected by declaration -- a stage that resolves a force is a
    ``ForcedChoiceStage`` or something that names itself one -- rather than by
    running it and seeing what it writes, because a stage that forces only
    under some conditions would compose cleanly and collide in production on
    the turn where both fire.
    """
    forcing = [
        getattr(stage, "name", type(stage).__name__)
        for stage in stages
        if _resolves_a_force(stage)
    ]
    if len(forcing) > 1:
        raise ConfigurationError(
            f"this pipeline composes {len(forcing)} stages that each resolve a "
            f"forced tool choice ({', '.join(forcing)}). The context holds one "
            "force, so the last one to run would win and the other would be "
            "silently inert -- an order-dependent answer to a question that "
            "must have one. Compose one, and express the rest of the policy as "
            "stages that narrow the candidate set."
        )

run_selection async

run_selection(pipeline: ToolSelectionPipeline, registered: Sequence[str], *, state: Any = None, messages: Sequence[Any] = (), query: str = '', role: str = '') -> SelectionResult

One turn through the composed stages.

state is the turn's own declared state, passed by identity because an adopter's resolver reads its own keys out of it -- the same contract the pre-kernel react node had.

Source code in src/symfonic/capabilities/tools/selection.py

async def run_selection(
    pipeline: ToolSelectionPipeline,
    registered: Sequence[str],
    *,
    state: Any = None,
    messages: Sequence[Any] = (),
    query: str = "",
    role: str = "",
) -> SelectionResult:
    """One turn through the composed stages.

    ``state`` is the turn's own declared state, passed by identity because an
    adopter's resolver reads its own keys out of it -- the same contract the
    pre-kernel react node had.
    """
    context = SelectionContext(
        state=dict(state or {}), messages=tuple(messages), query=query, role=role
    )
    return await pipeline.select(candidates_for(registered), context)