symfonic.capabilities.tools.stages

stages

The selection stages (T3.1.2).

Six levers that used to be six unrelated shapes — a pure function, an engine coroutine, a config tuple read in two places, two deps-registered Protocols, and a registry keyword set — now share one signature, so the composition order is something you can read instead of reconstruct.

AllowlistStage

AllowlistStage(*, always_include: Sequence[str] = (), catalog: ToolCatalog | None = None)

Re-add mandated tools that an earlier stage dropped.

A mandated name that is not in the catalogue is ignored rather than fabricated: the framework constrains the choice set, it never authors tools.

Source code in src/symfonic/capabilities/tools/stages/allowlist.py

def __init__(
    self,
    *,
    always_include: Sequence[str] = (),
    catalog: ToolCatalog | None = None,
) -> None:
    self._always_include = tuple(name for name in always_include if name)
    self._catalog = catalog

ForcedChoiceStage

ForcedChoiceStage(*, resolver: Any = None, state_key: str = 'forced_tool_choice', is_tool_message: Callable[[Any], bool] = _default_is_tool_message)

Resolve the forced tool_choice for this iteration.

Precedence, unchanged from v7.10:

1. an explicit state["forced_tool_choice"] stamp — tests and adopters pre-stamp directly;
2. the registered resolver, which re-runs every iteration so force and release are symmetric by construction;
3. None — let the model choose, the safe default.

The release check runs on both paths: once a ToolMessage naming the forced tool is in history the force is spent. A resolver that raises releases too, because a stalled loop is worse than an unforced turn.

Source code in src/symfonic/capabilities/tools/stages/forced.py

def __init__(
    self,
    *,
    resolver: Any = None,
    state_key: str = "forced_tool_choice",
    is_tool_message: Callable[[Any], bool] = _default_is_tool_message,
) -> None:
    self._resolver = resolver
    self._state_key = state_key
    self._is_tool_message = is_tool_message

IntentRoutingStage

IntentRoutingStage(*, verdict: Any = None, trigger_keywords: Mapping[str, Sequence[str]] | None = None, always_include: Sequence[str] = DEFAULT_ALWAYS_INCLUDE)

Narrow the turn's tools from an IntentVerdict-shaped object.

Decision matrix (unchanged from v7.0.1):

• no verdict, or ambiguous — abstain. A noisy signal must not cost recall.
• knowledge — keep only the always-include names.
• action — keep the verdict's matched tools, plus every tool the domain declares no trigger keywords for (pre-v6.1.8 semantics: an absent entry means "always include"), plus always-include.

Source code in src/symfonic/capabilities/tools/stages/intent.py

def __init__(
    self,
    *,
    verdict: Any = None,
    trigger_keywords: Mapping[str, Sequence[str]] | None = None,
    always_include: Sequence[str] = DEFAULT_ALWAYS_INCLUDE,
) -> None:
    self._verdict = verdict
    self._trigger_keywords = dict(trigger_keywords or {})
    self._always_include = tuple(name for name in always_include if name)

decide

decide(tools: Sequence[Any]) -> StageOutcome

The decision matrix, synchronously.

Split out because the v7.0.1 entry point (narrow_tools_for_intent) is synchronous and is called from inside a running event loop. One implementation, two callers — the alternative was asyncio.run from a coroutine, which raises, or a second copy of the matrix, which drifts.

Source code in src/symfonic/capabilities/tools/stages/intent.py

