diff --git a/README.md b/README.md
index 496b79dce..1e0e8303e 100644
--- a/README.md
+++ b/README.md
@@ -739,6 +739,8 @@ This is the difference between a chatbot with tool access and an agent with an a
 
 **Strict Tool-Call Recovery**: When a provider or middleware interrupts a tool-call loop, DeerFlow now strips provider-level raw tool-call metadata on forced-stop assistant messages and injects placeholder tool results for dangling calls before the next model invocation. This keeps OpenAI-compatible reasoning models that strictly validate `tool_call_id` sequences from failing with malformed history errors.
 
+**Visible Tool-Run Completion**: For interactive turns, DeerFlow retries an empty post-tool final response once, then surfaces a visible error instead of reporting a silent successful run.
+
 ### Long-Term Memory
 
 Most agents forget everything the moment a conversation ends. DeerFlow remembers.
diff --git a/backend/AGENTS.md b/backend/AGENTS.md
index d47025bd5..1942f9099 100644
--- a/backend/AGENTS.md
+++ b/backend/AGENTS.md
@@ -249,9 +249,10 @@ Lead-agent middlewares are assembled in strict order across three functions: the
 26. **SubagentLimitMiddleware** - *(optional, if `subagent_enabled`)* Truncates excess `task` tool calls to enforce the `MAX_CONCURRENT_SUBAGENTS` limit
 27. **LoopDetectionMiddleware** - *(optional, if `loop_detection.enabled`)* Detects repeated tool-call loops; hard-stop clears both structured `tool_calls` and raw provider tool-call metadata before forcing a final text answer; stamps `loop_capped` via `consume_stop_reason` (#3875 Phase 2), symmetric to `TokenBudgetMiddleware`
 28. **TokenBudgetMiddleware** - *(optional, if `token_budget.enabled`)* Enforces per-run token limits
-29. **Custom middlewares** - *(optional)* Any `custom_middlewares` passed to `build_middlewares` are injected here, before the safety/clarification tail
-30. **SafetyFinishReasonMiddleware** - *(optional, if `safety_finish_reason.enabled`)* Suppresses tool execution when the provider safety-terminated the response (e.g. `finish_reason=content_filter`); registered after custom middlewares so LangChain's reverse-order `after_model` dispatch runs it first
-31. **ClarificationMiddleware** - Intercepts `ask_clarification` tool calls, writes a readable `ToolMessage.content` fallback plus structured `ToolMessage.artifact.human_input` request payload, and interrupts via `Command(goto=END)` (must be last). Because this middleware can short-circuit tool execution before LangChain emits `on_tool_end`, `RunJournal` performs a root-run final reconciliation for allowlisted clarification `ToolMessage`s whose `tool_call_id` was produced by the current run, so human-input request cards remain recoverable from `run_events` after checkpoint compaction. Human Input Card replies are submitted as `hide_from_ui` `HumanMessage`s with `additional_kwargs.human_input_response`; `RunJournal` persists only allowlisted hidden response sources (currently `ask_clarification`) as `llm.human.input`, which preserves answered-card state after compaction without exposing generic internal hidden context.
+29. **Custom middlewares** - *(optional)* Any `custom_middlewares` passed to `build_middlewares` are injected here, before the terminal-response/safety/clarification tail
+30. **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
+31. **SafetyFinishReasonMiddleware** - *(optional, if `safety_finish_reason.enabled`)* Suppresses tool execution when the provider safety-terminated the response (e.g. `finish_reason=content_filter`); registered after terminal-response/custom middlewares so LangChain's reverse-order `after_model` dispatch runs it first
+32. **ClarificationMiddleware** - Intercepts `ask_clarification` tool calls, writes a readable `ToolMessage.content` fallback plus structured `ToolMessage.artifact.human_input` request payload, and interrupts via `Command(goto=END)` (must be last). Because this middleware can short-circuit tool execution before LangChain emits `on_tool_end`, `RunJournal` performs a root-run final reconciliation for allowlisted clarification `ToolMessage`s whose `tool_call_id` was produced by the current run, so human-input request cards remain recoverable from `run_events` after checkpoint compaction. Human Input Card replies are submitted as `hide_from_ui` `HumanMessage`s with `additional_kwargs.human_input_response`; `RunJournal` persists only allowlisted hidden response sources (currently `ask_clarification`) as `llm.human.input`, which preserves answered-card state after compaction without exposing generic internal hidden context.
 
 ### Configuration System
 
diff --git a/backend/packages/harness/deerflow/agents/lead_agent/agent.py b/backend/packages/harness/deerflow/agents/lead_agent/agent.py
index e113cdf8f..2e3ce5b3a 100644
--- a/backend/packages/harness/deerflow/agents/lead_agent/agent.py
+++ b/backend/packages/harness/deerflow/agents/lead_agent/agent.py
@@ -33,6 +33,7 @@ from deerflow.agents.middlewares.memory_middleware import MemoryMiddleware
 from deerflow.agents.middlewares.safety_finish_reason_middleware import SafetyFinishReasonMiddleware
 from deerflow.agents.middlewares.subagent_limit_middleware import SubagentLimitMiddleware
 from deerflow.agents.middlewares.summarization_middleware import DeerFlowSummarizationMiddleware, create_summarization_middleware
+from deerflow.agents.middlewares.terminal_response_middleware import TerminalResponseMiddleware
 from deerflow.agents.middlewares.title_middleware import TitleMiddleware
 from deerflow.agents.middlewares.todo_middleware import TodoMiddleware
 from deerflow.agents.middlewares.token_usage_middleware import TokenUsageMiddleware
@@ -369,11 +370,16 @@ def build_middlewares(
     if custom_middlewares:
         middlewares.extend(custom_middlewares)
 
+    # A provider may return an empty AIMessage after tool execution. Retry the
+    # final response once, then persist a visible error fallback rather than
+    # allowing LangChain's no-tool-call router to end a silent successful run.
+    middlewares.append(TerminalResponseMiddleware())
+
     # SafetyFinishReasonMiddleware — suppress tool execution when the provider
-    # safety-terminated the response. Registered after custom middlewares so
-    # that LangChain's reverse-order after_model dispatch runs Safety first;
-    # cleared tool_calls then flow through Loop/Subagent accounting without
-    # firing extra alarms. See safety_finish_reason_middleware.py docstring.
+    # safety-terminated the response. Registered after the terminal-response
+    # and custom middlewares so LangChain's reverse-order after_model dispatch
+    # runs Safety first; cleared tool_calls then flow through the remaining
+    # accounting/terminal guards without firing extra alarms.
     safety_config = resolved_app_config.safety_finish_reason
     if safety_config.enabled:
         middlewares.append(SafetyFinishReasonMiddleware.from_config(safety_config))
diff --git a/backend/packages/harness/deerflow/agents/middlewares/terminal_response_middleware.py b/backend/packages/harness/deerflow/agents/middlewares/terminal_response_middleware.py
new file mode 100644
index 000000000..8c0e69320
--- /dev/null
+++ b/backend/packages/harness/deerflow/agents/middlewares/terminal_response_middleware.py
@@ -0,0 +1,214 @@
+"""Ensure tool-using lead-agent turns end with a visible assistant response."""
+
+from __future__ import annotations
+
+import threading
+from collections.abc import Awaitable, Callable
+from typing import Any, override
+
+from langchain.agents import AgentState
+from langchain.agents.middleware import AgentMiddleware
+from langchain.agents.middleware.types import ModelCallResult, ModelRequest, ModelResponse, hook_config
+from langchain_core.messages import AIMessage, HumanMessage, RemoveMessage, ToolMessage
+from langgraph.runtime import Runtime
+
+from deerflow.agents.middlewares._bounded_dict import BoundedDict
+
+_RECOVERY_PROMPT = (
+    "<system_reminder>\n"
+    "Your previous response after the tool execution was empty. Review the tool results "
+    "already present in the conversation and provide a concise, user-visible final response. "
+    "Do not call another tool unless it is strictly necessary.\n"
+    "</system_reminder>"
+)
+
+_FALLBACK_CONTENT = "The model completed the tool run but returned no final response, including after one automatic retry. Please try again or use a different model."
+
+_TOOL_CALL_FINISH_REASONS = {"tool_calls", "function_call"}
+
+
+def _has_visible_content(message: AIMessage) -> bool:
+    """Return whether an AI message contains user-visible text."""
+    content = message.content
+    if isinstance(content, str):
+        return bool(content.strip())
+    if isinstance(content, list):
+        for block in content:
+            if isinstance(block, str) and block.strip():
+                return True
+            if isinstance(block, dict) and block.get("type") in {"text", "output_text"}:
+                text = block.get("text")
+                if isinstance(text, str) and text.strip():
+                    return True
+    return False
+
+
+def _has_tool_call_intent_or_error(message: AIMessage) -> bool:
+    """Keep tool routing and malformed tool-call handling out of this guard."""
+    if message.tool_calls or getattr(message, "invalid_tool_calls", None):
+        return True
+    additional_kwargs = message.additional_kwargs or {}
+    if additional_kwargs.get("tool_calls") or additional_kwargs.get("function_call"):
+        return True
+    response_metadata = message.response_metadata or {}
+    return response_metadata.get("finish_reason") in _TOOL_CALL_FINISH_REASONS
+
+
+def _tool_result_in_current_turn(messages: list[Any]) -> bool:
+    """Return whether a tool result follows the latest real user message."""
+    latest_user_index = -1
+    for index, message in enumerate(messages):
+        if not isinstance(message, HumanMessage):
+            continue
+        if (message.additional_kwargs or {}).get("hide_from_ui"):
+            continue
+        latest_user_index = index
+    # Scope: #4027 covers interactive post-tool turns. Scheduled/internal
+    # invocations without a real HumanMessage need a separate terminal-success
+    # invariant rather than being inferred from arbitrary historical tools.
+    if latest_user_index == -1:
+        return False
+    return any(isinstance(message, ToolMessage) for message in messages[latest_user_index + 1 :])
+
+
+class TerminalResponseMiddleware(AgentMiddleware[AgentState]):
+    """Retry one empty post-tool response, then persist a visible error fallback."""
+
+    def __init__(self) -> None:
+        super().__init__()
+        self._lock = threading.Lock()
+        self._retry_counts: BoundedDict[tuple[str, str], int] = BoundedDict(1000)
+        self._pending_prompts: BoundedDict[tuple[str, str], bool] = BoundedDict(1000)
+
+    @staticmethod
+    def _key(runtime: Runtime) -> tuple[str, str]:
+        context = getattr(runtime, "context", None)
+        if isinstance(context, dict):
+            thread_id = str(context.get("thread_id") or "unknown-thread")
+            run_id = str(context.get("run_id") or context.get("run_attempt_id") or id(runtime))
+            return thread_id, run_id
+        # Defensive fallback for tests/custom embeddings. Production Gateway
+        # runs always provide thread_id and run_id in Runtime.context.
+        return "unknown-thread", str(id(runtime))
+
+    def _clear(self, runtime: Runtime) -> None:
+        key = self._key(runtime)
+        with self._lock:
+            self._retry_counts.pop(key, None)
+            self._pending_prompts.pop(key, None)
+
+    def _clear_other_runs(self, runtime: Runtime) -> None:
+        thread_id, run_id = self._key(runtime)
+        with self._lock:
+            stale = [key for key in self._retry_counts if key[0] == thread_id and key[1] != run_id]
+            for key in stale:
+                self._retry_counts.pop(key, None)
+                self._pending_prompts.pop(key, None)
+
+    def _apply(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
+        messages = list(state.get("messages") or [])
+        if not messages or not isinstance(messages[-1], AIMessage):
+            return None
+
+        last = messages[-1]
+        if _has_visible_content(last) or _has_tool_call_intent_or_error(last):
+            return None
+        if not _tool_result_in_current_turn(messages):
+            return None
+
+        key = self._key(runtime)
+        with self._lock:
+            # The recovery budget is once per run, not once per empty message.
+            # A retry that calls another tool must not refresh the budget and
+            # create an unbounded empty -> retry -> tool loop.
+            retry_count = self._retry_counts.get(key, 0)
+            if retry_count == 0:
+                self._retry_counts[key] = 1
+                self._pending_prompts[key] = True
+
+        if retry_count == 0:
+            # The next model call gets a new message id. Remove this empty
+            # terminal message now so a successful recovery does not leave it
+            # in checkpoint history or future model context.
+            message_updates = [RemoveMessage(id=last.id)] if last.id else []
+            return {"messages": message_updates, "jump_to": "model"}
+
+        additional_kwargs = dict(last.additional_kwargs or {})
+        additional_kwargs.update(
+            {
+                "deerflow_error_fallback": True,
+                "error_reason": "Model returned an empty terminal response after one retry",
+            }
+        )
+        fallback = last.model_copy(
+            update={
+                "content": _FALLBACK_CONTENT,
+                "additional_kwargs": additional_kwargs,
+            }
+        )
+        return {"messages": [fallback]}
+
+    def _augment_request(self, request: ModelRequest) -> ModelRequest:
+        key = self._key(request.runtime)
+        with self._lock:
+            pending = key in self._pending_prompts
+            self._pending_prompts.pop(key, None)
+        if not pending:
+            return request
+        reminder = HumanMessage(
+            content=_RECOVERY_PROMPT,
+            name="terminal_response_recovery",
+            additional_kwargs={"hide_from_ui": True},
+        )
+        return request.override(messages=[*request.messages, reminder])
+
+    @override
+    def before_agent(self, state: AgentState, runtime: Runtime) -> dict | None:
+        self._clear_other_runs(runtime)
+        # A prior invocation can bypass after_agent via Command(goto=END).
+        # Reset the same run id here so resume starts with a fresh one-retry
+        # budget; internal jump_to=model loops do not re-run before_agent.
+        self._clear(runtime)
+        return None
+
+    @override
+    async def abefore_agent(self, state: AgentState, runtime: Runtime) -> dict | None:
+        self._clear_other_runs(runtime)
+        self._clear(runtime)
+        return None
+
+    @hook_config(can_jump_to=["model"])
+    @override
+    def after_model(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
+        return self._apply(state, runtime)
+
+    @hook_config(can_jump_to=["model"])
+    @override
+    async def aafter_model(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
+        return self._apply(state, runtime)
+
+    @override
+    def wrap_model_call(
+        self,
+        request: ModelRequest,
+        handler: Callable[[ModelRequest], ModelResponse],
+    ) -> ModelCallResult:
+        return handler(self._augment_request(request))
+
+    @override
+    async def awrap_model_call(
+        self,
+        request: ModelRequest,
+        handler: Callable[[ModelRequest], Awaitable[ModelResponse]],
+    ) -> ModelCallResult:
+        return await handler(self._augment_request(request))
+
+    @override
+    def after_agent(self, state: AgentState, runtime: Runtime) -> dict | None:
+        self._clear(runtime)
+        return None
+
+    @override
+    async def aafter_agent(self, state: AgentState, runtime: Runtime) -> dict | None:
+        self._clear(runtime)
+        return None
diff --git a/backend/tests/test_lead_agent_model_resolution.py b/backend/tests/test_lead_agent_model_resolution.py
index 4083cf576..5b3ef97f7 100644
--- a/backend/tests/test_lead_agent_model_resolution.py
+++ b/backend/tests/test_lead_agent_model_resolution.py
@@ -372,10 +372,18 @@ def test_build_middlewares_uses_resolved_model_name_for_vision(monkeypatch):
 
     assert any(isinstance(m, lead_agent_module.ViewImageMiddleware) for m in middlewares)
     # verify the custom middleware is injected correctly.
-    # Chain tail order after the custom middleware is:
-    #   ..., custom, SafetyFinishReasonMiddleware, ClarificationMiddleware
-    # so the custom mock sits at index [-3].
-    assert len(middlewares) > 0 and isinstance(middlewares[-3], MagicMock)
+    # With this test's default safety config enabled, the tail order is:
+    #   ..., custom, TerminalResponseMiddleware, SafetyFinishReasonMiddleware,
+    #   ClarificationMiddleware, so the custom mock sits at index [-4].
+    assert len(middlewares) > 0 and isinstance(middlewares[-4], MagicMock)
+
+    from deerflow.agents.middlewares.clarification_middleware import ClarificationMiddleware
+    from deerflow.agents.middlewares.safety_finish_reason_middleware import SafetyFinishReasonMiddleware
+    from deerflow.agents.middlewares.terminal_response_middleware import TerminalResponseMiddleware
+
+    assert isinstance(middlewares[-3], TerminalResponseMiddleware)
+    assert isinstance(middlewares[-2], SafetyFinishReasonMiddleware)
+    assert isinstance(middlewares[-1], ClarificationMiddleware)
 
 
 def test_build_middlewares_passes_explicit_app_config_to_shared_factory(monkeypatch):
diff --git a/backend/tests/test_terminal_response_middleware.py b/backend/tests/test_terminal_response_middleware.py
new file mode 100644
index 000000000..e8151e742
--- /dev/null
+++ b/backend/tests/test_terminal_response_middleware.py
@@ -0,0 +1,278 @@
+from __future__ import annotations
+
+from typing import Any
+
+import pytest
+from langchain.agents import create_agent
+from langchain_core.language_models import BaseChatModel
+from langchain_core.messages import AIMessage, HumanMessage, ToolMessage
+from langchain_core.outputs import ChatGeneration, ChatResult
+from langchain_core.tools import tool
+
+from deerflow.agents.middlewares.terminal_response_middleware import TerminalResponseMiddleware
+from deerflow.runtime.runs.worker import _extract_llm_error_fallback_message
+
+
+@tool
+def lookup_status() -> str:
+    """Return a deterministic tool result."""
+    return "tool completed"
+
+
+class _PostToolResponseModel(BaseChatModel):
+    responses: list[str]
+    call_count: int = 0
+    observed_messages: list[list[Any]] = []
+
+    @property
+    def _llm_type(self) -> str:
+        return "post-tool-response"
+
+    def bind_tools(self, tools, **kwargs):
+        return self
+
+    def _generate(self, messages, stop=None, run_manager=None, **kwargs):
+        self.observed_messages.append(list(messages))
+        self.call_count += 1
+        if self.call_count == 1:
+            message = AIMessage(
+                content="",
+                tool_calls=[{"id": "call-1", "name": "lookup_status", "args": {}}],
+                response_metadata={"finish_reason": "tool_calls"},
+            )
+        else:
+            message = AIMessage(
+                content=self.responses[self.call_count - 2],
+                response_metadata={"finish_reason": "stop"},
+            )
+        return ChatResult(generations=[ChatGeneration(message=message)])
+
+    async def _agenerate(self, messages, stop=None, run_manager=None, **kwargs):
+        return self._generate(messages, stop=stop, run_manager=run_manager, **kwargs)
+
+
+class _PerRunRetryBudgetModel(BaseChatModel):
+    call_count: int = 0
+    observed_messages: list[list[Any]] = []
+
+    @property
+    def _llm_type(self) -> str:
+        return "per-run-retry-budget"
+
+    def bind_tools(self, tools, **kwargs):
+        return self
+
+    def _generate(self, messages, stop=None, run_manager=None, **kwargs):
+        self.observed_messages.append(list(messages))
+        self.call_count += 1
+        if self.call_count == 1:
+            message = AIMessage(
+                content="",
+                tool_calls=[{"id": "call-budget-1", "name": "lookup_status", "args": {}}],
+                response_metadata={"finish_reason": "tool_calls"},
+            )
+        elif self.call_count == 2:
+            message = AIMessage(content="", response_metadata={"finish_reason": "stop"})
+        elif self.call_count == 3:
+            message = AIMessage(
+                content="I need one more status check.",
+                tool_calls=[{"id": "call-budget-2", "name": "lookup_status", "args": {}}],
+                response_metadata={"finish_reason": "tool_calls"},
+            )
+        else:
+            message = AIMessage(content="", response_metadata={"finish_reason": "stop"})
+        return ChatResult(generations=[ChatGeneration(message=message)])
+
+    async def _agenerate(self, messages, stop=None, run_manager=None, **kwargs):
+        return self._generate(messages, stop=stop, run_manager=run_manager, **kwargs)
+
+
+def _agent(model: BaseChatModel):
+    return create_agent(
+        model=model,
+        tools=[lookup_status],
+        middleware=[TerminalResponseMiddleware()],
+    )
+
+
+def _empty_terminal_messages(messages: list[Any]) -> list[AIMessage]:
+    return [message for message in messages if isinstance(message, AIMessage) and not message.tool_calls and not message.invalid_tool_calls and not str(message.content).strip()]
+
+
+def test_retries_empty_post_tool_response_once_and_returns_model_answer():
+    model = _PostToolResponseModel(responses=["", "The tool completed successfully."])
+
+    result = _agent(model).invoke(
+        {"messages": [HumanMessage(content="Check the status")]},
+        context={"thread_id": "thread-1", "run_id": "run-1"},
+    )
+
+    assert model.call_count == 3
+    final = result["messages"][-1]
+    assert isinstance(final, AIMessage)
+    assert final.content == "The tool completed successfully."
+    assert _empty_terminal_messages(result["messages"]) == []
+    assert any(isinstance(message, HumanMessage) and message.name == "terminal_response_recovery" and message.additional_kwargs.get("hide_from_ui") is True for message in model.observed_messages[-1])
+    assert not any(isinstance(message, HumanMessage) and message.name == "terminal_response_recovery" for message in result["messages"])
+
+
+def test_second_empty_post_tool_response_becomes_visible_error_fallback():
+    model = _PostToolResponseModel(responses=["", ""])
+
+    result = _agent(model).invoke(
+        {"messages": [HumanMessage(content="Check the status")]},
+        context={"thread_id": "thread-2", "run_id": "run-2"},
+    )
+
+    assert model.call_count == 3
+    final = result["messages"][-1]
+    assert isinstance(final, AIMessage)
+    assert "returned no final response" in str(final.content)
+    assert final.additional_kwargs["deerflow_error_fallback"] is True
+    assert _empty_terminal_messages(result["messages"]) == []
+    assert _extract_llm_error_fallback_message(result) == ("Model returned an empty terminal response after one retry")
+
+
+@pytest.mark.asyncio
+async def test_async_graph_retries_empty_post_tool_response_once():
+    model = _PostToolResponseModel(responses=["", "Recovered asynchronously."])
+
+    result = await _agent(model).ainvoke(
+        {"messages": [HumanMessage(content="Check the status")]},
+        context={"thread_id": "thread-async", "run_id": "run-async"},
+    )
+
+    assert model.call_count == 3
+    assert result["messages"][-1].content == "Recovered asynchronously."
+    assert _empty_terminal_messages(result["messages"]) == []
+
+
+def test_graph_with_thread_id_only_keeps_recovery_state_across_model_loop():
+    model = _PostToolResponseModel(responses=["", "Recovered without a run id."])
+
+    result = _agent(model).invoke(
+        {"messages": [HumanMessage(content="Check the status")]},
+        context={"thread_id": "thread-only"},
+    )
+
+    assert model.call_count == 3
+    assert result["messages"][-1].content == "Recovered without a run id."
+    assert _empty_terminal_messages(result["messages"]) == []
+
+
+def test_recovery_budget_is_once_per_run_even_when_retry_calls_another_tool():
+    model = _PerRunRetryBudgetModel()
+
+    result = _agent(model).invoke(
+        {"messages": [HumanMessage(content="Check the status twice")]},
+        context={"thread_id": "thread-budget", "run_id": "run-budget"},
+    )
+
+    assert model.call_count == 4
+    final = result["messages"][-1]
+    assert final.additional_kwargs["deerflow_error_fallback"] is True
+    assert _empty_terminal_messages(result["messages"]) == []
+    recovery_prompt_count = sum(1 for request_messages in model.observed_messages for message in request_messages if isinstance(message, HumanMessage) and message.name == "terminal_response_recovery")
+    assert recovery_prompt_count == 1
+
+
+def test_empty_response_without_tool_result_is_not_retried():
+    middleware = TerminalResponseMiddleware()
+    message = AIMessage(content="", response_metadata={"finish_reason": "stop"})
+    state = {"messages": [HumanMessage(content="Hello"), message]}
+    runtime = type("RuntimeStub", (), {"context": {"thread_id": "thread-3", "run_id": "run-3"}})()
+
+    assert middleware.after_model(state, runtime) is None
+
+
+def test_tool_call_intent_is_not_treated_as_empty_terminal_response():
+    middleware = TerminalResponseMiddleware()
+    message = AIMessage(
+        content="",
+        tool_calls=[{"id": "call-2", "name": "lookup_status", "args": {}}],
+        response_metadata={"finish_reason": "tool_calls"},
+    )
+    state = {"messages": [HumanMessage(content="Hello"), message]}
+    runtime = type("RuntimeStub", (), {"context": {"thread_id": "thread-4", "run_id": "run-4"}})()
+
+    assert middleware.after_model(state, runtime) is None
+
+
+@pytest.mark.parametrize(
+    "message",
+    [
+        AIMessage(content="", invalid_tool_calls=[{"id": "bad-1", "name": "lookup_status", "args": "{"}]),
+        AIMessage(content="", additional_kwargs={"function_call": {"name": "lookup_status", "arguments": "{}"}}),
+        AIMessage(content="", response_metadata={"finish_reason": "function_call"}),
+    ],
+)
+def test_invalid_or_legacy_tool_call_intent_is_not_treated_as_empty_terminal_response(message):
+    middleware = TerminalResponseMiddleware()
+    state = {"messages": [HumanMessage(content="Hello"), message]}
+    runtime = type("RuntimeStub", (), {"context": {"thread_id": "thread-5", "run_id": "run-5"}})()
+
+    assert middleware.after_model(state, runtime) is None
+
+
+def test_after_agent_clears_retry_state_for_the_run():
+    middleware = TerminalResponseMiddleware()
+    runtime = type("RuntimeStub", (), {"context": {"thread_id": "thread-6", "run_id": "run-6"}})()
+    empty_after_tool = {
+        "messages": [
+            HumanMessage(content="Check the status"),
+            ToolMessage(content="tool completed", tool_call_id="call-6"),
+            AIMessage(content="", response_metadata={"finish_reason": "stop"}),
+        ]
+    }
+
+    first = middleware.after_model(empty_after_tool, runtime)
+    assert first is not None and first["jump_to"] == "model"
+    middleware.after_agent(empty_after_tool, runtime)
+    second = middleware.after_model(empty_after_tool, runtime)
+    assert second is not None and second["jump_to"] == "model"
+
+
+def test_before_agent_clears_same_run_state_for_resumed_invocation():
+    middleware = TerminalResponseMiddleware()
+    runtime = type("RuntimeStub", (), {"context": {"thread_id": "thread-7", "run_id": "run-7"}})()
+    empty_after_tool = {
+        "messages": [
+            HumanMessage(content="Check the status"),
+            ToolMessage(content="tool completed", tool_call_id="call-7"),
+            AIMessage(content="", response_metadata={"finish_reason": "stop"}),
+        ]
+    }
+
+    first = middleware.after_model(empty_after_tool, runtime)
+    assert first is not None and first["jump_to"] == "model"
+    middleware.before_agent(empty_after_tool, runtime)
+    resumed = middleware.after_model(empty_after_tool, runtime)
+    assert resumed is not None and resumed["jump_to"] == "model"
+
+
+def test_tool_history_without_real_user_message_does_not_trigger_recovery():
+    middleware = TerminalResponseMiddleware()
+    runtime = type("RuntimeStub", (), {"context": {"thread_id": "thread-8", "run_id": "run-8"}})()
+    state = {
+        "messages": [
+            HumanMessage(content="internal", additional_kwargs={"hide_from_ui": True}),
+            ToolMessage(content="tool completed", tool_call_id="call-8"),
+            AIMessage(content="", response_metadata={"finish_reason": "stop"}),
+        ]
+    }
+
+    assert middleware.after_model(state, runtime) is None
+
+
+def test_abandoned_run_state_is_bounded():
+    middleware = TerminalResponseMiddleware()
+
+    for index in range(1001):
+        key = (f"thread-{index}", f"run-{index}")
+        middleware._retry_counts[key] = 1
+        middleware._pending_prompts[key] = True
+
+    assert len(middleware._retry_counts) == 1000
+    assert len(middleware._pending_prompts) == 1000
+    assert ("thread-0", "run-0") not in middleware._retry_counts
+    assert ("thread-0", "run-0") not in middleware._pending_prompts