diff --git a/README.md b/README.md index b75be4049..dbdacea9d 100644 --- a/README.md +++ b/README.md @@ -707,7 +707,7 @@ Use `/compact` in the Web UI composer to summarize older context for the current Complex tasks rarely fit in a single pass. DeerFlow decomposes them. -The lead agent can spawn sub-agents on the fly — each with its own scoped context, tools, and termination conditions. Sub-agents run in parallel when possible, report back structured results, and the lead agent synthesizes everything into a coherent output. When token usage tracking is enabled, completed sub-agent usage is attributed back to the dispatching step. +The lead agent can spawn sub-agents on the fly — each with its own scoped context, tools, and termination conditions. Sub-agents run in parallel when possible, report back structured results, and the lead agent synthesizes everything into a coherent output. Long-running sub-agents compact older history when summarization is enabled and re-inject the summary as guarded, hidden durable context before continuing, so recent assistant/tool activity remains grounded in the task. When token usage tracking is enabled, completed sub-agent usage is attributed back to the dispatching step. This is how DeerFlow handles tasks that take minutes to hours: a research task might fan out into a dozen sub-agents, each exploring a different angle, then converge into a single report — or a website — or a slide deck with generated visuals. One harness, many hands. diff --git a/backend/AGENTS.md b/backend/AGENTS.md index c8b39bd9d..d01b992f8 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -236,7 +236,7 @@ Lead-agent middlewares are assembled in strict order across three functions: the 14. **DynamicContextMiddleware** - Injects the current date (and optionally memory) as a `` into the first HumanMessage, keeping the base system prompt fully static for prefix-cache reuse 15. **SkillActivationMiddleware** - Detects strict `/skill-name task` syntax on the latest real user message, resolves only enabled and runtime-allowed skills, injects the `SKILL.md` body as hidden current-turn context, and records a `middleware:skill_activation` audit event -16. **DurableContextMiddleware** - Captures `task` delegations into `ThreadState.delegations` (including in-progress dispatches and terminal result summaries) and loaded skill-file references (name/path/description, parsed in-memory - not the body) into `ThreadState.skill_context` before summarization can compact the paired tool-call/result messages, then projects durable context into each model request. Static authority rules are injected as a `SystemMessage`; untrusted field values (`summary_text`, delegation results, skill descriptions) are injected separately as a hidden `HumanMessage` data block so compressed history, delegated work, and which skills are active stay visible without being stored as `messages` or promoted to system-role instructions. +16. **DurableContextMiddleware** - Captures `task` delegations into `ThreadState.delegations` (including in-progress dispatches and terminal result summaries) and loaded skill-file references (name/path/description, parsed in-memory - not the body) into `ThreadState.skill_context` before summarization can compact the paired tool-call/result messages, then projects durable context into each model request. Static authority rules are injected as a `SystemMessage`; untrusted field values (`summary_text`, delegation results, skill descriptions) are injected separately as a hidden `HumanMessage` data block so compressed history, delegated work, and which skills are active stay visible without being stored as `messages` or promoted to system-role instructions. `build_subagent_runtime_middlewares` also attaches this middleware immediately before subagent summarization so a compacted `summary_text` is projected ahead of a preserved assistant/tool tail instead of leaving strict providers with an assistant-first request. 17. **SummarizationMiddleware** - *(optional, if enabled)* Context reduction when approaching token limits 18. **TodoListMiddleware** - *(optional, if `is_plan_mode`)* Task tracking with the `write_todos` tool 19. **TokenUsageMiddleware** - *(optional, if `token_usage.enabled`)* Records token usage metrics; subagent usage is merged back into the dispatching AIMessage by message position @@ -380,7 +380,7 @@ Proxied through nginx: `/api/langgraph/*` → Gateway LangGraph-compatible runti **Flow**: `task()` tool → `SubagentExecutor` → background thread → poll 5s → SSE events → result **Events**: `task_started`, `task_running`, `task_completed`/`task_failed`/`task_timed_out` **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.) -**Context compaction (#3875 Phase 3)**: 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 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. +**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. `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` assembles deferral after policy filtering via the shared `assemble_deferred_tools` (fail-closed), 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(...)`. 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 **Checkpointer isolation**: Subagent graphs are compiled with `checkpointer=False` to avoid inheriting the parent run's checkpointer, since subagents are one-shot and never resume. diff --git a/backend/packages/harness/deerflow/agents/middlewares/tool_error_handling_middleware.py b/backend/packages/harness/deerflow/agents/middlewares/tool_error_handling_middleware.py index 38ba66e78..e50eae884 100644 --- a/backend/packages/harness/deerflow/agents/middlewares/tool_error_handling_middleware.py +++ b/backend/packages/harness/deerflow/agents/middlewares/tool_error_handling_middleware.py @@ -379,6 +379,22 @@ def build_subagent_runtime_middlewares( middlewares.append(SafetyFinishReasonMiddleware.from_config(safety_config)) + # DurableContextMiddleware (#4039) — summarization stores compacted history in the + # ``summary_text`` state channel instead of writing a summary message back + # into ``messages``. Mirror the lead chain so subagents project that summary + # into subsequent model requests; otherwise a message-count keep policy can + # leave an assistant tool-call + tool-result tail with no leading user + # context, which strict providers reject. The same middleware also keeps + # skill references durable when their original read results are compacted. + from deerflow.agents.middlewares.durable_context_middleware import DurableContextMiddleware + + middlewares.append( + DurableContextMiddleware( + skills_container_path=app_config.skills.container_path, + skill_file_read_tool_names=app_config.summarization.skill_file_read_tool_names, + ) + ) + # DeerFlowSummarizationMiddleware — subagents inherit none of the lead's # context compaction today (#3875 Phase 3): a deep-research subagent # (``max_turns`` up to 150) can accumulate >1M cumulative input before @@ -420,4 +436,20 @@ def build_subagent_runtime_middlewares( if summarization_middleware is not None: middlewares.append(summarization_middleware) + # SystemMessageCoalescingMiddleware (#4040) — DurableContextMiddleware above + # inserts a second ``SystemMessage(authority_contract)`` after the leading + # system prompt (subagents carry their prompt as a leading ``SystemMessage`` + # in ``messages``, not via ``create_agent(system_prompt=...)``). Two system + # messages — or a non-leading one — are exactly what the strict backends this + # targets (vLLM/SGLang/Qwen/Anthropic) reject, so the durable fix would trade + # #4039's assistant-first 400 for a duplicate-system 400. Mirror the lead + # chain: append the coalescer innermost so it merges every SystemMessage into + # one leading ``system_message`` on the outgoing request. It only rewrites the + # per-request payload (no ``after_model``/``consume_stop_reason``), so it is + # inert to the Phase 2 guard-cap channel, and must sit inner of + # DurableContextMiddleware to observe the injected system message. + from deerflow.agents.middlewares.system_message_coalescing_middleware import SystemMessageCoalescingMiddleware + + middlewares.append(SystemMessageCoalescingMiddleware()) + return middlewares diff --git a/backend/tests/test_tool_error_handling_middleware.py b/backend/tests/test_tool_error_handling_middleware.py index a7721913c..5c358c48f 100644 --- a/backend/tests/test_tool_error_handling_middleware.py +++ b/backend/tests/test_tool_error_handling_middleware.py @@ -1,3 +1,4 @@ +import posixpath import sys from types import ModuleType, SimpleNamespace @@ -148,18 +149,29 @@ def test_build_subagent_runtime_middlewares_threads_app_config_to_llm_middleware # ToolErrorHandling) # + 1 ReadBeforeWriteMiddleware + 1 LoopDetectionMiddleware # + 1 TokenBudgetMiddleware (subagents.token_budget enabled by default, #3875 Phase 2) - # + 1 SafetyFinishReasonMiddleware (all enabled by default). + # + 1 SafetyFinishReasonMiddleware + 1 DurableContextMiddleware + # + 1 SystemMessageCoalescingMiddleware (all enabled by default). + from deerflow.agents.middlewares.durable_context_middleware import DurableContextMiddleware from deerflow.agents.middlewares.safety_finish_reason_middleware import SafetyFinishReasonMiddleware + from deerflow.agents.middlewares.system_message_coalescing_middleware import SystemMessageCoalescingMiddleware from deerflow.agents.middlewares.token_budget_middleware import TokenBudgetMiddleware from deerflow.agents.middlewares.tool_output_budget_middleware import ToolOutputBudgetMiddleware - assert len(middlewares) == 13 + assert len(middlewares) == 15 assert isinstance(middlewares[0], FakeMiddleware) # InputSanitizationMiddleware stub assert isinstance(middlewares[1], ToolOutputBudgetMiddleware) assert any(isinstance(m, ToolErrorHandlingMiddleware) for m in middlewares) # The token-budget backstop is attached by default so the cap engages (#3875). assert any(isinstance(m, TokenBudgetMiddleware) for m in middlewares) - assert isinstance(middlewares[-1], SafetyFinishReasonMiddleware) + assert any(isinstance(m, SafetyFinishReasonMiddleware) for m in middlewares) + # DurableContextMiddleware is present but not last: the coalescer (#4040) is + # appended innermost so it can merge the SystemMessage DurableContext injects. + # The coalescer is appended unconditionally (after the optional summarization + # middleware), so it is the last element regardless of summarization.enabled — + # unlike DurableContextMiddleware, which is only last when summarization is off. + durable_idx = next(i for i, m in enumerate(middlewares) if isinstance(m, DurableContextMiddleware)) + assert isinstance(middlewares[-1], SystemMessageCoalescingMiddleware) + assert durable_idx < len(middlewares) - 1 def test_tool_progress_middleware_is_outer_relative_to_error_handling(monkeypatch: pytest.MonkeyPatch): @@ -628,14 +640,19 @@ def test_subagent_runtime_middlewares_place_loop_detection_before_safety_finish( assert loop_idx < safety_idx -def test_subagent_runtime_middlewares_attach_summarization_when_enabled(monkeypatch): - """Subagents must inherit the lead's DeerFlowSummarizationMiddleware so a - long-running deep-research subagent compacts its context before it - accumulates pathological input (#3875 Phase 3). Gated on the SAME - ``summarization.enabled`` switch the lead reads — one config covers both - chains. The factory is patched at its source module so the test does not - depend on a real chat model; only the gating + wiring is asserted.""" +def test_subagent_runtime_middlewares_attach_durable_context_before_summarization(monkeypatch): + """Subagents must project ``summary_text`` back into model requests after + compaction, just like the lead agent does. + + Without ``DurableContextMiddleware``, a message-count keep policy can + retain only an assistant tool-call plus its tool results. The summary is + stored in ``ThreadState.summary_text`` but never reaches the next request, + so strict providers reject the assistant-first history. The durable + context layer must use the same skill settings as the lead chain and run + before summarization. + """ from deerflow.agents.middlewares import summarization_middleware as sm + from deerflow.agents.middlewares.durable_context_middleware import DurableContextMiddleware sentinel = object() captured: dict[str, object] = {} @@ -662,7 +679,168 @@ def test_subagent_runtime_middlewares_attach_summarization_when_enabled(monkeypa # skip_memory_flush=True so subagent-internal turns are not flushed into the # PARENT thread's durable memory (#3875 Phase 3 review). assert captured["skip_memory_flush"] is True - assert sentinel in middlewares + durable = [middleware for middleware in middlewares if isinstance(middleware, DurableContextMiddleware)] + assert len(durable) == 1 + # ``_skills_root`` is ``posixpath.normpath(container_path)``, so compare against + # the normalized form — a trailing slash / ``.`` / ``..`` in config would fail + # a raw equality even though the wiring is correct. + assert durable[0]._skills_root == posixpath.normpath(app_config.skills.container_path) + assert durable[0]._skill_read_tool_names == frozenset(app_config.summarization.skill_file_read_tool_names) + assert middlewares.index(durable[0]) < middlewares.index(sentinel) + + +def test_subagent_compaction_injects_summary_before_assistant_tool_tail(monkeypatch): + """A three-tool turn with ``keep=4`` must remain provider-valid. + + This reproduces the production failure shape: compaction preserves an + assistant tool-call plus three tool results while removing the original + system/user messages. The subagent chain must inject the generated summary + as durable human context before that tail reaches the model. + """ + from langchain.agents import create_agent + from langchain_core.language_models import BaseChatModel + from langchain_core.messages import AIMessage, HumanMessage, SystemMessage, ToolMessage + from langchain_core.outputs import ChatGeneration, ChatResult + + from deerflow.agents.middlewares.durable_context_middleware import DurableContextMiddleware + from deerflow.agents.middlewares.summarization_middleware import DeerFlowSummarizationMiddleware + from deerflow.agents.middlewares.system_message_coalescing_middleware import SystemMessageCoalescingMiddleware + from deerflow.agents.thread_state import ThreadState + from deerflow.config.summarization_config import ContextSize, SummarizationConfig + + class _StaticModel(BaseChatModel): + text: str + require_durable_summary: bool = False + + @property + def _llm_type(self) -> str: + return "static" + + def bind_tools(self, tools, **kwargs): + return self + + def _generate(self, messages, stop=None, run_manager=None, **kwargs): + if self.require_durable_summary: + first_ai = next(i for i, message in enumerate(messages) if isinstance(message, AIMessage)) + durable = [(i, message) for i, message in enumerate(messages) if isinstance(message, HumanMessage) and message.additional_kwargs.get("durable_context_data")] + assert durable, "compacted summary must be injected into the subagent request" + assert durable[0][0] < first_ai, "durable summary must precede the assistant/tool tail" + assert "COMPRESSED_SUBAGENT_HISTORY" in durable[0][1].content + # DurableContext injects a SystemMessage(authority); without the + # coalescer the request would carry it as a second/non-leading + # system message, which strict providers reject (#4040). Assert the + # outgoing request is provider-valid: a single leading SystemMessage. + system_indices = [i for i, message in enumerate(messages) if isinstance(message, SystemMessage)] + assert system_indices == [0], f"request must have exactly one leading SystemMessage, got {system_indices}" + return ChatResult(generations=[ChatGeneration(message=AIMessage(content=self.text))]) + + summary_model = _StaticModel(text="COMPRESSED_SUBAGENT_HISTORY") + strict_model = _StaticModel(text="final answer", require_durable_summary=True) + monkeypatch.setattr( + "deerflow.agents.middlewares.summarization_middleware.create_chat_model", + lambda **kwargs: summary_model, + ) + + app_config = _make_app_config().model_copy( + update={ + "summarization": SummarizationConfig( + enabled=True, + trigger=ContextSize(type="messages", value=5), + keep=ContextSize(type="messages", value=4), + ) + } + ) + runtime_middlewares = build_subagent_runtime_middlewares( + app_config=app_config, + model_name="test-model", + agent_name="general-purpose", + ) + compaction_middlewares = [middleware for middleware in runtime_middlewares if isinstance(middleware, (DurableContextMiddleware, DeerFlowSummarizationMiddleware, SystemMessageCoalescingMiddleware))] + agent = create_agent( + model=strict_model, + tools=[], + middleware=compaction_middlewares, + state_schema=ThreadState, + ) + + tool_calls = [{"name": "web_search", "args": {"query": f"q{i}"}, "id": f"call_{i}", "type": "tool_call"} for i in range(3)] + seed = [ + SystemMessage(content="subagent instructions", id="system"), + HumanMessage(content="research three regions", id="human"), + AIMessage(content="searching", tool_calls=tool_calls, id="assistant"), + *[ToolMessage(content=f"result {i}", tool_call_id=f"call_{i}", id=f"tool_{i}") for i in range(3)], + ] + + result = agent.invoke({"messages": seed}) + + assert result["summary_text"] == "COMPRESSED_SUBAGENT_HISTORY" + assert result["messages"][-1].content == "final answer" + + +def test_subagent_chain_coalesces_durable_authority_system_message(monkeypatch): + """The durable-context authority SystemMessage must not survive as a second one. + + Subagents carry their system prompt as a leading ``SystemMessage`` in + ``messages`` (``create_agent(system_prompt=None)``), and + ``DurableContextMiddleware`` inserts ``SystemMessage(authority_contract)`` + directly after it whenever durable data (summary / delegations / skills) is + present. That leaves two adjacent system messages — the exact non-leading / + duplicate-system shape strict OpenAI-compatible providers reject and the + same #4039 failure class the durable fix set out to avoid. + + ``build_subagent_runtime_middlewares`` must therefore pair durable context + with ``SystemMessageCoalescingMiddleware`` (#4040). This drives the real + builder output through a strict model and asserts the outgoing request keeps + exactly one leading ``SystemMessage``. Remove the coalescer from the builder + and the model sees ``[System(base), System(authority), ...]`` and this fails. + """ + from langchain.agents import create_agent + from langchain_core.language_models import BaseChatModel + from langchain_core.messages import AIMessage, SystemMessage, ToolMessage + from langchain_core.outputs import ChatGeneration, ChatResult + + from deerflow.agents.middlewares.durable_context_middleware import DurableContextMiddleware + from deerflow.agents.middlewares.system_message_coalescing_middleware import SystemMessageCoalescingMiddleware + from deerflow.agents.thread_state import ThreadState + + seen: dict[str, list[int]] = {} + + class _StrictModel(BaseChatModel): + @property + def _llm_type(self) -> str: + return "strict" + + def bind_tools(self, tools, **kwargs): + return self + + def _generate(self, messages, stop=None, run_manager=None, **kwargs): + seen["system_indices"] = [i for i, message in enumerate(messages) if isinstance(message, SystemMessage)] + return ChatResult(generations=[ChatGeneration(message=AIMessage(content="ok"))]) + + app_config = _make_app_config() + runtime_middlewares = build_subagent_runtime_middlewares( + app_config=app_config, + model_name="test-model", + agent_name="general-purpose", + ) + # Isolate the two middlewares under test, preserving builder order. The + # coalescer must come after (inner of) durable context to observe the + # injected system message. + chain = [m for m in runtime_middlewares if isinstance(m, (DurableContextMiddleware, SystemMessageCoalescingMiddleware))] + assert [type(m).__name__ for m in chain] == ["DurableContextMiddleware", "SystemMessageCoalescingMiddleware"] + + agent = create_agent(model=_StrictModel(), tools=[], middleware=chain, state_schema=ThreadState) + + # A leading system prompt plus an assistant tool-call tail, with a summary + # already in state so durable context injects its authority SystemMessage. + seed = [ + SystemMessage(content="subagent instructions", id="system"), + AIMessage(content="searching", tool_calls=[{"name": "web_search", "args": {"query": "x"}, "id": "call_0", "type": "tool_call"}], id="assistant"), + ToolMessage(content="result", tool_call_id="call_0", id="tool_0"), + ] + agent.invoke({"messages": seed, "summary_text": "COMPRESSED_SUBAGENT_HISTORY"}) + + assert seen["system_indices"] == [0], f"request must have a single leading SystemMessage, got {seen['system_indices']}" def test_subagent_runtime_middlewares_omit_summarization_when_factory_returns_none(monkeypatch):