Skip to content

symfonic.capabilities.delegation

delegation

Sub-agent delegation capability (T3.4.2).

Delegation used to be spread across five places: two declaration types and a registry in one package, a tool factory in another, a child constructor and three context variables inside the parent agent's own class, and a teardown sweep in two of its dunder methods. Reading the rules meant reading all five, and three of them could only be reached by constructing an agent.

Everything delegation decides now lives here:

  • Declarations — :mod:.declarations. Two forms, one validation, one fact that distinguishes them: who owns the child's lifetime.
  • The child compiler — :mod:.compiler. Inheritance, sanitisation, ownership, and duplicate refusal, all before anything is constructed.
  • The roster — :mod:.roster. The registration checkpoint every construction path passes through, and the only one that sees pre-built children.
  • Depth — :mod:.depth. The ceiling, the arithmetic, and the refusal.
  • The run scope — :mod:.context. Depth, prompt-block snapshot, and delegated-to tally as one lifetime, isolated per task.
  • Tool exposure — :mod:.tools. run_agent / list_agents, where every refusal is a value the model can read.
  • Lockdown — :mod:.lockdown. No delegated child holds a prompt-block write surface, on any construction path.
  • Lifecycle — :mod:.lifecycle. Flush then close, for children this parent built and no others.

No recursive facade coupling. A delegated child is an agent, so the obvious implementation imports the class that composes this capability — to build a child, to type-check one, and to inherit a config from one. All three are ports in :mod:.contracts instead, implemented at the composition root. This package imports nothing but itself; a contract test asserts it, and a second asserts that importing it does not drag the agent package in behind it.

SELF_EDIT_FIELD module-attribute 

SELF_EDIT_FIELD = 'prompt_block_self_edit'

The config field naming the (always denied) child write surface.

WRITE_SURFACE_NAMESPACE module-attribute 

WRITE_SURFACE_NAMESPACE = 'memory_block_'

Reserved tool-name prefix. Fail-closed on the whole prefix, not a fixed set: a verb added under it tomorrow is caught without editing this module.

WRITE_SURFACE_TOOL_NAMES module-attribute 

WRITE_SURFACE_TOOL_NAMES: frozenset[str] = frozenset({'memory_block_append', 'memory_block_replace', 'memory_block_rewrite'})

The named write tools. Documentation, not the check — see the pattern.

ActiveRun 

ActiveRun(depth: int, identity: Any = None)

The delegation facts one run accumulates.

Handed out by :meth:DelegationContext.run_scope and readable after the scope closes — the caller stamping delegated_to onto a response reads it once the run has finished, and a value that evaporated with the scope would be unreadable exactly when it is needed.

Source code in src/symfonic/capabilities/delegation/context.py 
59
60
61
62
63
64
65
66
def __init__(self, depth: int, identity: Any = None) -> None:
    self.depth = depth
    self.run_id = str(getattr(identity, "run_id", "") or "")
    self.root_run_id = str(getattr(identity, "root_run_id", "") or self.run_id)
    self.parent_run_id = (
        str(getattr(identity, "parent_run_id", "") or "") or None
    )
    self._delegated: list[str] = []

delegated property 

delegated: bool

True if this run handed work to at least one child.

delegated_to property 

delegated_to: tuple[str, ...]

The children this run delegated to, in order, with repeats.

Repeats are kept. "The parent asked the researcher three times" is a different run from "the parent asked once", and de-duplicating would erase the loop an operator is usually looking for.

ChildBuildPlan dataclass 

ChildBuildPlan(*, name: str, description: str, config: Any, provider: Any, tools: Sequence[Any] = (), when_to_use: str | None = None, shared: dict[str, Any] = dict())

Everything a builder needs, and nothing about how to build it.

The compiler's whole output for one spec. Splitting the decision (this package) from the construction (a builder port at the composition root) is what lets inheritance, sanitisation, and ownership be tested without ever constructing an agent — and what keeps a capability from importing the facade whose children it compiles.

Attributes:

Name Type Description
shared dict[str, Any]

Wiring the parent hands every child it builds — the memory orchestrator, for instance. A dict rather than named fields because the set is the composition root's business, not this package's: a capability that enumerated the parent's collaborators would be coupled to all of them.

ChildBuilder

Bases: Protocol

Turns a :class:~.values.ChildBuildPlan into a runner.

The one port that must live at the composition root, because it is the one that names a concrete agent class. Everything the builder needs to decide has already been decided: inheritance resolved, config sanitised, tools fixed, shared wiring attached.

ChildCompiler 

ChildCompiler(*, builder: Any, inheritance: Any)

Resolves declarations into children, via ports it does not implement.

Parameters:

Name Type Description Default
builder Any

Constructs a child from a plan. The one port that names a concrete agent type, which is why it lives at the composition root.

 required
inheritance Any

Derives a child config from the parent's. A port because the shipped primitive is a classmethod on the facade's config class, and reaching for it directly would drag the whole configuration module into a capability.

 required
Source code in src/symfonic/capabilities/delegation/compiler.py 
74
75
76
def __init__(self, *, builder: Any, inheritance: Any) -> None:
    self._builder = builder
    self._inheritance = inheritance

compile 

compile(declarations: Iterable[Any], *, parent_config: Any, parent_provider: Any = None, shared: dict[str, Any] | None = None) -> CompiledChildren

Resolve declarations into roster entries and owned children.

Parameters:

Name Type Description Default
parent_config Any

Inherited by every spec child that declares no config of its own.

 required
parent_provider Any

Inherited by every spec child that declares no provider. Sharing one instance is safe — providers are stateless — and building a second would double the connection pools for no behavioural difference.

 None
shared dict[str, Any] | None

