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 means disabled. No router, no stage — a deployment that never routed
compiles the plan it always did, and the kernel's None case returns the
binding the plan already made.
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(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, ...] = ())
Narrow the palette a turn offers the model, from the turn's own cue.
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)
outcome: Any = _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 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:
return no_change(reason or "selection narrowed nothing")
return applied(
ResolvedInput(
capability=TOOLS_CAPABILITY,
value=palette,
provenance=_provenance(cue, selection),
)
)
# 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:
# 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.routed"}),
),
),
handlers={ROUTING_STAGE: handle},
effect_grants=effects,
)
|