`symfonic.capabilities.tools.execution`¶

execution ¶

Tool-execution capability — the lifecycle of one tool call (T3.1.3).

Selection (T3.1.2) decides which tools are offered. This package owns what happens next, as separate, individually-testable phases:

binding — the chosen tools become a provider schema, with a named answer for every way that can degrade.

preconditions — the two facts the registry has recorded since v7 and nothing enforced: a tool's required capability and its per-invocation call budget.

steering — what the model is told about a call that did not run, under the invariant that every call is answered exactly once.

execution, progress, retries — the executor: order, containment, attempt accounting, and a progress channel a non-streaming tool never has to know about.

termination — when the loop stops, and which of the three reasons stopped it.

The capability imports nothing but its own package (LAY-ADR §2): chat models, tools, guards and messages are all read structurally, and every adapter to a concrete transport is injected.

BindingResult `dataclass` ¶

BindingResult(model: Any, bound: bool, reason: str, forced_choice: str | None = None, degraded: bool = False, notes: tuple[str, ...] = ())

The bound model plus a stated reason for whatever happened.

CallBudget ¶

CallBudget(descriptors: Mapping[str, ToolDescriptor])

Enforce max_calls_per_invocation, per tool, per invocation.

Counting is stateful by necessity — a budget is a fact about the invocation, not about the call — so the object is scoped to one invocation and :meth:reset starts the next one.

Source code in src/symfonic/capabilities/tools/execution/preconditions.py

def __init__(self, descriptors: Mapping[str, ToolDescriptor]) -> None:
    self._descriptors = dict(descriptors)
    self._counts: Counter[str] = Counter()

CapabilityRequirement ¶

CapabilityRequirement(descriptors: Mapping[str, ToolDescriptor], *, available: frozenset[type] = frozenset())

Block a tool whose declared requires capability is not present.

A tool the catalogue has never heard of is admitted: "the catalogue does not know this tool" is a wiring bug and must stay distinguishable from "a precondition forbids this tool". That is T3.1.2's rule for the policy stage, applied at the execution seam.

Source code in src/symfonic/capabilities/tools/execution/preconditions.py

def __init__(
    self,
    descriptors: Mapping[str, ToolDescriptor],
    *,
    available: frozenset[type] = frozenset(),
) -> None:
    self._descriptors = dict(descriptors)
    self._available = available

ExecutionReport `dataclass` ¶

ExecutionReport(outcomes: tuple[ToolOutcome, ...] = (), trace: tuple[PhaseRecord, ...] = ())

Every call's outcome, in call order, plus the phase trace.

LoopTerminationPolicy `dataclass` ¶

LoopTerminationPolicy(window: int = 10, signature_threshold: int = 3, name_only_threshold: int | None = None)

Decide whether the tool loop should run another hop.

from_mapping `classmethod` ¶

from_mapping(mapping: Mapping[str, Any] | None) -> LoopTerminationPolicy

Read settings structurally, so a state dict and a typed config reach the policy through one call site.

Source code in src/symfonic/capabilities/tools/execution/termination.py

@classmethod
def from_mapping(cls, mapping: Mapping[str, Any] | None) -> LoopTerminationPolicy:
    """Read settings structurally, so a state dict and a typed config
    reach the policy through one call site."""
    source = mapping if isinstance(mapping, Mapping) else {}
    defaults = cls()
    name_only = source.get("name_only_threshold")
    return cls(
        window=_int_or(source.get("window"), defaults.window),
        signature_threshold=_int_or(
            source.get("signature_threshold"), defaults.signature_threshold,
        ),
        name_only_threshold=name_only if isinstance(name_only, int) else None,
    )

NullProgressSink ¶

The default sink: progress is built by nobody and goes nowhere.

PhaseRecord `dataclass` ¶

PhaseRecord(phase: str, call_id: str, action: str, detail: str = '')

What one lifecycle phase did to one call, for the trace.

Precondition ¶

Bases: Protocol

A named check that may object to one call.

PreconditionSet ¶

PreconditionSet(checks: Sequence[Precondition])

Run checks in order; the first objection wins.

Source code in src/symfonic/capabilities/tools/execution/preconditions.py

def __init__(self, checks: Sequence[Precondition]) -> None:
    self._checks = tuple(checks)

PreconditionVerdict `dataclass` ¶

PreconditionVerdict(admitted: bool, content: str, reason: str, is_error: bool = True, undetermined: bool = False)

One precondition's answer for one call.

Three answers, and the third is narrow on purpose. A check that is happy returns None, so "admitted" is the absence of an objection rather than a claim any single check is entitled to make. A check that objects returns :meth:block. A check that could not reach the thing it needed in order to decide returns :meth:unavailable -- and only that case.

Why the third is not "anything went wrong". An enforcement fault must not become a permission. A predicate that is present but malformed, an evaluator that raised, a state the rule cannot read: each of those is a gate that failed while a rule existed, and admitting on them turns a broken gate into an open one. What unavailable covers is the narrower fact that the store of rules could not be reached at all -- there may be no rule for this call, and refusing every tool because a database is down takes an agent offline for a reason unrelated to what it was asked to do.