Wiring handed to every built child, verbatim. The parent's memory orchestrator goes here: a child built with its own would resolve a different store than the parent, which silently breaks any child whose prompt reads a memory-backed block.

 None

Raises:

Type Description
DuplicateChildError

If two declarations claim one name. Checked across the whole list first, so a failure constructs nothing.
TypeError

If a declaration is neither form.
Source code in src/symfonic/capabilities/delegation/compiler.py 
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
def compile(
    self,
    declarations: Iterable[Any],
    *,
    parent_config: Any,
    parent_provider: Any = None,
    shared: dict[str, Any] | None = None,
) -> CompiledChildren:
    """Resolve ``declarations`` into roster entries and owned children.

    Args:
        parent_config: Inherited by every spec child that declares no
            config of its own.
        parent_provider: Inherited by every spec child that declares no
            provider. Sharing one instance is safe — providers are
            stateless — and building a second would double the connection
            pools for no behavioural difference.
        shared: Wiring handed to every built child, verbatim. The parent's
            memory orchestrator goes here: a child built with its own would
            resolve a different store than the parent, which silently
            breaks any child whose prompt reads a memory-backed block.

    Raises:
        DuplicateChildError: If two declarations claim one name. Checked
            across the whole list first, so a failure constructs nothing.
        TypeError: If a declaration is neither form.
    """
    declared = tuple(declarations)
    self._refuse_duplicates(declared)
    entries: list[RosterEntry] = []
    owned: list[Any] = []
    for declaration in declared:
        entry = self._resolve(
            declaration,
            parent_config=parent_config,
            parent_provider=parent_provider,
            shared=shared or {},
        )
        entries.append(entry)
        if entry.owned:
            owned.append(entry.runner)
    return CompiledChildren(entries=tuple(entries), owned=tuple(owned))

ChildDeclaration dataclass 

ChildDeclaration(*, name: str, description: str, when_to_use: str | None = None)

The vocabulary both declaration forms share.

Keyword-only on purpose. A positional (name, agent, description) and a positional (name, description, ...) in the same package is a transposition waiting to happen, and the two forms are meant to be interchangeable in a single list.

Attributes:

Name Type Description
name str

The routing key the parent's model passes to run_agent. Non-empty and whitespace-free — it is matched literally against what a model typed, so a name with a space in it is a name the model cannot reliably reproduce.
description str

The one-liner shown to the parent model. Required, because a roster entry nobody can choose between is not a roster.
when_to_use str | None

Optional longer guidance appended to the roster listing.

owned property 

owned: bool

True when whoever compiles this declaration owns the child.

ChildDeclarationError

Bases: DelegationError, ValueError

A declared child is not describable: no name, no description, no runner.

Also a :class:ValueError because that is what the shipped declaration types raised, and an adopter's except ValueError around agent construction is a reasonable thing to have written.

ChildDefinition dataclass 

ChildDefinition(*, name: str, description: str, version: str)

The catalogue shape an agent store answers list/read with.

ChildLifecycle 

ChildLifecycle(owned: Sequence[Any])

Owns the teardown of the children a compilation constructed.

Source code in src/symfonic/capabilities/delegation/lifecycle.py 
34
35
36
37
38
def __init__(self, owned: Sequence[Any]) -> None:
    # Copied at construction: a caller's list that grew afterwards would
    # hand this object children it never agreed to own.
    self._owned: tuple[Any, ...] = tuple(owned)
    self._closed = False

owned property 

owned: tuple[Any, ...]

The children this lifecycle will release.

Still readable after teardown. Idempotency is a flag, not an emptied list, because "what did this parent own?" is a question asked during an investigation of a shutdown that went wrong.

aclose async 

aclose() -> TeardownReport

Release each owned child's resources. Idempotent.

Source code in src/symfonic/capabilities/delegation/lifecycle.py 
62
63
64
65
66
67
async def aclose(self) -> TeardownReport:
    """Release each owned child's resources. Idempotent."""
    if self._closed:
        return TeardownReport(verb="aclose")
    self._closed = True
    return await self._sweep("aclose", "aclose")

flush async 

flush() -> TeardownReport

Await each child's pending background work.

Never raises for a child's sake. One child whose consolidation is wedged must not strand the others' — including their aclose, which runs after this.

Source code in src/symfonic/capabilities/delegation/lifecycle.py 
53
54
55
56
57
58
59
60
async def flush(self) -> TeardownReport:
    """Await each child's pending background work.

    Never raises for a child's sake. One child whose consolidation is
    wedged must not strand the others' — including their ``aclose``, which
    runs after this.
    """
    return await self._sweep("flush", "flush_background_tasks")

shutdown async 

shutdown() -> TeardownReport

Flush, then close, whatever the flush did.

The inner finally is the contract: a raising flush cannot skip the close. Both phases report into one record so a caller sees everything that went wrong, not only whatever failed last.

Source code in src/symfonic/capabilities/delegation/lifecycle.py 
69
70
71
72
73
74
75
76
77
78
79
80
81
async def shutdown(self) -> TeardownReport:
    """Flush, then close, whatever the flush did.

    The inner ``finally`` is the contract: a raising flush cannot skip the
    close. Both phases report into one record so a caller sees everything
    that went wrong, not only whatever failed last.
    """
    flushed = TeardownReport(verb="flush")
    try:
        flushed = await self.flush()
    finally:
        closed = await self.aclose()
    return flushed.merge(closed)

ChildLockdownError

Bases: DelegationError, ValueError

A child would have held a prompt-block write surface.

A prompt block has exactly one writer. A delegated child that can write one is a second writer, and one the parent never sees — the child runs its own loop with its own palette. Raised at construction or registration, never mid-run: by the time a run starts, the palette has already been advertised.

ChildRunner

