symfonic.agent.cutover.delegate

delegate

The delegate: one legacy call, served by the compiler and the kernel.

This module is deliberately thin, and thinness is the whole point. It holds no loop, no round budget, no event numbering and no result assembly — those live once, in symfonic.kernel, and reaching them through :class:~symfonic.agent.backend.AgentPlanFactory is what keeps the legacy facade from becoming IPL-2's forbidden second pipeline. What is here is translation: the legacy engine's construction inputs into a compile request, the caller's arguments into a :class:~symfonic.kernel.contracts.TurnRequest, and — through :mod:~symfonic.agent.cutover.projection — the kernel's result and event stream back into AgentResponse and StreamChunk, which are the shapes legacy callers already hold.

The outward projection is next door rather than here because the two directions are separable and only one of them touches the kernel: this module compiles and invokes, projection only reads a finished value. They were one module until TA8.10 gave this one an inbound conversion to carry as well. Two more neighbours followed for the same line-budget reason and are named where they are used: rounds, authorised and delegation.

KernelDelegate

KernelDelegate(*, model_provider: Any, instructions: str | None = None, tools: Sequence[Any] = (), recursion_limit: int | None = None, model: Any = None, role_models: Any = None, bundle: Any = None, max_conversation_messages: int | None = None, observability: ObservabilitySuite | None = None)

Serves one legacy run/stream call through the migrated path.

Bind one plan factory for the life of this agent.

bundle is the composition root's authorised :class:~.bundle.RetrievalBundle, transported unchanged. The delegate does not fold it and does not inspect what it contains: folding is the host's decision about what it trusts, and a delegate that composed one would be deciding that on the host's behalf. None is the ordinary case — an agent that does not hydrate compiles the same plan it always did, with no stages.

model is config.agent.model. Threading it is what lets ALLOWED_AGENT_FIELDS name the field: an allowlist entry asserts the migrated path honours a field, so admitting model without passing it here would turn a refusal into a silent substitution — the agent would be admitted to the kernel and then answered by whichever model the provider happened to declare.

max_conversation_messages is config.agent.max_conversation_messages — the cap replayed history is trimmed at. On the constructor rather than the call because it is agent configuration, not a property of the turn, the same reason recursion_limit is here. None means "not stated" and falls back to the stock value, which is what an admitted turn always carries anyway: the envelope refuses the field. Threading it is nonetheless what makes the trim honour the configuration rather than a constant, so admitting the field later is a decision about evidence instead of a change of behaviour.

bundle.tools are merged into this agent's own tools rather than kept beside them (TA8.12) -- one normalized set, because a second collection only some of the four tool readers consult is how a contributed tool becomes bindable and not callable. bundle.delegation is held for the other half: :meth:_delegation_scope opens a run scope on it, which is how the turn's agent_depth reaches the ceiling those tools enforce. See :mod:~symfonic.agent.cutover.delegation.

observability is the agent-lifetime half of TA8.20's migration: the config, the metrics_collector and the OTEL handles the engine holds, folded into one object that can compose a per-run event sink. None means "nobody is watching", which is both the default and the state every agent was in before the migration, and it binds no sink at all. Like bundle, it is transported rather than derived — deciding what observability a deployment buys is the host's decision that symfonic.services.observability.suite already owns.

Source code in src/symfonic/agent/cutover/delegate.py

