symfonic.core.nodes.elicitation

elicitation

ElicitationNode — interrupt node for ask_user tool calls.

This node is inserted into the graph topology between the react node and the tools node when ask_user_enabled=True. It runs AFTER the react node has returned normally (and its AIMessage has been checkpointed), so the interrupt() call here can never orphan a ToolMessage.

Graph flow when ask_user is called

react -> (conditional: _ask_user_pending set?) yes -> elicitation [interrupt() here] [on resume: inject ToolMessage] no -> tools | END (normal tool/finish routing)

ask_user_pending_is_current

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

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

A user who never answers leaves the marker set forever: nothing clears it but the elicitation node, and that node is only reached by the very router this predicate feeds. Every later turn on the thread would re-route to a question the user already walked away from.

Two independent discriminators are checked, and either can condemn the marker as stale:

run_id _build_input_state mints a fresh run_id per turn only when the caller does not supply one (core/runtime.py: run_id = run_id or uuid.uuid4().hex[:12]). run_id is a PUBLIC kwarg on run()/stream()/stream_typed() -- an adopter threading a stable conversation id through it for tracing makes every turn share a run_id, so this check alone always passes for that deployment. A resume enters through a Command whose update carries only deps and _callback_manager, so the checkpointed run_id correctly survives a resume either way.

Markers written before this field existed carry no ``run_id``;
those pass this check (treated as current), so an in-flight
checkpoint keeps working.

minted_at (wall-clock TTL) Stamped alongside run_id where the marker is built (core/nodes/react.py). When ttl_seconds is supplied and the marker is older than that, it is stale -- independent of whether run_id matches. This is what actually bounds abandonment when a caller-supplied run_id defeats the check above.

