symfonic.capabilities.delegation.roster

roster

The roster: who the parent may delegate to, and how a child is reached.

Registration is the last point at which a child can be refused, and it is the only point every construction path passes through — a spec the parent compiled and a child the caller handed over arrive here identically. That is why the lockdown lands here rather than in the compiler: a guard on the build path alone would cover exactly the children that were already sanitised, and miss the ones that could not be.

DelegationRoster

DelegationRoster(entries: Sequence[RosterEntry])

An immutable, ordered set of registered children.

Ordered by declaration, because the order is what the parent's model reads in the roster listing, and an adopter who put the general-purpose child last meant it to be read last.

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

def __init__(self, entries: Sequence[RosterEntry]) -> None:
    by_name: dict[str, RosterEntry] = {}
    for entry in entries:
        if entry.name in by_name:
            raise DuplicateChildError(f"duplicate sub-agent name: {entry.name!r}")
        # A pre-built child holding a write surface cannot be sanitised
        # after construction, and registering it anyway would leave the
        # caller trusting a lock that did not hold.
        assert_no_write_surface(entry.name, entry.runner)
        by_name[entry.name] = entry
    self._entries: tuple[RosterEntry, ...] = tuple(entries)
    self._by_name = by_name
    self._listing = self._render_listing()

definitions

definitions() -> tuple[ChildDefinition, ...]

The catalogue an agent-store binding answers list with.

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

def definitions(self) -> tuple[ChildDefinition, ...]:
    """The catalogue an agent-store binding answers ``list`` with."""
    return tuple(
        ChildDefinition(
            name=entry.name,
            description=entry.description,
            version=self._DEFINITION_VERSION,
        )
        for entry in self._entries
    )

describe

describe() -> str

The roster listing shown to the parent's model.

Rendered once at construction. It goes into a tool description that is sent on every turn, and re-deriving it per call would put string formatting on the hot path for a value that cannot change: the roster freezes with the graph.

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

def describe(self) -> str:
    """The roster listing shown to the parent's model.

    Rendered once at construction. It goes into a tool description that is
    sent on every turn, and re-deriving it per call would put string
    formatting on the hot path for a value that cannot change: the roster
    freezes with the graph.
    """
    return self._listing

find

find(name: str) -> RosterEntry | None

The entry registered under name, or None.

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

def find(self, name: str) -> RosterEntry | None:
    """The entry registered under ``name``, or ``None``."""
    return self._by_name.get(name)

run async

run(name: str, task: Any, *, scope: Any = None, depth: int = 1, run_id: str | None = None, root_run_id: str | None = None, parent_run_id: str | None = None) -> Any

Run child name on task and return whatever it answers.

Raises rather than returning a refusal, and raises the child's own exception untouched. The roster is the mechanism; deciding that a failed child should become a tool message the parent's model can read is a policy, and it lives one layer up in :class:~.tools.DelegationTools. A store bound into a graph needs the exception — it has its own error contract.

Parameters:

Name	Type	Description	Default
task	Any	Coerced to str. A model can produce a structured argument for a field typed as text, and dropping it would silently delegate an empty task.	required
depth	int	The depth to stamp on the child — the parent's, plus one. Passed rather than re-derived so the child's own ceiling check sees the tree it is actually in.	1

Raises:

Type	Description
UnknownChildError	If name is not registered.

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

async def run(
    self,
    name: str,
    task: Any,
    *,
    scope: Any = None,
    depth: int = 1,
    run_id: str | None = None,
    root_run_id: str | None = None,
    parent_run_id: str | None = None,
) -> Any:
    """Run child ``name`` on ``task`` and return whatever it answers.

    Raises rather than returning a refusal, and raises the child's own
    exception untouched. The roster is the mechanism; deciding that a
    failed child should become a tool message the parent's model can read
    is a policy, and it lives one layer up in
    :class:`~.tools.DelegationTools`. A store bound into a graph needs the
    exception — it has its own error contract.

    Args:
        task: Coerced to ``str``. A model can produce a structured argument
            for a field typed as text, and dropping it would silently
            delegate an empty task.
        depth: The depth to stamp on the child — the parent's, plus one.
            Passed rather than re-derived so the child's own ceiling check
            sees the tree it is actually in.

    Raises:
        UnknownChildError: If ``name`` is not registered.
    """
    entry = self._by_name.get(name)
    if entry is None:
        raise UnknownChildError(f"unknown sub-agent: {name!r}")
    inherited = {
        "scope": scope,
        "run_id": run_id,
        "agent_depth": depth,
    }
    # Lineage is withheld from the one signature known to refuse it: a
    # ``**state_overrides`` catch-all rejects unknown keys instead of
    # forwarding them, so sending lineage there turned a working
    # delegation into an error string standing where the child's answer
    # belonged. Asked of the signature, never found out by catching the
    # call -- see ``accepts_run_lineage`` for why TypeError is the wrong
    # detector.
    if accepts_run_lineage(entry.runner):
        if root_run_id is not None:
            inherited["root_run_id"] = root_run_id
        if parent_run_id is not None:
            inherited["parent_run_id"] = parent_run_id
    return await entry.runner.run(query=str(task), **inherited)