def decide(self, tools: Sequence[Any]) -> StageOutcome:
    """The decision matrix, synchronously.

    Split out because the v7.0.1 entry point
    (``narrow_tools_for_intent``) is synchronous and is called from
    inside a running event loop. One implementation, two callers —
    the alternative was ``asyncio.run`` from a coroutine, which
    raises, or a second copy of the matrix, which drifts.
    """
    tools = tuple(tools)
    if not tools:
        return StageOutcome.abstain("no candidate tools")

    verdict = self._verdict
    label = getattr(verdict, "label", None)
    if verdict is None or label == "ambiguous":
        return StageOutcome.abstain("no actionable verdict")

    always = set(self._always_include)

    if label == "knowledge":
        # Unnamed tools survive here exactly as they do on the action
        # path: they cannot be gated by a name-keyed policy, and the
        # v7.0.1 helper kept them.
        return StageOutcome.select(
            filter_by_names(tools, always),
            "knowledge turn: always-include only",
        )

    keep = set(always)
    keep.update(getattr(verdict, "matched_tool_keywords", None) or [])
    for tool in tools:
        name = tool_name(tool)
        if name is None:
            continue
        if not self._trigger_keywords.get(name):
            keep.add(name)

    return StageOutcome.select(
        filter_by_names(tools, keep), f"{label} turn: keyword intersection",
    )

LazyRoutingStage

LazyRoutingStage(*, allowed_names: Sequence[str] | None = None)

Narrow to the tool names procedural routing resolved.

Source code in src/symfonic/capabilities/tools/stages/lazy.py

def __init__(self, *, allowed_names: Sequence[str] | None = None) -> None:
    self._allowed = None if allowed_names is None else tuple(allowed_names)

PolicyStage

PolicyStage(*, catalog: ToolCatalog, max_safety_level: Any = None, budget_usd: float | None = None, include_unknown_cost: bool = True, available_capabilities: set[type] | None = None, agent_visible_only: bool = False, categories: Sequence[Any] | None = None)

Drop tools the configured policy forbids.

A tool absent from the catalogue is kept. Its metadata is unknown, and denying it would make "the registry has never heard of this tool" indistinguishable from "policy forbids this tool" — the first is a wiring bug that has to stay visible.

Source code in src/symfonic/capabilities/tools/stages/policy.py

def __init__(
    self,
    *,
    catalog: ToolCatalog,
    max_safety_level: Any = None,
    budget_usd: float | None = None,
    include_unknown_cost: bool = True,
    available_capabilities: set[type] | None = None,
    agent_visible_only: bool = False,
    categories: Sequence[Any] | None = None,
) -> None:
    self._catalog = catalog
    self._max_safety_rank = _safety_rank(max_safety_level)
    self._budget_usd = budget_usd
    self._include_unknown_cost = include_unknown_cost
    self._available_capabilities = available_capabilities
    self._agent_visible_only = agent_visible_only
    self._categories = (
        None if categories is None else {_category_value(c) for c in categories}
    )

RolePaletteStage

RolePaletteStage(*, resolver: Any = None, role: str = '')

Narrow to the palette a role resolver returns.

Three answers, three meanings:

• None — no policy. Abstain (the safe default for an empty role_tools map or an unmapped role).
• a non-empty list — the palette. It can only narrow; a resolver returning tools that were not offered is not honoured, because the framework constrains the choice set and never authors it.
• [] — abstain with a warning. The Protocol has always documented that the consumer "will detect this and abstain to all_tools with a WARN"; before T3.1.2 nothing did, and an empty palette reached _bind_tools, which binds nothing for an empty list — silently dropping every tool AND any forced tool_choice.

Source code in src/symfonic/capabilities/tools/stages/role.py

def __init__(self, *, resolver: Any = None, role: str = "") -> None:
    self._resolver = resolver
    self._role = role

SelectionStage

Bases: Protocol

One narrowing (or widening) lever in the selection pipeline.

name labels the stage in the decision trace.

protects_forced declares whether the pipeline reinstates a forced tool this stage dropped. It is True for stages that run after force resolution and express a preference (a role palette, a policy ceiling); a stage that must be able to drop a forced tool — a hard security filter, say — sets it False and says so.

apply never raises for a caller's benefit: the pipeline contains failures either way, but a stage that can degrade meaningfully should return StageOutcome.abstain(degraded=True) and log why.