def __init__(
    self,
    *,
    model_provider: Any,
    instructions: str | None = None,
    tools: Sequence[Any] = (),
    recursion_limit: int | None = None,
    model: Any = None,
    role_models: Any = None,
    bundle: Any = None,
    max_conversation_messages: int | None = None,
    observability: ObservabilitySuite | None = None,
) -> None:
    """Bind one plan factory for the life of this agent.

    ``bundle`` is the composition root's authorised
    :class:`~.bundle.RetrievalBundle`, transported unchanged. The delegate
    does not fold it and does not inspect what it contains: folding is the
    host's decision about what it trusts, and a delegate that composed one
    would be deciding that on the host's behalf. ``None`` is the ordinary
    case — an agent that does not hydrate compiles the same plan it always
    did, with no stages.

    ``model`` is ``config.agent.model``. Threading it is what lets
    ``ALLOWED_AGENT_FIELDS`` name the field: an allowlist entry asserts the
    migrated path *honours* a field, so admitting ``model`` without passing
    it here would turn a refusal into a silent substitution — the agent
    would be admitted to the kernel and then answered by whichever model the
    provider happened to declare.

    ``max_conversation_messages`` is ``config.agent.max_conversation_messages``
    — the cap replayed history is trimmed at. On the constructor rather than
    the call because it is agent configuration, not a property of the turn,
    the same reason ``recursion_limit`` is here. ``None`` means "not stated"
    and falls back to the stock value, which is what an admitted turn always
    carries anyway: the envelope refuses the field. Threading it is
    nonetheless what makes the trim honour the configuration rather than a
    constant, so admitting the field later is a decision about evidence
    instead of a change of behaviour.

    ``bundle.tools`` are merged into this agent's own tools rather than
    kept beside them (TA8.12) -- one normalized set, because a second
    collection only some of the four tool readers consult is how a
    contributed tool becomes bindable and not callable. ``bundle.delegation``
    is held for the other half: :meth:`_delegation_scope` opens a run scope
    on it, which is how the turn's ``agent_depth`` reaches the ceiling those
    tools enforce. See :mod:`~symfonic.agent.cutover.delegation`.

    ``observability`` is the agent-lifetime half of TA8.20's migration: the
    config, the ``metrics_collector`` and the OTEL handles the *engine*
    holds, folded into one object that can compose a per-run event sink.
    ``None`` means "nobody is watching", which is both the default and the
    state every agent was in before the migration, and it binds no sink at
    all. Like ``bundle``, it is transported rather than derived — deciding
    what observability a deployment buys is the host's decision that
    ``symfonic.services.observability.suite`` already owns.
    """
    self._observability = observability
    self._delegation = getattr(bundle, "delegation", None)
    self._history_cap = history_cap(max_conversation_messages)
    self._plans = AgentPlanFactory(
        model_provider=model_provider,
        instructions=instructions,
        tools=merge_capability_tools(tools, getattr(bundle, "tools", ())),
        max_model_rounds=rounds_for_recursion_limit(recursion_limit),
        model=model,
        # TA8.60. Not ``dict(role_models or {})``: that raises on a foreign
        # value, on only the doors that build a delegate, so they disagree.
        role_models=dict(role_models) if isinstance(role_models, Mapping) else {},
        stages=getattr(bundle, "stages", ()),
        stage_handlers=getattr(bundle, "stage_handlers", None) or {},
        authorized_effects=authorised_effects(bundle),
        capability_names=getattr(bundle, "capability_names", ()),
    )

kernel_typed_stream

kernel_typed_stream(plan: Any, request: TurnRequest) -> Any

Enter the kernel's typed projection for one compiled plan.

The one door this module's typed route goes through, and the reason it is here rather than in typed_route: IPL-1 declares the pipeline heads that may enter the invocation kernel, and this delegate is one of them. A second module reaching InvocationKernel directly would be a second entry point into the single invocation path, which is exactly what that rule exists to prevent.

Source code in src/symfonic/agent/cutover/delegate.py

def kernel_typed_stream(self, plan: Any, request: TurnRequest) -> Any:
    """Enter the kernel's typed projection for one compiled plan.

    The **one** door this module's typed route goes through, and the reason
    it is here rather than in ``typed_route``: IPL-1 declares the pipeline
    heads that may enter the invocation kernel, and this delegate is one of
    them. A second module reaching ``InvocationKernel`` directly would be a
    second entry point into the single invocation path, which is exactly
    what that rule exists to prevent.
    """
    return _KERNEL.stream_typed(plan, request)

run async

run(query: str, *, run_id: str, session_id: str = '', response_model: type[Any] | None = None, scope: Any = None, history: Sequence[Any] | None = None, attachments: Sequence[Any] | None = None, tenant_id: str | None = None, agent_depth: int | None = None) -> AgentResponse

One non-streaming turn, compiled once and run once.

scope rides on the request, never on the plan: the plan is compiled once for the agent's life and a scope burned into it would make one agent answer for one tenant. The bundle stays immutable configuration and nothing shared is mutated per call.

history and attachments ride there for the same reason — see :func:~.turn_request.turn_request.

agent_depth is the turn's delegation depth and :meth:_delegation_scope is what reads it. Per-call for the reason scope is: a depth fixed at construction would be a parent that could only ever be a root.

tenant_id is telemetry identity and nothing else: the legacy FrameworkTenantScope.tenant_id, from the same expression _legacy_run_impl hands to _otel_run_span, so an admitted run is attributed to the tenant the replaced path attributed it to. Separate from scope, which the engine builds through a translation answering None for a scope it cannot read — right for recall, wrong for billing.

Source code in src/symfonic/agent/cutover/delegate.py

