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 src/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 src/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 src/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()

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

        async for event in self._workflow.astream(input_state, config=config):
            if deadline is not None and asyncio.get_running_loop().time() > deadline:
                raise TimeoutError(
                    f"stream() exceeded timeout of {self._timeout_seconds}s"
                )
            yield event
        exhausted = True
    finally:
        duration_ms = (time.monotonic() - start) * 1000
        end_event = AgentEndEvent(
            run_id=run_id,
            final_response=None,
            duration_ms=duration_ms,
            node_count=None,
        )
        if exhausted:
            # Normal completion -- safe to await directly.
            await mgr.on_agent_end(end_event)
        else:
            # Early break (GeneratorExit) or CancelledError --
            # cannot reliably await inside an async generator
            # finally block, so schedule as a background task.
            # Keep a strong reference to prevent GC before completion.
            task = asyncio.get_running_loop().create_task(
                mgr.on_agent_end(end_event)
            )
            _background_tasks.add(task)
            task.add_done_callback(_background_tasks.discard)

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 src/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

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

    transpiler = EventTranspiler()
    exhausted = False
    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
        )
        async for timestamped in transpiler.transpile(raw_stream):
            if deadline is not None and asyncio.get_running_loop().time() > deadline:
                raise TimeoutError(
                    f"stream_events() exceeded timeout of {self._timeout_seconds}s"
                )
            yield timestamped.event
        exhausted = True
    finally:
        duration_ms = (time.monotonic() - start) * 1000
        end_event = AgentEndEvent(
            run_id=run_id,
            final_response=None,
            duration_ms=duration_ms,
            node_count=None,
        )
        if exhausted:
            await mgr.on_agent_end(end_event)
        else:
            task = asyncio.get_running_loop().create_task(
                mgr.on_agent_end(end_event)
            )
            _background_tasks.add(task)
            task.add_done_callback(_background_tasks.discard)