Bases: Protocol

Anything a parent can delegate to.

The keyword arguments are the inheritance surface: the child is told which tenant it is acting for, which run it belongs to, and how deep it already is. A runner that ignores agent_depth cannot enforce its own ceiling, which is why it is passed rather than re-derived — the child's context is not the parent's, and a nested run that started its depth count from zero would make the ceiling unreachable.

ChildSpec dataclass 

ChildSpec(*, name: str, description: str, when_to_use: str | None = None, tools: Sequence[Any] = tuple(), domain_description: str | None = None, model_name: str | None = None, temperature: float | None = None, max_tokens: int | None = None, provider: Any | None = None, config: Any | None = None)

Bases: ChildDeclaration

A child described rather than built: the parent constructs it.

The declarative form. Everything unset is inherited from the parent — provider, behaviour flags, model settings — because a child that silently reverted to framework defaults would drift from its parent on every flag the adopter had deliberately changed. What is not inherited is the domain: the child gets a fresh one scoped to this spec, so its tool manifest derives from its own tools rather than leaking the parent's.

Attributes:

Name Type Description
tools Sequence[Any]

The child's whole palette. Normalised to a tuple so a caller's list cannot be appended to after the child is compiled.
domain_description str | None

The child's domain directive. Defaults to :attr:~ChildDeclaration.description — see :attr:effective_domain_description.
model_name str | None

Optional child model override; inherits the parent's.
temperature float | None

Optional child sampling override.
max_tokens int | None

Optional child output-ceiling override.
provider Any | None

Optional model provider. Unset inherits the parent's, which is safe because providers are stateless.
config Any | None

A complete child config, bypassing inheritance entirely. It is sanitised rather than trusted — this is the path an adopter reaches for precisely when they want something non-default, and a lock skipped on that path is decorative.

effective_domain_description property 

effective_domain_description: str

The domain directive this spec actually resolves to.

Read here rather than at the compiler so the fallback is a property of the declaration — the thing the adopter reads — instead of a rule two construction branches each have to remember.

owned property 

owned: bool

Always. Whoever compiles a spec built the child and must close it.

CompiledChildren dataclass 

CompiledChildren(entries: tuple[RosterEntry, ...] = (), owned: tuple[Any, ...] = ())

What compiling a declaration list produced.

Attributes:

Name Type Description
entries tuple[RosterEntry, ...]

Roster entries in declaration order.
owned tuple[Any, ...]

The children this compilation constructed, and only those. The list a lifecycle closes.

ConfigInheritance

Bases: Protocol

Derives a child config from a parent config.

A port rather than a direct call for a subtle reason: the shipped inheritance primitive is a classmethod on the facade's own config class, so calling it directly would couple the capability to the whole configuration module — and through it, transitively, to most of the framework.

Implementations receive only None-meaning-inherit overrides. An implementation that treats an explicit None as "set this to None" would silently clear the parent's model settings on every child.

DelegationCapability 

DelegationCapability(*, roster: DelegationRoster, tools: DelegationTools, lifecycle: ChildLifecycle, context: DelegationContext, depth: DepthPolicy)

Roster, tools, run scope, and lifecycle for one parent agent.

Source code in src/symfonic/capabilities/delegation/capability.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
def __init__(
    self,
    *,
    roster: DelegationRoster,
    tools: DelegationTools,
    lifecycle: ChildLifecycle,
    context: DelegationContext,
    depth: DepthPolicy,
) -> None:
    self.roster = roster
    self.tools = tools
    self.lifecycle = lifecycle
    self.context = context
    self.depth = depth

active property 

active: bool

True when at least one child is declared.

A parent that declared none must not be offered run_agent: a tool whose whole roster is "(none declared)" is an invitation to attempt a delegation that can only ever be refused.

compile classmethod 

compile(declarations: Iterable[Any], *, builder: Any, inheritance: Any, parent_config: Any, parent_provider: Any = None, max_depth: int = 3, snapshots: Any | None = None, resolve_scope: Callable[[], Any] | None = None, shared: dict[str, Any] | None = None, record_delegations: bool = True) -> DelegationCapability

Compile declarations into a wired capability.

Every argument after declarations is either a port or a policy. There is no argument that is a piece of the agent being wired, which is what keeps this a capability rather than a second constructor for the thing that composes it.

Source code in src/symfonic/capabilities/delegation/capability.py 
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
@classmethod
def compile(
    cls,
    declarations: Iterable[Any],
    *,
    builder: Any,
    inheritance: Any,
    parent_config: Any,
    parent_provider: Any = None,
    max_depth: int = 3,
    snapshots: Any | None = None,
    resolve_scope: Callable[[], Any] | None = None,
    shared: dict[str, Any] | None = None,
    record_delegations: bool = True,
) -> DelegationCapability:
    """Compile ``declarations`` into a wired capability.

    Every argument after ``declarations`` is either a port or a policy.
    There is no argument that is a piece of the agent being wired, which is
    what keeps this a capability rather than a second constructor for the
    thing that composes it.
    """
    compiled: CompiledChildren = ChildCompiler(
        builder=builder, inheritance=inheritance
    ).compile(
        declarations,
        parent_config=parent_config,
        parent_provider=parent_provider,
        shared=shared,
    )
    context = DelegationContext(snapshots=snapshots, resolve_scope=resolve_scope)
    roster = compiled.roster()
    return cls(
        roster=roster,
        tools=DelegationTools(
            roster=roster,
            depth=DepthPolicy(max_depth=max_depth),
            context=context,
            record=record_delegations,
        ),
        lifecycle=ChildLifecycle(compiled.owned),
        context=context,
        depth=DepthPolicy(max_depth=max_depth),
    )

contribute 

