symfonic.capabilities.tools.capability

capability

Lazy tool routing as something Agent can be handed.

The resolver in :mod:~symfonic.capabilities.tools.routing decides which tools a turn may use; :mod:symfonic.kernel.palette binds that subset at the model-call boundary. This is the twenty lines that hand the declaration and the handler to the kernel at once, the way MemoryCapability did for memory.

A resolution stage, and it must run before the model. PROMPT_ASSEMBLY is the phase that does, and the palette is the one thing here that a later phase could not supply: by POST_MODEL the model has already been offered whatever it was going to be offered.

Absence remains disabled. Without a router or an explicit unavailable report, nothing starts routing implicitly. An opt-in unavailable report retains the full compiled palette before selection policy, warns at composition, and resolves a reasoned outcome. Alone it is not compatibility admission evidence.

Selection stages, if any, run here. stages= composes the public selection pipeline — an allowlist, a role palette, a policy ceiling, a forced choice — and its answer is what this contributes. Before that argument existed the seven stages were public, constructible, and unreachable: nothing in the kernel ran them, so procedural_force_tool_choice='hard' had a class and no path to the model. The router and the stages compose rather than compete: the router says which tools this turn is about and the stages apply the deployment's policy to that set.

ToolsCapability dataclass

ToolsCapability(entries_for: Callable[[Any], Any] | None = None, registered: tuple[str, ...] = (), cue_from: Callable[[Any], str] = lambda request: str(getattr(request, 'prompt', '') or ''), stages: tuple[Any, ...] = (), preconditions: tuple[Any, ...] = (), routing_unavailable: RoutingUnavailable | None = None, warning_already_issued: bool = False, count_unavailable: bool = True)

Narrow the palette a turn offers the model, from the turn's own cue.

contribute

contribute(request: CapabilityRequest) -> CapabilityContribution

Declare the routing stage and the handler that answers it.

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

def contribute(self, request: CapabilityRequest) -> CapabilityContribution:
    """Declare the routing stage and the handler that answers it."""

    async def handle(context: Any) -> StageResult[Any]:
        turn_request = getattr(context, "request", None)
        if turn_request is None:  # pragma: no cover - defensive
            return no_change("no turn request on the stage context")
        cue = self.cue_from(turn_request)
        # The compiled plan decides (CORE-227 owner decision): no tool, no
        # unavailable count; any tool, counted. Without a plan, the field.
        tools = getattr(getattr(context, "plan", None), "tool_manifest", None)
        counted = self.count_unavailable if tools is None else bool(tools)
        counted = counted and self.routing_unavailable is not None
        outcome: Any = self.routing_unavailable or _NO_ROUTER
        if self.entries_for is not None:
            entries_for = self.entries_for
            outcome = await resolve_palette(
                entries=lambda: entries_for(turn_request),
                registered=self.registered,
            )
        # Every routing refusal means *allow every tool*. It is not the end
        # of the turn's resolution any more: a forced choice is a decision
        # about which call must happen, and it survives a router that had
        # nothing to say about which calls exist.
        routed = outcome.names if isinstance(outcome, Routed) else None
        # Three states, not two: routed, refused-with-a-reason, and no
        # router composed at all. Collapsing the last two reads a reason
        # off a sentinel.
        reason = getattr(outcome, "reason", "") or "no router composed"

        if not self.stages:
            if self.routing_unavailable is not None:
                return applied(
                    ResolvedInput(
                        capability=TOOLS_CAPABILITY,
                        value=ToolPalette(names=None),
                        provenance=reason,
                    ),
                    counts={"tools.routing_unavailable": 1} if counted else None,
                )
            if routed is None:
                # Reported rather than silent: "why is that tool never
                # offered?" is answered by this reason.
                return no_change(reason)
            return applied(
                ResolvedInput(
                    capability=TOOLS_CAPABILITY,
                    value=routed,
                    # The cue that produced this palette, so a reader of a
                    # turn can get from "these tools" back to "for this
                    # question".
                    provenance=cue[:120],
                )
            )

        from symfonic.capabilities.tools.pipeline import (
            ToolSelectionPipeline,
        )
        from symfonic.capabilities.tools.selection import run_selection

        selection = await run_selection(
            ToolSelectionPipeline(self.stages),
            sorted(routed) if routed is not None else self.registered,
            state=getattr(turn_request, "properties", None),
            messages=getattr(turn_request, "history", ()),
            query=cue,
        )
        names = frozenset(selection.names)
        palette = ToolPalette(
            names=names if routed is not None or names else None,
            forced=selection.forced_choice,
        )
        if (
            palette.names is None
            and palette.forced is None
            and self.routing_unavailable is None
        ):
            return no_change(reason or "selection narrowed nothing")
        return applied(
            ResolvedInput(
                capability=TOOLS_CAPABILITY,
                value=palette,
                provenance=(
                    reason + "; " + _provenance(cue, selection)
                    if self.routing_unavailable is not None
                    else _provenance(cue, selection)
                ),
            ),
            counts={"tools.routing_unavailable": 1} if counted else None,
        )

    # Declared from what this composition actually does. The router reads
    # the procedural layer; the stages read the turn. A capability that
    # asked for ``memory-read`` in order to run an allowlist would be
    # holding authority it never exercises, which is the shape of every
    # over-grant an audit later has to reason about.
    effects = frozenset({"memory-read"}) if self.entries_for is not None else frozenset()
    if not self.entries_for and not self.stages and self.routing_unavailable is None:
        # Preconditions only: nothing to resolve before the model, so no
        # stage is declared. The gate still runs, because it hangs off the
        # dispatch point rather than off a stage -- which is the whole
        # reason it can judge the amended call.
        return CapabilityContribution(
            capability=TOOLS_CAPABILITY,
            preconditions=self.preconditions,
        )
    return CapabilityContribution(
        capability=TOOLS_CAPABILITY,
        preconditions=self.preconditions,
        stages=(
            StageDescriptor(
                stage_id=ROUTING_STAGE,
                phase=Phase.PROMPT_ASSEMBLY,
                capability=TOOLS_CAPABILITY,
                priority=_ROUTING_PRIORITY,
                effects=effects,
                kind=StageKind.RESOLUTION,
                emits=frozenset(
                    {
                        "tools.routing_unavailable"
                        if self.routing_unavailable is not None
                        else "tools.routed"
                    }
                ),
            ),
        ),
        handlers={ROUTING_STAGE: handle},
        effect_grants=effects,
    )