The distinction is declared by the check rather than inferred by the kernel, because only the check knows which of the two happened.

unavailable `classmethod` ¶

unavailable(*, reason: str) -> PreconditionVerdict

No rules could be consulted, so no rule was applied.

Admits, and says it admitted without deciding. content is empty because nothing is fed back to the model: the call runs, and the fact that it ran unchecked belongs in the diagnostic rather than in the conversation.

Source code in src/symfonic/capabilities/tools/execution/values.py

@classmethod
def unavailable(cls, *, reason: str) -> PreconditionVerdict:
    """No rules could be consulted, so no rule was applied.

    Admits, and says it admitted without deciding. ``content`` is empty
    because nothing is fed back to the model: the call runs, and the fact
    that it ran unchecked belongs in the diagnostic rather than in the
    conversation.
    """
    return cls(admitted=True, content="", reason=reason, undetermined=True)

ProgressSink ¶

Bases: Protocol

Where a :class:ToolProgress goes once the executor builds one.

RetrySchedule `dataclass` ¶

RetrySchedule(max_attempts: int = 1, backoff_seconds: float = 0.0, multiplier: float = 2.0, retry_on: tuple[type[BaseException], ...] = (Exception,))

How many attempts a tool call gets, and how long between them.

delay_for ¶

delay_for(attempt: int) -> float

Seconds to wait before attempt attempt + 1 (1-based).

Source code in src/symfonic/capabilities/tools/execution/retry.py

def delay_for(self, attempt: int) -> float:
    """Seconds to wait before attempt ``attempt + 1`` (1-based)."""
    if self.backoff_seconds <= 0:
        return 0.0
    return self.backoff_seconds * (self.multiplier ** max(0, attempt - 1))

should_retry ¶

should_retry(attempt: int, exc: BaseException) -> bool

Is attempt (1-based) allowed another try after exc?

Source code in src/symfonic/capabilities/tools/execution/retry.py

def should_retry(self, attempt: int, exc: BaseException) -> bool:
    """Is ``attempt`` (1-based) allowed another try after ``exc``?"""
    if attempt >= max(1, self.max_attempts):
        return False
    return isinstance(exc, self.retry_on)

SchemaBinder ¶

SchemaBinder(*, annotate_cache: Callable[[Any, str], None] | None = None)

Bind a tool set onto a chat model, structurally.

The binder never imports a chat-model class: it calls bind_tools on whatever it is handed and reads the exceptions the provider contract documents. annotate_cache is injected because cache annotation is provider-wire-format work that belongs on the legacy side of the seam, not in a capability.

Source code in src/symfonic/capabilities/tools/execution/binding.py

def __init__(
    self, *, annotate_cache: Callable[[Any, str], None] | None = None,
) -> None:
    self._annotate_cache = annotate_cache

SkillPrecondition ¶

SkillPrecondition(skills: Callable[[], Awaitable[Any]], state: Callable[[], Any])

Block a call whose governing skill says its state does not hold.

Parameters:

Name	Type	Description	Default
`skills`	`Callable[[], Awaitable[Any]]`	awaited for this turn's procedural hits -- the same callable shape the router takes, so a deployment wires one source of skills rather than two.	required
`state`	`Callable[[], Any]`	the turn's state, read when a call is being judged rather than captured at construction. A precondition about what has happened must be evaluated against what has happened.	required

Source code in src/symfonic/capabilities/tools/execution/skill_gate.py

def __init__(
    self,
    skills: Callable[[], Awaitable[Any]],
    state: Callable[[], Any],
) -> None:
    """
    Args:
        skills: awaited for this turn's procedural hits -- the same
            callable shape the router takes, so a deployment wires one
            source of skills rather than two.
        state: the turn's state, read when a call is being judged rather
            than captured at construction. A precondition about what has
            happened must be evaluated against what has happened.
    """
    self._skills = skills
    self._state = state

SteeringGate ¶

SteeringGate(consult: Consultant | None)

Consult a guard per call and partition into admitted vs answered.

The consultant is injected rather than imported: the legacy seam builds a BeforeToolCallEvent and calls a callback manager, a test passes a lambda, and the capability needs to know about neither.

The verdict contract is structural and deliberately narrow — a mapping with action == "skip", plus content and an optional is_error. Anything else admits the call, including a consultant that raises: a broken guard must not stall dispatch.

Source code in src/symfonic/capabilities/tools/execution/steering.py

def __init__(self, consult: Consultant | None) -> None:
    self._consult = consult

SteeringResult `dataclass` ¶

SteeringResult(call_id: str, tool_name: str, content: str, is_error: bool = True, origin: str = 'steering')

The answer given to a call that was not dispatched.

It carries call_id because an unanswered tool call permanently bricks a provider thread: every blocked call owes the transcript exactly one result.

TerminationVerdict `dataclass` ¶