contribute(request: CapabilityRequest) -> CapabilityContribution

Offer run_agent / list_agents to the turn being compiled.

Tools only, and no stage. Delegation contributes nothing to the prompt and performs no effect of its own before the model runs: what it offers is a pair of callables the model may reach for, and the child's own run is the child's effect, not this capability's. Declaring a stage that did nothing would be the declared-and-never-read shape the fold exists to refuse -- validate() would even accept it, because an empty handler answers an empty stage.

Contributing nothing at all when no child is declared is the same rule :attr:tool_specs states, applied one layer out: a run_agent whose whole roster reads "(none declared)" invites the model to attempt a delegation that can only be refused. The contribution still carries the capability's name, so a plan records that delegation was folded and found nothing rather than that it was never folded.

request is read for its grants and found to need none: delegation declares no :class:~symfonic.kernel.contracts.effects.EffectFamily, because every effect a child performs is granted to the child's plan by whoever composed it. A parent that could widen its children's authority by delegating to them would make the grant list a description rather than a bound.

Source code in src/symfonic/capabilities/delegation/capability.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
def contribute(self, request: CapabilityRequest) -> CapabilityContribution:
    """Offer ``run_agent`` / ``list_agents`` to the turn being compiled.

    **Tools only, and no stage.** Delegation contributes nothing to the
    prompt and performs no effect of its own before the model runs: what it
    offers is a pair of callables the model may reach for, and the child's
    own run is the child's effect, not this capability's. Declaring a stage
    that did nothing would be the declared-and-never-read shape the fold
    exists to refuse -- ``validate()`` would even accept it, because an
    empty handler answers an empty stage.

    Contributing *nothing at all* when no child is declared is the same
    rule :attr:`tool_specs` states, applied one layer out: a ``run_agent``
    whose whole roster reads "(none declared)" invites the model to attempt
    a delegation that can only be refused. The contribution still carries
    the capability's name, so a plan records that delegation was folded and
    found nothing rather than that it was never folded.

    ``request`` is read for its grants and found to need none: delegation
    declares no :class:`~symfonic.kernel.contracts.effects.EffectFamily`,
    because every effect a child performs is granted to the *child's* plan
    by whoever composed it. A parent that could widen its children's
    authority by delegating to them would make the grant list a description
    rather than a bound.
    """
    return CapabilityContribution(
        capability=CAPABILITY_NAME,
        tools=self.tool_specs(),
    )

run_scope async 

run_scope(*, depth: Any = 0) -> AsyncIterator[ActiveRun]

Enter one run's delegation scope.

Entered by every entry point, not only the one that happens to delegate. The scope is where a delegated run's prompt-block snapshot and its delegation tally live, and an entry point that skipped it gave the same agent different behaviour depending on how it was called.

Source code in src/symfonic/capabilities/delegation/capability.py 
158
159
160
161
162
163
164
165
166
167
168
@asynccontextmanager
async def run_scope(self, *, depth: Any = 0) -> AsyncIterator[ActiveRun]:
    """Enter one run's delegation scope.

    Entered by *every* entry point, not only the one that happens to
    delegate. The scope is where a delegated run's prompt-block snapshot
    and its delegation tally live, and an entry point that skipped it gave
    the same agent different behaviour depending on how it was called.
    """
    async with self.context.run_scope(depth=depth) as run:
        yield run

shutdown async 

shutdown() -> TeardownReport

Flush and release the children this capability built.

Source code in src/symfonic/capabilities/delegation/capability.py 
170
171
172
async def shutdown(self) -> TeardownReport:
    """Flush and release the children this capability built."""
    return await self.lifecycle.shutdown()

tool_specs 

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

The tools to bind, or nothing at all when no child is declared.

Source code in src/symfonic/capabilities/delegation/capability.py 
154
155
156
def tool_specs(self) -> tuple[DelegationToolSpec, ...]:
    """The tools to bind, or nothing at all when no child is declared."""
    return self.tools.specs() if self.active else ()

DelegationContext 

DelegationContext(*, snapshots: Any | None = None, resolve_scope: Callable[[], Any] | None = None)

Opens and closes the run-local scope delegation needs.

Parameters:

Name Type Description Default
snapshots Any | None

The prompt-block snapshot port, or None. A deployment that renders no blocks passes nothing and no slot is ever opened — supported, not degraded.

 None
resolve_scope Callable[[], Any] | None

Returns the tenant scope a delegated child should inherit. A callable rather than a value because the scope belongs to the run in flight, not to the wiring: the capability is built once and serves every tenant that arrives.

 None
Source code in src/symfonic/capabilities/delegation/context.py 
102
103
104
105
106
107
108
109
def __init__(
    self,
    *,
    snapshots: Any | None = None,
    resolve_scope: Callable[[], Any] | None = None,
) -> None:
    self._snapshots = snapshots
    self._resolve_scope = resolve_scope

current_depth 

current_depth() -> int

The depth of the run in flight; 0 outside any run.

Source code in src/symfonic/capabilities/delegation/context.py 
113
114
115
def current_depth(self) -> int:
    """The depth of the run in flight; ``0`` outside any run."""
    return _active_depth.get()

current_run 

current_run() -> ActiveRun | None

The run in flight, or None when nothing has opened a scope.

Source code in src/symfonic/capabilities/delegation/context.py 
117
118
119
def current_run(self) -> ActiveRun | None:
    """The run in flight, or ``None`` when nothing has opened a scope."""
    return _active_run.get()

current_scope 

current_scope() -> Any

The tenant scope a child of this run should inherit.

Source code in src/symfonic/capabilities/delegation/context.py 
121
122
123
def current_scope(self) -> Any:
    """The tenant scope a child of this run should inherit."""
    return self._resolve_scope() if self._resolve_scope is not None else None

