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

@@ -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.
 

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`.
 

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,

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('<tool_call><invoke name="write_file">')
     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"]

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]

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()

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