def create_react_node(
config: AgentConfig,
tools: list[Any],
prompt_builder: Callable[..., str] | None = None,
) -> Callable[..., Any]:
"""Factory for ReactNode.
prompt_builder is an injection point for Phase 3.
When provided, it receives state and returns the system prompt string,
completely replacing _build_prompt.
"""
@observed_node("react")
async def react(state: dict[str, Any]) -> dict[str, Any]:
from ..callbacks.manager import CallbackManager
from ..contracts.callbacks import LLMEndEvent, LLMPreCallEvent, LLMStartEvent
from ..deps import BaseAgentDeps
from ..observability.protocol import ObservabilityHook
deps: BaseAgentDeps = state["deps"]
provider = deps.require(ModelProvider)
hook = deps.require(ObservabilityHook)
run_id: str = state.get("run_id", "") or ""
# Prefer merged manager from state (per-invocation), fall back to deps.
callback_mgr: CallbackManager | None = (
state.get("_callback_manager") or deps.get(CallbackManager)
)
# v7.23 T-7.23.6: per-iteration model dispatch via the adopter-
# supplied ``role_model_resolver``. When the adapter is
# registered (the engine always registers one; ``resolver=None``
# there is the abstain path), build the DispatchContext for THIS
# iteration and ask the adapter to resolve the model. ``None``
# from the adopter falls through to the snapshot default.
#
# The resolved ``ModelConfig`` flows through TWO downstream
# consumers in this node:
# 1. ``provider.get_chat_model(resolved_model)`` -- the LLM
# that processes THIS iteration.
# 2. ``_detect_provider_family(provider, resolved_model)``
# below (Path E co-fix, T-7.23.7) -- the family detector
# descends into the router's route map for the leaf that
# actually serves the request, so the cache_control
# annotation targets the right wire dialect.
#
# Caching contract: per-iteration memoization (R-V7.23-A
# mitigation). The force-resolver introspection at
# engine.py:2649 ALSO consults the same adapter via
# ``_AgentModelResolver.resolve``. Without memoization the
# adopter's classifier would run twice per turn. We stash the
# resolved config on ``state["_resolved_action_model"]`` so the
# next consumer can read it directly.
from ...agent.engine import _AgentModelResolver
_mr_adapter = (
deps.get(_AgentModelResolver) if deps is not None else None
)
resolved_model = config.model # snapshot default fallback
if _mr_adapter is not None:
# Build a snapshot DispatchContext for THIS iteration via the
# v8.1.0 C2 shared helper so a ``ToolCallPolicy.match`` sees the
# SAME context shape here (PRE-model) and at the POST-model
# redirect seam below. cache_state is the live engine view at
# this seam (the POST-model seam passes an empty mapping).
dispatch_ctx = _build_dispatch_context(
state, run_id, cache_state=_mr_adapter.cache_state_view(),
)
try:
resolved_model = _mr_adapter.resolve(dispatch_ctx)
except Exception:
logger.exception(
"v7.23 ModelResolver.resolve raised; degrading to "
"snapshot default for this iteration",
)
resolved_model = config.model
# v7.23 R-V7.23-A memoization: stash the resolved config so
# the force-introspection site at engine.py:2649 (T-7.23.8)
# reads the SAME object instead of re-invoking the adopter's
# classifier. Single resolve call per iteration.
state["_resolved_action_model"] = resolved_model
llm = provider.get_chat_model(resolved_model)
resolved_tools = state.get("resolved_tools") or tools
# v7.8.3 Lever 1: forced ``tool_choice`` resolution.
# v7.10 contract: the resolver runs PER ITERATION via
# :class:`ForcedToolChoiceResolver` registered on deps. This
# makes force AND release symmetric by construction (the
# resolver's idempotency clause re-evaluates the live
# ``messages`` slice every time) and closes the v7.8.3-v7.9.5
# bug where the engine assemble-time stamp never released on
# iterations 2..N -- the bug Jarvio surfaced after v7.9.5
# made the HARD lever actually fire. As a side effect this
# also activates force on ``stream()`` / ``stream_typed()``
# entry points (pre-v7.10 the assemble-time stamp lived only
# in ``run()``).
#
# Precedence:
# 1. Adopter override at ``state["forced_tool_choice"]`` --
# tests and custom adopters can pre-stamp directly. The
# v7.9.6 defense-in-depth release check below still
# applies to this path.
# 2. ``ForcedToolChoiceResolver`` from deps -- the engine
# registers an adapter at __init__ that wraps
# ``SymfonicAgent._maybe_resolve_forced_tool_choice``.
# The resolver already releases internally via the same
# ``ToolMessage.name`` equality check.
# 3. ``None`` -- pre-v7.8.3 default-safe (let the model
# choose).
_forced_tc = state.get("forced_tool_choice")
if _forced_tc is None:
from ..force_resolver import ForcedToolChoiceResolver
_resolver = (
deps.get(ForcedToolChoiceResolver)
if deps is not None else None
)
if _resolver is not None:
try:
_forced_tc = await _resolver.resolve(
state, state.get("messages") or [],
)
except Exception:
# Defensive: a misbehaving custom resolver must
# not stall the react loop. Treat resolver
# failure as "abstain" -- the safe default.
logger.exception(
"ForcedToolChoiceResolver.resolve raised; "
"degrading to no-force for this iteration",
)
_forced_tc = None
# v7.9.6 defense-in-depth release check. The resolver path
# already releases internally, so this is a no-op there. For
# the adopter-override path (precedence 1 above) this is the
# canonical release mechanism.
if _forced_tc is not None:
from langchain_core.messages import ToolMessage
for _m in state.get("messages") or ():
if isinstance(_m, ToolMessage) and getattr(_m, "name", None) == _forced_tc:
_forced_tc = None
break
# v7.11.0 role-aware tool palette filter. Applied AFTER the
# v7.0.1 intent-routing narrowing (``state["resolved_tools"]``)
# and BEFORE ``_bind_tools`` so palette ∩ resolved_tools is
# the final set seen by the model. Resolver returns ``None``
# to mean "no policy; pass through unchanged" (the safe
# default for empty role_tools / unmapped roles).
#
# Force-lever interaction (v7.11.0 contract): if the forced
# tool is excluded from the role palette, add it back to the
# bound list and emit a one-time WARN. Force is a structural
# compliance constraint (procedural skill metadata); silently
# dropping it would resurrect the v7.8.3 zero-tool-call probe
# bug class.
from ..roles import ACTION as _ACTION_ROLE
from ..tool_palette_resolver import ToolPaletteResolver
_palette_resolver = (
deps.get(ToolPaletteResolver) if deps is not None else None
)
if _palette_resolver is not None:
try:
_palette = await _palette_resolver.resolve(
state, _ACTION_ROLE, resolved_tools,
)
except Exception:
logger.exception(
"ToolPaletteResolver.resolve raised; "
"degrading to no-filter for this iteration",
)
_palette = None
if _palette is not None:
# Force-wins: if a tool is forced but the palette
# excluded it, append it back so bind_tools can still
# honour tool_choice=<forced_tc>.
if _forced_tc is not None and not any(
getattr(t, "name", None) == _forced_tc for t in _palette
):
for t in resolved_tools:
if getattr(t, "name", None) == _forced_tc:
logger.warning(
"procedural_force_first_action_tool "
"forces %r but role_tools[%r] excludes "
"it; appending forced tool to the bound "
"palette. Resolve by aligning the "
"palette allowlist with the procedural "
"skill's precondition / action_tool.",
_forced_tc, _ACTION_ROLE,
)
_palette = [*_palette, t]
break
resolved_tools = _palette
# v7.24.0 §4: forward ``tools_cache_ttl`` from FrameworkConfig so
# ``_bind_tools`` post-processes the bound runnable to stamp
# ``cache_control`` on the last tool definition. ``None`` default
# = no annotation; ``"5m"`` / ``"1h"`` make the wire marker
# explicit. The knob lives on ``FrameworkConfig`` (Pydantic),
# so we surface it via state when the engine threads it through
# OR fall back to ``getattr(config, ...)`` for adopter direct-
# construction paths (tests that build AgentConfig directly).
_tools_cache_ttl: str | None = (
state.get("_tools_cache_ttl")
if isinstance(state, dict)
else None
)
if _tools_cache_ttl is None:
_tools_cache_ttl = getattr(config, "tools_cache_ttl", None)
bound = _bind_tools(
llm,
resolved_tools,
tool_choice=_forced_tc,
tools_cache_ttl=_tools_cache_ttl,
)
if prompt_builder is not None:
prompt = prompt_builder(state)
else:
prompt = _build_prompt(state, config)
messages = list(state["messages"])
# v7.12.0 tool-result compaction. Swap verbose ToolMessage
# content for compact stubs once the result ages past
# ``keep_last_n`` AND exceeds the size threshold. The
# ``tool_call_id`` is preserved verbatim so Anthropic's
# tool_use/tool_result pairing invariant holds (any
# modification = HTTP 400). Operates on the snapshot
# ``messages`` only -- LangGraph state and checkpoint history
# stay lossless. The hook below sees the wire-accurate
# (compacted) view, which is what adopters want for cost
# observability.
from ..tool_result_ledger import ToolResultLedger
_ledger = deps.get(ToolResultLedger) if deps is not None else None
if _ledger is not None:
try:
# v8.6.0: stamp the RESOLVED model name into the
# compaction cfg so the offload net-saving gate prices
# from the correct ``MODEL_PRICING`` row for the model
# that will actually serve this turn (router-aware).
_tc = state.get("_tool_compaction") if isinstance(state, dict) else None
if isinstance(_tc, dict) and _tc.get("offload_enabled"):
_tc.setdefault(
"model_name",
getattr(resolved_model, "model_name", "") or "",
)
messages = await _maybe_compact_tool_results(
messages, _ledger, state,
)
except Exception:
logger.exception(
"ToolResultLedger compaction raised; degrading "
"to full-replay for this iteration (lossless "
"fallback)",
)
# v7.14.0: report the resolved model SKU (what the provider
# actually returned) instead of the requested ModelConfig string.
# Adopters with routing providers (MultiProviderRouter,
# Jarvio's JarvioModelProvider) can swap models server-side;
# stamping the requested name under-attributes cost. Falls
# back to the requested name when the chat model exposes no
# ``.model`` / ``.model_name`` attribute (MockChatModel).
from ..callbacks.emit import resolve_model_name as _resolve_model_name
# v7.23: report the model_name from the per-iteration resolved
# config (not the snapshot default) so cost attribution and
# observability hooks reflect which model actually served the
# turn. ``resolved_model`` is the snapshot when no adopter
# resolver is set; this preserves pre-v7.23 attribution.
_requested_model_name = getattr(
resolved_model, "model_name", str(resolved_model),
)
model_name: str = _resolve_model_name(llm, _requested_model_name)
try:
await hook.on_llm_start(model_name, messages, run_id)
except Exception:
logger.exception("ObservabilityHook.on_llm_start failed")
# v8.6.2: 1-based user-turn index (count of HumanMessages so far)
# so the OTel bridge can stamp ``symfonic.turn.index`` and adopters
# can break LLM cost down per-turn. Computed from state["messages"]
# (same source as ``iteration_index_v7150`` below). Shared by the
# start and end events so both spans agree on the turn.
_state_messages_turn = (
state.get("messages", ()) if isinstance(state, dict) else ()
)
turn_index_v862 = _compute_turn_index(_state_messages_turn)
if callback_mgr is not None and not callback_mgr.is_noop:
await callback_mgr.on_llm_start(
LLMStartEvent(
model=model_name,
messages=messages,
run_id=run_id,
system_prompt=prompt or "",
node_name="react",
turn_index=turn_index_v862,
)
)
# v7.13.0 Path E: assemble the wire-accurate message list ONCE
# (consolidate + messages-region cache breakpoint annotation),
# then share that exact list between the LLMPreCallEvent (if
# any handler subscribes) and the actual ``ainvoke`` call. The
# annotation is byte-deterministic per message-list shape, so
# sharing keeps the event payload truthful to the wire.
#
# Path E adds a SECOND ``cache_control`` breakpoint in the
# messages region when (a) provider is Anthropic, (b) prefix
# exceeds the model-family threshold (C1), and (c) a stable
# boundary exists (C2: closed AIMessage / HumanMessage). Off
# by default for non-Anthropic providers and below-threshold
# prefixes -- wire is byte-identical to v7.12.2 in those cases.
_full_unannotated = _consolidate_messages(messages, prompt)
# v7.23 T-7.23.6+T-7.23.7: read the resolved model for THIS turn
# so Path E's family detector descends into a router's route map
# for the LEAF that will actually serve the request (not the
# wrapper's ``_default``). Bundled per R-V7.23-B: splitting
# T-7.23.6 from T-7.23.7 ships per-turn switching with the
# wrong cache annotation.
_model_id = getattr(resolved_model, "model_name", "") or ""
_thinking_enabled = bool(getattr(resolved_model, "thinking", None))
from ..providers import ModelProvider as _ModelProvider # noqa: F401
_provider_family = "unknown"
try:
from ...agent.engine import _detect_provider_family
_provider_family = _detect_provider_family(
provider, resolved_model,
)
except Exception:
# Defensive: provider-family detection lives on the engine
# to keep this node decoupled. When unavailable (rare),
# degrade to no-annotation by leaving family as "unknown".
pass
# v8.7.0: count the cache markers the system prefix + tools array
# already claimed so the rolling ladder knows how many of
# Anthropic's 4-marker/request budget remain for the messages
# region. ``prompt`` is the (possibly stratigraphic-JSON) system
# string; ``_tools_cache_ttl`` was resolved above for _bind_tools.
_prefix_markers = _count_prefix_cache_markers(
prompt or "", _tools_cache_ttl,
)
full_messages = _apply_messages_cache_breakpoint(
_full_unannotated,
deps=deps,
provider_family=_provider_family,
model_id=_model_id,
thinking_enabled=_thinking_enabled,
prefix_markers_used=_prefix_markers,
# v8.7.1 (H1): isolate the per-conversation cache-marker state
# by the run_id so interleaved React loops on different
# conversations (one shared agent in create_agent_router) do
# not clobber each other's sticky/ladder anchor.
conversation_id=run_id or None,
)
# Emit LLMPreCallEvent (optional hook -- zero cost when no handler
# implements it). Built only when at least one registered handler
# defines on_llm_pre_call; constructing the event eagerly would
# otherwise inflate the hot path for handlers that don't care.
if callback_mgr is not None and callback_mgr.has_hook("on_llm_pre_call"):
from ._llm_pre_call import describe_tools, extract_invocation_params
tool_defs = describe_tools(resolved_tools)
invocation_params = extract_invocation_params(bound, full_messages)
await callback_mgr.on_llm_pre_call(
LLMPreCallEvent(
model=model_name,
messages=full_messages,
tools=tool_defs,
run_id=run_id,
invocation_params=invocation_params,
node_name="react",
)
)
# v7.15.0: compute the 0-indexed iteration BEFORE the ainvoke so
# the LLMEndEvent carries the count of prior AIMessages. First
# iteration of a turn -> 0. Subsequent iterations increment as
# the react loop appends AIMessages to state["messages"].
from langchain_core.messages import AIMessage as _AIMessage
_state_messages = state.get("messages", ()) if isinstance(state, dict) else ()
iteration_index_v7150 = sum(
1 for _m in _state_messages if isinstance(_m, _AIMessage)
)
# v7.15.0: capture wall-clock duration and start timestamp via
# the canonical context manager. Exception-safe -- duration_ms
# is populated even if _invoke_llm raises (caller-side will
# still emit nothing on exception, but the timing is honest).
from ..callbacks.emit import llm_timing
async with llm_timing() as _llm_timing_v7150:
ai_message = await _invoke_llm(
bound, messages, prompt, pre_annotated_full=full_messages,
)
# -- Elicitation Interrupt (ask_user) ----------------------------------
# v7.1.0: If the LLM called the built-in ``ask_user`` tool AND
# ask_user_enabled is True, we must NOT call interrupt() here.
# Calling interrupt() before returning would prevent the AIMessage
# from being checkpointed, leaving the graph in a state where the
# subsequent ToolMessage arrives orphaned (no preceding AIMessage
# with a matching tool_call_id).
#
# Instead, we return NORMALLY with ``_ask_user_pending`` set.
# A conditional edge routes the graph to the dedicated ``elicitation``
# node, which calls interrupt() AFTER the AIMessage is checkpointed.
tool_calls = getattr(ai_message, "tool_calls", []) or []
# v8.1.0 C4: POST-model ToolCallPolicy redirect seam. Runs BEFORE
# the adopter ``on_tool_call_dispatch`` block below so the adopter
# callback sees the policy's rewrite as input and remains the final
# escape hatch (policies-first precedence, locked §5). The policy
# rewrite rides the SAME palette gate (an unrouted target drops
# with a WARNING; the original dispatches). Empty policy tuple
# (the default) short-circuits here -> byte-identical to v8.0.1.
_policies: tuple[Any, ...] = tuple(
state.get("_tool_call_policies") or ()
)
if tool_calls and _policies:
_palette_names_p: tuple[str, ...] = tuple(
getattr(t, "name", "") for t in resolved_tools
if getattr(t, "name", "") != ""
)
_policy_ctx = _build_dispatch_context(state, run_id)
_p_rewritten: list[dict[str, Any]] = []
_p_any = False
for _tc in tool_calls:
_tc_name = _tc.get("name", "")
_tc_args = _tc.get("args", {}) or {}
_winner = None
for _policy in _policies:
if getattr(_policy, "redirect_to", None) is None:
continue
if not _policy.matches_tool(_tc_name):
continue
try:
if not _policy.match(_policy_ctx):
continue
except Exception:
logger.exception(
"v8.1 ToolCallPolicy %r match raised at "
"POST-model seam; skipping",
getattr(_policy, "name", "<unknown>"),
)
continue
_guard = getattr(_policy, "guard", None)
if _guard is not None:
try:
if not _guard(_policy_ctx):
# guard gates the redirect: precondition
# not satisfiable -> this policy abstains,
# try the next.
continue
except Exception:
logger.exception(
"v8.1 ToolCallPolicy %r guard raised at "
"POST-model seam; skipping (no redirect)",
getattr(_policy, "name", "<unknown>"),
)
continue
_winner = _policy
break # first-match-wins within the policy list
if _winner is None:
_p_rewritten.append(_tc)
continue
_new_name = _winner.redirect_to
_new_args = _tc_args
_xform = getattr(_winner, "args_transform", None)
if _xform is not None:
try:
_new_args = _xform(_tc_args, _policy_ctx)
except Exception:
logger.exception(
"v8.1 ToolCallPolicy %r args_transform raised; "
"redirecting with original args",
getattr(_winner, "name", "<unknown>"),
)
_new_args = _tc_args
# Palette gate: a redirect to an unrouted tool is dropped
# with a WARNING; the original call dispatches.
if _new_name not in _palette_names_p:
logger.warning(
"v8.1 ToolCallPolicy %r redirect targets "
"unregistered tool %r (available=%r); dropping "
"rewrite, dispatching original %r (call_id=%r)",
getattr(_winner, "name", "<unknown>"),
_new_name,
_palette_names_p,
_tc_name,
_tc.get("id", ""),
)
_p_rewritten.append(_tc)
continue
_p_new_tc: dict[str, Any] = {
"id": _tc.get("id", ""),
"name": _new_name,
"args": _new_args,
}
if "type" in _tc:
_p_new_tc["type"] = _tc["type"]
_p_rewritten.append(_p_new_tc)
_p_any = True
if _p_any:
ai_message.tool_calls = _p_rewritten
tool_calls = _p_rewritten
# v7.19.0 on_tool_call_dispatch -- state-conditioned tool-call
# rewriting (Jarvio SL-02 CSV-export -> slack_upload_file,
# SL-04 per-user memory title shape). Fires INSIDE the React
# node (not at engine post-chain) because LangGraph owns the
# dispatch loop -- there is no symfonic-owned seam between
# react and ToolNode. Mutating one node earlier IS the right
# seam; firing post-react in the engine would defeat the
# purpose because LangGraph already dispatched. See
# docs/guides/13-tool-call-dispatch-rewriter.md for the
# asymmetry rationale.
#
# Per-call dispatch + chained handler composition + validation
# gate (rewrites targeting unregistered tools are dropped with
# a WARNING). Zero-cost when no handler subscribes -- the
# has_hook gate skips event construction entirely.
if (
tool_calls
and callback_mgr is not None
and callback_mgr.has_hook("on_tool_call_dispatch")
):
from ..contracts.callbacks import ToolCallDispatchEvent
_palette_names: tuple[str, ...] = tuple(
getattr(t, "name", "") for t in resolved_tools
if getattr(t, "name", "") != ""
)
_rewritten_calls: list[dict[str, Any]] = []
_any_rewrite = False
for _tc in tool_calls:
_tc_id = _tc.get("id", "")
_tc_name = _tc.get("name", "")
_tc_args = _tc.get("args", {}) or {}
_event = ToolCallDispatchEvent(
run_id=run_id,
iteration_index=iteration_index_v7150,
node_name="react",
call_id=_tc_id,
tool_name=_tc_name,
args=_tc_args,
available_tools=_palette_names,
)
try:
_rewrite = await callback_mgr.on_tool_call_dispatch(
_event, state,
)
except Exception:
logger.exception(
"on_tool_call_dispatch: dispatch raised; "
"delivering unrewritten tool_call (call_id=%r)",
_tc_id,
)
_rewrite = None
if _rewrite is None:
_rewritten_calls.append(_tc)
continue
_new_name = _rewrite["tool_name"]
_new_args = _rewrite["args"]
# Validation gate: drop rewrites that target a tool
# outside the post-routing palette. Closes the
# "rewrite to unregistered tool -> hard LangGraph
# error several frames later" footgun. The original
# call is dispatched so the React loop stays live.
if _new_name not in _palette_names:
logger.warning(
"on_tool_call_dispatch: rewrite targets "
"unregistered tool %r (available=%r); "
"dropping rewrite, dispatching original call "
"%r (call_id=%r)",
_new_name,
_palette_names,
_tc_name,
_tc_id,
)
_rewritten_calls.append(_tc)
continue
# Build the rewritten LangChain tool_call dict.
# ``id`` is engine-owned and preserved verbatim;
# ``type`` (if present on the original) is preserved
# so non-Anthropic tool-call shapes round-trip.
_new_tc: dict[str, Any] = {
"id": _tc_id,
"name": _new_name,
"args": _new_args,
}
if "type" in _tc:
_new_tc["type"] = _tc["type"]
_rewritten_calls.append(_new_tc)
_any_rewrite = True
if _any_rewrite:
# Mutate ai_message.tool_calls in place so LangGraph's
# dispatcher sees the rewritten calls. The original
# ``tool_calls`` local was already used for
# ``ask_user_call`` lookup below, so we re-read from
# the rewritten list to keep ask_user routing
# consistent with the rewritten dispatch.
ai_message.tool_calls = _rewritten_calls
tool_calls = _rewritten_calls
ask_user_call = next(
(tc for tc in tool_calls if tc.get("name") == "ask_user"), None
)
usage = _extract_usage(ai_message)
output_text: str = (
ai_message.content if isinstance(ai_message.content, str) else ""
)
# v7.23 T-7.23.5: stamp per-engine cache_state for the
# adopter's per-iteration ``role_model_resolver`` to consult.
# Looks up the agent via the registered ``_AgentModelResolver``
# adapter (carries the cache_state provider closure that points
# at ``SymfonicAgent._cache_state``). Off-path when the adapter
# is missing (degraded test fixtures, bare react-node use).
try:
from ...agent.engine import _AgentModelResolver
_mr_adapter = (
deps.get(_AgentModelResolver) if deps is not None else None
)
if _mr_adapter is not None:
_on_resp = getattr(
_mr_adapter, "_on_llm_response_proxy", None,
)
if _on_resp is None:
# Common case: surface via the agent backref --
# adapters constructed at agent init carry a
# ``cache_state_provider`` returning the live dict.
# We mutate it via the agent's ``_on_llm_response``
# method, located by walking the provider closure.
pass
# The adapter's cache_state_provider returns the live
# dict by closure; stamp it directly when cache_creation
# is non-zero.
cache_creation = (
usage.get("cache_creation_input_tokens")
or usage.get("cache_creation_tokens")
or 0
)
if cache_creation and int(cache_creation) > 0 and model_name:
from datetime import UTC as _UTC
from datetime import datetime as _dt
_live = _mr_adapter._cache_state_provider()
if isinstance(_live, dict):
_live[model_name] = _dt.now(_UTC)
except Exception:
logger.exception(
"v7.23 cache_state stamp raised; degrading silently",
)
try:
await hook.on_llm_end(model_name, output_text, usage, run_id)
except Exception:
logger.exception("ObservabilityHook.on_llm_end failed")
if callback_mgr is not None and not callback_mgr.is_noop:
await callback_mgr.on_llm_end(
LLMEndEvent(
model=model_name,
output=output_text,
usage=usage,
run_id=run_id,
node_name="react",
duration_ms=_llm_timing_v7150.duration_ms,
started_at_utc=_llm_timing_v7150.started_at_utc,
iteration_index=iteration_index_v7150,
turn_index=turn_index_v862,
)
)
if ask_user_call and getattr(config, "ask_user_enabled", False):
# Return normally — the AIMessage will be checkpointed by LangGraph.
# The elicitation node (wired via conditional edge) will call
# interrupt() after the AIMessage is safe in the checkpoint.
raw_request = ask_user_call["args"].get("request", {})
return {
"messages": [ai_message],
"_ask_user_pending": {
"tool_call_id": ask_user_call["id"],
"request_input": raw_request,
},
}
# v7.17.0 React terminator contract -- salvage textual content
# even when the terminal AIMessage carries ``tool_calls``.
#
# Pre-v7.17 the extraction was Anthropic / OpenAI-shape biased:
# "AIMessage with no tool_calls -> done; otherwise -> None". Both
# providers consistently emit a tool-call iteration, then a
# ToolMessage, then a text-only final iteration, so the predicate
# held empirically. Three forced-termination paths break it:
#
# 1. OpenAI-compatible providers (qwen3-max via DashScope) may
# emit text content AND tool_calls in the SAME terminal
# message. ``tools_condition`` still routes to ``END`` per
# its loop-detection / single-turn shape, but the pre-v7.17
# extractor wrote ``None`` to ``final_response``, throwing
# away content the model authored.
# 2. ``tools_condition`` forces ``"finish"`` when the loop
# detector sees the same ``name:args`` signature >= 3 times
# in the last 10 messages (see
# ``core/edges/tool_condition.py:38-46``). The last
# AIMessage HAS tool_calls in this case, by definition.
# 3. LangGraph's ``recursion_limit`` cap (default 25, set via
# ``runtime.py:228``) terminates the graph because the loop
# kept calling tools -- last AIMessage HAS tool_calls.
#
# Architectural principle: ``tools_condition`` (NOT the react
# node) owns "should the loop continue?". If the routing
# condition decides to terminate while tool_calls are present,
# that's a forced-termination signal -- surface whatever text
# the model authored instead of returning ``""`` downstream.
# Engine still receives the AIMessage in ``state["messages"]``,
# so adopters that reconstruct the trace lose nothing.
has_tool_calls = bool(tool_calls)
# Reuse the existing engine helper for list-shape content
# (Anthropic extended thinking can interleave thinking / text /
# tool_use blocks in a single AIMessage). ``_flatten_content_blocks``
# already drops thinking and tool_use blocks; only ``text`` blocks
# contribute. Imported lazily to avoid a top-level engine import
# in the core node module (core -> agent would invert dependency).
from symfonic.agent.engine import _flatten_content_blocks
flattened = _flatten_content_blocks(ai_message.content)
final = flattened if flattened else None
# v7.17.0 diagnostic INFO log on empty-final exit. Adopters
# debugging silent-bail incidents (Jarvio's qwen3-max followup
# 2026-06-03) need to know WHY the loop produced no text. The
# structured payload covers all three forced-termination paths
# so the same log line fires whether the model emitted nothing
# (genuine empty) or emitted only tool_calls (the bug we're
# fixing). ``ReactLoopEndEvent`` carries the same payload for
# callback-system consumers (e.g. OTel bridges).
if not final:
content_type = type(ai_message.content).__name__
termination_reason = (
"tool_calls_with_no_text"
if has_tool_calls
else "no_content_no_tool_calls"
)
logger.info(
"react: empty final_response on terminator "
"(iteration_index=%d, content_type=%s, has_tool_calls=%s, "
"tool_call_count=%d, termination_reason=%s)",
iteration_index_v7150,
content_type,
has_tool_calls,
len(tool_calls),
termination_reason,
)
if callback_mgr is not None and callback_mgr.has_hook(
"on_react_loop_end"
):
from ..contracts.callbacks import ReactLoopEndEvent
await callback_mgr.on_react_loop_end(
ReactLoopEndEvent(
run_id=run_id,
iteration_index=iteration_index_v7150,
content_type=content_type,
has_tool_calls=has_tool_calls,
tool_call_count=len(tool_calls),
termination_reason=termination_reason,
)
)
return {
"messages": [ai_message],
"final_response": final,
}
return react