symfonic.capabilities.delegation.context

context

One run-local scope for depth, snapshot, and delegated metadata.

These three used to be three context variables and two nearly-identical decorators, opened and closed by hand at every entry point. The streaming entry points opened one of the three, which is how a streamed delegated child ended up re-resolving its prompt blocks every turn while the identical agent driven through the non-streaming path did not.

They are one object here because they have one lifetime — the run — and one trigger. depth > 0 is "this run is delegated"; it decides the snapshot, and the snapshot is the only reason a top-level run must not open a slot (with no slot open, block resolution is byte-for-byte what it was, so the feature cannot change what a parent sees).

Concurrency

Every piece of run state is a :class:~contextvars.ContextVar, so isolation between concurrent runs is inherited from the task that started them rather than maintained by this class. A parent delegating to four children concurrently gets four independent tallies, four independent depths, and four independent snapshot slots, and a cancelled sibling unwinds only its own — the finally runs in the task that opened it.

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

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.

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

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

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

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

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

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

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

@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)