delegated_to 

delegated_to() -> tuple[str, ...]

What the run in flight has delegated to so far.

Source code in src/symfonic/capabilities/delegation/context.py 
125
126
127
128
def delegated_to(self) -> tuple[str, ...]:
    """What the run in flight has delegated to so far."""
    run = _active_run.get()
    return () if run is None else run.delegated_to

record_delegation 

record_delegation(name: str) -> None

Note a completed hand-off on the run in flight.

A no-op outside a run rather than an error. The tool surface is reachable from a direct call in a test or a script that never opened a scope, and refusing there would make the observability feature able to break a delegation that otherwise worked.

Source code in src/symfonic/capabilities/delegation/context.py 
132
133
134
135
136
137
138
139
140
141
142
def record_delegation(self, name: str) -> None:
    """Note a completed hand-off on the run in flight.

    A no-op outside a run rather than an error. The tool surface is
    reachable from a direct call in a test or a script that never opened a
    scope, and refusing there would make the observability feature able to
    break a delegation that otherwise worked.
    """
    run = _active_run.get()
    if run is not None:
        run.record(name)

run_scope async 

run_scope(*, depth: Any = 0, identity: Any = None) -> AsyncIterator[ActiveRun]

Open the scope one run needs; close it whatever happens.

Yields the run's :class:ActiveRun. A delegated run (depth > 0) also holds a snapshot slot for its whole lifetime; a top-level run does not open one at all.

Source code in src/symfonic/capabilities/delegation/context.py 
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
@asynccontextmanager
async def run_scope(
    self, *, depth: Any = 0, identity: Any = None
) -> AsyncIterator[ActiveRun]:
    """Open the scope one run needs; close it whatever happens.

    Yields the run's :class:`ActiveRun`. A delegated run (``depth > 0``)
    also holds a snapshot slot for its whole lifetime; a top-level run does
    not open one at all.
    """
    resolved = coerce_depth(depth)
    if identity is None:
        from symfonic.kernel.contracts.run_identity import current_run_identity

        identity = current_run_identity()
    run = ActiveRun(resolved, identity)
    depth_token = _active_depth.set(resolved)
    run_token = _active_run.set(run)
    snapshot_token = self._open_snapshot(resolved)
    try:
        yield run
    finally:
        self._close_snapshot(snapshot_token)
        _active_run.reset(run_token)
        _active_depth.reset(depth_token)

DelegationError

Bases: Exception

Base for everything this capability raises.

DelegationOutcome

Bases: StrEnum

What happened when the parent's model asked to delegate.

Four values, and only one of them is a delegation. The other three are the reasons the shipped tool returned prose for — depth reached, name unknown, child raised — promoted from strings a caller would have to pattern-match into a value it can branch on. The message stays; what changes is that the message is no longer the only record.

delivered property 

delivered: bool

True only for a hand-off a child actually completed.

Read by the recording path: a refused, misrouted, or failed delegation is not something the parent delegated to, and stamping it on the response would make an operator reading delegated_to believe a child ran.

DelegationRecord dataclass 

DelegationRecord(*, name: str, outcome: DelegationOutcome, depth: int, message: str, run_id: str = '', root_run_id: str = '', parent_run_id: str | None = None)

One attempted hand-off, as a value.

Attributes:

Name Type Description
name str

The routing key the model asked for — kept even when it matched nothing, because "which name did it guess?" is the question an operator asks about an unknown-child outcome.
depth int

The depth the child would have run at, whether or not it ran.
message str

Exactly what the tool returns to the model.

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 
41
42
43
44
45
46
47
48
49
50
51
52
53
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 
67
68
69
70
71
72
73
74
75
76
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 
78
79
80
81
82
83
84
85
86
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 
63
64
65
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 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
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)

DelegationToolSpec dataclass 

DelegationToolSpec(*, name: str, description: str, coroutine: Callable[..., Any], parameters: Sequence[str] = ())

A tool this capability offers, described rather than constructed.

The shipped code built framework tool objects here, which put a third-party tool library on the import path of a capability that has no other use for one. A spec carries the same four facts — name, description, parameter names, coroutine — and the composition root wraps it in whatever tool type the runtime actually uses.

Attributes:

Name Type Description
parameters Sequence[str]

The coroutine's argument names, in order — the declared signature a binder holds the coroutine to. It is read: the composition root's binder compares it against the coroutine it was handed and refuses the pair when the two disagree, so a spec cannot describe a tool the runtime would not bind. Leave it empty to declare nothing and be checked against nothing.

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 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
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 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
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 
92
93
94
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 
88
89
90
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 
82
83
84
def specs(self) -> tuple[DelegationToolSpec, ...]:
    """The tools to bind into the parent's palette."""
    return self._specs

DepthPolicy dataclass 

DepthPolicy(max_depth: int)

The delegation ceiling, and the arithmetic around it.

Attributes:

Name Type Description
max_depth int

The deepest a child may run at. 0 disables delegation entirely — a supported configuration, not a misconfiguration: an operator switching delegation off for a tenant should not have to unwire the roster to do it.

admits 

admits(parent_depth: Any) -> bool

True if a run at parent_depth may delegate one level down.

Source code in src/symfonic/capabilities/delegation/depth.py 
64
65
66
def admits(self, parent_depth: Any) -> bool:
    """``True`` if a run at ``parent_depth`` may delegate one level down."""
    return self.child_depth(parent_depth) <= self.max_depth

child_depth 

child_depth(parent_depth: Any) -> int

The depth a child of a run at parent_depth would run at.

Source code in src/symfonic/capabilities/delegation/depth.py 
60
61
62
def child_depth(self, parent_depth: Any) -> int:
    """The depth a child of a run at ``parent_depth`` would run at."""
    return coerce_depth(parent_depth) + 1

