symfonic.core.nodes.interrupt

interrupt

Generic InterruptNode — Roadmap Item 9 (v7.2-bound).

Parallel to elicitation.py: receives a _interrupt_pending state marker carrying a registered interrupt name + payload, calls interrupt(), and on resume injects the typed response back into the graph so the next react step can act on it.

Topology

The node is wired into the LangGraph workflow when FrameworkConfig.experimental_interrupt=True. It sits beside the elicitation node (which still handles the built-in ask_user interrupt unchanged) so the two pause paths share no code: a regression in the generic path cannot influence ask_user, and vice-versa.

Pending state shape

A tool/node triggers an interrupt by returning::

{"_interrupt_pending": {
    "name": "approval_required",       # registered interrupt name
    "interrupt_id": "i-<12hex>",       # uuid4.hex[:12]; client correlation
    "payload": {...},                  # JSON-serialised payload
    "tool_call_id": "...",            # optional; only set when called from a tool
}}

The :class:SymfonicAgent.interrupt helper builds this marker so callers do not assemble it by hand.

INTERRUPT_ACK_SENTINEL module-attribute

INTERRUPT_ACK_SENTINEL = '__symfonic_interrupt_ack__'

Carried by Command(resume=...) when the response dumps to {}.

langgraph treats an empty resume map as nothing to resume with, so the node would re-fire its interrupt() forever. Stripped here so the sentinel never reaches the model.

build_interrupt_marker

build_interrupt_marker(*, name: str, interrupt_id: str, payload: dict[str, Any], tool_call_id: str | None = None, run_id: str | None = None) -> dict[str, Any]

Construct the state-update dict a tool/node returns to trigger an interrupt.

Public helper used by :meth:SymfonicAgent.interrupt and any tool that wants to wire a generic interrupt by hand. Keeping this in one place avoids the marker shape drifting between callers.

