Merge #3823 (tool iteration limit terminal state) onto master

# Conflicts:
#	CHANGELOG.md

CHANGELOG.md

@@ -3,6 +3,12 @@
 
 ## [Unreleased]
 
+## [v0.51.380] — 2026-06-13 — Release MS (tool-iteration-limit stops surfaced explicitly, #3821)
+
+### Fixed
+
+- **Tool iteration-limit stops are now surfaced explicitly instead of looking like a normal user turn (#3821).** When Hermes Agent stops a turn at the max tool-calling iteration budget, WebUI filters the agent's synthetic max-iteration summary prompt out of both the visible transcript and the model-facing context (so it isn't persisted or replayed as if the human typed it), marks a usable final answer with a `tool_limit_reached` terminal state, and shows a no-final terminal error when the limit fires before any assistant answer is available. Detection reads current-turn result metadata only (not conversation history), and the status annotation is gated behind the terminal-failure check so a genuine failure still takes the error path. (#3821)
+
 ## [v0.51.379] — 2026-06-13 — Release MR (Worklog detail collapse survives live refresh, #4062)
 
 ### Fixed
@@ -106,6 +112,7 @@
 ### Added
 
 - **Slice 2 of the Stable Assistant Turn Anchors foundation: a source-event normalizer (#3980, #3926).** Adds `normalizeAssistantTurnAnchorSourceEvent` / `normalizeAssistantTurnAnchorSourceEvents` to the frozen `HermesAssistantTurnAnchors` global, converting live SSE-style, replay/journal, and settled-message events into a single anchor-normalized shape with stable identity and dedupe keys. The helper is **inert** — no `send()` / `attachLiveStream()` / `renderMessages()` / settlement / `S.messages` / `INFLIGHT` / DOM path consumes it yet — so there is zero rendering change in this release; it lands the model that Compact Worklog and the in-progress Transparent Stream work will later consume from one normalizer. Identity reads are own-property-only and payload shaping uses `Object.create(null)` with unsafe-key skipping, so prototype-pollution and inherited-identity payloads can't leak through. (#3980)
+
 ## [v0.51.365] — 2026-06-11 — Release MD (lineage-segment open + reasoning chip fixes)
 
 ### Fixed

api/run_journal.py

@@ -84,11 +84,17 @@ def _next_seq(path: Path) -> int:
 def _terminal_state_for_event(event_name: str, payload) -> str | None:
     name = str(event_name or "")
     if name == "done" or name == "stream_end":
+        if isinstance(payload, dict):
+            explicit_state = str(payload.get("terminal_state") or "").strip().lower()
+            if explicit_state in {"tool_limit_reached"}:
+                return explicit_state
         return "completed"
     if name == "cancel":
         return "interrupted-by-user"
     if name in {"apperror", "error"}:
         err_type = str((payload or {}).get("type") or "").strip().lower() if isinstance(payload, dict) else ""
+        if err_type == "tool_limit_reached":
+            return "tool_limit_reached"
         if err_type in {"cancelled", "canceled"}:
             return "interrupted-by-user"
         if err_type == "interrupted":

api/streaming.py

@@ -1020,6 +1020,86 @@ def _provider_error_payload(message: str, err_type: str, hint: str = '') -> dict
     return payload
 
 
+_MAX_ITERATION_SUMMARY_REQUEST = (
+    "You've reached the maximum number of tool-calling iterations allowed. "
+    "Please provide a final response summarizing what you've found and accomplished "
+    "so far, without calling any more tools."
+)
+
+
+def _is_synthetic_max_iteration_summary_request(message) -> bool:
+    """Return True for Hermes Agent's internal max-iteration summary prompt."""
+    if not isinstance(message, dict) or message.get('role') != 'user':
+        return False
+    text = " ".join(_message_text(message.get('content', '')).split())
+    expected = " ".join(_MAX_ITERATION_SUMMARY_REQUEST.split())
+    return text == expected
+
+
+def _drop_synthetic_max_iteration_summary_requests(messages, *, enabled: bool = True):
+    """Remove Agent-internal max-iteration summary prompts from WebUI state."""
+    if not enabled:
+        return list(messages or [])
+    return [
+        msg
+        for msg in list(messages or [])
+        if not _is_synthetic_max_iteration_summary_request(msg)
+    ]
+
+
+def _agent_result_tool_limit_reached(result) -> bool:
+    """Return True when current-turn metadata says the tool iteration cap fired."""
+    if not isinstance(result, dict):
+        return False
+    fields = [
+        result.get('turn_exit_reason'),
+        result.get('terminal_reason'),
+        result.get('status'),
+        result.get('state'),
+        result.get('error'),
+    ]
+    haystack = " ".join(str(value or '') for value in fields).lower()
+    if (
+        'max_iterations_reached' in haystack
+        or 'maximum number of tool-calling iterations' in haystack
+        or ('tool-calling iterations' in haystack and 'maximum' in haystack)
+    ):
+        return True
+    return False
+
+
+def _mark_latest_assistant_tool_limit_status(messages) -> bool:
+    """Annotate the latest usable assistant final answer as limit-stopped."""
+    for msg in reversed(list(messages or [])):
+        if not isinstance(msg, dict):
+            continue
+        if msg.get('_error') or msg.get('role') != 'assistant':
+            continue
+        content = msg.get('content')
+        if isinstance(content, list):
+            text = '\n'.join(
+                str(part.get('text') or part.get('content') or '')
+                for part in content
+                if isinstance(part, dict)
+            )
+        else:
+            text = str(content or '')
+        if msg.get('tool_calls') or not text.strip():
+            continue
+        msg['_terminal_state'] = 'tool_limit_reached'
+        msg['_terminal_reason'] = 'max_iterations'
+        msg.setdefault('_statusCard', {
+            'title': 'Tool iteration limit reached',
+            'subtitle': 'Stopped because the tool iteration limit was reached.',
+            'rows': [
+                {'label': 'State', 'value': 'Limit reached'},
+                {'label': 'Next step', 'value': 'Start a new turn to continue.'},
+            ],
+        })
+        return True
+    return False
+
+
 def _session_has_cancel_marker(session) -> bool:
     """Return True if a visible cancel/interrupted marker is already persisted."""
     for msg in reversed(getattr(session, 'messages', None) or []):
@@ -6599,7 +6679,12 @@ def _run_agent_streaming(
                             getattr(s, 'active_stream_id', None),
                         )
                         return
+                _tool_limit_reached = _agent_result_tool_limit_reached(result)
                 _result_messages = result.get('messages') or _previous_context_messages
+                _result_messages = _drop_synthetic_max_iteration_summary_requests(
+                    _result_messages,
+                    enabled=_tool_limit_reached,
+                )
                 if cancel_event.is_set():
                     _finalize_cancelled_turn(s, ephemeral=False)
                     try:
@@ -6647,7 +6732,6 @@ def _run_agent_streaming(
                             for _part in _raw_content:
                                 if isinstance(_part, dict) and isinstance(_part.get('text'), str):
                                     _part['text'] = _strip_xml_tool_calls(_part['text'])
-
                 # ── Handle context compression side effects ──
                 # If compression fired inside run_conversation, the agent may have
                 # rotated its session_id. Detect and fix the mismatch before any
@@ -6790,6 +6874,10 @@ def _run_agent_streaming(
                 )
                 _terminal_failure = (
                     _agent_result_terminal_failure(result)
+                    or (
+                        _tool_limit_reached
+                        and _session_lacks_final_assistant_answer(_all_result_messages)
+                    )
                     or (
                         not _token_sent
                         and _session_lacks_final_assistant_answer(_all_result_messages)
@@ -6797,6 +6885,8 @@ def _run_agent_streaming(
                 )
                 if _terminal_failure:
                     _assistant_added = False
+                elif _tool_limit_reached and not _session_lacks_final_assistant_answer(s.messages):
+                    _mark_latest_assistant_tool_limit_status(s.messages)
                 # _token_sent tracks whether on_token() was called (any streamed text)
                 if _terminal_failure or (not _assistant_added and not _token_sent):
                     if cancel_event.is_set():
@@ -6898,6 +6988,10 @@ def _run_agent_streaming(
                                 # Since we're in a flat block, directly run the
                                 # post-result merge logic here.
                                 _result_messages = result.get('messages') or _previous_context_messages
+                                _result_messages = _drop_synthetic_max_iteration_summary_requests(
+                                    _result_messages,
+                                    enabled=_agent_result_tool_limit_reached(result),
+                                )
                                 _next_context_messages = _restore_reasoning_metadata(
                                     _previous_context_messages,
                                     _result_messages,
@@ -6935,6 +7029,17 @@ def _run_agent_streaming(
                             'your API key is invalid. Run `hermes model` in your terminal to '
                             'update credentials, then restart the WebUI.'
                         )
+                    elif _tool_limit_reached:
+                        _err_label = 'Tool iteration limit reached'
+                        _err_type = 'tool_limit_reached'
+                        _err_hint = (
+                            'The agent reached its configured tool iteration limit before producing '
+                            'a final answer. Start a narrower follow-up or increase agent.max_turns.'
+                        )
+                        _err_str = (
+                            'The agent reached its configured tool iteration limit before producing '
+                            'a final answer.'
+                        )
                     else:
                         _err_label = _classification['label']
                         _err_type = _classification['type']
@@ -6972,6 +7077,8 @@ def _run_agent_streaming(
                             _error_message['provider_details_label'] = 'Cancellation details'
                         elif _err_type == 'interrupted':
                             _error_message['provider_details_label'] = 'Interruption details'
+                        elif _err_type == 'tool_limit_reached':
+                            _error_message['provider_details_label'] = 'Terminal state details'
                         s.messages.append(_error_message)
                         try:
                             s.save()
@@ -6985,6 +7092,9 @@ def _run_agent_streaming(
                         if _compression_continuation_session_id is not None:
                             _error_payload['new_session_id'] = _compression_continuation_session_id
                             _error_payload['continuation_session_id'] = _compression_continuation_session_id
+                        if _err_type == 'tool_limit_reached':
+                            _error_payload['terminal_state'] = 'tool_limit_reached'
+                            _error_payload['terminal_reason'] = 'max_iterations'
                         put('apperror', _error_payload)
                         # Legacy #373 source tests and clients look for the
                         # no_response type; #1765 keeps that type but improves
@@ -7679,7 +7789,11 @@ def _run_agent_streaming(
             except Exception as _goal_exc:
                 logger.debug("Goal continuation hook failed for session %s: %s", session_id, _goal_exc)
             raw_session = s.compact() | {'messages': s.messages, 'tool_calls': tool_calls}
-            put('done', {'session': redact_session_data(raw_session), 'usage': usage})
+            _done_payload = {'session': redact_session_data(raw_session), 'usage': usage}
+            if _tool_limit_reached:
+                _done_payload['terminal_state'] = 'tool_limit_reached'
+                _done_payload['terminal_reason'] = 'max_iterations'
+            put('done', _done_payload)
             # Emit one last metering packet for the live message-header TPS label.
             meter_stats = meter().get_stats()
             meter_stats['session_id'] = session_id

static/messages.js

@@ -3515,12 +3515,13 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){
           const isCancelled=d.type==='cancelled';
           const isInterrupted=d.type==='interrupted';
           const isCompressionExhausted=d.type==='compression_exhausted';
+          const isToolLimitReached=d.type==='tool_limit_reached';
           isRecoveryControlMessage=isInterrupted && (d.recovery_control===true || _streamRecoveryControlMessageText(d.message));
           const isNoResponse=d.type==='no_response'||d.type==='silent_failure';
-          const label=isCancelled?'Task cancelled':isInterrupted?'Response interrupted':isCompressionExhausted?'Context compression exhausted':isQuotaExhausted?'Out of credits':isRateLimit?'Rate limit reached':isGatewayAuthError?(typeof t==='function'?t('gateway_auth_label'):'Gateway authentication failed'):isAuthMismatch?(typeof t==='function'?t('provider_mismatch_label'):'Provider mismatch'):isModelNotFound?(typeof t==='function'?t('model_not_found_label'):'Model not found'):isNoResponse?'No response from provider':'Error';
+          const label=isCancelled?'Task cancelled':isInterrupted?'Response interrupted':isCompressionExhausted?'Context compression exhausted':isToolLimitReached?'Tool iteration limit reached':isQuotaExhausted?'Out of credits':isRateLimit?'Rate limit reached':isGatewayAuthError?(typeof t==='function'?t('gateway_auth_label'):'Gateway authentication failed'):isAuthMismatch?(typeof t==='function'?t('provider_mismatch_label'):'Provider mismatch'):isModelNotFound?(typeof t==='function'?t('model_not_found_label'):'Model not found'):isNoResponse?'No response from provider':'Error';
           const hint=d.hint?`\n\n*${d.hint}*`:'';
           const details=d.details?String(d.details).replace(/```/g,'`\u200b``'):'';
-          const detailsLabel=isCancelled?'Cancellation details':isInterrupted?'Interruption details':undefined;
+          const detailsLabel=isCancelled?'Cancellation details':isInterrupted?'Interruption details':isToolLimitReached?'Terminal state details':undefined;
           window._compressionUi=null;
           if(typeof clearCompressionUi==='function') clearCompressionUi();
           if(isRecoveryControlMessage){

tests/test_run_journal.py

@@ -108,11 +108,17 @@ def test_terminal_state_classification_distinguishes_crash_from_user_cancel(tmp_
     append_run_event("session_1", "run_cancelled", "cancel", {"message": "Cancelled by user"}, session_dir=tmp_path)
     append_run_event("session_1", "run_crashed", "apperror", {"type": "interrupted"}, session_dir=tmp_path)
     append_run_event("session_1", "run_failed", "apperror", {"type": "auth_mismatch"}, session_dir=tmp_path)
+    append_run_event("session_1", "run_tool_limit", "apperror", {"type": "tool_limit_reached"}, session_dir=tmp_path)
+    append_run_event("session_1", "run_tool_limit_done", "done", {"terminal_state": "tool_limit_reached"}, session_dir=tmp_path)
+    append_run_event("session_1", "run_unknown_done", "done", {"terminal_state": "future_unknown_state"}, session_dir=tmp_path)
     append_run_event("session_1", "run_done", "done", {"session": {}}, session_dir=tmp_path)
 
     assert latest_run_summary("session_1", "run_cancelled", session_dir=tmp_path)["terminal_state"] == "interrupted-by-user"
     assert latest_run_summary("session_1", "run_crashed", session_dir=tmp_path)["terminal_state"] == "interrupted-by-crash"
     assert latest_run_summary("session_1", "run_failed", session_dir=tmp_path)["terminal_state"] == "errored"
+    assert latest_run_summary("session_1", "run_tool_limit", session_dir=tmp_path)["terminal_state"] == "tool_limit_reached"
+    assert latest_run_summary("session_1", "run_tool_limit_done", session_dir=tmp_path)["terminal_state"] == "tool_limit_reached"
+    assert latest_run_summary("session_1", "run_unknown_done", session_dir=tmp_path)["terminal_state"] == "completed"
     assert latest_run_summary("session_1", "run_done", session_dir=tmp_path)["terminal_state"] == "completed"
 
 

tests/test_tool_limit_terminal_state.py

@@ -0,0 +1,349 @@
+import json
+import queue
+import sys
+import types
+from pathlib import Path
+
+from api import models
+from api import streaming
+from api.models import Session
+
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+def _run_streaming_with_fake_agent(
+    tmp_path,
+    monkeypatch,
+    agent_result,
+    *,
+    prior_messages=None,
+    prior_context_messages=None,
+):
+    session_dir = tmp_path / "sessions"
+    session_dir.mkdir()
+    monkeypatch.setattr(models, "SESSION_DIR", session_dir)
+    monkeypatch.setattr(models, "SESSION_INDEX_FILE", session_dir / "_index.json")
+    monkeypatch.setattr(streaming, "SESSION_DIR", session_dir)
+    models.SESSIONS.clear()
+    streaming.SESSIONS.clear()
+    streaming.STREAMS.clear()
+    streaming.AGENT_INSTANCES.clear()
+    streaming.SESSION_AGENT_LOCKS.clear()
+    streaming.PENDING_GOAL_CONTINUATION.clear()
+    try:
+        from api.config import SESSION_AGENT_CACHE
+
+        SESSION_AGENT_CACHE.clear()
+    except Exception:
+        pass
+
+    session_id = "tool_limit_session"
+    stream_id = "stream-tool-limit"
+    session = Session(
+        session_id=session_id,
+        title="Tool limit test",
+        workspace=str(tmp_path),
+        model="gpt-4o",
+        messages=list(prior_messages or []),
+        context_messages=list(prior_context_messages or []),
+    )
+    session.active_stream_id = stream_id
+    session.pending_user_message = "Do the long task."
+    session.pending_started_at = 1.0
+    session.save()
+    models.SESSIONS[session_id] = session
+    streaming.SESSIONS[session_id] = session
+    event_queue = queue.Queue()
+    streaming.STREAMS[stream_id] = event_queue
+
+    class FakeAgent:
+        def __init__(self, **kwargs):
+            self.session_id = kwargs.get("session_id")
+            self.stream_delta_callback = kwargs.get("stream_delta_callback")
+            self.context_compressor = None
+            self.session_prompt_tokens = 0
+            self.session_completion_tokens = 0
+            self.session_estimated_cost_usd = None
+            self.session_cache_read_tokens = 0
+            self.session_cache_write_tokens = 0
+            self.reasoning_config = None
+            self.ephemeral_system_prompt = None
+            self._last_error = None
+
+        def run_conversation(self, **kwargs):
+            return agent_result
+
+        def interrupt(self, _message):
+            return None
+
+    fake_hermes_state = types.ModuleType("hermes_state")
+    fake_hermes_state.SessionDB = lambda *_args, **_kwargs: object()
+
+    with monkeypatch.context() as m:
+        m.setattr(streaming, "get_session", lambda _sid: session)
+        m.setattr(streaming, "_get_ai_agent", lambda: FakeAgent)
+        m.setattr(streaming, "resolve_model_provider", lambda *_args, **_kwargs: ("gpt-4o", "openai", None))
+        m.setattr("api.config.get_config", lambda *_args, **_kwargs: {})
+        m.setattr("api.config._resolve_cli_toolsets", lambda *_args, **_kwargs: [])
+        m.setitem(sys.modules, "hermes_state", fake_hermes_state)
+        streaming._run_agent_streaming(
+            session_id=session_id,
+            msg_text="Do the long task.",
+            model="gpt-4o",
+            workspace=str(tmp_path),
+            stream_id=stream_id,
+        )
+
+    events = []
+    while not event_queue.empty():
+        events.append(event_queue.get_nowait())
+    payload = json.loads((session_dir / f"{session_id}.json").read_text(encoding="utf-8"))
+    return events, payload
+
+
+def test_synthetic_max_iteration_summary_request_is_dropped_from_agent_result():
+    synthetic = {
+        "role": "user",
+        "content": streaming._MAX_ITERATION_SUMMARY_REQUEST,
+    }
+    messages = [
+        {"role": "user", "content": "Do the long task."},
+        {"role": "assistant", "content": "", "tool_calls": [{"id": "call_1"}]},
+        {"role": "tool", "tool_call_id": "call_1", "content": "result"},
+        synthetic,
+        {"role": "assistant", "content": "I reached the limit; here is the summary."},
+    ]
+    result = {
+        "turn_exit_reason": "max_iterations_reached(30/30)",
+        "messages": messages,
+    }
+
+    assert streaming._agent_result_tool_limit_reached(result) is True
+
+    cleaned = streaming._drop_synthetic_max_iteration_summary_requests(
+        result["messages"],
+        enabled=streaming._agent_result_tool_limit_reached(result),
+    )
+
+    assert synthetic not in cleaned
+    assert cleaned[-1]["role"] == "assistant"
+    assert "here is the summary" in cleaned[-1]["content"]
+
+
+def test_tool_limit_detection_uses_explicit_boolean_grouping():
+    streaming_py = (ROOT / "api" / "streaming.py").read_text(encoding="utf-8")
+
+    assert "or ('tool-calling iterations' in haystack and 'maximum' in haystack)" in streaming_py
+
+
+def test_historical_synthetic_summary_prompt_does_not_mark_normal_result_as_tool_limit():
+    result = {
+        "messages": [
+            {"role": "user", "content": "Earlier task."},
+            {"role": "user", "content": streaming._MAX_ITERATION_SUMMARY_REQUEST},
+            {"role": "user", "content": "Current normal task."},
+            {"role": "assistant", "content": "Current task completed normally."},
+        ],
+    }
+
+    assert streaming._agent_result_tool_limit_reached(result) is False
+
+
+def test_tool_limit_with_final_answer_marks_latest_assistant_status_card():
+    messages = [
+        {"role": "user", "content": "Do the long task."},
+        {"role": "assistant", "content": "I reached the limit; here is the summary."},
+    ]
+
+    assert streaming._session_lacks_final_assistant_answer(messages) is False
+    assert streaming._mark_latest_assistant_tool_limit_status(messages) is True
+
+    assistant = messages[-1]
+    assert assistant["_terminal_state"] == "tool_limit_reached"
+    assert assistant["_terminal_reason"] == "max_iterations"
+    assert assistant["_statusCard"]["title"] == "Tool iteration limit reached"
+
+
+def test_tool_limit_without_final_answer_is_no_final_terminal_state_after_filtering():
+    messages = [
+        {"role": "user", "content": "Do the long task."},
+        {"role": "assistant", "content": "", "tool_calls": [{"id": "call_1"}]},
+        {"role": "tool", "tool_call_id": "call_1", "content": "result"},
+        {"role": "user", "content": streaming._MAX_ITERATION_SUMMARY_REQUEST},
+    ]
+
+    cleaned = streaming._drop_synthetic_max_iteration_summary_requests(messages)
+
+    assert all(
+        not streaming._is_synthetic_max_iteration_summary_request(message)
+        for message in cleaned
+    )
+    assert streaming._session_lacks_final_assistant_answer(cleaned) is True
+
+
+def test_display_merge_does_not_render_synthetic_summary_prompt():
+    previous_display = [{"role": "user", "content": "Do the long task."}]
+    previous_context = [{"role": "user", "content": "Do the long task."}]
+    result_messages = previous_context + [
+        {"role": "assistant", "content": "", "tool_calls": [{"id": "call_1"}]},
+        {"role": "tool", "tool_call_id": "call_1", "content": "result"},
+        {"role": "user", "content": streaming._MAX_ITERATION_SUMMARY_REQUEST},
+        {"role": "assistant", "content": "I reached the limit; here is the summary."},
+    ]
+    result_messages = streaming._drop_synthetic_max_iteration_summary_requests(
+        result_messages,
+        enabled=True,
+    )
+
+    merged = streaming._merge_display_messages_after_agent_result(
+        previous_display,
+        previous_context,
+        result_messages,
+        "Do the long task.",
+    )
+
+    assert all(
+        message.get("content") != streaming._MAX_ITERATION_SUMMARY_REQUEST
+        for message in merged
+    )
+    assert merged[-1]["role"] == "assistant"
+    assert "here is the summary" in merged[-1]["content"]
+
+
+def test_frontend_handles_tool_limit_apperror_label():
+    messages_js = (ROOT / "static" / "messages.js").read_text(encoding="utf-8")
+    start = messages_js.find("source.addEventListener('apperror'")
+    end = messages_js.find("source.addEventListener('warning'", start)
+    assert start != -1 and end != -1
+    block = messages_js[start:end]
+
+    assert "const isToolLimitReached=d.type==='tool_limit_reached';" in block
+    assert "Tool iteration limit reached" in block
+    assert "Terminal state details" in block
+
+
+def test_streaming_tool_limit_with_final_answer_persists_clean_done_state(tmp_path, monkeypatch):
+    result = {
+        "turn_exit_reason": "max_iterations_reached(30/30)",
+        "messages": [
+            {"role": "user", "content": "Do the long task."},
+            {"role": "assistant", "content": "", "tool_calls": [{"id": "call_1"}]},
+            {"role": "tool", "tool_call_id": "call_1", "content": "result"},
+            {"role": "user", "content": streaming._MAX_ITERATION_SUMMARY_REQUEST},
+            {"role": "assistant", "content": "I reached the limit; here is the summary."},
+        ],
+    }
+
+    events, payload = _run_streaming_with_fake_agent(tmp_path, monkeypatch, result)
+
+    done_payloads = [payload for event, payload in events if event == "done"]
+    assert done_payloads, "expected done SSE payload"
+    assert done_payloads[-1]["terminal_state"] == "tool_limit_reached"
+    assert done_payloads[-1]["terminal_reason"] == "max_iterations"
+    assert all(
+        message.get("content") != streaming._MAX_ITERATION_SUMMARY_REQUEST
+        for message in payload["messages"]
+    )
+    assert all(
+        message.get("content") != streaming._MAX_ITERATION_SUMMARY_REQUEST
+        for message in payload["context_messages"]
+    )
+    assistant = payload["messages"][-1]
+    assert assistant["role"] == "assistant"
+    assert assistant["_terminal_state"] == "tool_limit_reached"
+    assert assistant["_statusCard"]["title"] == "Tool iteration limit reached"
+
+
+def test_streaming_tool_limit_without_final_answer_emits_no_final_apperror(tmp_path, monkeypatch):
+    result = {
+        "turn_exit_reason": "max_iterations_reached(30/30)",
+        "messages": [
+            {"role": "user", "content": "Do the long task."},
+            {"role": "assistant", "content": "", "tool_calls": [{"id": "call_1"}]},
+            {"role": "tool", "tool_call_id": "call_1", "content": "result"},
+            {"role": "user", "content": streaming._MAX_ITERATION_SUMMARY_REQUEST},
+        ],
+    }
+
+    events, payload = _run_streaming_with_fake_agent(tmp_path, monkeypatch, result)
+
+    apperror_payloads = [payload for event, payload in events if event == "apperror"]
+    assert apperror_payloads, "expected apperror SSE payload"
+    assert apperror_payloads[-1]["type"] == "tool_limit_reached"
+    assert apperror_payloads[-1]["terminal_state"] == "tool_limit_reached"
+    assert payload["messages"][-1]["_error"] is True
+    assert "Tool iteration limit reached" in payload["messages"][-1]["content"]
+    assert all(
+        message.get("content") != streaming._MAX_ITERATION_SUMMARY_REQUEST
+        for message in payload["messages"]
+    )
+
+
+def test_streaming_tool_limit_terminal_failure_does_not_mark_final_answer(tmp_path, monkeypatch):
+    result = {
+        "status": "partial",
+        "turn_exit_reason": "max_iterations_reached(30/30)",
+        "messages": [
+            {"role": "user", "content": "Do the long task."},
+            {"role": "assistant", "content": "I reached the limit; here is the summary."},
+        ],
+    }
+
+    events, payload = _run_streaming_with_fake_agent(tmp_path, monkeypatch, result)
+
+    apperror_payloads = [payload for event, payload in events if event == "apperror"]
+    assert apperror_payloads, "expected terminal-failure apperror"
+    assert apperror_payloads[-1]["type"] == "tool_limit_reached"
+    assert not [payload for event, payload in events if event == "done"]
+    assistant = next(
+        message
+        for message in payload["messages"]
+        if message.get("role") == "assistant"
+        and message.get("content") == "I reached the limit; here is the summary."
+    )
+    assert "_terminal_state" not in assistant
+    assert "_statusCard" not in assistant
+    assert payload["messages"][-1]["_error"] is True
+
+
+def test_streaming_historical_synthetic_prompt_normal_result_does_not_emit_tool_limit(tmp_path, monkeypatch):
+    result = {
+        "messages": [
+            {"role": "user", "content": "Earlier task."},
+            {"role": "user", "content": streaming._MAX_ITERATION_SUMMARY_REQUEST},
+            {"role": "user", "content": "Do the long task."},
+            {"role": "assistant", "content": "Current task completed normally."},
+        ],
+    }
+
+    events, payload = _run_streaming_with_fake_agent(tmp_path, monkeypatch, result)
+
+    assert not [payload for event, payload in events if event == "apperror"]
+    done_payloads = [payload for event, payload in events if event == "done"]
+    assert done_payloads, "expected normal done SSE payload"
+    assert "terminal_state" not in done_payloads[-1]
+    assert payload["messages"][-1]["role"] == "assistant"
+    assert payload["messages"][-1]["content"] == "Current task completed normally."
+
+
+def test_streaming_empty_result_messages_do_not_treat_prior_assistant_as_current_answer(tmp_path, monkeypatch):
+    prior = [
+        {"role": "user", "content": "Earlier task."},
+        {"role": "assistant", "content": "Earlier answer."},
+    ]
+    result = {"messages": []}
+
+    events, payload = _run_streaming_with_fake_agent(
+        tmp_path,
+        monkeypatch,
+        result,
+        prior_messages=prior,
+        prior_context_messages=prior,
+    )
+
+    apperror_payloads = [payload for event, payload in events if event == "apperror"]
+    assert apperror_payloads, "expected silent-failure apperror"
+    assert apperror_payloads[-1]["type"] == "no_response"
+    assert not [payload for event, payload in events if event == "done"]
+    assert payload["messages"][-1]["_error"] is True