async def run(
    self,
    query: str,
    *,
    run_id: str,
    session_id: str = "",
    response_model: type[Any] | None = None,
    scope: Any = None,
    history: Sequence[Any] | None = None,
    attachments: Sequence[Any] | None = None,
    tenant_id: str | None = None,
    agent_depth: int | None = None,
) -> AgentResponse:
    """One non-streaming turn, compiled once and run once.

    ``scope`` rides on the *request*, never on the plan: the plan is
    compiled once for the agent's life and a scope burned into it would
    make one agent answer for one tenant. The bundle stays immutable
    configuration and nothing shared is mutated per call.

    ``history`` and ``attachments`` ride there for the same reason — see
    :func:`~.turn_request.turn_request`.

    ``agent_depth`` is the turn's delegation depth and
    :meth:`_delegation_scope` is what reads it. Per-call for the reason
    ``scope`` is: a depth fixed at construction would be a parent that could
    only ever be a root.

    ``tenant_id`` is telemetry identity and nothing else: the *legacy*
    ``FrameworkTenantScope.tenant_id``, from the same expression
    ``_legacy_run_impl`` hands to ``_otel_run_span``, so an admitted run is
    attributed to the tenant the replaced path attributed it to. Separate
    from ``scope``, which the engine builds through a translation answering
    ``None`` for a scope it cannot read — right for recall, wrong for
    billing.
    """
    request = turn_request(
        query, scope, history, attachments, cap=self._history_cap, run_id=run_id
    )
    observed = for_turn(
        self._observability,
        run_id=run_id,
        session_id=session_id,
        tenant_id=tenant_id,
        entry_point="run",
        prompt=query,
        request=request,
    )
    plan = self._plans.compile(response_model, sink_factory=observed)
    try:
        # Around the *kernel* call, not the compile: the plan is compiled
        # once for the agent's life, so a depth burned into it would make
        # one agent answer at one depth forever.
        async with self._delegation_scope(agent_depth):
            result = await _KERNEL.run(plan, request)
    finally:
        # Idempotent and safe after a normal terminal, so it is unconditional
        # rather than reached only on the error path: the case it exists for
        # is the run that never reached a terminal at all, and that run does
        # not announce itself.
        if observed is not None:
            await observed.release()
    return as_response(result, run_id=run_id, session_id=session_id)

stream async

stream(query: str, *, run_id: str, session_id: str = '', response_model: type[Any] | None = None, scope: Any = None, history: Sequence[Any] | None = None, attachments: Sequence[Any] | None = None, tenant_id: str | None = None, agent_depth: int | None = None) -> AsyncIterator[StreamChunk]

The streaming projection of the same invocation, not a second one.

scope is threaded for the same reason it is on :meth:run, and its absence here was the sharper bug of the two: streaming admission already accepted a caller scope, so a multi-tenant stream was admitted to the kernel and then hydrated from the bundle's default scope. "The same invocation, projected" has to include what the invocation was for.

tenant_id is threaded for the reason given on :meth:run, and the release below is the half that matters more here: a consumer who stops iterating ends the run with no terminal event, which is precisely the abandoned run ObservabilityBridge.aclose exists for.

Source code in src/symfonic/agent/cutover/delegate.py

async def stream(
    self,
    query: str,
    *,
    run_id: str,
    session_id: str = "",
    response_model: type[Any] | None = None,
    scope: Any = None,
    history: Sequence[Any] | None = None,
    attachments: Sequence[Any] | None = None,
    tenant_id: str | None = None,
    agent_depth: int | None = None,
) -> AsyncIterator[StreamChunk]:
    """The streaming projection of the same invocation, not a second one.

    ``scope`` is threaded for the same reason it is on :meth:`run`, and its
    absence here was the sharper bug of the two: streaming admission already
    accepted a caller scope, so a multi-tenant stream was admitted to the
    kernel and then hydrated from the bundle's default scope. "The same
    invocation, projected" has to include what the invocation was *for*.

    ``tenant_id`` is threaded for the reason given on :meth:`run`, and the
    release below is the half that matters more here: a consumer who stops
    iterating ends the run with no terminal event, which is precisely the
    abandoned run ``ObservabilityBridge.aclose`` exists for.
    """
    request = turn_request(
        query, scope, history, attachments, cap=self._history_cap, run_id=run_id
    )
    observed = for_turn(
        self._observability,
        run_id=run_id,
        session_id=session_id,
        tenant_id=tenant_id,
        entry_point="stream",
        prompt=query,
        request=request,
    )
    plan = self._plans.compile(response_model, sink_factory=observed)
    # ``InvocationKernel`` is binding-agnostic and therefore exposes its
    # projected stream as kernel events. This delegate always compiles an
    # ``AgentPlanFactory`` response binding, whose ``build_event`` output
    # is the public ``AgentEvent`` consumed by ``_as_chunk`` below.
    stream = cast(
        AsyncIterator[AgentEvent],
        _KERNEL.stream(plan, request),
    )
    try:
        # Held open for the whole drain: a hand-off happens mid-stream, and
        # a scope closed after the first chunk would read depth ``0``.
        async with self._delegation_scope(agent_depth):
            async for event in stream:
                chunk = as_chunk(event, run_id=run_id)
                if chunk is not None:
                    yield chunk
    finally:
        await closing(stream)
        if observed is not None:
            await observed.release()

