symfonic.core.runtime

runtime

AgentRuntime -- graph execution for a single agent invocation.

Per ADR-PFX-008 and TDD SS2.9: Eager compile in init, run() for execution, stream() for streaming. Callbacks dispatched via CallbackManager.

AgentRuntime

AgentRuntime(graph: AgentGraph, deps: BaseAgentDeps, config: AgentConfig | None = None, timeout_seconds: float | None = None, checkpointer: Any | None = None)

Execute a compiled graph for a single agent invocation.

Eagerly compiles the graph in init -- raises immediately on misconfiguration (missing capabilities, invalid preset, etc.).

Source code in symfonic/core/runtime.py

def __init__(
    self,
    graph: AgentGraph,
    deps: BaseAgentDeps,
    config: AgentConfig | None = None,
    timeout_seconds: float | None = None,
    checkpointer: Any | None = None,
) -> None:
    self._deps = deps
    self._config = config or AgentConfig()
    self._recursion_limit = self._config.recursion_limit
    self._timeout_seconds = timeout_seconds
    self._checkpointer = checkpointer
    self._workflow = graph.compile(deps, self._config, checkpointer=checkpointer)

run async

run(query: str, *, callbacks: Sequence[CallbackHandler] | None = None, **state_overrides: Any) -> dict[str, Any]

Execute the graph and return final state dict with final_response.

v7.18.0 (Option B emission-ordering): the caller may pass _suppress_agent_end=True in state_overrides to disable the success-path on_agent_end dispatch in this method. The engine layer (SymfonicAgent.run) sets this flag so it can emit on_response_render AFTER the full transformation chain (synthesize-on-empty-final salvage -> fabrication -> metacog -> extraction-stripping) and THEN emit on_agent_end with the rewritten content. Composability rule (locked):

• salvage -> fabrication -> metacog -> extraction-stripping -> on_response_render -> on_agent_end.
• The rewriter runs at most ONCE per turn.

The error-path on_agent_end STILL fires from runtime -- engine cannot intercept the BaseException, and adopters relying on a forced-failure terminator event must still receive it. _suppress_agent_end is a runtime-internal flag and is NOT part of the public state_overrides surface; adopter code that passes arbitrary keys to runtime.run() ignores it per the legacy default-False contract.

Source code in symfonic/core/runtime.py

async def run(
    self,
    query: str,
    *,
    callbacks: Sequence[CallbackHandler] | None = None,
    **state_overrides: Any,
) -> dict[str, Any]:
    """Execute the graph and return final state dict with final_response.

    v7.18.0 (Option B emission-ordering): the caller may pass
    ``_suppress_agent_end=True`` in ``state_overrides`` to disable
    the success-path ``on_agent_end`` dispatch in this method.  The
    engine layer (``SymfonicAgent.run``) sets this flag so it can
    emit ``on_response_render`` AFTER the full transformation chain
    (synthesize-on-empty-final salvage -> fabrication -> metacog ->
    extraction-stripping) and THEN emit ``on_agent_end`` with the
    rewritten content.  Composability rule (locked):

    * salvage -> fabrication -> metacog -> extraction-stripping ->
      on_response_render -> on_agent_end.
    * The rewriter runs at most ONCE per turn.

    The error-path ``on_agent_end`` STILL fires from runtime --
    engine cannot intercept the BaseException, and adopters relying
    on a forced-failure terminator event must still receive it.
    ``_suppress_agent_end`` is a runtime-internal flag and is NOT
    part of the public ``state_overrides`` surface; adopter code
    that passes arbitrary keys to ``runtime.run()`` ignores it
    per the legacy default-False contract.
    """
    suppress_agent_end = bool(
        state_overrides.pop("_suppress_agent_end", False),
    )
    # v8.6.4 Ask 2(b): recursion-exhaustion salvage opt-in.  The engine
    # sets this flag when ``synthesize_on_empty_final`` is True so the
    # runtime can retain the in-flight state and recover gathered
    # ToolMessages when LangGraph hits ``recursion_limit`` -- instead of
    # propagating GraphRecursionError (which bypassed the engine's
    # normal-return salvage).  Popped here so it never reaches the graph
    # input state.  Default False -> the legacy ``ainvoke`` path runs
    # unchanged (byte-identical).
    recursion_salvage = bool(
        state_overrides.pop("_recursion_salvage", False),
    )
    input_state = self._build_input_state(query, **state_overrides)
    config = self._build_runnable_config(state_overrides)
    run_id = input_state["run_id"]
    mgr = self._resolve_callback_manager(callbacks)

    # Propagate merged manager to nodes via state so per-invocation
    # callbacks reach @observed_node (Finding 1).
    input_state["_callback_manager"] = mgr

    await mgr.on_agent_start(AgentStartEvent(query=query, run_id=run_id))
    start = time.monotonic()

    try:
        if recursion_salvage:
            # Drive via values-streaming so the last accumulated state
            # (including gathered ToolMessages) is retained when
            # ``recursion_limit`` is hit.  Equivalent final result to
            # ``ainvoke`` on the happy path -- values-mode yields the
            # full state after each super-step.
            result = await self._run_with_recursion_salvage(
                input_state, config,
            )
        else:
            coro = self._workflow.ainvoke(input_state, config=config)
            if self._timeout_seconds is not None:
                result = await asyncio.wait_for(
                    coro, timeout=self._timeout_seconds,
                )
            else:
                result = await coro
        duration_ms = (time.monotonic() - start) * 1000
        final_response = result.get("final_response")
        node_log = result.get("node_execution_log", [])

        if not suppress_agent_end:
            await mgr.on_agent_end(
                AgentEndEvent(
                    run_id=run_id,
                    final_response=final_response,
                    duration_ms=duration_ms,
                    node_count=len(node_log),
                )
            )

        return {
            "final_response": final_response,
            "messages": result.get("messages", []),
            "node_execution_log": node_log,
            # v7.18.0: expose runtime-leg latency + node_count so
            # the engine caller (which suppresses the runtime
            # ``on_agent_end``) can stamp them onto the
            # AgentEndEvent it emits post-transformation-chain.
            "_runtime_duration_ms": duration_ms,
            "_runtime_node_count": len(node_log),
            # v8.6.4 Ask 2(b): propagate the recursion-exhaustion
            # marker so the engine runs its synthesis salvage over the
            # gathered ToolMessages and emits the recursion_exhausted
            # ReactLoopEndEvent.  Absent (falsy) on the normal path.
            "_recursion_exhausted": result.get(
                "_recursion_exhausted", False,
            ),
        }
    except BaseException:
        duration_ms = (time.monotonic() - start) * 1000
        # Error-path ``on_agent_end`` fires unconditionally.  Even
        # when the caller asked to suppress success-path emission,
        # a forced-failure terminator must reach observability
        # consumers.  Engine cannot intercept BaseException so this
        # is the only reliable dispatch site for failure events.
        await asyncio.shield(
            mgr.on_agent_end(
                AgentEndEvent(
                    run_id=run_id,
                    final_response=None,
                    duration_ms=duration_ms,
                    node_count=None,
                )
            )
        )
        raise