Stamps minted_at unconditionally so :func:interrupt_pending_is_current can apply a TTL even when no run_id is available (the common case here -- see that function's docstring). run_id is optional: pass it when the caller has one in scope (e.g. a node with access to graph state); a tool body generally does not.

Source code in src/symfonic/core/nodes/interrupt.py

def build_interrupt_marker(
    *,
    name: str,
    interrupt_id: str,
    payload: dict[str, Any],
    tool_call_id: str | None = None,
    run_id: str | None = None,
) -> dict[str, Any]:
    """Construct the state-update dict a tool/node returns to trigger an interrupt.

    Public helper used by :meth:`SymfonicAgent.interrupt` and any tool
    that wants to wire a generic interrupt by hand. Keeping this in one
    place avoids the marker shape drifting between callers.

    Stamps ``minted_at`` unconditionally so
    :func:`interrupt_pending_is_current` can apply a TTL even when no
    ``run_id`` is available (the common case here -- see that
    function's docstring). ``run_id`` is optional: pass it when the
    caller has one in scope (e.g. a node with access to graph
    ``state``); a tool body generally does not.
    """
    return {
        "_interrupt_pending": {
            "name": name,
            "interrupt_id": interrupt_id,
            "payload": payload,
            "tool_call_id": tool_call_id,
            "run_id": run_id,
            "minted_at": time.time(),
        }
    }

create_interrupt_node

create_interrupt_node() -> Any

Return the generic interrupt node function.

The node reads state["_interrupt_pending"], calls interrupt(), and on resume injects the typed response back into the messages channel as a structured ToolMessage (when the interrupt was triggered from a tool) or as plain state (otherwise).

Source code in src/symfonic/core/nodes/interrupt.py

def create_interrupt_node() -> Any:
    """Return the generic interrupt node function.

    The node reads ``state["_interrupt_pending"]``, calls ``interrupt()``,
    and on resume injects the typed response back into the messages
    channel as a structured ``ToolMessage`` (when the interrupt was
    triggered from a tool) or as plain state (otherwise).
    """

    async def interrupt_node(state: dict[str, Any]) -> dict[str, Any]:
        from langchain_core.messages import ToolMessage
        from langgraph.types import interrupt

        pending: dict[str, Any] = state["_interrupt_pending"]
        name: str = pending["name"]
        interrupt_id: str = pending["interrupt_id"]
        payload: dict[str, Any] = pending.get("payload", {})
        tool_call_id: str | None = pending.get("tool_call_id") or None

        # Call interrupt() — LangGraph surfaces this as an
        # ExtensionEvent(type="on_interrupt") in astream_events(v2).
        # The engine's stream_typed() intercepts it, sees
        # ``interrupt_data["type"] == "interrupt"``, and routes to
        # ``_mint_interrupt_token`` (the generic minter) instead of
        # ``_mint_pause_token`` (the ask_user-specific one).
        resume_value: Any = interrupt(
            {
                "type": "interrupt",
                "name": name,
                "interrupt_id": interrupt_id,
                "payload": payload,
                "tool_call_id": tool_call_id,
            }
        )

        # --- Resume path ---------------------------------------------------
        # LangGraph re-runs the node body from the top with the resume
        # payload returned by ``interrupt()``. Clear the pending marker
        # and inject the response into messages so downstream react
        # turns see it as conversational context.
        if resume_value is not None:
            response_dict: dict[str, Any]
            if isinstance(resume_value, dict):
                # Drop the empty-resume sentinel the engine adds when a
                # field-less response model dumps to {} -- langgraph
                # cannot resume on an empty map, and the sentinel must
                # not leak into the ToolMessage the model reads.
                response_dict = {
                    k: v
                    for k, v in resume_value.items()
                    if k != INTERRUPT_ACK_SENTINEL
                }
            elif hasattr(resume_value, "model_dump"):
                response_dict = resume_value.model_dump()
            else:
                response_dict = {"value": str(resume_value)}

            updates: dict[str, Any] = {"_interrupt_pending": None}
            if tool_call_id:
                # Tool-call-originated interrupt: emit a ToolMessage so
                # the downstream react node sees a normal tool result
                # for the originating call. The content is the
                # JSON-serialised response.
                import json as _json

                updates["messages"] = [
                    ToolMessage(
                        content=_json.dumps(response_dict, default=str),
                        tool_call_id=tool_call_id,
                        name=f"interrupt:{name}",
                    )
                ]
            else:
                # Non-tool interrupt: stash on state for the caller
                # to read via ``state.get("_interrupt_response")``.
                updates["_interrupt_response"] = {
                    "name": name,
                    "interrupt_id": interrupt_id,
                    "response": response_dict,
                }
            return updates

        # Should not reach here in normal operation; log and fail safely.
        logger.warning(
            "interrupt_node: interrupt() returned None unexpectedly "
            "for name=%s interrupt_id=%s",
            name,
            interrupt_id,
        )
        return {"_interrupt_pending": None}

    return interrupt_node

interrupt_condition

interrupt_condition(state: dict[str, Any], ttl_seconds: int | None = None) -> str

Conditional-edge router shared with the elicitation condition.

Returns:

Type	Description
str	"interrupt" when _interrupt_pending is current (see :func:interrupt_pending_is_current)
str	"elicitation" when _ask_user_pending is current (preserves the existing ask_user routing)
str	"continue" when normal tool calls are present
str	"finish" otherwise

Wired in by presets.py when experimental_interrupt=True so a graph that mixes ask_user AND generic interrupts routes through the right pause node deterministically. ttl_seconds is bound at graph-wiring time from FrameworkConfig.ask_user_pause_ttl_seconds; callers that omit it get the pre-TTL, run_id-only behaviour.

Source code in src/symfonic/core/nodes/interrupt.py

def interrupt_condition(
    state: dict[str, Any], ttl_seconds: int | None = None,
) -> str:
    """Conditional-edge router shared with the elicitation condition.

    Returns:
      - ``"interrupt"``   when ``_interrupt_pending`` is current (see
        :func:`interrupt_pending_is_current`)
      - ``"elicitation"`` when ``_ask_user_pending`` is current (preserves
        the existing ask_user routing)
      - ``"continue"``    when normal tool calls are present
      - ``"finish"``      otherwise

    Wired in by ``presets.py`` when ``experimental_interrupt=True`` so a
    graph that mixes ``ask_user`` AND generic interrupts routes through
    the right pause node deterministically. ``ttl_seconds`` is bound at
    graph-wiring time from ``FrameworkConfig.ask_user_pause_ttl_seconds``;
    callers that omit it get the pre-TTL, run_id-only behaviour.
    """
    if interrupt_pending_is_current(state, ttl_seconds):
        return "interrupt"
    from symfonic.core.nodes.elicitation import ask_user_pending_is_current

    if ask_user_pending_is_current(state, ttl_seconds):
        return "elicitation"

    from symfonic.core.edges.tool_condition import tools_condition

    return tools_condition(state)

interrupt_pending_is_current

interrupt_pending_is_current(state: dict[str, Any], ttl_seconds: int | None = None) -> bool

Is _interrupt_pending from THIS run, or left over from an old one?

Symmetric with :func:symfonic.core.nodes.elicitation.ask_user_pending_is_current -- deliberately duplicated rather than shared, matching this module's existing "the two pause paths share no code" isolation (see module docstring): a regression in one predicate cannot influence the other.

Nothing clears _interrupt_pending but the interrupt node, and that node is only reached by the very router this predicate feeds. An abandoned generic interrupt would otherwise re-route every later turn on the thread forever.

run_id is checked when the marker carries one, with the same back-compat rule as ask_user: a marker with no run_id passes this check (treated as current). In practice run_id is rarely available where the generic marker is built (:func:build_interrupt_marker is called from arbitrary tool bodies via SymfonicAgent.interrupt(), which has no graph state to read it from) -- minted_at is the discriminator that actually bounds abandonment for this path.

minted_at (wall-clock TTL) is stamped unconditionally by :func:build_interrupt_marker; when ttl_seconds is supplied and the marker is older than that, it is stale independent of the run_id outcome.

Source code in src/symfonic/core/nodes/interrupt.py

def interrupt_pending_is_current(
    state: dict[str, Any], ttl_seconds: int | None = None,
) -> bool:
    """Is ``_interrupt_pending`` from THIS run, or left over from an old one?

    Symmetric with :func:`symfonic.core.nodes.elicitation.ask_user_pending_is_current`
    -- deliberately duplicated rather than shared, matching this module's
    existing "the two pause paths share no code" isolation (see module
    docstring): a regression in one predicate cannot influence the other.

    Nothing clears ``_interrupt_pending`` but the interrupt node, and
    that node is only reached by the very router this predicate feeds.
    An abandoned generic interrupt would otherwise re-route every later
    turn on the thread forever.

    ``run_id`` is checked when the marker carries one, with the same
    back-compat rule as ask_user: a marker with no ``run_id`` passes
    this check (treated as current). In practice ``run_id`` is rarely
    available where the generic marker is built (:func:`build_interrupt_marker`
    is called from arbitrary tool bodies via ``SymfonicAgent.interrupt()``,
    which has no graph ``state`` to read it from) -- ``minted_at`` is
    the discriminator that actually bounds abandonment for this path.

    ``minted_at`` (wall-clock TTL) is stamped unconditionally by
    :func:`build_interrupt_marker`; when ``ttl_seconds`` is supplied and
    the marker is older than that, it is stale independent of the
    ``run_id`` outcome.
    """
    pending = state.get("_interrupt_pending")
    if not pending:
        return False
    if not isinstance(pending, dict):
        return True

    minted_run = pending.get("run_id")
    if minted_run is not None and minted_run != state.get("run_id"):
        return False

    if ttl_seconds is not None:
        minted_at = pending.get("minted_at")
        if isinstance(minted_at, int | float) and (
            time.time() - minted_at > ttl_seconds
        ):
            return False

    return True