refusal 

refusal() -> str

The message a model gets when the ceiling stops it.

Prose, and returned rather than raised, because this is a conversational fact: the model asked for something the deployment does not allow, and it needs to read that and route around it. An exception here would end the parent's run over a decision the parent made.

Source code in src/symfonic/capabilities/delegation/depth.py 
68
69
70
71
72
73
74
75
76
77
78
79
def refusal(self) -> str:
    """The message a model gets when the ceiling stops it.

    Prose, and returned rather than raised, because this is a
    conversational fact: the model asked for something the deployment does
    not allow, and it needs to read that and route around it. An exception
    here would end the parent's run over a decision the parent made.
    """
    return (
        f"Delegation refused: maximum sub-agent depth ({self.max_depth}) "
        "reached; cannot delegate further."
    )

DuplicateChildError

Bases: ChildDeclarationError

Two children claim the same routing key.

Raised before any child is constructed. A roster that answers one name with two children has no defensible resolution order, and picking one silently would route a task to a child the operator never chose.

PrebuiltChild dataclass 

PrebuiltChild(*, name: str, description: str, when_to_use: str | None = None, agent: Any)

Bases: ChildDeclaration

A child the caller constructed and hands over.

The escape hatch, and deliberately the unsanitisable one: the child is already built, its palette is already frozen, and there is nothing left to clear. Registration can therefore only accept or refuse it — see :func:~.lockdown.assert_no_write_surface.

Attributes:

Name Type Description
agent Any

Anything exposing async run(query, ...). Structural rather than nominal: a test double, a remote proxy, and an adapter around somebody else's agent are all legitimate children, and none of them can be expected to inherit from a class this package cannot import.

owned property 

owned: bool

Never. The caller built this child and controls its lifetime.

RosterEntry dataclass 

RosterEntry(*, name: str, description: str, runner: Any, when_to_use: str | None = None, owned: bool = False)

One registered child: how to reach it, and who owns it.

The declaration form is deliberately not carried here. By the time an entry exists the compiler has already answered every question the two forms differed on, and keeping the spec around would invite a later stage to re-decide inheritance behind the compiler's back.

listing 

listing() -> str

The one line the parent's model reads for this child.

Source code in src/symfonic/capabilities/delegation/values.py 
41
42
43
44
def listing(self) -> str:
    """The one line the parent's model reads for this child."""
    extra = f" ({self.when_to_use})" if self.when_to_use else ""
    return f"  - {self.name}: {self.description}{extra}"

RunSnapshots

Bases: Protocol

Opens and closes a run-local prompt-block snapshot slot.

Two verbs, both synchronous, because opening a slot is a context-variable write and the delegation path must be able to do it inside a finally.

A deployment that renders no prompt blocks supplies nothing at all and the delegation context skips the slot entirely — see :class:~.context.DelegationContext.

ScopedChild dataclass 

ScopedChild(*, name: str, description: str, when_to_use: str | None = None, build: Callable[[Any], Any])

Bases: ChildDeclaration

A child the composer builds for the scope it is serving.

The declaration that closes a leak PrebuiltChild cannot. A prebuilt child is a finished object, so a composition root has two places to build one and the convenient place is wrong: construct the children once beside the provider, close over them in compose(scope, shared), and every tenant shares them. Give such a child memory and it answers one tenant's delegation out of another tenant's recollections -- measured, not feared.

build is called once per scope, at composition time, with the scope the parent is being composed for. Deliberately not a per-call rebinding: a finished agent cannot be safely re-pointed at another tenant halfway through a turn, and a contract that pretended otherwise would be the same leak with more machinery.

Attributes:

Name Type Description
build Callable[[Any], Any]

scope -> agent. Whatever it returns must satisfy the same contract PrebuiltChild.agent does -- the ChildRunner protocol, or an Agent.

owned property 

owned: bool

True: the composer built it, so the composer tears it down.

for_scope 

for_scope(scope: Any) -> Any

Build this child for scope, refusing an absent one.

Building against None would produce exactly the process-wide child this declaration exists to prevent, and it would look like it worked.

Source code in src/symfonic/capabilities/delegation/declarations.py 
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
def for_scope(self, scope: Any) -> Any:
    """Build this child for ``scope``, refusing an absent one.

    Building against ``None`` would produce exactly the process-wide
    child this declaration exists to prevent, and it would look like it
    worked.
    """
    if scope is None:
        raise ChildDeclarationError(
            f"ScopedChild {self.name!r} needs the scope its parent is "
            "composed for: pass scope= to delegated_children(). Building "
            "it without one produces a child shared by every tenant, "
            "which is the leak this declaration exists to close."
        )
    return self.build(scope)

TeardownReport dataclass 

TeardownReport(*, verb: str, attempted: int = 0, failures: tuple[str, ...] = ())

What a flush or close managed, and what it did not.

clean is derived from :attr:failures rather than set, so there is no field an unlucky caller can pass to make a teardown that lost a pool look successful.

merge 

merge(other: TeardownReport) -> TeardownReport

Combine two phases of one shutdown into a single report.

Source code in src/symfonic/capabilities/delegation/values.py 
155
156
157
158
159
160
161
def merge(self, other: TeardownReport) -> TeardownReport:
    """Combine two phases of one shutdown into a single report."""
    return TeardownReport(
        verb=f"{self.verb}+{other.verb}",
        attempted=self.attempted + other.attempted,
        failures=self.failures + other.failures,
    )

UnknownChildError

Bases: DelegationError, KeyError

Nothing is registered under that routing key.

Raised by the roster, which has no opinion about how a caller should react. The tool surface catches it and answers the model with the roster instead, because a model that guessed a name needs the list, not a stack trace.

