From 3f86f071f3d26cb3fcf91841523f6f7be2796914 Mon Sep 17 00:00:00 2001 From: linkangjun Date: Mon, 2 Nov 2026 10:46:59 +0800 Subject: [PATCH] fix(middlewares): prevent todo retries after text-only output caps (#7605) * fix(middlewares): prevent todo retries after text-only output caps * docs(middlewares): clarify text-only output cap markers * test(subagents): isolate cleanup test from shared assembly pool * docs(middlewares): keep cap guidance within chain budget --------- Co-authored-by: linkangjun <331227811+linkangjun@users.noreply.github.com> --- README.md | 2 +- .../deerflow/agents/middlewares/AGENTS.md | 2 +- .../model_length_finish_reason_middleware.py | 7 +++- ...t_model_length_finish_reason_middleware.py | 25 +++++++++--- backend/tests/test_model_length_responses.py | 7 +++- backend/tests/test_subagent_batch_service.py | 13 ++++++- backend/tests/test_todo_middleware.py | 39 +++++++++++++++++++ 7 files changed, 83 insertions(+), 12 deletions(-) diff --git a/README.md b/README.md index 93856f54bc..aeed218580 100644 --- a/README.md +++ b/README.md @@ -395,7 +395,7 @@ fences, including unclosed fences, in complete and streamed replies. To route OpenAI models through `/v1/responses`, keep using `langchain_openai:ChatOpenAI` and set `use_responses_api: true` with `output_version: responses/v1`. - The lead agent also recognizes Responses output-token caps (`status: incomplete` with `incomplete_details.reason: max_output_tokens`), including streaming responses. It suppresses tool execution, removes the suppressed calls from later model requests, and preserves text and reasoning. A capped tool-call response includes a visible notice; other incomplete reasons retain their existing handling. + The lead agent also recognizes Responses output-token caps (`status: incomplete` with `incomplete_details.reason: max_output_tokens`), including streaming responses. It suppresses tool execution, removes the suppressed calls from later model requests, and preserves text and reasoning. A capped tool-call response includes a visible notice. Text-only output caps preserve the response and end the turn without unfinished-todo retries; other incomplete reasons retain their existing handling. When thinking is disabled on a legacy profile, blank `extra_body`, `thinking`, and nested `extra_body.thinking` keys in a `when_thinking_enabled` template are treated as absent during dialect inference. When thinking is enabled, template values are forwarded as configured. diff --git a/backend/packages/harness/deerflow/agents/middlewares/AGENTS.md b/backend/packages/harness/deerflow/agents/middlewares/AGENTS.md index ed8147d528..8395525388 100644 --- a/backend/packages/harness/deerflow/agents/middlewares/AGENTS.md +++ b/backend/packages/harness/deerflow/agents/middlewares/AGENTS.md @@ -188,7 +188,7 @@ Before changing a later authorization phase, read the [authorization RFC](../../ 34. **Custom middlewares** - *(optional)* Any `custom_middlewares` passed to `build_middlewares` are injected here, before config-declared extensions and the terminal-response/safety/clarification tail 35. **Configured extension middlewares** - `extensions.middlewares` in `config.yaml` or `extensions_config.json` optionally accepts `module.path:ClassName` strings or `{class, kwargs, apply_to}` objects. `deerflow.reflection.resolve_class` loads `AgentMiddleware` classes; import, class, and constructor errors fail agent creation. `kwargs` must be JSON-compatible; YAML dates/timestamps become ISO strings. Order: built-ins/custom and loop/token guards → extensions → terminal-response/safety/clarification tail. Optional `apply_to` selects both chains (default), lead, or subagent before import/construction. Trusted operator config only: paths instantiate arbitrary code. Gateway skill/MCP toggles preserve it in raw JSON; adding an API write path requires explicit trust-boundary review. 36. **TerminalResponseMiddleware** - When a provider returns an empty terminal `AIMessage` after tool execution, injects a hidden recovery prompt and retries the model once; a second empty response is replaced in checkpoint state by a visible error fallback marked for the run worker, so the run finishes as an error instead of a silent success -37. **ModelLengthFinishReasonMiddleware** - Responses: same-container `status=incomplete` + `incomplete_details.reason=max_output_tokens`. Suppress calls via shared cloning after clearing invalid calls; keep non-call blocks/metadata. Stamp cap markers; add notice. Also registered on subagent chains (before Safety), where `consume_stop_reason` surfaces `model_length_capped` to the executor. +37. **ModelLengthFinishReasonMiddleware** - Responses: same-container `status=incomplete` + `incomplete_details.reason=max_output_tokens`. Suppress calls via shared cloning after clearing invalid calls; keep non-call blocks/metadata. Stamp every cap; visible text-only caps get no notice. Add a notice only for suppressed calls or no visible content. Also registered on subagent chains (before Safety), where `consume_stop_reason` surfaces `model_length_capped` to the executor. 38. **SafetyFinishReasonMiddleware** - `safety_finish_reason.enabled`: suppresses valid/invalid safety-stopped calls. Responses: `status=incomplete` + `incomplete_details.reason=content_filter`. Clear invalid calls before shared cloning. Place after terminal/custom/configured hooks; `after_model` runs in reverse. 39. **ClarificationMiddleware** - Intercepts `ask_clarification`, writes a readable `ToolMessage.content` fallback plus a structured `ToolMessage.artifact.human_input` payload, and interrupts via `Command(goto=END)` (must be last). `after_model` drops same-turn sibling tool calls so they cannot run before the user answers; a malformed `ask_clarification` parked on `invalid_tool_calls` is the same stop signal. `disable_clarification` runs keep the siblings. Payloads are versioned — legacy `free_text`/`choice_with_other` stay `version: 1`; the v2 `form` mode (from `fields`) is `version: 2` so older frontends reject it and fall back to plain text. Field normalization is deterministic and lives in the middleware (it short-circuits before tool execution, so tool-arg typing gives no runtime validation), and it is atomic: any structurally broken entry — non-dict, bad/duplicate name, a name colliding with a JS `Object.prototype` member (`__proto__`/`constructor`), or exceeding the caps (16 fields / 24 options per field / 200 chars per text / `MAX_FORM_SERIALIZED_BYTES` = 16KB UTF-8, the per-item caps alone admitting forms whose IM text fallback overruns channel limits) — degrades the whole form to the legacy option/free-text modes, so a card never renders "complete" while missing a field. Benign issues degrade locally (unknown types — incl. unhashable JSON like `type: []` — and option-less selects become `text`); options are trimmed/deduped with blanks dropped (form- and top-level) since the frontend rejects blank labels. XML-to-dict option payloads are recursively flattened from dict/list containers in source order, scalar leaves kept, residual XML tags stripped before that trimming. Checkboxes are booleans defaulting to "no"; `required` on one means consent semantics. The response protocol is unchanged (v1 `text`/`option`): form cards submit a text summary as `response_kind: "text"`, so journal persistence needs no new allowlist entries. `RunJournal` reconciles visible, unpersisted `ToolMessage`s for current-run lead-agent calls at the next lead-agent `on_chat_model_start` after consumption, or at successful root `on_chain_end`. This preserves any middleware short-circuit, including blocked writes, across later model errors and checkpoint compaction (#4666). `_remember_current_run_tool_calls` excludes subagents; their results stay in `subagent.step`. Human Input Card replies are `hide_from_ui` `HumanMessage`s with `additional_kwargs.human_input_response`; `RunJournal` persists only allowlisted hidden sources (currently `ask_clarification`) as `llm.human.input`. diff --git a/backend/packages/harness/deerflow/agents/middlewares/model_length_finish_reason_middleware.py b/backend/packages/harness/deerflow/agents/middlewares/model_length_finish_reason_middleware.py index 8a085d4912..a287e4161d 100644 --- a/backend/packages/harness/deerflow/agents/middlewares/model_length_finish_reason_middleware.py +++ b/backend/packages/harness/deerflow/agents/middlewares/model_length_finish_reason_middleware.py @@ -103,6 +103,7 @@ class ModelLengthFinishReasonMiddleware(AgentMiddleware[AgentState]): return { "suppress_truncated_tool_calls": True, + "mark_text_only_length_termination": True, "length_notice_hash": canonical_hash(_MODEL_LENGTH_CAPPED_CONTENT), } @@ -157,7 +158,7 @@ class ModelLengthFinishReasonMiddleware(AgentMiddleware[AgentState]): content_source = clone_ai_message_with_tool_calls(last.model_copy(update={"invalid_tool_calls": []}), []) contains_tool_call = has_tool_call_intent(last) or content_source.content is not last.content contains_visible_content = has_visible_content(content_source) - if not contains_tool_call and contains_visible_content: + if not contains_tool_call and contains_visible_content and (last.additional_kwargs or {}).get("model_length_termination"): return None additional_kwargs = dict(last.additional_kwargs or {}) @@ -175,7 +176,9 @@ class ModelLengthFinishReasonMiddleware(AgentMiddleware[AgentState]): } replacement = content_source.model_copy( update={ - "content": append_visible_text(content_source, _MODEL_LENGTH_CAPPED_CONTENT), + # Text-only caps still need the terminal marker for downstream + # guards, but preserve their visible response without a notice. + "content": append_visible_text(content_source, _MODEL_LENGTH_CAPPED_CONTENT) if contains_tool_call or not contains_visible_content else content_source.content, "tool_calls": [], "invalid_tool_calls": [], "additional_kwargs": additional_kwargs, diff --git a/backend/tests/test_model_length_finish_reason_middleware.py b/backend/tests/test_model_length_finish_reason_middleware.py index 0128b70101..af5a8835a0 100644 --- a/backend/tests/test_model_length_finish_reason_middleware.py +++ b/backend/tests/test_model_length_finish_reason_middleware.py @@ -21,6 +21,19 @@ def _runtime(run_id: str = "run-1"): return runtime +def _assert_text_only_cap_preserved(result, original): + assert result is not None + replacement = result["messages"][0] + assert replacement.content == original.content + assert replacement.response_metadata == original.response_metadata + assert replacement.id == original.id + assert replacement.tool_calls == original.tool_calls + assert replacement.invalid_tool_calls == original.invalid_tool_calls + assert all(replacement.additional_kwargs[key] == value for key, value in original.additional_kwargs.items()) + assert replacement.additional_kwargs["model_length_termination"]["suppressed_tool_call_count"] == 0 + assert "model_length_termination" not in original.additional_kwargs + + def test_finish_reason_length_records_stop_reason_without_rewriting_textual_tool_call(): mw = ModelLengthFinishReasonMiddleware() runtime = _runtime() @@ -33,7 +46,7 @@ def test_finish_reason_length_records_stop_reason_without_rewriting_textual_tool result = mw._apply({"messages": [msg]}, runtime) - assert result is None + _assert_text_only_cap_preserved(result, msg) assert runtime.context["stop_reason"] == MODEL_LENGTH_CAPPED_STOP_REASON assert msg.content.startswith('') assert msg.tool_calls == [] @@ -60,7 +73,7 @@ def test_additional_kwargs_finish_reason_length_records_stop_reason(): additional_kwargs={"finish_reason": "length"}, ) - assert mw._apply({"messages": [msg]}, runtime) is None + _assert_text_only_cap_preserved(mw._apply({"messages": [msg]}, runtime), msg) assert runtime.context["stop_reason"] == MODEL_LENGTH_CAPPED_STOP_REASON @@ -72,7 +85,7 @@ def test_gemini_max_tokens_finish_reason_records_stop_reason(): response_metadata={"finish_reason": "MAX_TOKENS"}, ) - assert mw._apply({"messages": [msg]}, runtime) is None + _assert_text_only_cap_preserved(mw._apply({"messages": [msg]}, runtime), msg) assert runtime.context["stop_reason"] == MODEL_LENGTH_CAPPED_STOP_REASON @@ -84,7 +97,7 @@ def test_anthropic_max_tokens_stop_reason_records_stop_reason(): response_metadata={"stop_reason": "max_tokens"}, ) - assert mw._apply({"messages": [msg]}, runtime) is None + _assert_text_only_cap_preserved(mw._apply({"messages": [msg]}, runtime), msg) assert runtime.context["stop_reason"] == MODEL_LENGTH_CAPPED_STOP_REASON @@ -98,7 +111,7 @@ def test_length_cap_detection_logs_observability_fields(caplog): ) with caplog.at_level(logging.INFO, logger=_MW_LOGGER): - assert mw._apply({"messages": [msg]}, runtime) is None + _assert_text_only_cap_preserved(mw._apply({"messages": [msg]}, runtime), msg) records = [record for record in caplog.records if record.message == "Provider model length cap detected"] assert len(records) == 1 @@ -327,7 +340,7 @@ def test_existing_stop_reason_is_not_overwritten(caplog): msg = AIMessage(content="partial", response_metadata={"finish_reason": "length"}) with caplog.at_level(logging.INFO, logger=_MW_LOGGER): - assert mw._apply({"messages": [msg]}, runtime) is None + _assert_text_only_cap_preserved(mw._apply({"messages": [msg]}, runtime), msg) assert runtime.context["stop_reason"] == "token_capped" records = [record for record in caplog.records if record.message == "Provider model length cap detected"] diff --git a/backend/tests/test_model_length_responses.py b/backend/tests/test_model_length_responses.py index e9c7d66a09..5d3b73f039 100644 --- a/backend/tests/test_model_length_responses.py +++ b/backend/tests/test_model_length_responses.py @@ -277,7 +277,12 @@ def test_native_length_cap_preserves_content_metadata_and_earlier_stop_reason(co update = middleware.after_model({"messages": [message]}, SimpleNamespace(context=context)) assert context["stop_reason"] == "token_capped" if isinstance(content, str): - assert update is None + assert update is not None + replacement = update["messages"][0] + assert replacement.content == content + assert replacement.additional_kwargs["model_length_termination"]["suppressed_tool_call_count"] == 0 + assert getattr(replacement, container)["incomplete_details"] == metadata["incomplete_details"] + assert middleware.after_model({"messages": [replacement]}, SimpleNamespace(context=context)) is None else: assert update is not None replacement = update["messages"][0] diff --git a/backend/tests/test_subagent_batch_service.py b/backend/tests/test_subagent_batch_service.py index bf4073179e..6617332de7 100644 --- a/backend/tests/test_subagent_batch_service.py +++ b/backend/tests/test_subagent_batch_service.py @@ -1,5 +1,6 @@ import asyncio import logging +from concurrent.futures import ThreadPoolExecutor from enum import Enum from types import SimpleNamespace from unittest.mock import AsyncMock @@ -13,10 +14,20 @@ from deerflow.subagents import batch_service as service_module from deerflow.subagents.batch_runtime import BatchSubmitRequest from deerflow.subagents.batch_service import SubagentBatchService from deerflow.subagents.capacity import SubagentExecutionCapacity +from deerflow.utils import assembly_io _MISSING = object() +@pytest.fixture +def isolated_assembly_pool(monkeypatch): + # Lifecycle deadlines must not include work queued by other tests, while + # retaining the real off-loop assembly and cancellation behavior. + with ThreadPoolExecutor(max_workers=assembly_io._ASSEMBLY_WORKERS, thread_name_prefix="test-batch-assembly") as pool: + monkeypatch.setattr(assembly_io, "_ASSEMBLY_EXECUTOR", pool) + yield + + class FakeStatus(Enum): PENDING = "pending" RUNNING = "running" @@ -626,7 +637,7 @@ async def test_stopped_item_releases_its_execution_once_it_turns_terminal(monkey @pytest.mark.asyncio -async def test_item_error_cancels_and_releases_an_unfinished_execution(monkeypatch) -> None: +async def test_item_error_cancels_and_releases_an_unfinished_execution(monkeypatch, isolated_assembly_pool) -> None: # A repository error while polling fails the durable item; the subagent # must not keep running unsupervised, nor stay in the registry. registry = _Registry() diff --git a/backend/tests/test_todo_middleware.py b/backend/tests/test_todo_middleware.py index e2656cc3af..57c5c26155 100644 --- a/backend/tests/test_todo_middleware.py +++ b/backend/tests/test_todo_middleware.py @@ -762,6 +762,45 @@ class TestTodoMiddlewareAgentGraphIntegration: assert [len(_todo_completion_reminders(messages)) for messages in model.seen_messages] == [0, 0, 1, 1, 1] assert result["messages"][-1].content == "premature final 3" + @pytest.mark.parametrize("invoke_async", [False, True], ids=["sync", "async"]) + @pytest.mark.parametrize( + "content,response_metadata,additional_kwargs", + [ + ("partial answer", {"finish_reason": "length"}, {}), + ("partial answer", {"stop_reason": "max_tokens"}, {}), + ([{"type": "text", "text": "partial answer"}], {"finish_reason": "MAX_TOKENS"}, {}), + ("partial answer", {"status": "incomplete", "incomplete_details": {"reason": "max_output_tokens"}}, {}), + ("partial answer", {}, {"finish_reason": "length"}), + ], + ids=["openai", "anthropic", "gemini-blocks", "responses", "additional-kwargs"], + ) + def test_text_only_length_cap_does_not_reengage_todos(self, invoke_async, content, response_metadata, additional_kwargs): + todo_mw = TodoMiddleware() + capped_message = AIMessage(content=content, response_metadata=response_metadata, additional_kwargs=additional_kwargs) + model = _CapturingFakeMessagesListChatModel(responses=[capped_message]) + graph = create_agent( + model=model, + tools=[], + middleware=[todo_mw, ModelLengthFinishReasonMiddleware()], + state_schema=ThreadState, + ) + todos = [{"content": "Finish the report", "status": "in_progress"}] + state = {"messages": [("user", "write the report")], "todos": todos} + context = {"thread_id": "text-cap-thread", "run_id": "text-cap-run"} + + result = asyncio.run(graph.ainvoke(state, context=context)) if invoke_async else graph.invoke(state, context=context) + + assert len(model.seen_messages) == 1 + final_ai = result["messages"][-1] + assert final_ai.content == content + assert final_ai.response_metadata == response_metadata + assert final_ai.additional_kwargs["model_length_termination"]["suppressed_tool_call_count"] == 0 + assert all(final_ai.additional_kwargs[key] == value for key, value in additional_kwargs.items()) + assert "model_length_termination" not in capped_message.additional_kwargs + assert result["todos"] == todos + assert todo_mw._pending_completion_reminders == {} + assert not any(_todo_completion_reminders(messages) for messages in model.seen_messages) + def test_length_capped_write_file_does_not_reengage_todos(self): """Reproduces the incident (thread b1723286): model emits a write_file call with finish_reason=length, ModelLength suppresses it, and