stream async

stream(query: str, *, callbacks: Sequence[CallbackHandler] | None = None, **state_overrides: Any) -> AsyncGenerator[Any, None]

Yields stream events. Emits AgentStartEvent/AgentEndEvent via callbacks.

Source code in symfonic/core/runtime.py

async def stream(
    self,
    query: str,
    *,
    callbacks: Sequence[CallbackHandler] | None = None,
    **state_overrides: Any,
) -> AsyncGenerator[Any, None]:
    """Yields stream events. Emits AgentStartEvent/AgentEndEvent via callbacks."""
    input_state = self._build_input_state(query, **state_overrides)
    config = self._build_runnable_config(state_overrides)
    run_id = input_state["run_id"]
    mgr = self._resolve_callback_manager(callbacks)

    # Propagate merged manager to nodes via state (Finding 1).
    input_state["_callback_manager"] = mgr

    await mgr.on_agent_start(AgentStartEvent(query=query, run_id=run_id))
    start = time.monotonic()

    end_dispatched = False
    try:
        if self._timeout_seconds is not None:
            deadline = asyncio.get_running_loop().time() + self._timeout_seconds
        else:
            deadline = None

        stream = self._workflow.astream(input_state, config=config)
        async with _closing_if_supported(stream):
            async for event in stream:
                if (
                    deadline is not None
                    and asyncio.get_running_loop().time() > deadline
                ):
                    raise TimeoutError(
                        f"stream() exceeded timeout of {self._timeout_seconds}s"
                    )
                try:
                    yield event
                except GeneratorExit:
                    # ``async for ...: break`` does not explicitly await
                    # ``aclose()``. Dispatch the callback as soon as the
                    # loop finalizer reaches this yield while the enclosing
                    # context continues closing the workflow iterator.
                    end_event = AgentEndEvent(
                        run_id=run_id,
                        final_response=None,
                        duration_ms=(time.monotonic() - start) * 1000,
                        node_count=None,
                    )
                    task = asyncio.get_running_loop().create_task(
                        mgr.on_agent_end(end_event)
                    )
                    _background_tasks.add(task)
                    task.add_done_callback(_background_tasks.discard)
                    end_dispatched = True
                    raise
    finally:
        if not end_dispatched:
            duration_ms = (time.monotonic() - start) * 1000
            end_event = AgentEndEvent(
                run_id=run_id,
                final_response=None,
                duration_ms=duration_ms,
                node_count=None,
            )
            await mgr.on_agent_end(end_event)