UnsanitisableChildConfigError

Bases: ChildLockdownError, TypeError

A config asked for the write surface and cannot be copied to clear it.

Also a :class:TypeError so the shipped behaviour of the engine's deny_child_self_edit — which raised exactly that — survives the move. A lock that gives up quietly on an input shape it did not expect is not a lock, so this is a refusal rather than a pass-through.

assert_no_write_surface 

assert_no_write_surface(name: str, agent: Any) -> Any

Refuse pre-built child agent if it can write a prompt block.

Returns the child, so a caller can register in one expression and a reader can see that registration passed through the lock rather than around it.

Raises:

Type Description
ChildLockdownError

If the child carries a write tool, or a config requesting one.
Source code in src/symfonic/capabilities/delegation/lockdown.py 
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
def assert_no_write_surface(name: str, agent: Any) -> Any:
    """Refuse pre-built child ``agent`` if it can write a prompt block.

    Returns the child, so a caller can register in one expression and a reader
    can see that registration passed through the lock rather than around it.

    Raises:
        ChildLockdownError: If the child carries a write tool, or a config
            requesting one.
    """
    tools = registered_write_surface_tools(agent)
    if tools:
        raise ChildLockdownError(
            f"sub-agent {name!r} was built with the block-edit tool(s) "
            f"{list(tools)!r} registered. A delegated child holds no block-edit "
            "tool on any construction path: a prompt block has one writer, and "
            "a child that can write one is a second writer the parent never "
            "sees. Remove the tool from the child, or declare the child as a "
            "ChildSpec -- specs are built by the parent, which clears the write "
            "surface for you."
        )
    if wants_block_self_edit(getattr(agent, "_config", None)):
        raise ChildLockdownError(
            f"sub-agent {name!r} was built with a config setting "
            f"{SELF_EDIT_FIELD}=True. A delegated child never receives the "
            "block self-edit surface, and a pre-built child cannot be sanitised "
            "after the fact -- its tool registry is already built. Set "
            f"{SELF_EDIT_FIELD}=False on the child's config, or declare the "
            "child as a ChildSpec so the parent builds it with the flag cleared."
        )
    return agent

child_roster 

child_roster(children: Sequence[Any], *, scope: Any = None, context: Any = None) -> DelegationRoster

The roster children form, refusing a duplicate name.

A duplicate is a tool the model cannot address unambiguously. Refused here rather than resolved by last-wins, because last-wins leaves the deployment's list and the model's palette disagreeing with nothing to say so.

Source code in src/symfonic/capabilities/delegation/composition.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
def child_roster(
    children: Sequence[Any], *, scope: Any = None, context: Any = None
) -> DelegationRoster:
    """The roster ``children`` form, refusing a duplicate name.

    A duplicate is a tool the model cannot address unambiguously. Refused here
    rather than resolved by last-wins, because last-wins leaves the
    deployment's list and the model's palette disagreeing with nothing to say
    so.
    """
    seen: set[str] = set()
    entries: list[RosterEntry] = []
    for child in children:
        if child.name in seen:
            raise ValueError(
                f"two children are named {child.name!r}. The model addresses a "
                "child by name, so a duplicate is a tool it cannot call "
                "unambiguously."
            )
        seen.add(child.name)
        # A ScopedChild is built here, once, for the scope the parent is being
        # composed for -- never rebound per call. A PrebuiltChild is taken as
        # it comes, which is what it is for.
        agent = child.for_scope(scope) if isinstance(child, ScopedChild) else child.agent
        entries.append(
            RosterEntry(
                name=child.name,
                description=child.description,
                # Adapted, not passed through: an ``Agent`` speaks
                # ``run(prompt, ...)`` and the roster calls
                # ``run(query=..., ...)``. Handing the child over raw made
                # every hand-off raise, and ``DelegationTools`` turns a
                # child's failure into a tool message by design -- so the
                # parent's turn succeeded carrying the TypeError where the
                # answer belonged.
                runner=adapt_child(child.name, agent, context),
                when_to_use=child.when_to_use,
            )
        )
    return DelegationRoster(tuple(entries))

coerce_depth 

coerce_depth(value: Any) -> int

Read value as a run depth: a non-negative int, always.

Anything unparseable is zero. Failing the other way — raising, or keeping the raw value — turns an adopter's typo in a state override into either a crashed run or a comparison whose result nobody can predict.

Source code in src/symfonic/capabilities/delegation/depth.py 
28
29
30
31
32
33
34
35
36
37
38
def coerce_depth(value: Any) -> int:
    """Read ``value`` as a run depth: a non-negative ``int``, always.

    Anything unparseable is zero. Failing the other way — raising, or keeping
    the raw value — turns an adopter's typo in a state override into either a
    crashed run or a comparison whose result nobody can predict.
    """
    try:
        return max(0, int(value))
    except (TypeError, ValueError):
        return 0

delegated_children 

delegated_children(children: Sequence[Any], *, max_depth: int = DEFAULT_MAX_DEPTH, scope: Any = None) -> DelegationCapability

One capability that offers children as tools the model may call.

Raises:

Type Description
ValueError

if children is empty, or two share a name.

An empty roster is refused because the capability would contribute a tool that can call nobody: the model is told it may delegate, every attempt fails, and it reads as a runtime fault rather than as a deployment with no children.

