symfonic.capabilities.delegation.tools

tools

Tool exposure: what the parent's model is offered, and what it reads back.

Two tools, and one rule that governs both: a refusal is a return value. Depth reached, name unknown, child raised — each is a message the model reads and routes around, never an exception that ends the parent's run. The parent asked a question; the framework answered it. That the answer was "no" does not make it a failure of the parent.

The one exception to the exception rule is cancellation. A cancelled child means the run is going away, not that the delegation failed, and turning that into a chatty tool message would keep a model talking inside a run somebody already stopped.

What is new here is that the refusal is also a value. The shipped tool returned prose and nothing else, so anything wanting to know whether delegation had been refused had to pattern-match English. :meth:DelegationTools.delegate returns the record; :meth:DelegationTools.run_agent returns its message.

DelegationTools

DelegationTools(*, roster: DelegationRoster, depth: DepthPolicy, context: DelegationContext | None = None, record: bool = True)

The run_agent / list_agents surface over a roster.

Parameters:

Name	Type	Description	Default
record	bool	Whether a delivered hand-off is noted on the active run. On by default; switchable because the tally is observability, and a deployment must be able to turn observability off without losing the capability it observes.	True

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

def __init__(
    self,
    *,
    roster: DelegationRoster,
    depth: DepthPolicy,
    context: DelegationContext | None = None,
    record: bool = True,
) -> None:
    self._roster = roster
    self._depth = depth
    self._context = context if context is not None else DelegationContext()
    self._record = record
    self._specs = (
        DelegationToolSpec(
            name="run_agent",
            description=_RUN_AGENT_PREAMBLE + roster.describe(),
            coroutine=self.run_agent,
            parameters=("name", "task"),
        ),
        DelegationToolSpec(
            name="list_agents",
            description=_LIST_AGENTS_DESCRIPTION,
            coroutine=self.list_agents,
        ),
    )

delegate async

delegate(name: str, task: Any) -> DelegationRecord

Attempt one hand-off and report what happened.

The order of the checks is load-bearing. Depth is checked before the name, so a run at the ceiling is refused for the reason that actually applies rather than being told the child does not exist; and the child is never reached at all, which is the point of a ceiling.

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

async def delegate(self, name: str, task: Any) -> DelegationRecord:
    """Attempt one hand-off and report what happened.

    The order of the checks is load-bearing. Depth is checked before the
    name, so a run at the ceiling is refused for the reason that actually
    applies rather than being told the child does not exist; and the child
    is never reached at all, which is the point of a ceiling.
    """
    parent_depth = self._context.current_depth()
    child_depth = self._depth.child_depth(parent_depth)
    if not self._depth.admits(parent_depth):
        return DelegationRecord(
            name=name,
            outcome=DelegationOutcome.REFUSED_DEPTH,
            depth=child_depth,
            message=self._depth.refusal(),
        )
    try:
        # Deferred for the import-footprint gate: a delegation needs
        # this, resolving the facade does not.
        from symfonic.kernel.contracts.run_identity import (
            current_run_identity,
        )

        identity = current_run_identity()
        response = await self._roster.run(
            name,
            task,
            scope=self._context.current_scope(),
            depth=child_depth,
            root_run_id=identity.root_run_id if identity is not None else None,
            parent_run_id=identity.run_id if identity is not None else None,
        )
    except UnknownChildError:
        return DelegationRecord(
            name=name,
            outcome=DelegationOutcome.UNKNOWN_CHILD,
            depth=child_depth,
            message=(
                f"Unknown sub-agent {name!r}. Available:\n"
                f"{self._roster.describe()}"
            ),
        )
    except Exception as exc:  # noqa: BLE001 - contract: surface, don't raise
        # A child's failure is the parent's information, not the parent's
        # death. ``Exception`` and not ``BaseException``: a cancelled child
        # means the run is going away.
        return DelegationRecord(
            name=name,
            outcome=DelegationOutcome.CHILD_FAILED,
            depth=child_depth,
            message=f"Sub-agent {name!r} failed: {exc}",
        )
    message = str(getattr(response, "final_response", None) or "").strip()
    if not message:
        # A model may spend its completion budget on reasoning and return
        # no answer.  The call completed, but the delegation delivered
        # nothing and must not be recorded as successful.
        return DelegationRecord(
            name=name,
            outcome=DelegationOutcome.CHILD_FAILED,
            depth=child_depth,
            message=f"Sub-agent {name!r} failed: it returned no final response",
            run_id=str(getattr(response, "run_id", "") or ""),
            root_run_id=str(getattr(response, "root_run_id", "") or ""),
            parent_run_id=(
                str(getattr(response, "parent_run_id", "") or "") or None
            ),
        )
    if self._record:
        self._context.record_delegation(name)
    return DelegationRecord(
        name=name,
        outcome=DelegationOutcome.DELIVERED,
        depth=child_depth,
        message=message,
        run_id=str(getattr(response, "run_id", "") or ""),
        root_run_id=str(getattr(response, "root_run_id", "") or ""),
        parent_run_id=str(getattr(response, "parent_run_id", "") or "") or None,
    )

list_agents async

list_agents() -> str

List the sub-agents available for delegation.

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

async def list_agents(self) -> str:
    """List the sub-agents available for delegation."""
    return f"Available sub-agents:\n{self._roster.describe()}"

run_agent async

run_agent(name: str, task: str) -> str

Delegate a self-contained task to a named sub-agent.

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

async def run_agent(self, name: str, task: str) -> str:
    """Delegate a self-contained task to a named sub-agent."""
    return (await self.delegate(name, task)).message

specs

specs() -> tuple[DelegationToolSpec, ...]

The tools to bind into the parent's palette.

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

def specs(self) -> tuple[DelegationToolSpec, ...]:
    """The tools to bind into the parent's palette."""
    return self._specs