diff --git a/CHANGELOG.md b/CHANGELOG.md index d9d59b996..9f1cb7e44 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -582,6 +582,18 @@ This section accumulates work toward the **2.1.0** milestone ### Fixed +- **subagents:** Give `max_turns` the meaning operators read it as. It was + handed to LangGraph as `recursion_limit`, which counts super-steps — one per + graph node — while `create_agent` compiles a node for every middleware + lifecycle hook, so one turn cost seven to eight steps through the subagent + chain and the built-in `general-purpose` agent's `max_turns=150` bought about + 18 tool-using turns before failing as `turn_capped`. Every middleware added + to the chain shrank the effective budget again. The executor now scales the + configured turn count by the per-turn node count of the chain it actually + assembled, so raising `max_turns` buys the turns it names. No config keys + changed; existing `max_turns` values now grant their full budget, which can + make a previously truncated subagent run longer, bounded as before by + `subagents.timeout_seconds` and `subagents.token_budget`. - **scheduler:** Enforce the global `max_concurrent_runs` budget on SQLite, which previously only held on Postgres. Claiming a queued occurrence counts the executing rows and then promotes one row to `launching`, and Postgres diff --git a/CHANGELOG_zh.md b/CHANGELOG_zh.md index cfad49cb4..adf47711f 100644 --- a/CHANGELOG_zh.md +++ b/CHANGELOG_zh.md @@ -397,6 +397,14 @@ ### 修复 +- **子智能体：** `max_turns` 现在真正表示运维人员理解的"轮次"。此前它被直接当作 LangGraph 的 + `recursion_limit` 传入，而后者统计的是 super-step——每个图节点一步，且 `create_agent` 会为每个 + 中间件生命周期钩子编译出一个节点，因此在子智能体的中间件链上一轮要花掉 7~8 步：内置 + `general-purpose` 的 `max_turns=150` 实际只买到约 18 轮带工具调用的轮次，随后以 `turn_capped` + 结束；每往链上加一个中间件，有效预算还会再缩水一次。现在执行器会按实际组装出的中间件链的 + 每轮节点数来换算配置的轮次，调高 `max_turns` 就能得到它所声明的轮次。没有配置项变化；既有的 + `max_turns` 取值现在会获得完整预算，因此原先被截断的子智能体运行可能变长，其上界仍由 + `subagents.timeout_seconds` 与 `subagents.token_budget` 约束。 - **调度器：** 在 SQLite 上同样强制执行全局 `max_concurrent_runs`，此前该上限只在 Postgres 上成立。 认领排队中的 occurrence 时，会先统计正在执行的行，再把其中一行提升为 `launching`，Postgres 用 advisory lock 将这两步串行化。而 SQLite 的 deferred 事务直到那条提升用的 UPDATE 才占用 writer， diff --git a/backend/packages/harness/deerflow/subagents/AGENTS.md b/backend/packages/harness/deerflow/subagents/AGENTS.md index 8b42cf184..b4119a226 100644 --- a/backend/packages/harness/deerflow/subagents/AGENTS.md +++ b/backend/packages/harness/deerflow/subagents/AGENTS.md @@ -29,7 +29,7 @@ executions are not checked, and acceptance never changes automatic retry policy. **Acceptance checklist path portability**: Paths are host-independent and raw `..` fails closed. Drive/UNC absolutes retain their class across `ntpath` normalization and cannot cross the root; drive-relative, shell-dependent provider/PSDrive, and POSIX-rooted `cd` on Windows are unprovable. Selection overlap separates pytest node IDs, normalizes safe `.`/duplicate separators, rejects `..`, and applies Windows casing to drive, UNC, PSDrive, or Windows-context paths; provider-qualified drive/UNC roots remain paths until a later node-ID split. POSIX paths and all node IDs keep their required case sensitivity. Ambiguous trailing-dot/space or 8.3 components, different volume IDs, cross-family pairs, and absolute/relative PSDrive pairs fail closed without execution/filesystem provenance; only same-volume, same-root-form paths compare lexically. Without shell provenance, raw commands reject backslashes; cmd `%VAR%`/`!VAR!`, `^`, `#`; Bash braces/tilde; and PowerShell splatting, typographic quotes, or unquoted parentheses. A lone `%` remains eligible. POSIX-only markers must short-circuit on Windows because pytest evaluates them at import. **Events**: `task_started`, `task_running`, `task_completed`/`task_failed`/`task_timed_out` **Handled LLM failures**: `LLMErrorHandlingMiddleware` deliberately converts provider/model exceptions into an `AIMessage` so the graph can end cleanly, stamping `additional_kwargs.deerflow_error_fallback=true` plus error metadata. Clean graph termination does not imply subagent success: `SubagentExecutor` inspects the last assistant message at terminalization and maps a marked fallback to `SubagentStatus.FAILED`, which then emits `task_failed` and the existing structured `subagent_error`. Only the marker is authoritative — error-looking assistant prose without it remains a normal completed result, so neither the executor nor frontend parses display text as a status protocol. -**Guardrail caps & `stop_reason` (#3875 Phase 2)**: three independent axes can end a subagent run early, and all now surface *why* through one additive field rather than a new status enum. **Turn axis**: `recursion_limit` on the subagent `run_config` equals `max_turns`, so exhausting the turn budget raises `GraphRecursionError` from `agent.astream`; `executor.py::_aexecute` catches it specifically (before the generic `except Exception`). **Token axis**: `TokenBudgetMiddleware` is attached per-agent via `build_subagent_runtime_middlewares` from `subagents.token_budget` (default `max_tokens` **coupled to `summarization.enabled`** — 1,000,000 when subagent summarization is on, 2,000,000 when off, warn at 0.7, hard-stop at 1.0; a user-set budget always wins regardless of the switch — #3875 Phase 3; a backstop against a subagent that burns tokens on trivial work). It does *not* raise: at the hard-stop threshold it strips the in-flight turn's tool calls, forces `finish_reason="stop"`, and lets the run complete naturally with a final answer. **Loop axis**: `LoopDetectionMiddleware` (attached at the same point) catches repeated identical tool-call sets — or one tool *type* called many times with varying args — and its hard-stop likewise strips `tool_calls` and forces a final answer without raising, recording `loop_capped`. Each guard exposes its cap on a per-`run_id` `consume_stop_reason(run_id)` accessor; `_aexecute` collects **every** middleware with that method (duck-typed via `hasattr`, so the executor has no import coupling to the guard classes) and surfaces the first non-`None` reason — adding a future guard needs no executor change. **Surfacing**: whichever axis fired, `_aexecute` stamps a normal status plus an additive reason — `completed` + `stop_reason=token_capped|turn_capped|loop_capped` when a usable final answer (or partial recovered from the last streamed chunk via `_extract_final_result` → `utils/messages.py::message_content_to_text`, returning a `"No response Generated"` sentinel when no text survived) was produced; `failed` + `stop_reason=turn_capped` when nothing usable survived. `SubagentResult.stop_reason` flows through `task_tool.py::_task_result_command` → `format_subagent_result_message` (renders `Task Succeeded (capped: ...)` / `Task failed (capped: ...)`) and `make_subagent_additional_kwargs`, which stamps the additive `subagent_stop_reason` key alongside the normal `subagent_status`. **Why additive, not an enum**: a new status value would break v1 consumers; an optional field is ignored by older frontends and ledger readers, so the cross-language contract (`contracts/subagent_status_contract.json` v2 + `subagents/status_contract.py` + `frontend/.../subtask-result.ts`, pinned by `test_status_values_match_contract` / `test_stop_reason_values_match_contract`) stays backward-compatible. The durable delegation ledger captures `stop_reason` onto the entry and renders model-facing guidance ("hit a guardrail cap with a partial result; reuse it, retry tighter, or raise the per-agent budget (`max_turns` / `token_budget`)") so the lead reuses a capped completion knowingly instead of mistaking it for a clean one. (Phase 1 shipped this surfacing as a `MAX_TURNS_REACHED` status enum in #3949; Phase 2 replaced that enum with the additive `stop_reason` field per the agreed design — the `max_turns_reached` status value and `SubagentStatus.MAX_TURNS_REACHED` are gone.) +**Guardrail caps & `stop_reason` (#3875 Phase 2)**: three independent axes can end a subagent run early, and all now surface *why* through one additive field rather than a new status enum. **Turn axis**: `max_turns` counts turns, `recursion_limit` counts super-steps — one per graph node, and `create_agent` compiles a node per middleware hook — so `turn_budget.py::resolve_recursion_limit` scales the turn budget by the assembled chain's depth (`test_subagent_turn_budget.py` pins it against the real compiled graph); a verbatim hand-off bought `general-purpose` ~18 of its 150 turns. Exhausting it raises `GraphRecursionError` from `agent.astream`; `executor.py::_aexecute` catches it specifically (before the generic `except Exception`). **Token axis**: `TokenBudgetMiddleware` is attached per-agent via `build_subagent_runtime_middlewares` from `subagents.token_budget` (default `max_tokens` **coupled to `summarization.enabled`** — 1,000,000 when subagent summarization is on, 2,000,000 when off, warn at 0.7, hard-stop at 1.0; a user-set budget always wins regardless of the switch — #3875 Phase 3; a backstop against a subagent that burns tokens on trivial work). It does *not* raise: at the hard-stop threshold it strips the in-flight turn's tool calls, forces `finish_reason="stop"`, and lets the run complete naturally with a final answer. **Loop axis**: `LoopDetectionMiddleware` (attached at the same point) catches repeated identical tool-call sets — or one tool *type* called many times with varying args — and its hard-stop likewise strips `tool_calls` and forces a final answer without raising, recording `loop_capped`. Each guard exposes its cap on a per-`run_id` `consume_stop_reason(run_id)` accessor; `_aexecute` collects **every** middleware with that method (duck-typed via `hasattr`, so the executor has no import coupling to the guard classes) and surfaces the first non-`None` reason — adding a future guard needs no executor change. **Surfacing**: whichever axis fired, `_aexecute` stamps a normal status plus an additive reason — `completed` + `stop_reason=token_capped|turn_capped|loop_capped` when a usable final answer (or partial recovered from the last streamed chunk via `_extract_final_result` → `utils/messages.py::message_content_to_text`, returning a `"No response Generated"` sentinel when no text survived) was produced; `failed` + `stop_reason=turn_capped` when nothing usable survived. `SubagentResult.stop_reason` flows through `task_tool.py::_task_result_command` → `format_subagent_result_message` (renders `Task Succeeded (capped: ...)` / `Task failed (capped: ...)`) and `make_subagent_additional_kwargs`, which stamps the additive `subagent_stop_reason` key alongside the normal `subagent_status`. **Why additive, not an enum**: a new status value would break v1 consumers; an optional field is ignored by older frontends and ledger readers, so the cross-language contract (`contracts/subagent_status_contract.json` v2 + `subagents/status_contract.py` + `frontend/.../subtask-result.ts`, pinned by `test_status_values_match_contract` / `test_stop_reason_values_match_contract`) stays backward-compatible. The durable delegation ledger captures `stop_reason` onto the entry and renders model-facing guidance ("hit a guardrail cap with a partial result; reuse it, retry tighter, or raise the per-agent budget (`max_turns` / `token_budget`)") so the lead reuses a capped completion knowingly instead of mistaking it for a clean one. (Phase 1 shipped this surfacing as a `MAX_TURNS_REACHED` status enum in #3949; Phase 2 replaced that enum with the additive `stop_reason` field per the agreed design — the `max_turns_reached` status value and `SubagentStatus.MAX_TURNS_REACHED` are gone.) **Context compaction (#3875 Phase 3, #4039)**: subagents inherit `DeerFlowSummarizationMiddleware` via `build_subagent_runtime_middlewares`, gated on the **same** `summarization.enabled` switch the lead reads (one config covers both chains; trigger/keep/model/prompt come from the shared `summarization` config so they cannot drift). The subagent builder attaches `DurableContextMiddleware` immediately before summarization, using the same skills path/read-tool settings as the lead chain. Compaction stores the generated summary in `ThreadState.summary_text` rather than as a `messages` item; the durable-context wrapper therefore projects it into the next model request as guarded hidden human data. This is required when a message-count keep policy preserves only an assistant tool-call plus its tool results: without the injected summary the next request begins with assistant/tool history and strict OpenAI-compatible providers can reject it. Because `DurableContextMiddleware` inserts a second `SystemMessage(authority_contract)` after the subagent's leading system prompt, the builder also appends `SystemMessageCoalescingMiddleware` innermost (mirroring the lead chain, appended after the optional summarization middleware so it is unconditionally last) to merge every `SystemMessage` into one leading `system_message` — otherwise the durable fix would trade #4039's assistant-first HTTP 400 for a duplicate-system 400 on the same strict backends (#4040). The factory is called with `skip_memory_flush=True` on the subagent path: the lead's `memory_flush_hook` (attached when `memory.enabled`) flushes pre-compaction messages into durable memory keyed by `thread_id`, and subagents share the parent's `thread_id`, so without skipping the hook a subagent's internal turns would pollute the **parent** thread's durable memory. Placement differs from the lead chain (lead appends summarization *before* the guard trio; subagent appends it *after*) — benign because the middleware implements only `before_model` (compaction) with no `after_model`/`consume_stop_reason`, so it cannot disturb the Phase 2 guard-cap stop-reason channel. Compaction rewrites the messages channel via `RemoveMessage(id=REMOVE_ALL_MESSAGES)`, which shrinks `len(messages)` below the step-capture cursor mid-run; `capture_new_step_messages` (see Step capture below) resets the cursor to the new tail on contraction so steps appended after the compaction point are not silently dropped. **Step capture & persistence (#3779)**: `executor.py` captures both assistant turns (`AIMessage`) **and** tool outputs (`ToolMessage`) via `subagents/step_events.py::capture_new_step_messages`, which walks the *newly-appended tail* of each `stream_mode="values"` chunk (not just `messages[-1]`) so a multi-tool-call turn — where LangGraph's `ToolNode` appends several `ToolMessage`s in one super-step — keeps every tool output instead of dropping all but the last. `runtime/runs/worker.py::_SubagentEventBuffer` additionally persists these `task_*` custom events to the `RunEventStore` as `subagent.start`/`subagent.step`/`subagent.end` (`category="subagent"`, `task_id` in `metadata`). It **batches** writes via `put_batch` (flushing on a terminal `subagent.end`, at `FLUSH_THRESHOLD` events, and in the worker's `finally`) rather than one `put()` per step, since `put()` is a documented low-frequency path (per-thread advisory lock per call) and a deep subagent (`max_turns=150`) emits hundreds of steps on the hot stream loop. `subagent_run_event` rejects malformed chunks that lack a non-empty `task_id`; running chunks additionally require a non-negative integer `message_index` and a message object, so persisted records always satisfy the required lifecycle envelope. `build_subagent_step` caps both the per-step `text` and each tool call's serialized `args` at `SUBAGENT_STEP_MAX_CHARS` (flagged `truncated` / `args_truncated`) so a large `write_file`/`bash` payload can't produce an unbounded row. The dedicated category keeps them out of `list_messages` (the thread feed) while `list_events` returns them for the frontend's fetch-on-expand backfill. `list_events` accepts `task_id` (filters on `metadata["task_id"]` — SQL-side in `DbRunEventStore` via `event_metadata["task_id"].as_string()`, in-memory in the JSONL/memory stores) plus an `after_seq` forward cursor, so the card pages through one subagent's steps without the run-wide `limit` truncating the tail (no schema migration: the filter rides the existing run-scoped index). `step_events.py` is a pure, unit-tested layer (`build_subagent_step` / `subagent_run_event`). **History contraction (#3875 Phase 3)**: `capture_new_step_messages` assumes append-only growth, but `DeerFlowSummarizationMiddleware` rewrites the messages channel via `RemoveMessage(id=REMOVE_ALL_MESSAGES)`, shrinking `len(messages)` below the cursor mid-run. On contraction (`total < processed_count`) the cursor resets to the new tail; `capture_step_message`'s id/content dedup prevents re-emitting pre-compaction steps, so steps appended after the compaction point are still captured instead of being dropped until `total` overtakes the stale cursor. **Deferred MCP tools** (if `tool_search.enabled`): `SubagentExecutor._build_initial_state` applies the subagent name allow/deny list and assembly-time authorization before calling the shared `assemble_deferred_tools`, appends the `tool_search` tool, injects the `` section into the subagent's `SystemMessage`, and threads the setup to `_create_agent`, which attaches `McpRoutingMiddleware` (when PR1 routing metadata matches deferred tools) before `DeferredToolFilterMiddleware` through `build_subagent_runtime_middlewares(...)`. Runtime skill policy is intentionally later and dynamic: `tool_search` may disclose/promote catalog metadata, but `SkillToolPolicyMiddleware` still removes or blocks any promoted business tool omitted by the active skill. Subagents thus withhold full MCP schemas until promotion, same as the lead agent; each task run gets a fresh `ThreadState` so promotion is isolated per run diff --git a/backend/packages/harness/deerflow/subagents/config.py b/backend/packages/harness/deerflow/subagents/config.py index 2d3de029c..b16c4e88e 100644 --- a/backend/packages/harness/deerflow/subagents/config.py +++ b/backend/packages/harness/deerflow/subagents/config.py @@ -22,9 +22,13 @@ class SubagentConfig: disabled for this subagent. Skill bodies and their allowed-tools policies take effect only after activation/loading at runtime. model: Model to use - 'inherit' uses parent's model. - max_turns: Maximum agent turns before stopping. Built-in agents use the - value set here (general-purpose=150, bash=60) unless the global - ``subagents.max_turns`` is set. + max_turns: Maximum agent turns — model call plus the tools it runs — + before stopping. Built-in agents use the value set here + (general-purpose=150, bash=60) unless the global + ``subagents.max_turns`` is set. ``turn_budget.py`` converts this + into the LangGraph ``recursion_limit`` that buys that many turns + through the assembled middleware chain; it is not passed through as + a super-step count. timeout_seconds: Bare fallback execution-time cap. For built-in agents the effective limit is the global ``subagents.timeout_seconds`` (default 1800 = 30 min), layered on by the registry; this 900 only applies diff --git a/backend/packages/harness/deerflow/subagents/executor.py b/backend/packages/harness/deerflow/subagents/executor.py index 0e2bd98b3..2766e3729 100644 --- a/backend/packages/harness/deerflow/subagents/executor.py +++ b/backend/packages/harness/deerflow/subagents/executor.py @@ -52,6 +52,7 @@ from deerflow.subagents.report_contract import ( ) from deerflow.subagents.step_events import capture_new_step_messages from deerflow.subagents.token_collector import SubagentTokenCollector +from deerflow.subagents.turn_budget import find_jumping_hooks, resolve_recursion_limit from deerflow.trace_context import DEERFLOW_TRACE_METADATA_KEY, ensure_trace_context, resolve_trace_id from deerflow.tracing import build_tracing_callbacks, inject_langfuse_metadata from deerflow.utils.messages import message_content_to_text @@ -921,6 +922,12 @@ class SubagentExecutor: # not just the first — because the v2 contract advertises more than one # cap reason. self._stop_reason_middlewares: list[Any] = [] + # LangGraph super-step budget that buys ``config.max_turns`` turns, + # resolved in ``_create_agent`` once the middleware chain — and with it + # the compiled graph's per-turn node count — is known. Stays ``None`` + # until then; ``_aexecute`` falls back to the raw turn count so a test + # double replacing ``_create_agent`` still produces a runnable config. + self._recursion_limit: int | None = None # What this subagent was assembled from, published to extension # observers at the end of ``_create_agent``. The prompt and skill set # are captured while ``_build_initial_state`` renders them because @@ -992,6 +999,7 @@ class SubagentExecutor: # a list (not ``next(...)``) so every guard is checked and a later one # is picked up automatically. self._stop_reason_middlewares = [m for m in middlewares if hasattr(m, "consume_stop_reason")] + self._recursion_limit = self._resolve_recursion_limit(middlewares) # system_prompt is included in initial state messages (see _build_initial_state) # to avoid multiple SystemMessages which some LLM APIs don't support. @@ -1013,6 +1021,44 @@ class SubagentExecutor: ) return agent + def _resolve_recursion_limit(self, middlewares: list[Any]) -> int: + """Translate ``max_turns`` into the super-step budget it actually means. + + ``max_turns`` is the operator-facing policy ("how many times may this + agent think and act"), while LangGraph's ``recursion_limit`` counts + graph nodes. ``create_agent`` compiles one node per middleware + lifecycle hook, so the two differ by the depth of the assembled chain — + see ``turn_budget.py`` for the arithmetic. Resolving it here, from the + chain this subagent was actually built with, keeps the budget stable as + middlewares are added and lets a per-agent chain differ. + """ + recursion_limit = resolve_recursion_limit(self.config.max_turns, middlewares) + logger.debug( + "[trace=%s] Subagent %s turn budget: max_turns=%s -> recursion_limit=%s (%d middlewares)", + self.trace_id, + self.config.name, + self.config.max_turns, + recursion_limit, + len(middlewares), + ) + # A hook that declares ``can_jump_to`` — agent-level hooks included — + # can leave the straight path through the graph, spending super-steps + # the flat per-turn cost does not model, so the budget silently becomes + # a lower bound. No middleware in the subagent chain declares one today; + # say so loudly if that changes, rather than letting runs quietly cap + # short again. + jumping_hooks = find_jumping_hooks(middlewares) + if jumping_hooks: + logger.warning( + "[trace=%s] Subagent %s has jump-declaring middleware hooks (%s); recursion_limit=%s is a lower bound for max_turns=%s, so the run may cap early", + self.trace_id, + self.config.name, + ", ".join(f"{name}.{hook}" for name, hook in jumping_hooks), + recursion_limit, + self.config.max_turns, + ) + return recursion_limit + def _describe_assembly( self, *, @@ -1450,7 +1496,10 @@ class SubagentExecutor: # namespace. Business consumers receive thread_id via ``context`` # below instead. run_config: RunnableConfig = { - "recursion_limit": self.config.max_turns, + # Super-steps, not turns: ``_create_agent`` (just above) scaled + # ``max_turns`` by the compiled chain's per-turn node count. + # Unset only when that method was replaced by a test double. + "recursion_limit": self._recursion_limit if self._recursion_limit is not None else self.config.max_turns, "callbacks": [collector], "tags": [collector_caller], } @@ -1604,14 +1653,15 @@ class SubagentExecutor: ) except GraphRecursionError: - # ``recursion_limit`` on run_config == ``self.config.max_turns`` - # (set above). Hitting it means the subagent exhausted its turn - # budget. Route into the additive ``stop_reason`` channel (#3875 - # Phase 2) rather than a dedicated status enum (which would break v1 - # contract consumers). If the run streamed usable partial work, - # surface it as ``completed``; otherwise ``failed``. Either way the - # lead can tell "out of budget" from "broken subagent" without - # parsing result text. + # ``recursion_limit`` on run_config is ``self.config.max_turns`` + # scaled into super-steps (set above), so hitting it means the + # subagent exhausted its turn budget — report the turn count, which + # is the number the operator configured. Route into the additive + # ``stop_reason`` channel (#3875 Phase 2) rather than a dedicated + # status enum (which would break v1 contract consumers). If the run + # streamed usable partial work, surface it as ``completed``; + # otherwise ``failed``. Either way the lead can tell "out of budget" + # from "broken subagent" without parsing result text. # # Prefer a guard's stop reason if one already fired this run: a # token-budget / loop hard-stop strips tool_calls to force a final diff --git a/backend/packages/harness/deerflow/subagents/turn_budget.py b/backend/packages/harness/deerflow/subagents/turn_budget.py new file mode 100644 index 000000000..87f6e82e8 --- /dev/null +++ b/backend/packages/harness/deerflow/subagents/turn_budget.py @@ -0,0 +1,138 @@ +"""Translating a subagent's turn budget into a LangGraph recursion limit. + +``max_turns`` is a per-agent policy an operator writes in ``config.yaml``: +how many times this agent may think and act before it is cut off. LangGraph's +``recursion_limit`` counts something else — super-steps, one per graph node +executed — and ``create_agent`` compiles every middleware lifecycle hook into +its own node (``{middleware}.before_model``, ``{middleware}.after_model``), so +one turn costs + + before_model nodes + ``model`` + after_model nodes + ``tools`` + +super-steps. The agent-level hooks add ``before_agent + after_agent`` on top, +once per invocation rather than once per turn. + +Passing ``max_turns`` straight through as ``recursion_limit`` therefore divides +the operator's budget by the depth of the middleware chain — the subagent chain +compiles seven to eight loop nodes, so ``max_turns=150`` bought roughly eighteen +turns — and silently shrinks every agent's budget again each time a middleware +is added. This module does the translation instead, deriving the multiplier from +the chain that was actually assembled. + +Hook participation is read the way LangChain's own factory reads it: a +class-level identity check against :class:`AgentMiddleware`'s base +implementation. A middleware that leaves a hook alone costs nothing, and an +extension middleware wrapped by ``IsolatedMiddleware`` — which mirrors that +identity precisely so LangChain sees the wrapper as the middleware it wraps — +is counted exactly like its inner middleware. + +The per-turn cost is a flat multiplier, which models the straight +``before_agent -> (before_model -> model -> tools)* -> after_agent`` path and +nothing else. A hook that declares ``can_jump_to`` and returns +``{"jump_to": ...}`` leaves that path: out of a model hook it re-enters the loop +without traversing ``tools``, spending another ``before_model + model + +after_model`` pass that buys no tool result; out of ``after_agent`` it re-enters +the loop after it has finished (or, for ``end``, reruns the ``after_agent`` +chain), and the hook runs again on the next exit; out of ``before_agent`` a +``tools`` jump runs a ``tools`` step no turn paid for. Any of these makes the +resolved limit a lower bound rather than an exact budget. How often a jump fires +is data-dependent and unbounded, so it cannot be folded into the arithmetic; +:func:`find_jumping_hooks` exposes the condition instead, and the subagent +executor warns when a hook declares one. No middleware in today's subagent +chain does. +""" + +from __future__ import annotations + +from collections.abc import Sequence +from typing import Any + +from langchain.agents.middleware import AgentMiddleware + +# The nodes every turn traverses whatever the middleware chain looks like: +# LangChain's ``model`` node, and the ``tools`` node the loop returns through. +# The turn that answers without tool calls skips ``tools`` but instead pays the +# graph's entry step, so the per-turn cost is the same either way — which is why +# this is a flat multiplier and not a multiplier plus a one-off correction. +_LOOP_NODES_PER_TURN = 2 + +# Sync/async hook pairs, in the grouping LangChain compiles them: one node per +# pair, whichever side (or both) a middleware overrides. +_LOOP_HOOK_PAIRS = (("before_model", "abefore_model"), ("after_model", "aafter_model")) +_INVOCATION_HOOK_PAIRS = (("before_agent", "abefore_agent"), ("after_agent", "aafter_agent")) + +# Where LangChain's ``hook_config(can_jump_to=...)`` decorator leaves its +# declaration; the factory reads the same attribute off the same hook methods. +_JUMP_DECLARATION_ATTR = "__can_jump_to__" + + +def _implements(middleware: Any, hook_pair: tuple[str, str]) -> bool: + """Whether *middleware* overrides either side of one sync/async hook pair. + + An object that does not carry the hook at all compiles to no node, so it + counts as not implementing it — LangChain's own check assumes an + :class:`AgentMiddleware` subclass and would raise on anything else. + """ + middleware_type = type(middleware) + for hook in hook_pair: + implementation = getattr(middleware_type, hook, None) + if implementation is not None and implementation is not getattr(AgentMiddleware, hook, None): + return True + return False + + +def _count_nodes(middlewares: Sequence[Any], hook_pairs: tuple[tuple[str, str], ...]) -> int: + """Nodes *middlewares* contribute across *hook_pairs* (one per implemented pair).""" + return sum(1 for middleware in middlewares for hook_pair in hook_pairs if _implements(middleware, hook_pair)) + + +def count_turn_steps(middlewares: Sequence[Any]) -> int: + """Super-steps one agent turn costs with *middlewares* in the chain.""" + return _count_nodes(middlewares, _LOOP_HOOK_PAIRS) + _LOOP_NODES_PER_TURN + + +def count_invocation_steps(middlewares: Sequence[Any]) -> int: + """Super-steps spent once per invocation, outside the agent loop.""" + return _count_nodes(middlewares, _INVOCATION_HOOK_PAIRS) + + +def find_jumping_hooks(middlewares: Sequence[Any]) -> list[tuple[str, str]]: + """Hooks that declare a jump, as ``(middleware, hook)`` name pairs. + + A jump spends super-steps the flat multiplier does not model, so a non-empty + result means :func:`resolve_recursion_limit` is a lower bound. Every + node-creating hook is inspected — the four LangChain's factory wires jump + edges for — because none is safe to exempt: ``after_agent`` can re-enter the + loop unboundedly, and even a once-per-invocation ``before_agent`` jump to + ``tools`` costs one unpriced step, which is enough to cap the last turn of an + exact budget. It is the declaration that is reported, not the destinations: + ``end`` is not a safe exit either, since out of ``after_agent`` LangChain + routes it back to the head of the ``after_agent`` chain. The declaration is + read off the same attribute, on the same overridden methods, that LangChain's + factory reads it from. + """ + jumping: list[tuple[str, str]] = [] + for middleware in middlewares: + middleware_type = type(middleware) + for hook_pair in _LOOP_HOOK_PAIRS + _INVOCATION_HOOK_PAIRS: + for hook in hook_pair: + # No identity check against the base method is needed here, unlike + # in the node counting above: only ``hook_config`` writes this + # attribute, and it is never on ``AgentMiddleware``'s own hooks, so + # a middleware that leaves the hook alone reads back as no jump. + if getattr(getattr(middleware_type, hook, None), _JUMP_DECLARATION_ATTR, None): + jumping.append((middleware_type.__name__, hook)) + return jumping + + +def resolve_recursion_limit(max_turns: int, middlewares: Sequence[Any]) -> int: + """The ``recursion_limit`` that buys *max_turns* turns through this chain. + + A non-positive ``max_turns`` is clamped to one turn: LangGraph rejects a + ``recursion_limit`` below 1, and a misconfigured budget should still let the + agent answer once rather than fail the run before it starts. + + The result is exact for a chain with no jump-declaring hooks and a lower bound + otherwise; see :func:`find_jumping_hooks`. + """ + return max(1, max_turns) * count_turn_steps(middlewares) + count_invocation_steps(middlewares) diff --git a/backend/tests/test_subagent_executor.py b/backend/tests/test_subagent_executor.py index 4ed653f5b..19fd2fd14 100644 --- a/backend/tests/test_subagent_executor.py +++ b/backend/tests/test_subagent_executor.py @@ -390,6 +390,103 @@ class TestAgentConstruction: assert captured["agent"]["tools"] == [] assert captured["agent"]["system_prompt"] is None # system_prompt is merged into initial state messages + def test_create_agent_scales_max_turns_into_a_super_step_budget( + self, + classes, + base_config, + monkeypatch: pytest.MonkeyPatch, + ): + """Regression: ``max_turns`` used to be passed to LangGraph verbatim. + + ``recursion_limit`` counts graph nodes and ``create_agent`` compiles one + per middleware lifecycle hook, so a verbatim hand-off bought roughly + ``max_turns / chain_depth`` turns — about 18 of the built-in + ``general-purpose`` agent's 150. + """ + from langchain.agents.middleware import AgentMiddleware + + from deerflow.subagents import executor as executor_module + + SubagentExecutor = classes["SubagentExecutor"] + + class _AfterModel(AgentMiddleware): + def after_model(self, state, runtime): + return None + + middlewares = [_AfterModel(), _AfterModel()] + + monkeypatch.setattr(executor_module, "create_chat_model", lambda **kwargs: object()) + monkeypatch.setattr(executor_module, "create_agent", lambda **kwargs: object()) + monkeypatch.setitem( + sys.modules, + "deerflow.agents.middlewares.tool_error_handling_middleware", + _module( + "deerflow.agents.middlewares.tool_error_handling_middleware", + build_subagent_runtime_middlewares=lambda **kwargs: middlewares, + ), + ) + + executor = SubagentExecutor( + config=base_config, + tools=[], + app_config=SimpleNamespace(models=[SimpleNamespace(name="default-model")]), + parent_model="parent-model", + ) + executor._create_agent() + + # model + tools + two after_model nodes, once per turn. + assert executor._recursion_limit == base_config.max_turns * 4 + + def test_create_agent_warns_when_a_counted_hook_can_jump( + self, + classes, + base_config, + monkeypatch: pytest.MonkeyPatch, + caplog, + ): + """A jump re-enters the loop without traversing ``tools``. + + The flat per-turn cost does not model that, so the resolved limit becomes + a lower bound. Nothing in today's subagent chain declares a jump; if one + ever does, the run must not quietly cap short the way it did before this + translation existed. + """ + import logging + + from langchain.agents.middleware import AgentMiddleware, hook_config + + from deerflow.subagents import executor as executor_module + + SubagentExecutor = classes["SubagentExecutor"] + + class _Jumper(AgentMiddleware): + @hook_config(can_jump_to=["model"]) + def after_model(self, state, runtime): + return None + + monkeypatch.setattr(executor_module, "create_chat_model", lambda **kwargs: object()) + monkeypatch.setattr(executor_module, "create_agent", lambda **kwargs: object()) + monkeypatch.setitem( + sys.modules, + "deerflow.agents.middlewares.tool_error_handling_middleware", + _module( + "deerflow.agents.middlewares.tool_error_handling_middleware", + build_subagent_runtime_middlewares=lambda **kwargs: [_Jumper()], + ), + ) + + executor = SubagentExecutor( + config=base_config, + tools=[], + app_config=SimpleNamespace(models=[SimpleNamespace(name="default-model")]), + parent_model="parent-model", + ) + with caplog.at_level(logging.WARNING, logger=executor_module.logger.name): + executor._create_agent() + + assert "_Jumper.after_model" in caplog.text + assert "lower bound" in caplog.text + @pytest.mark.anyio async def test_load_skills_uses_explicit_app_config_for_skill_storage( self, @@ -2122,6 +2219,56 @@ class TestAsyncExecutionPath: assert "regression-skill" in system_messages[0].content assert "Skill instruction text" not in system_messages[0].content + @pytest.mark.anyio + async def test_aexecute_sends_the_resolved_recursion_limit_to_the_graph(self, classes, base_config): + """The run config carries the scaled super-step budget ``_create_agent`` resolved.""" + from langchain_core.messages import AIMessage + + SubagentExecutor = classes["SubagentExecutor"] + captured_configs: list[dict] = [] + + async def capturing_astream(state, *, config, **kwargs): + captured_configs.append(config) + yield {"messages": [AIMessage(content="Done", id="msg-1")]} + + mock_agent = MagicMock() + mock_agent.astream = capturing_astream + + executor = SubagentExecutor(config=base_config, tools=[], thread_id="test-thread") + + def create_agent_resolving_budget(*args, **kwargs): + executor._recursion_limit = 77 + return mock_agent + + with patch.object(executor, "_create_agent", side_effect=create_agent_resolving_budget): + await executor._aexecute("Do something") + + assert captured_configs[0]["recursion_limit"] == 77 + + @pytest.mark.anyio + async def test_aexecute_falls_back_to_the_turn_count_when_the_chain_is_unknown(self, classes, base_config, mock_agent, msg): + """A test double replacing ``_create_agent`` leaves no chain to measure. + + The run must still get a usable limit rather than ``None``, which + LangGraph would reject. + """ + SubagentExecutor = classes["SubagentExecutor"] + captured_configs: list[dict] = [] + + async def capturing_astream(state, *, config, **kwargs): + captured_configs.append(config) + yield {"messages": [msg.ai("Done", msg_id="msg-1")]} + + mock_agent.astream = capturing_astream + + executor = SubagentExecutor(config=base_config, tools=[], thread_id="test-thread") + + with patch.object(executor, "_create_agent", return_value=mock_agent): + await executor._aexecute("Do something") + + assert executor._recursion_limit is None + assert captured_configs[0]["recursion_limit"] == base_config.max_turns + class TestSkillAllowedTools: @pytest.mark.anyio diff --git a/backend/tests/test_subagent_turn_budget.py b/backend/tests/test_subagent_turn_budget.py new file mode 100644 index 000000000..d7f2031db --- /dev/null +++ b/backend/tests/test_subagent_turn_budget.py @@ -0,0 +1,281 @@ +"""Tests for translating a subagent turn budget into a LangGraph recursion limit. + +Covers: +- Per-turn and per-invocation node counting from a middleware chain +- Sync-only, async-only, and both-sided hooks each costing exactly one node +- ``resolve_recursion_limit`` arithmetic, including a non-positive budget +- A pin test that compiles real ``create_agent`` graphs and binary-searches the + smallest ``recursion_limit`` that completes N turns, so the formula is checked + against the installed LangChain rather than against its documentation +- Jump detection across every hook LangChain wires a jump edge for, and the + shortfall each kind of jump causes against a compiled graph +""" + +import pytest +from langchain.agents import create_agent +from langchain.agents.middleware import AgentMiddleware, hook_config +from langchain_core.language_models import BaseChatModel +from langchain_core.messages import AIMessage +from langchain_core.outputs import ChatGeneration, ChatResult +from langchain_core.tools import tool +from langgraph.errors import GraphRecursionError + +from deerflow.subagents.turn_budget import count_invocation_steps, count_turn_steps, find_jumping_hooks, resolve_recursion_limit + + +def _middleware(name: str, *hooks: str) -> AgentMiddleware: + """Build a no-op middleware overriding exactly *hooks*.""" + namespace = {hook: (lambda self, state, runtime: None) for hook in hooks} + return type(name, (AgentMiddleware,), namespace)() + + +class TestNodeCounting: + def test_bare_chain_costs_model_plus_tools(self): + assert count_turn_steps([]) == 2 + assert count_invocation_steps([]) == 0 + + def test_each_model_hook_adds_one_node(self): + chain = [_middleware("Before", "before_model"), _middleware("After", "after_model")] + + assert count_turn_steps(chain) == 4 + + def test_one_middleware_implementing_both_model_hooks_adds_two_nodes(self): + chain = [_middleware("Both", "before_model", "after_model")] + + assert count_turn_steps(chain) == 4 + + def test_async_only_hook_costs_the_same_as_sync(self): + """LangChain compiles one node per sync/async pair, whichever side exists.""" + sync_only = [_middleware("Sync", "after_model")] + async_only = [_middleware("Async", "aafter_model")] + both_sides = [_middleware("Both", "after_model", "aafter_model")] + + assert count_turn_steps(sync_only) == count_turn_steps(async_only) == count_turn_steps(both_sides) == 3 + + def test_agent_hooks_are_charged_once_per_invocation_not_per_turn(self): + chain = [_middleware("Lifecycle", "before_agent", "after_agent")] + + assert count_turn_steps(chain) == 2 + assert count_invocation_steps(chain) == 2 + + def test_middleware_without_lifecycle_hooks_is_free(self): + """A wrap-only middleware runs inside the model node, so it adds none.""" + + class WrapOnly(AgentMiddleware): + def wrap_model_call(self, request, handler): + return handler(request) + + assert count_turn_steps([WrapOnly()]) == 2 + assert count_invocation_steps([WrapOnly()]) == 0 + + def test_object_without_hooks_contributes_nothing(self): + """A duck-typed entry compiles to no node, so a missing hook is not an override.""" + assert count_turn_steps([object()]) == 2 + assert count_invocation_steps([object()]) == 0 + + +class TestResolveRecursionLimit: + def test_scales_turns_by_the_assembled_chain(self): + """Regression: passing ``max_turns`` through divided the budget by chain depth.""" + chain = [ + _middleware("Before", "before_model"), + _middleware("After1", "after_model"), + _middleware("After2", "after_model"), + _middleware("After3", "after_model"), + _middleware("After4", "after_model"), + ] + + assert resolve_recursion_limit(150, chain) == 150 * 7 + + def test_adds_invocation_steps_on_top_of_the_turn_budget(self): + chain = [_middleware("Loop", "before_model"), _middleware("Lifecycle", "before_agent", "after_agent")] + + assert resolve_recursion_limit(10, chain) == 10 * 3 + 2 + + def test_non_positive_budget_still_buys_one_turn(self): + """LangGraph rejects a limit below 1; a misconfigured budget must not fail the run.""" + assert resolve_recursion_limit(0, []) == 2 + assert resolve_recursion_limit(-5, []) == 2 + + +class TestFindJumpingHooks: + """A jump re-enters the loop without traversing ``tools``. + + The flat per-turn multiplier does not model that, and how often a jump fires + is data-dependent, so the condition is reported rather than priced in. + """ + + def test_reports_the_middleware_and_hook_that_declares_a_jump(self): + class Jumper(AgentMiddleware): + @hook_config(can_jump_to=["model"]) + def after_model(self, state, runtime): + return None + + assert find_jumping_hooks([Jumper()]) == [("Jumper", "after_model")] + + def test_reports_an_async_hook_declaration(self): + class AsyncJumper(AgentMiddleware): + @hook_config(can_jump_to=["end"]) + async def abefore_model(self, state, runtime): + return None + + assert find_jumping_hooks([AsyncJumper()]) == [("AsyncJumper", "abefore_model")] + + def test_plain_hooks_declare_no_jump(self): + chain = [_middleware("Before", "before_model"), _middleware("After", "after_model")] + + assert find_jumping_hooks(chain) == [] + + @pytest.mark.parametrize("hook", ["before_agent", "abefore_agent", "after_agent", "aafter_agent"]) + def test_agent_level_hooks_are_reported(self, hook): + """They run once per invocation, but a jump out of them is not free. + + ``after_agent`` can re-enter the loop after it finished, and a + ``before_agent`` jump to ``tools`` runs a step no turn paid for. + """ + + async def async_hook(self, state, runtime): + return None + + def sync_hook(self, state, runtime): + return None + + implementation = async_hook if hook in {"abefore_agent", "aafter_agent"} else sync_hook + LifecycleJumper = type("LifecycleJumper", (AgentMiddleware,), {hook: hook_config(can_jump_to=["model"])(implementation)}) + + assert find_jumping_hooks([LifecycleJumper()]) == [("LifecycleJumper", hook)] + + def test_object_without_hooks_is_not_reported(self): + assert find_jumping_hooks([object()]) == [] + + +class _ScriptedModel(BaseChatModel): + """Emits ``turns - 1`` tool calls, then a final text answer.""" + + turns: int + calls: int = 0 + + @property + def _llm_type(self) -> str: + return "scripted" + + def bind_tools(self, tools, **kwargs): + return self + + def _generate(self, messages, stop=None, run_manager=None, **kwargs) -> ChatResult: + self.calls += 1 + if self.calls < self.turns: + message = AIMessage(content="", tool_calls=[{"name": "ping", "args": {}, "id": f"call-{self.calls}"}]) + else: + message = AIMessage(content="done") + return ChatResult(generations=[ChatGeneration(message=message)]) + + +@tool +def ping() -> str: + """Return a fixed token.""" + return "pong" + + +def _smallest_limit_completing(chain: list[AgentMiddleware], turns: int) -> int: + """Binary-search the smallest ``recursion_limit`` that completes *turns* turns.""" + low, high = 1, 512 + while low < high: + candidate = (low + high) // 2 + agent = create_agent(model=_ScriptedModel(turns=turns), tools=[ping], middleware=chain, checkpointer=False) + try: + agent.invoke({"messages": [("user", "go")]}, {"recursion_limit": candidate}) + except GraphRecursionError: + low = candidate + 1 + else: + high = candidate + return low + + +class TestFormulaMatchesCompiledGraph: + """Pins the arithmetic against the installed LangChain, not against its docs. + + If LangChain changes how it compiles middleware hooks into nodes, the + subagent turn budget silently drifts again. These cases fail instead. + """ + + @pytest.mark.parametrize( + "hooks", + [ + (), + (("Before", "before_model"),), + (("After", "after_model"),), + (("Before", "before_model"), ("After1", "after_model"), ("After2", "after_model")), + (("Lifecycle", "before_agent", "after_agent"), ("After", "after_model")), + ], + ) + @pytest.mark.parametrize("turns", [1, 3]) + def test_resolved_limit_is_exactly_what_the_graph_needs(self, hooks, turns): + chain = [_middleware(*spec) for spec in hooks] + + assert resolve_recursion_limit(turns, chain) == _smallest_limit_completing(chain, turns) + + @pytest.mark.parametrize( + ("hook", "destination", "count_model_calls", "jump_cost"), + [ + # Re-enters the model without traversing ``tools``: another + # before_model + model + after_model pass, here model + after_model. + pytest.param("after_model", "model", False, 2, id="after_model-to-model"), + # Re-enters the loop after it finished: a model pass, then the + # after_agent node again on the way out. + pytest.param("after_agent", "model", False, 2, id="after_agent-to-model"), + # ``end`` routes back to the head of the after_agent chain, so even + # the exit is not free — why detection ignores destinations. + pytest.param("after_agent", "end", False, 1, id="after_agent-to-end"), + # Once per invocation, but the staged tool call runs a ``tools`` + # step outside any model turn. + pytest.param("before_agent", "tools", True, 1, id="before_agent-to-tools"), + ], + ) + def test_a_jumping_hook_makes_the_limit_a_lower_bound(self, hook, destination, count_model_calls, jump_cost): + """Why ``find_jumping_hooks`` exists, pinned against a real graph. + + The loop cases count progress off the ToolMessages actually in state, so + a jump that re-enters the model buys no progress and is pure overhead, + the retry/repair shape. The ``before_agent`` case stages a tool call the + model never made, so there the model counts its own calls instead — + otherwise the staged tool result would pass for one of its turns. + """ + + class _ToolProgressModel(_ScriptedModel): + def _generate(self, messages, stop=None, run_manager=None, **kwargs) -> ChatResult: + executed = sum(1 for message in messages if getattr(message, "type", None) == "tool") + if executed < self.turns: + message = AIMessage(content="", tool_calls=[{"name": "ping", "args": {}, "id": f"call-{executed}"}]) + else: + message = AIMessage(content="done") + return ChatResult(generations=[ChatGeneration(message=message)]) + + def jump_once(self, state, runtime): + if self.jumped: + return None + self.jumped = True + update = {"jump_to": destination} + if destination == "tools": + update["messages"] = [AIMessage(content="", tool_calls=[{"name": "ping", "args": {}, "id": "staged"}])] + return update + + _JumpOnce = type( + "_JumpOnce", + (AgentMiddleware,), + {"jumped": False, hook: hook_config(can_jump_to=[destination])(jump_once)}, + ) + + tool_turns = 3 + budget_turns = tool_turns + 1 # the tool turns plus the turn that answers + + def run(limit): + model = _ScriptedModel(turns=budget_turns) if count_model_calls else _ToolProgressModel(turns=tool_turns) + create_agent(model=model, tools=[ping], middleware=[_JumpOnce()], checkpointer=False).invoke({"messages": [("user", "go")]}, {"recursion_limit": limit}) + + resolved = resolve_recursion_limit(budget_turns, [_JumpOnce()]) + with pytest.raises(GraphRecursionError): + run(resolved) + run(resolved + jump_cost) + + assert find_jumping_hooks([_JumpOnce()]) == [("_JumpOnce", hook)] diff --git a/config.example.yaml b/config.example.yaml index 1b8a10103..61e725398 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -1708,7 +1708,9 @@ subagent_runtime: # # Default timeout (seconds) for built-in subagents (default: 1800 = 30 min). # # Custom agents use their own timeout_seconds (default 900) unless overridden. # timeout_seconds: 1800 -# # Optional global max-turn override for all subagents. +# # Optional global max-turn override for all subagents. One turn is one +# # model call plus the tools it runs; the turn count is converted into +# # LangGraph's step budget, so this is the number of turns you actually get. # # Built-in defaults: general-purpose=150, bash=60. Leave unset to keep them. # # max_turns: 120 #