stream_typed

stream_typed(query: str, *, run_id: str, **options: Any) -> AsyncIterator[Any]

The typed projection of the same invocation -- ST2's kernel route.

The body, and the full keyword list options carries, live next door in :func:~.typed_route.typed_route, for the line-budget reason rounds, authorised and delegation already moved out. Named here because this is the address a caller holds.

Source code in src/symfonic/agent/cutover/delegate.py

def stream_typed(
    self, query: str, *, run_id: str, **options: Any
) -> AsyncIterator[Any]:
    """The typed projection of the same invocation -- ST2's kernel route.

    The body, and the full keyword list ``options`` carries, live next door
    in :func:`~.typed_route.typed_route`, for the line-budget reason
    ``rounds``, ``authorised`` and ``delegation`` already moved out. Named
    here because this is the address a caller holds.
    """
    from symfonic.agent.cutover.typed_route import typed_route

    return typed_route(self, query, run_id=run_id, **options)

rounds_for_recursion_limit

rounds_for_recursion_limit(recursion_limit: Any) -> int

Translate the legacy loop budget into the kernel's round budget.

The legacy body bounds the ReAct loop with LangGraph's recursion_limit, which counts graph super-steps; the kernel bounds it with PlanLimits.max_model_rounds, which counts model rounds. Two steps per round — one model node, one tool node — is the conversion, and the graph spends _ENTRY_STEPS before the first of them, so the entry cost comes off the budget before the division rather than after.

Both corrections point the same way: never more rounds than the plain react_loop graph. A conversion that only divided would give 50 -> 25 where that graph reaches 24, and the extra round is a billed model call the path being replaced would not have made. Integer division is the floor for the remainder, for the same reason.

The invariant is stated against react_loop and not "legacy" in general because the two constants above are read off that one wiring. presets adds a super-step per round for elicitation, for the experimental interrupt and for the precondition gate, so those graphs reach fewer rounds than (recursion_limit - 1) // 2 and this conversion would over-budget them. None of them can reach here: ask_user_enabled, experimental_interrupt and procedural_enforce_preconditions are all named refusals in :mod:~symfonic.agent.cutover.envelope, and a non-ReAct graph_preset is refused there too. If one of those switches flips, this conversion has to grow the extra step with it.

Leaving the constant in place instead would bound every migrated turn at ten rounds while the legacy body allowed twenty-four, which truncates a long tool loop and returns the empty answer as if it were the real one.

Source code in src/symfonic/agent/cutover/rounds.py

def rounds_for_recursion_limit(recursion_limit: Any) -> int:
    """Translate the legacy loop budget into the kernel's round budget.

    The legacy body bounds the ReAct loop with LangGraph's ``recursion_limit``,
    which counts graph *super-steps*; the kernel bounds it with
    ``PlanLimits.max_model_rounds``, which counts *model rounds*. Two steps per
    round — one model node, one tool node — is the conversion, and the graph
    spends ``_ENTRY_STEPS`` before the first of them, so the entry cost comes
    off the budget before the division rather than after.

    Both corrections point the same way: *never more rounds than the plain
    ``react_loop`` graph*. A conversion that only divided would give
    ``50 -> 25`` where that graph reaches 24, and the extra round is a billed
    model call the path being replaced would not have made. Integer division is
    the floor for the remainder, for the same reason.

    The invariant is stated against ``react_loop`` and not "legacy" in general
    because the two constants above are read off that one wiring. ``presets``
    adds a super-step per round for elicitation, for the experimental interrupt
    and for the precondition gate, so those graphs reach *fewer* rounds than
    ``(recursion_limit - 1) // 2`` and this conversion would over-budget them.
    None of them can reach here: ``ask_user_enabled``,
    ``experimental_interrupt`` and ``procedural_enforce_preconditions`` are all
    named refusals in :mod:`~symfonic.agent.cutover.envelope`, and a non-ReAct
    ``graph_preset`` is refused there too. If one of those switches flips, this
    conversion has to grow the extra step with it.

    Leaving the constant in place instead would bound every migrated turn at
    ten rounds while the legacy body allowed twenty-four, which truncates a
    long tool loop and returns the empty answer as if it were the real one.
    """
    if not isinstance(recursion_limit, int) or isinstance(recursion_limit, bool):
        return MAX_TOOL_ITERATIONS
    return max(1, (recursion_limit - _ENTRY_STEPS) // _STEPS_PER_ROUND)