TerminationVerdict(terminate: bool, reason: str, detail: str = '')

Stop or continue, and why.

ToolCall `dataclass` ¶

ToolCall(id: str, name: str, args: Mapping[str, Any] = dict())

One call the model asked for: an id, a name, and arguments.

id may be empty. A call with no id cannot be answered, and the framework keeps it rather than discarding it so the provider bug surfaces as an unanswered call rather than as a call that vanished.

signature `property` ¶

signature: str

Name plus arguments — the repeat key the loop cut counts.

ToolExecutor ¶

ToolExecutor(*, preconditions: PreconditionSet | None = None, steering: SteeringGate | None = None, retries: RetrySchedule | None = None, progress: ProgressSink | None = None)

Run a batch of tool calls through the lifecycle, in order.

Source code in src/symfonic/capabilities/tools/execution/executor.py

def __init__(
    self,
    *,
    preconditions: PreconditionSet | None = None,
    steering: SteeringGate | None = None,
    retries: RetrySchedule | None = None,
    progress: ProgressSink | None = None,
) -> None:
    self._preconditions = preconditions or PreconditionSet([])
    self._steering = steering or SteeringGate(None)
    self._retries = retries or RetrySchedule()
    self._progress = progress or NullProgressSink()

ToolOutcome `dataclass` ¶

ToolOutcome(call: ToolCall, content: str, is_error: bool = False, attempts: int = 0, phase: str = 'dispatch', reason: str = '', value: Any = None)

The single result of one call, whatever produced it.

ToolProgress `dataclass` ¶

ToolProgress(call_id: str, tool_name: str, sequence: int, payload: Any = None)

One intermediate value a still-running tool emitted.

descriptors_by_name ¶

descriptors_by_name(catalog: Any) -> dict[str, ToolDescriptor]

Read a :class:~symfonic.capabilities.tools.catalog.ToolCatalog (or anything iterable of descriptors) into the mapping the checks want.

Source code in src/symfonic/capabilities/tools/execution/preconditions.py

def descriptors_by_name(catalog: Any) -> dict[str, ToolDescriptor]:
    """Read a :class:`~symfonic.capabilities.tools.catalog.ToolCatalog`
    (or anything iterable of descriptors) into the mapping the checks want."""
    return {d.name: d for d in catalog}

drain_with_progress `async` ¶

drain_with_progress(values: AsyncIterator[Any], emit: Callable[[int, Any], Awaitable[None]]) -> Any

Drain values, emitting all but the last, and return the last.

empty iterator -> None (what a plain tool returning None would surface, so the result round-trip stays consistent);
one value -> returned directly, no progress emitted;
N values -> N - 1 progress emissions numbered 0..N-2.

Source code in src/symfonic/capabilities/tools/execution/progress.py

async def drain_with_progress(
    values: AsyncIterator[Any],
    emit: Callable[[int, Any], Awaitable[None]],
) -> Any:
    """Drain ``values``, emitting all but the last, and return the last.

    * empty iterator -> ``None`` (what a plain tool returning ``None``
      would surface, so the result round-trip stays consistent);
    * one value -> returned directly, no progress emitted;
    * N values -> ``N - 1`` progress emissions numbered ``0..N-2``.
    """
    last: Any = _EMPTY
    sequence = 0
    async for value in values:
        if last is not _EMPTY:
            await emit(sequence, last)
            sequence += 1
        last = value
    if last is _EMPTY:
        return None
    return last

symfonic.capabilities.tools.execution¶

execution ¶

BindingResult dataclass ¶

CallBudget ¶

CapabilityRequirement ¶

ExecutionReport dataclass ¶

LoopTerminationPolicy dataclass ¶

from_mapping classmethod ¶

NullProgressSink ¶

PhaseRecord dataclass ¶

Precondition ¶

PreconditionSet ¶

PreconditionVerdict dataclass ¶

unavailable classmethod ¶

ProgressSink ¶

RetrySchedule dataclass ¶

delay_for ¶

should_retry ¶

SchemaBinder ¶

SkillPrecondition ¶

SteeringGate ¶

SteeringResult dataclass ¶

TerminationVerdict dataclass ¶

ToolCall dataclass ¶

signature property ¶

ToolExecutor ¶

ToolOutcome dataclass ¶

ToolProgress dataclass ¶

descriptors_by_name ¶

drain_with_progress async ¶

`symfonic.capabilities.tools.execution`¶

BindingResult `dataclass` ¶

ExecutionReport `dataclass` ¶

LoopTerminationPolicy `dataclass` ¶

from_mapping `classmethod` ¶

PhaseRecord `dataclass` ¶

PreconditionVerdict `dataclass` ¶

unavailable `classmethod` ¶

RetrySchedule `dataclass` ¶

SteeringResult `dataclass` ¶

TerminationVerdict `dataclass` ¶

ToolCall `dataclass` ¶

signature `property` ¶

ToolOutcome `dataclass` ¶

ToolProgress `dataclass` ¶

drain_with_progress `async` ¶