Used by all three routers that read the marker (:func:elicitation_condition, interrupt_condition, and the precondition gate's router) -- guarding only one of them would leave the others routing to the dead question.

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

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

    A user who never answers leaves the marker set forever: nothing
    clears it but the elicitation node, and that node is only reached by
    the very router this predicate feeds. Every later turn on the thread
    would re-route to a question the user already walked away from.

    Two independent discriminators are checked, and either can condemn
    the marker as stale:

    ``run_id``
        ``_build_input_state`` mints a fresh ``run_id`` per turn
        *only when the caller does not supply one* (``core/runtime.py``:
        ``run_id = run_id or uuid.uuid4().hex[:12]``). ``run_id`` is a
        PUBLIC kwarg on ``run()``/``stream()``/``stream_typed()`` -- an
        adopter threading a stable conversation id through it for
        tracing makes every turn share a ``run_id``, so this check alone
        always passes for that deployment. A resume enters through a
        ``Command`` whose update carries only ``deps`` and
        ``_callback_manager``, so the checkpointed ``run_id`` correctly
        survives a resume either way.

        Markers written before this field existed carry no ``run_id``;
        those pass this check (treated as current), so an in-flight
        checkpoint keeps working.

    ``minted_at`` (wall-clock TTL)
        Stamped alongside ``run_id`` where the marker is built
        (``core/nodes/react.py``). When ``ttl_seconds`` is supplied and
        the marker is older than that, it is stale -- independent of
        whether ``run_id`` matches. This is what actually bounds
        abandonment when a caller-supplied ``run_id`` defeats the check
        above.

    Used by all three routers that read the marker
    (:func:`elicitation_condition`, ``interrupt_condition``, and the
    precondition gate's router) -- guarding only one of them would leave
    the others routing to the dead question.
    """
    pending = state.get("_ask_user_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

create_elicitation_node

create_elicitation_node() -> Any

Return the elicitation node function.

The node reads state["_ask_user_pending"], calls interrupt(), and on resume builds a ToolMessage and clears _ask_user_pending.

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

def create_elicitation_node() -> Any:
    """Return the elicitation node function.

    The node reads ``state["_ask_user_pending"]``, calls ``interrupt()``,
    and on resume builds a ``ToolMessage`` and clears ``_ask_user_pending``.
    """

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

        from symfonic.core.contracts.elicitation import (
            AskUserQuestion,
            AskUserRequest,
        )

        pending: dict[str, Any] = state["_ask_user_pending"]
        tool_call_id: str = pending["tool_call_id"]
        raw_request: dict[str, Any] = pending.get("request_input", {})

        # Server-stamp question IDs (uuid4.hex[:12]) — the LLM never supplies ids.
        #
        # ``q_data`` is MODEL-SUPPLIED and therefore untrusted: the schema
        # sent to the provider carries ``minItems: 2`` on ``options``, but
        # nothing forces a model to honour it, and observed models do emit
        # single-option questions. Validating unguarded let a pydantic
        # ValidationError escape the graph node and kill the entire run --
        # a misbehaving model must not be able to terminate the agent.
        # Feed the error back as a ToolMessage instead so the react loop
        # can retry with a well-formed call.
        try:
            questions = [
                AskUserQuestion(**q_data)
                for q_data in raw_request.get("questions", [])
            ]
            stamped_request = AskUserRequest(questions=questions)
        except (ValidationError, TypeError) as exc:
            logger.warning(
                "ask_user call rejected as malformed; returning the error to "
                "the model for retry: %s", exc,
            )
            return {
                "messages": [
                    ToolMessage(
                        content=(
                            "ask_user call was rejected as malformed and no "
                            "question was shown to the user. Fix and call "
                            f"again. Validation error: {exc}"
                        ),
                        tool_call_id=tool_call_id,
                    )
                ],
                "_ask_user_pending": None,
            }

        # Call interrupt() — LangGraph surfaces this as an ExtensionEvent
        # with type="on_interrupt" in astream_events(v2).  The engine's
        # stream_typed() intercepts it and calls _mint_pause_token.
        # On resume, interrupt() returns the value passed to graph.invoke()
        # as the Command.resume payload.
        resume_value: Any = interrupt(
            {
                "type": "ask_user",
                "tool_call_id": tool_call_id,
                "request": stamped_request.model_dump(),
            }
        )

        # --- Resume path (re-invocation after graph.update_state / Command) ---
        # LangGraph re-runs this node body from the top; interrupt() returns
        # the resume payload this time instead of raising GraphInterrupt.
        if resume_value is not None:
            # The resume payload is either the raw AskUserResponse dict or
            # an already-parsed object.  The engine's resume() method builds
            # the ToolMessage and injects it via state_overrides["messages"],
            # so on the resume path we just clear the pending marker and let
            # the normal tool->react loop continue.
            tool_msg = ToolMessage(
                content=(
                    resume_value
                    if isinstance(resume_value, str)
                    else str(resume_value)
                ),
                tool_call_id=tool_call_id,
                name="ask_user",
            )
            return {
                "messages": [tool_msg],
                "_ask_user_pending": None,
            }

        # Should not reach here in normal operation, but be safe. Clearing
        # the marker without emitting a ToolMessage leaves the ask_user
        # call unanswered in the checkpoint, and every later turn replaying
        # that history is rejected by providers that validate tool-call
        # pairing. Close the call out explicitly instead.
        logger.warning(
            "elicitation node: interrupt() returned None unexpectedly for "
            "tool_call_id=%s", tool_call_id
        )
        return {
            "messages": [
                ToolMessage(
                    content=(
                        "ask_user was aborted before the user answered; no "
                        "question was shown. Continue without their input, or "
                        "ask again."
                    ),
                    tool_call_id=tool_call_id,
                    name="ask_user",
                )
            ],
            "_ask_user_pending": None,
        }

    return elicitation

elicitation_condition

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

Conditional-edge router: routes to 'elicitation' when ask_user is pending.

Used as the conditional edge from 'react'. Returns: - "elicitation" when _ask_user_pending is set (truthy) - "continue" when normal tool calls are present (handled by the existing tools_condition edge) - "finish" when there are no tool calls at all

ttl_seconds is bound at graph-wiring time (core/presets.py) from FrameworkConfig.ask_user_pause_ttl_seconds; callers that omit it (e.g. direct unit tests) get the pre-TTL, run_id-only behaviour.

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

def elicitation_condition(
    state: dict[str, Any], ttl_seconds: int | None = None,
) -> str:
    """Conditional-edge router: routes to 'elicitation' when ask_user is pending.

    Used as the conditional edge from 'react'.  Returns:
      - ``"elicitation"`` when ``_ask_user_pending`` is set (truthy)
      - ``"continue"``    when normal tool calls are present (handled by
                          the existing tools_condition edge)
      - ``"finish"``      when there are no tool calls at all

    ``ttl_seconds`` is bound at graph-wiring time (``core/presets.py``)
    from ``FrameworkConfig.ask_user_pause_ttl_seconds``; callers that
    omit it (e.g. direct unit tests) get the pre-TTL, run_id-only
    behaviour.
    """
    if ask_user_pending_is_current(state, ttl_seconds):
        return "elicitation"

    # Fall through to the standard tools_condition logic
    from symfonic.core.edges.tool_condition import tools_condition

    return tools_condition(state)