stream_events async

stream_events(query: str, *, callbacks: Sequence[CallbackHandler] | None = None, **state_overrides: Any) -> AsyncGenerator[StreamEvent, None]

Yield typed StreamEvent objects from astream_events(v2).

Calls _workflow.astream_events and pipes each raw LangGraph event through EventTranspiler, yielding typed StreamEvent objects (TextDeltaEvent, ToolCallStartEvent, ThinkingDeltaEvent, etc.). AgentStartEvent / AgentEndEvent are dispatched via the callback manager as in stream().

Source code in symfonic/core/runtime.py

async def stream_events(
    self,
    query: str,
    *,
    callbacks: Sequence[CallbackHandler] | None = None,
    **state_overrides: Any,
) -> AsyncGenerator[StreamEvent, None]:
    """Yield typed StreamEvent objects from astream_events(v2).

    Calls ``_workflow.astream_events`` and pipes each raw LangGraph
    event through ``EventTranspiler``, yielding typed StreamEvent
    objects (TextDeltaEvent, ToolCallStartEvent, ThinkingDeltaEvent,
    etc.).  AgentStartEvent / AgentEndEvent are dispatched via the
    callback manager as in ``stream()``.
    """
    # v8.7.1 C3: a resume MUST re-enter the graph via a LangGraph
    # ``Command(resume=...)`` so the paused ``interrupt()`` returns the
    # payload instead of re-firing. When a ``_resume_command`` is present
    # we feed the Command to the graph as its input (LangGraph resumes
    # from the checkpoint) rather than a fresh input-state dict, while
    # still refreshing the per-invocation callback manager on state.
    resume_command = state_overrides.pop("_resume_command", None)
    input_state = self._build_input_state(query, **state_overrides)
    config = self._build_runnable_config(state_overrides)
    run_id = input_state["run_id"]
    mgr = self._resolve_callback_manager(callbacks)

    input_state["_callback_manager"] = mgr

    if resume_command is not None:
        # A Command resumes FROM THE CHECKPOINT, so ``input_state`` --
        # and with it ``deps`` and the per-invocation callback manager
        # -- is discarded as graph input. Neither survives
        # checkpointing (both hold live objects), so every node
        # reading ``state["deps"]`` saw None on the resume path and
        # the observability decorator raised before the node body
        # ran. Re-inject them as a state update carried by the
        # Command itself.
        from langgraph.types import Command

        existing_update = getattr(resume_command, "update", None) or {}
        resume_command = Command(
            resume=resume_command.resume,
            update={
                **existing_update,
                "deps": self._deps,
                "_callback_manager": mgr,
            },
        )

    await mgr.on_agent_start(AgentStartEvent(query=query, run_id=run_id))
    start = time.monotonic()

    transpiler = EventTranspiler()
    # TA8.29 -- four outcomes, never collapsed. This body used to dispatch
    # the *normal* ``AgentEndEvent`` unconditionally from ``finally``, so a
    # consumer that broke out mid-stream and a consumer that read the turn
    # to its terminal were indistinguishable at the callback seam, and the
    # react node and LLM call the abandoned turn left open were never
    # closed at all. ``outcome`` records which of the four actually
    # happened and ``finish_turn`` closes the pairs the run still holds.
    outcome: TerminationOutcome = "completed"
    try:
        if self._timeout_seconds is not None:
            deadline = asyncio.get_running_loop().time() + self._timeout_seconds
        else:
            deadline = None

        graph_input = (
            resume_command if resume_command is not None else input_state
        )
        raw_stream = self._workflow.astream_events(
            graph_input, version="v2", config=config
        )
        typed_stream = transpiler.transpile(raw_stream)
        async with _closing_if_supported(typed_stream):
            async for timestamped in typed_stream:
                if (
                    deadline is not None
                    and asyncio.get_running_loop().time() > deadline
                ):
                    raise TimeoutError(
                        "stream_events() exceeded timeout of "
                        f"{self._timeout_seconds}s"
                    )
                yield timestamped.event
    except GeneratorExit:
        # The consumer stopped reading. Never swallowed -- re-raised below
        # so the generator finalises exactly once, the way the async
        # generator protocol requires.
        outcome = "abandoned"
        raise
    except asyncio.CancelledError:
        # A cancellation is not an error and must not be reported as one,
        # and it propagates untouched (CXL-3).
        outcome = "cancelled"
        raise
    except BaseException:
        # An ordinary provider/runtime error: the turn FAILED and the
        # exception reaches the consumer. Observability handler errors are
        # a different family entirely -- ``_dispatch`` logs and isolates
        # those, and they never convert this outcome.
        outcome = "failed"
        raise
    finally:
        await mgr.finish_turn(
            run_id=run_id,
            duration_ms=(time.monotonic() - start) * 1000,
            outcome=outcome,
        )