Source code in src/symfonic/capabilities/delegation/composition.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
def delegated_children(
    children: Sequence[Any],
    *,
    max_depth: int = DEFAULT_MAX_DEPTH,
    scope: Any = None,
) -> DelegationCapability:
    """One capability that offers ``children`` as tools the model may call.

    Raises:
        ValueError: if ``children`` is empty, or two share a name.

    An empty roster is refused because the capability would contribute a tool
    that can call nobody: the model is told it may delegate, every attempt
    fails, and it reads as a runtime fault rather than as a deployment with no
    children.
    """
    if not children:
        raise ValueError(
            "delegation with no children contributes a tool that can call "
            "nobody. Compose no delegation capability instead -- an agent "
            "without helpers is the ordinary case, not a degraded one."
        )
    context = DelegationContext()
    roster = child_roster(children, scope=scope, context=context)
    depth = DepthPolicy(max_depth=max_depth)
    return DelegationCapability(
        roster=roster,
        tools=DelegationTools(roster=roster, depth=depth, context=context),
        # The children were built by the caller, so the caller closes them.
        # Handing them to the lifecycle here would make this function's return
        # value own objects its caller still holds a reference to, and both
        # would close them.
        lifecycle=ChildLifecycle(()),
        context=context,
        depth=depth,
    )

deny_child_self_edit 

deny_child_self_edit(config: Any) -> Any

Return config with the write-surface request cleared.

The common path allocates nothing and preserves object identity: a config that never asked is returned unchanged, which matters for callers that compare configs by identity.

Raises:

Type Description
UnsanitisableChildConfigError

If the flag is set and the object exposes no model_copy to clear it through. Also a TypeError, matching what the shipped engine raised.
Source code in src/symfonic/capabilities/delegation/lockdown.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
def deny_child_self_edit(config: Any) -> Any:
    """Return ``config`` with the write-surface request cleared.

    The common path allocates nothing and preserves object identity: a config
    that never asked is returned unchanged, which matters for callers that
    compare configs by identity.

    Raises:
        UnsanitisableChildConfigError: If the flag is set and the object
            exposes no ``model_copy`` to clear it through. Also a
            ``TypeError``, matching what the shipped engine raised.
    """
    if not wants_block_self_edit(config):
        return config
    copier = getattr(config, "model_copy", None)
    if not callable(copier):
        raise UnsanitisableChildConfigError(
            f"a delegated child was given a config with {SELF_EDIT_FIELD}=True "
            f"of type {type(config).__name__!r}, which exposes no model_copy to "
            "clear it through. A child never holds a block-edit tool, and a "
            "config that cannot be sanitised cannot be used for one -- pass a "
            "framework config, or set the flag False yourself."
        )
    return copier(update={SELF_EDIT_FIELD: False})

is_write_surface_tool_name 

is_write_surface_tool_name(name: Any) -> bool

True if a tool called name may write a prompt block.

Matched case-insensitively against a stripped name. Nothing on the path from a registered tool to this predicate normalises either, so a guard that a shift key defeats would be no guard: MEMORY_BLOCK_APPEND reaches a child exactly as easily as the lowercase spelling.

This is a name heuristic and says so. It cannot detect a write tool given a name outside the vocabulary entirely (grant_edit). No name check can — the reserved namespace exists so that the framework's own tools are always inside it, and the verb family covers the near-misses an adopter wrapping the write API would plausibly reach for.

Source code in src/symfonic/capabilities/delegation/lockdown.py 
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
def is_write_surface_tool_name(name: Any) -> bool:
    """``True`` if a tool called ``name`` may write a prompt block.

    Matched case-insensitively against a stripped name. Nothing on the path
    from a registered tool to this predicate normalises either, so a guard that
    a shift key defeats would be no guard: ``MEMORY_BLOCK_APPEND`` reaches a
    child exactly as easily as the lowercase spelling.

    This is a name heuristic and says so. It cannot detect a write tool given a
    name outside the vocabulary entirely (``grant_edit``). No name check can —
    the reserved namespace exists so that the framework's own tools are always
    inside it, and the verb family covers the near-misses an adopter wrapping
    the write API would plausibly reach for.
    """
    if not isinstance(name, str):
        return False
    return bool(_WRITE_SURFACE_PATTERN.match(name.strip()))

registered_write_surface_tools 

registered_write_surface_tools(agent: Any) -> tuple[str, ...]

Names of prompt-block write tools registered on agent.

The authoritative check, because it reads what the model will actually be offered. A tool handed straight to a child's constructor never passed through :data:SELF_EDIT_FIELD, so a config-only guard waves it through.

Source code in src/symfonic/capabilities/delegation/lockdown.py 
183
184
185
186
187
188
189
190
191
192
193
def registered_write_surface_tools(agent: Any) -> tuple[str, ...]:
    """Names of prompt-block write tools registered on ``agent``.

    The authoritative check, because it reads what the model will actually be
    offered. A tool handed straight to a child's constructor never passed
    through :data:`SELF_EDIT_FIELD`, so a config-only guard waves it through.
    """
    direct = _direct_write_surface_tools(agent)
    if direct is not None:
        return direct
    return _delegated_write_surface_tools(agent)

wants_block_self_edit 

wants_block_self_edit(config: Any) -> bool

True if config explicitly requests the block write surface.

Compared with is True rather than coerced. A duck-typed child that synthesises attributes on access — a mock, a lazy or remote proxy — answers any attribute read with a truthy object, and coercion reported such a child as requesting a surface it had never heard of.

Source code in src/symfonic/capabilities/delegation/lockdown.py 
91
92
93
94
95
96
97
98
99
def wants_block_self_edit(config: Any) -> bool:
    """``True`` if ``config`` explicitly requests the block write surface.

    Compared with ``is True`` rather than coerced. A duck-typed child that
    synthesises attributes on access — a mock, a lazy or remote proxy — answers
    any attribute read with a truthy object, and coercion reported such a child
    as requesting a surface it had never heard of.
    """
    return getattr(config, SELF_EDIT_FIELD, False) is True