fix(agent): hint repeated tool results

nanobot/agent/loop.py

@@ -55,14 +55,13 @@ from nanobot.security.workspace_access import (
     bind_workspace_scope,
     reset_workspace_scope,
 )
-from nanobot.session import turn_continuation, turn_history
+from nanobot.session import turn_continuation
 from nanobot.session.automation_turns import automation_history_overrides
 from nanobot.session.goal_state import (
     goal_state_runtime_lines,
     runner_wall_llm_timeout_s,
     sustained_goal_active,
 )
-from nanobot.session.history_visibility import HIDDEN_HISTORY_META
 from nanobot.session.keys import UNIFIED_SESSION_KEY, session_key_for_channel
 from nanobot.session.manager import (
     Session,
@@ -71,6 +70,8 @@ from nanobot.session.manager import (
 )
 from nanobot.triggers.local_turns import LocalTriggerTurnCoordinator
 from nanobot.utils.document import extract_documents, reference_non_image_attachments
+from nanobot.utils.helpers import image_placeholder_text
+from nanobot.utils.helpers import truncate_text as truncate_text_fn
 from nanobot.utils.image_generation_intent import image_generation_prompt
 from nanobot.utils.llm_runtime import LLMRuntime
 from nanobot.utils.runtime import (
@@ -173,8 +174,8 @@ class AgentLoop:
         self._refresh_provider_snapshot()
         return LLMRuntime(self.provider, self.model)
 
-    _RUNTIME_CHECKPOINT_KEY = turn_history.RUNTIME_CHECKPOINT_KEY
+    _RUNTIME_CHECKPOINT_KEY = "runtime_checkpoint"
-    _PENDING_USER_TURN_KEY = turn_history.PENDING_USER_TURN_KEY
+    _PENDING_USER_TURN_KEY = "pending_user_turn"
 
     # Event-driven state transition table.
     # Handlers return an event string; the driver looks up the next state here.
@@ -794,20 +795,7 @@ class AgentLoop:
                     content, media = self._prepare_message_media(content, media)
                     media = media or None
                 user_content = self.context._build_user_content(content, media)
-                row: dict[str, Any] = {"role": "user", "content": user_content}
+                return {"role": "user", "content": user_content}
-                metadata = pending_msg.metadata if isinstance(pending_msg.metadata, dict) else {}
-                if (
-                    pending_msg.sender_id == "subagent"
-                    and metadata.get("injected_event") == "subagent_result"
-                ):
-                    marker: dict[str, Any] = {"kind": "subagent_result"}
-                    task_id = metadata.get("subagent_task_id")
-                    if isinstance(task_id, str) and task_id:
-                        marker["subagent_task_id"] = task_id
-                        row["subagent_task_id"] = task_id
-                    row[HIDDEN_HISTORY_META] = marker
-                    row["injected_event"] = "subagent_result"
-                return row
 
             items: list[dict[str, Any]] = []
             while len(items) < limit:
@@ -1652,13 +1640,38 @@ class AgentLoop:
         should_truncate_text: bool = False,
         drop_runtime: bool = False,
     ) -> list[dict[str, Any]]:
-        return turn_history.sanitize_persisted_blocks(
+        """Strip volatile multimodal payloads before writing session history."""
-            content,
+        filtered: list[dict[str, Any]] = []
-            max_tool_result_chars=self.max_tool_result_chars,
+        for block in content:
-            runtime_context_tag=ContextBuilder._RUNTIME_CONTEXT_TAG,
+            if not isinstance(block, dict):
-            should_truncate_text=should_truncate_text,
+                filtered.append(block)
-            drop_runtime=drop_runtime,
+                continue
-        )
+
+            if (
+                drop_runtime
+                and block.get("type") == "text"
+                and isinstance(block.get("text"), str)
+                and block["text"].startswith(ContextBuilder._RUNTIME_CONTEXT_TAG)
+            ):
+                continue
+
+            if block.get("type") == "image_url" and block.get("image_url", {}).get(
+                "url", ""
+            ).startswith("data:image/"):
+                path = (block.get("_meta") or {}).get("path", "")
+                filtered.append({"type": "text", "text": image_placeholder_text(path)})
+                continue
+
+            if block.get("type") == "text" and isinstance(block.get("text"), str):
+                text = block["text"]
+                if should_truncate_text and len(text) > self.max_tool_result_chars:
+                    text = truncate_text_fn(text, self.max_tool_result_chars)
+                filtered.append({**block, "text": text})
+                continue
+
+            filtered.append(block)
+
+        return filtered
 
     def _save_turn(
         self,
@@ -1668,36 +1681,193 @@ class AgentLoop:
         *,
         turn_latency_ms: int | None = None,
     ) -> None:
-        turn_history.save_turn(
+        """Save new-turn messages into session, truncating large tool results."""
-            session,
+        from datetime import datetime
-            messages,
+
-            skip,
+        declared_tool_call_ids = {
-            max_tool_result_chars=self.max_tool_result_chars,
+            str(tc["id"])
-            runtime_context_tag=ContextBuilder._RUNTIME_CONTEXT_TAG,
+            for m in session.messages
-            turn_latency_ms=turn_latency_ms,
+            if m.get("role") == "assistant"
-        )
+            for tc in m.get("tool_calls") or []
+            if isinstance(tc, dict) and tc.get("id")
+        }
+        last_assistant_idx: int | None = None
+        for m in messages[skip:]:
+            entry = dict(m)
+            role, content = entry.get("role"), entry.get("content")
+            if role == "assistant" and not content and not entry.get("tool_calls"):
+                continue  # skip empty assistant messages — they poison session context
+            if role == "tool":
+                tool_call_id = entry.get("tool_call_id")
+                if not tool_call_id or str(tool_call_id) not in declared_tool_call_ids:
+                    # Undeclared tool results corrupt future provider requests.
+                    logger.warning(
+                        "Dropping orphaned tool result {} from session {} during persistence",
+                        tool_call_id or "(missing id)",
+                        session.key,
+                    )
+                    continue
+                if isinstance(content, str) and len(content) > self.max_tool_result_chars:
+                    entry["content"] = truncate_text_fn(content, self.max_tool_result_chars)
+                elif isinstance(content, list):
+                    filtered = self._sanitize_persisted_blocks(content, should_truncate_text=True)
+                    if not filtered:
+                        # Preserve the tool_call/result pair after block filtering.
+                        filtered = [
+                            {"type": "text", "text": "[tool result omitted during persistence]"}
+                        ]
+                    entry["content"] = filtered
+            elif role == "user":
+                if isinstance(content, str) and ContextBuilder._RUNTIME_CONTEXT_TAG in content:
+                    # Strip the runtime-context block appended at the end.
+                    tag_pos = content.find(ContextBuilder._RUNTIME_CONTEXT_TAG)
+                    before = content[:tag_pos].rstrip("\n ")
+                    if before:
+                        entry["content"] = before
+                    else:
+                        continue
+                if isinstance(content, list):
+                    filtered = self._sanitize_persisted_blocks(content, drop_runtime=True)
+                    if not filtered:
+                        continue
+                    entry["content"] = filtered
+            entry.setdefault("timestamp", datetime.now().isoformat())
+            session.messages.append(entry)
+            if role == "assistant":
+                last_assistant_idx = len(session.messages) - 1
+                declared_tool_call_ids.update(
+                    str(tc["id"])
+                    for tc in entry.get("tool_calls") or []
+                    if isinstance(tc, dict) and tc.get("id")
+                )
+        if turn_latency_ms is not None and last_assistant_idx is not None:
+            session.messages[last_assistant_idx]["latency_ms"] = int(turn_latency_ms)
+        session.updated_at = datetime.now()
 
     def _persist_subagent_followup(self, session: Session, msg: InboundMessage) -> bool:
-        return turn_history.persist_subagent_followup(session, msg)
+        """Persist subagent follow-ups before prompt assembly so history stays durable.
+
+        Returns True if a new entry was appended; False if the follow-up was
+        deduped (same ``subagent_task_id`` already in session) or carries no
+        content worth persisting.
+        """
+        if not msg.content:
+            return False
+        task_id = msg.metadata.get("subagent_task_id") if isinstance(msg.metadata, dict) else None
+        if task_id and any(
+            m.get("injected_event") == "subagent_result" and m.get("subagent_task_id") == task_id
+            for m in session.messages
+        ):
+            return False
+        session.add_message(
+            "assistant",
+            msg.content,
+            sender_id=msg.sender_id,
+            injected_event="subagent_result",
+            subagent_task_id=task_id,
+        )
+        return True
 
     def _set_runtime_checkpoint(self, session: Session, payload: dict[str, Any]) -> None:
-        turn_history.set_runtime_checkpoint(session, payload)
+        """Persist the latest in-flight turn state into session metadata."""
+        session.metadata[self._RUNTIME_CHECKPOINT_KEY] = payload
         self.sessions.save(session)
 
     def _mark_pending_user_turn(self, session: Session) -> None:
-        turn_history.mark_pending_user_turn(session)
+        session.metadata[self._PENDING_USER_TURN_KEY] = True
 
     def _clear_pending_user_turn(self, session: Session) -> None:
-        turn_history.clear_pending_user_turn(session)
+        session.metadata.pop(self._PENDING_USER_TURN_KEY, None)
 
     def _clear_runtime_checkpoint(self, session: Session) -> None:
-        turn_history.clear_runtime_checkpoint(session)
+        if self._RUNTIME_CHECKPOINT_KEY in session.metadata:
+            session.metadata.pop(self._RUNTIME_CHECKPOINT_KEY, None)
+
+    @staticmethod
+    def _checkpoint_message_key(message: dict[str, Any]) -> tuple[Any, ...]:
+        return (
+            message.get("role"),
+            message.get("content"),
+            message.get("tool_call_id"),
+            message.get("name"),
+            message.get("tool_calls"),
+            message.get("reasoning_content"),
+            message.get("thinking_blocks"),
+        )
 
     def _restore_runtime_checkpoint(self, session: Session) -> bool:
-        return turn_history.restore_runtime_checkpoint(session)
+        """Materialize an unfinished turn into session history before a new request."""
+        from datetime import datetime
+
+        checkpoint = session.metadata.get(self._RUNTIME_CHECKPOINT_KEY)
+        if not isinstance(checkpoint, dict):
+            return False
+
+        assistant_message = checkpoint.get("assistant_message")
+        completed_tool_results = checkpoint.get("completed_tool_results") or []
+        pending_tool_calls = checkpoint.get("pending_tool_calls") or []
+
+        restored_messages: list[dict[str, Any]] = []
+        if isinstance(assistant_message, dict):
+            restored = dict(assistant_message)
+            restored.setdefault("timestamp", datetime.now().isoformat())
+            restored_messages.append(restored)
+        for message in completed_tool_results:
+            if isinstance(message, dict):
+                restored = dict(message)
+                restored.setdefault("timestamp", datetime.now().isoformat())
+                restored_messages.append(restored)
+        for tool_call in pending_tool_calls:
+            if not isinstance(tool_call, dict):
+                continue
+            tool_id = tool_call.get("id")
+            name = ((tool_call.get("function") or {}).get("name")) or "tool"
+            restored_messages.append(
+                {
+                    "role": "tool",
+                    "tool_call_id": tool_id,
+                    "name": name,
+                    "content": "Error: Task interrupted before this tool finished.",
+                    "timestamp": datetime.now().isoformat(),
+                }
+            )
+
+        overlap = 0
+        max_overlap = min(len(session.messages), len(restored_messages))
+        for size in range(max_overlap, 0, -1):
+            existing = session.messages[-size:]
+            restored = restored_messages[:size]
+            if all(
+                self._checkpoint_message_key(left) == self._checkpoint_message_key(right)
+                for left, right in zip(existing, restored)
+            ):
+                overlap = size
+                break
+        session.messages.extend(restored_messages[overlap:])
+
+        self._clear_pending_user_turn(session)
+        self._clear_runtime_checkpoint(session)
+        return True
 
     def _restore_pending_user_turn(self, session: Session) -> bool:
-        return turn_history.restore_pending_user_turn(session)
+        """Close a turn that only persisted the user message before crashing."""
+        from datetime import datetime
+
+        if not session.metadata.get(self._PENDING_USER_TURN_KEY):
+            return False
+
+        if session.messages and session.messages[-1].get("role") == "user":
+            session.messages.append(
+                {
+                    "role": "assistant",
+                    "content": "Error: Task interrupted before a response was generated.",
+                    "timestamp": datetime.now().isoformat(),
+                }
+            )
+            session.updated_at = datetime.now()
+
+        self._clear_pending_user_turn(session)
+        return True
 
     async def process_direct(
         self,

nanobot/agent/runner.py

@@ -20,7 +20,6 @@ from nanobot.agent.context_governance import (
 from nanobot.agent.hook import AgentHook, AgentHookContext, AgentRunHookContext
 from nanobot.agent.tools.registry import ToolRegistry, is_tool_error_result
 from nanobot.providers.base import LLMProvider, LLMResponse, ToolCallRequest
-from nanobot.session.history_visibility import is_hidden_history_message
 from nanobot.utils.file_edit_events import (
     StreamingFileEditTracker,
     build_file_edit_end_event,
@@ -53,6 +52,7 @@ from nanobot.utils.runtime import (
     build_length_recovery_message,
     is_blank_text,
     repeated_external_lookup_error,
+    repeated_tool_result_hint,
     repeated_workspace_violation_error,
 )
 
@@ -156,8 +156,6 @@ class AgentRunner:
                 messages
                 and injection.get("role") == "user"
                 and messages[-1].get("role") == "user"
-                and not is_hidden_history_message(injection)
-                and not is_hidden_history_message(messages[-1])
             ):
                 merged = dict(messages[-1])
                 merged["content"] = cls._merge_message_content(
@@ -354,6 +352,7 @@ class AgentRunner:
         stop_reason = "completed"
         tool_events: list[dict[str, str]] = []
         external_lookup_counts: dict[str, int] = {}
+        repeated_result_counts: dict[str, int] = {}
         # Per-turn throttle for repeated attempts against the same outside target.
         workspace_violation_counts: dict[str, int] = {}
         empty_content_retries = 0
@@ -471,17 +470,29 @@ class AgentRunner:
                 context.tool_results = list(results)
                 context.tool_events = list(new_events)
                 completed_tool_results: list[dict[str, Any]] = []
-                for tool_call, result in zip(response.tool_calls, results):
+                for tool_call, result, event in zip(response.tool_calls, results, new_events):
+                    content = self.context_governor.normalize_tool_result(
+                        governance_config,
+                        tool_call.id,
+                        tool_call.name,
+                        result,
+                    )
+                    if event.get("status") == "ok":
+                        result_hint = repeated_tool_result_hint(
+                            tool_call.name,
+                            content,
+                            repeated_result_counts,
+                        )
+                        if result_hint:
+                            if isinstance(content, str):
+                                content = content + result_hint
+                            elif isinstance(content, list):
+                                content = [*content, {"type": "text", "text": result_hint.strip()}]
                     tool_message = {
                         "role": "tool",
                         "tool_call_id": tool_call.id,
                         "name": tool_call.name,
-                        "content": self.context_governor.normalize_tool_result(
+                        "content": content,
-                            governance_config,
-                            tool_call.id,
-                            tool_call.name,
-                            result,
-                        ),
                     }
                     messages.append(tool_message)
                     completed_tool_results.append(tool_message)
@@ -1138,7 +1149,10 @@ class AgentRunner:
             if spec.concurrent_tools and len(batch) > 1:
                 batch_results = await asyncio.gather(*(
                     self._run_tool(
-                        spec, tool_call, external_lookup_counts, workspace_violation_counts,
+                        spec,
+                        tool_call,
+                        external_lookup_counts,
+                        workspace_violation_counts,
                     )
                     for tool_call in batch
                 ))
@@ -1147,7 +1161,10 @@ class AgentRunner:
                 batch_results = []
                 for tool_call in batch:
                     result = await self._run_tool(
-                        spec, tool_call, external_lookup_counts, workspace_violation_counts,
+                        spec,
+                        tool_call,
+                        external_lookup_counts,
+                        workspace_violation_counts,
                     )
                     tool_results.append(result)
                     batch_results.append(result)

nanobot/agent/tools/exec_session.py

@@ -128,15 +128,7 @@ class _ExecSession:
     ) -> _SessionPoll:
         self.last_access = time.monotonic()
         if yield_time_ms > 0 and self.process.returncode is None:
-            wait_s = min(yield_time_ms, MAX_YIELD_MS) / 1000
+            await asyncio.sleep(min(yield_time_ms, MAX_YIELD_MS) / 1000)
-            remaining_s = self.deadline - time.monotonic()
-            if remaining_s <= 0:
-                wait_s = 0
-            else:
-                wait_s = min(wait_s, remaining_s)
-            if wait_s > 0:
-                with suppress(asyncio.TimeoutError):
-                    await asyncio.wait_for(self.process.wait(), timeout=wait_s)
 
         if self.process.returncode is None and time.monotonic() >= self.deadline:
             self._timed_out = True

nanobot/session/history_visibility.py

@@ -1,22 +0,0 @@
-"""Visibility helpers for persisted session history messages."""
-
-from __future__ import annotations
-
-from collections.abc import Mapping
-from typing import Any
-
-from nanobot.session.automation_turns import is_automation_history_message
-
-HIDDEN_HISTORY_META = "_hidden_history"
-
-
-def _has_hidden_history_marker(message: Mapping[str, Any] | None) -> bool:
-    if not message:
-        return False
-    marker = message.get(HIDDEN_HISTORY_META)
-    return marker is True or isinstance(marker, Mapping)
-
-
-def is_hidden_history_message(message: Mapping[str, Any] | None) -> bool:
-    """True for persisted messages that should not be shown as chat turns."""
-    return _has_hidden_history_marker(message) or is_automation_history_message(message)

nanobot/session/manager.py

@@ -15,7 +15,6 @@ from typing import Any
 from loguru import logger
 
 from nanobot.config.paths import get_legacy_sessions_dir
-from nanobot.session.turn_history import PENDING_USER_TURN_KEY, RUNTIME_CHECKPOINT_KEY
 from nanobot.utils.helpers import (
     ensure_dir,
     estimate_message_tokens,
@@ -38,8 +37,8 @@ _SESSION_LIST_PREVIEW_MAX_RECORDS = 200
 _SESSION_LIST_PREVIEW_MAX_CHARS = 1_000_000
 _FORK_VOLATILE_METADATA_KEYS = {
     "goal_state",
-    PENDING_USER_TURN_KEY,
+    "pending_user_turn",
-    RUNTIME_CHECKPOINT_KEY,
+    "runtime_checkpoint",
     "thread_goal",
     "title",
     "title_user_edited",

nanobot/session/turn_history.py

@@ -1,266 +0,0 @@
-"""Turn history persistence and interrupted-turn recovery."""
-
-from __future__ import annotations
-
-from datetime import datetime
-from typing import TYPE_CHECKING, Any
-
-from loguru import logger
-
-from nanobot.utils.helpers import image_placeholder_text
-from nanobot.utils.helpers import truncate_text as truncate_text_fn
-
-if TYPE_CHECKING:
-    from nanobot.session.manager import Session
-
-RUNTIME_CHECKPOINT_KEY = "runtime_checkpoint"
-PENDING_USER_TURN_KEY = "pending_user_turn"
-
-
-def sanitize_persisted_blocks(
-    content: list[dict[str, Any]],
-    *,
-    max_tool_result_chars: int,
-    runtime_context_tag: str,
-    should_truncate_text: bool = False,
-    drop_runtime: bool = False,
-) -> list[dict[str, Any]]:
-    """Strip volatile multimodal payloads before writing session history."""
-    filtered: list[dict[str, Any]] = []
-    for block in content:
-        if not isinstance(block, dict):
-            filtered.append(block)
-            continue
-
-        if (
-            drop_runtime
-            and block.get("type") == "text"
-            and isinstance(block.get("text"), str)
-            and block["text"].startswith(runtime_context_tag)
-        ):
-            continue
-
-        if block.get("type") == "image_url" and block.get("image_url", {}).get(
-            "url", ""
-        ).startswith("data:image/"):
-            path = (block.get("_meta") or {}).get("path", "")
-            filtered.append({"type": "text", "text": image_placeholder_text(path)})
-            continue
-
-        if block.get("type") == "text" and isinstance(block.get("text"), str):
-            text = block["text"]
-            if should_truncate_text and len(text) > max_tool_result_chars:
-                text = truncate_text_fn(text, max_tool_result_chars)
-            filtered.append({**block, "text": text})
-            continue
-
-        filtered.append(block)
-
-    return filtered
-
-
-def save_turn(
-    session: Session,
-    messages: list[dict],
-    skip: int,
-    *,
-    max_tool_result_chars: int,
-    runtime_context_tag: str,
-    turn_latency_ms: int | None = None,
-) -> None:
-    """Save new-turn messages into session, truncating large tool results."""
-    declared_tool_call_ids = {
-        str(tc["id"])
-        for m in session.messages
-        if m.get("role") == "assistant"
-        for tc in m.get("tool_calls") or []
-        if isinstance(tc, dict) and tc.get("id")
-    }
-    last_assistant_idx: int | None = None
-    for m in messages[skip:]:
-        entry = dict(m)
-        role, content = entry.get("role"), entry.get("content")
-        if role == "assistant" and not content and not entry.get("tool_calls"):
-            continue  # skip empty assistant messages - they poison session context
-        if role == "tool":
-            tool_call_id = entry.get("tool_call_id")
-            if not tool_call_id or str(tool_call_id) not in declared_tool_call_ids:
-                # Undeclared tool results corrupt future provider requests.
-                logger.warning(
-                    "Dropping orphaned tool result {} from session {} during persistence",
-                    tool_call_id or "(missing id)",
-                    session.key,
-                )
-                continue
-            if isinstance(content, str) and len(content) > max_tool_result_chars:
-                entry["content"] = truncate_text_fn(content, max_tool_result_chars)
-            elif isinstance(content, list):
-                filtered = sanitize_persisted_blocks(
-                    content,
-                    max_tool_result_chars=max_tool_result_chars,
-                    runtime_context_tag=runtime_context_tag,
-                    should_truncate_text=True,
-                )
-                if not filtered:
-                    # Preserve the tool_call/result pair after block filtering.
-                    filtered = [
-                        {"type": "text", "text": "[tool result omitted during persistence]"}
-                    ]
-                entry["content"] = filtered
-        elif role == "user":
-            if isinstance(content, str) and runtime_context_tag in content:
-                # Strip the runtime-context block appended at the end.
-                tag_pos = content.find(runtime_context_tag)
-                before = content[:tag_pos].rstrip("\n ")
-                if before:
-                    entry["content"] = before
-                else:
-                    continue
-            if isinstance(content, list):
-                filtered = sanitize_persisted_blocks(
-                    content,
-                    max_tool_result_chars=max_tool_result_chars,
-                    runtime_context_tag=runtime_context_tag,
-                    drop_runtime=True,
-                )
-                if not filtered:
-                    continue
-                entry["content"] = filtered
-        entry.setdefault("timestamp", datetime.now().isoformat())
-        session.messages.append(entry)
-        if role == "assistant":
-            last_assistant_idx = len(session.messages) - 1
-            declared_tool_call_ids.update(
-                str(tc["id"])
-                for tc in entry.get("tool_calls") or []
-                if isinstance(tc, dict) and tc.get("id")
-            )
-    if turn_latency_ms is not None and last_assistant_idx is not None:
-        session.messages[last_assistant_idx]["latency_ms"] = int(turn_latency_ms)
-    session.updated_at = datetime.now()
-
-
-def persist_subagent_followup(session: Session, msg: Any) -> bool:
-    """Persist subagent follow-ups before prompt assembly so history stays durable.
-
-    Returns True if a new entry was appended; False if the follow-up was
-    deduped (same ``subagent_task_id`` already in session) or carries no
-    content worth persisting.
-    """
-    if not msg.content:
-        return False
-    task_id = msg.metadata.get("subagent_task_id") if isinstance(msg.metadata, dict) else None
-    if task_id and any(
-        m.get("injected_event") == "subagent_result" and m.get("subagent_task_id") == task_id
-        for m in session.messages
-    ):
-        return False
-    session.add_message(
-        "assistant",
-        msg.content,
-        sender_id=msg.sender_id,
-        injected_event="subagent_result",
-        subagent_task_id=task_id,
-    )
-    return True
-
-
-def set_runtime_checkpoint(session: Session, payload: dict[str, Any]) -> None:
-    """Persist the latest in-flight turn state into session metadata."""
-    session.metadata[RUNTIME_CHECKPOINT_KEY] = payload
-
-
-def mark_pending_user_turn(session: Session) -> None:
-    session.metadata[PENDING_USER_TURN_KEY] = True
-
-
-def clear_pending_user_turn(session: Session) -> None:
-    session.metadata.pop(PENDING_USER_TURN_KEY, None)
-
-
-def clear_runtime_checkpoint(session: Session) -> None:
-    session.metadata.pop(RUNTIME_CHECKPOINT_KEY, None)
-
-
-def checkpoint_message_key(message: dict[str, Any]) -> tuple[Any, ...]:
-    return (
-        message.get("role"),
-        message.get("content"),
-        message.get("tool_call_id"),
-        message.get("name"),
-        message.get("tool_calls"),
-        message.get("reasoning_content"),
-        message.get("thinking_blocks"),
-    )
-
-
-def restore_runtime_checkpoint(session: Session) -> bool:
-    """Materialize an unfinished turn into session history before a new request."""
-    checkpoint = session.metadata.get(RUNTIME_CHECKPOINT_KEY)
-    if not isinstance(checkpoint, dict):
-        return False
-
-    assistant_message = checkpoint.get("assistant_message")
-    completed_tool_results = checkpoint.get("completed_tool_results") or []
-    pending_tool_calls = checkpoint.get("pending_tool_calls") or []
-
-    restored_messages: list[dict[str, Any]] = []
-    if isinstance(assistant_message, dict):
-        restored = dict(assistant_message)
-        restored.setdefault("timestamp", datetime.now().isoformat())
-        restored_messages.append(restored)
-    for message in completed_tool_results:
-        if isinstance(message, dict):
-            restored = dict(message)
-            restored.setdefault("timestamp", datetime.now().isoformat())
-            restored_messages.append(restored)
-    for tool_call in pending_tool_calls:
-        if not isinstance(tool_call, dict):
-            continue
-        tool_id = tool_call.get("id")
-        name = ((tool_call.get("function") or {}).get("name")) or "tool"
-        restored_messages.append(
-            {
-                "role": "tool",
-                "tool_call_id": tool_id,
-                "name": name,
-                "content": "Error: Task interrupted before this tool finished.",
-                "timestamp": datetime.now().isoformat(),
-            }
-        )
-
-    overlap = 0
-    max_overlap = min(len(session.messages), len(restored_messages))
-    for size in range(max_overlap, 0, -1):
-        existing = session.messages[-size:]
-        restored = restored_messages[:size]
-        if all(
-            checkpoint_message_key(left) == checkpoint_message_key(right)
-            for left, right in zip(existing, restored)
-        ):
-            overlap = size
-            break
-    session.messages.extend(restored_messages[overlap:])
-
-    clear_pending_user_turn(session)
-    clear_runtime_checkpoint(session)
-    return True
-
-
-def restore_pending_user_turn(session: Session) -> bool:
-    """Close a turn that only persisted the user message before crashing."""
-    if not session.metadata.get(PENDING_USER_TURN_KEY):
-        return False
-
-    if session.messages and session.messages[-1].get("role") == "user":
-        session.messages.append(
-            {
-                "role": "assistant",
-                "content": "Error: Task interrupted before a response was generated.",
-                "timestamp": datetime.now().isoformat(),
-            }
-        )
-        session.updated_at = datetime.now()
-
-    clear_pending_user_turn(session)
-    return True

nanobot/session/webui_turns.py

@@ -31,8 +31,8 @@ from nanobot.bus.runtime_events import (
     TurnRunStatusChanged,
 )
 from nanobot.providers.base import LLMProvider
+from nanobot.session.automation_turns import is_automation_history_message
 from nanobot.session.goal_state import goal_state_ws_blob
-from nanobot.session.history_visibility import is_hidden_history_message
 from nanobot.session.manager import Session, SessionManager
 from nanobot.utils.helpers import strip_think, truncate_text
 from nanobot.utils.llm_runtime import LLMRuntime
@@ -77,7 +77,7 @@ def _title_inputs(session: Session) -> tuple[str, str]:
     for message in session.messages:
         if message.get("_command") is True:
             continue
-        if is_hidden_history_message(message):
+        if is_automation_history_message(message):
             continue
         role = message.get("role")
         content = message.get("content")

nanobot/utils/runtime.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import hashlib
 import re
 from pathlib import Path
 from typing import Any
@@ -10,7 +11,7 @@ from loguru import logger
 
 from nanobot.utils.helpers import stringify_text_blocks
 
-_MAX_REPEAT_EXTERNAL_LOOKUPS = 2
+_MAX_REPEAT_ATTEMPTS = 2
 
 # Third same-target workspace violation in a turn escalates to "stop retrying".
 _MAX_REPEAT_WORKSPACE_VIOLATIONS = 2
@@ -103,6 +104,14 @@ def external_lookup_signature(tool_name: str, arguments: Any) -> str | None:
     return None
 
 
+def _over_repeat_budget(signature: str | None, seen_counts: dict[str, int]) -> int | None:
+    if signature is None:
+        return None
+    count = seen_counts.get(signature, 0) + 1
+    seen_counts[signature] = count
+    return count if count > _MAX_REPEAT_ATTEMPTS else None
+
+
 def repeated_external_lookup_error(
     tool_name: str,
     arguments: Any,
@@ -110,11 +119,8 @@ def repeated_external_lookup_error(
 ) -> str | None:
     """Block repeated external lookups after a small retry budget."""
     signature = external_lookup_signature(tool_name, arguments)
-    if signature is None:
+    count = _over_repeat_budget(signature, seen_counts)
-        return None
+    if count is None:
-    count = seen_counts.get(signature, 0) + 1
-    seen_counts[signature] = count
-    if count <= _MAX_REPEAT_EXTERNAL_LOOKUPS:
         return None
     logger.warning(
         "Blocking repeated external lookup {} on attempt {}",
@@ -127,6 +133,33 @@ def repeated_external_lookup_error(
     )
 
 
+def repeated_tool_result_hint(
+    tool_name: str,
+    result: Any,
+    seen_counts: dict[str, int],
+) -> str | None:
+    """Hint when a successful tool keeps returning the exact same text in one turn."""
+    if isinstance(result, str):
+        text = result
+    elif isinstance(result, list):
+        text = stringify_text_blocks(result)
+    else:
+        text = None
+    if text is None:
+        return None
+    digest = hashlib.sha256(text.encode("utf-8", errors="replace")).hexdigest()
+    signature = f"tool_result:{tool_name}:{len(text)}:{digest}"
+    count = _over_repeat_budget(signature, seen_counts)
+    if count is None:
+        return None
+    logger.warning("Hinting repeated {} result on attempt {}", tool_name, count)
+    return (
+        f"\n\n[Repeated {tool_name} result: this exact output has already been "
+        "returned in this turn. Use the existing evidence, or change the tool input "
+        "if you need new information.]"
+    )
+
+
 # Workspace-boundary violations are soft errors, with per-target throttling.
 
 _OUTSIDE_PATH_PATTERN = re.compile(r"(?:^|[\s|>'\"])((?:/[^\s\"'>;|<]+)|(?:~[^\s\"'>;|<]+))")

nanobot/webui/session_automations.py

@@ -6,7 +6,7 @@ from collections.abc import Collection
 from typing import Any, Protocol
 
 from nanobot.cron.types import CronJob
-from nanobot.session.history_visibility import is_hidden_history_message
+from nanobot.session.automation_turns import is_automation_history_message
 from nanobot.session.manager import _message_preview_text
 from nanobot.triggers.local_types import LocalTrigger
 
@@ -328,7 +328,7 @@ def _session_preview(messages: Any) -> str:
     for message in messages:
         if not isinstance(message, dict):
             continue
-        if is_hidden_history_message(message):
+        if is_automation_history_message(message):
             continue
         text = _message_preview_text(message)
         if not text:

nanobot/webui/session_list_index.py

@@ -16,7 +16,7 @@ from typing import Any
 from loguru import logger
 
 from nanobot.config.paths import get_webui_dir
-from nanobot.session.history_visibility import is_hidden_history_message
+from nanobot.session.automation_turns import is_automation_history_message
 from nanobot.session.manager import (
     _SESSION_LIST_PREVIEW_MAX_CHARS,
     _SESSION_LIST_PREVIEW_MAX_RECORDS,
@@ -154,7 +154,7 @@ def _preview_from_messages(messages: list[dict[str, Any]]) -> str:
             or scanned_chars > _SESSION_LIST_PREVIEW_MAX_CHARS
         ):
             break
-        if is_hidden_history_message(item):
+        if is_automation_history_message(item):
             continue
         text = _message_preview_text(item)
         if not text:
@@ -216,7 +216,7 @@ def _latest_updated_at(stored: str | None, activity: str | None) -> str | None:
 
 
 def _visible_message_timestamp(item: dict[str, Any]) -> str | None:
-    if is_hidden_history_message(item):
+    if is_automation_history_message(item):
         return None
     if item.get("role") not in _VISIBLE_TRANSCRIPT_ROLES:
         return None
@@ -298,7 +298,7 @@ def _scan_session_row(session_manager: SessionManager, path: Path) -> dict[str,
                         continue
                     if item.get("_type") == "metadata":
                         continue
-                    if is_hidden_history_message(item):
+                    if is_automation_history_message(item):
                         continue
                     text = _message_preview_text(item)
                     if not text:

nanobot/webui/transcript.py

@@ -17,8 +17,7 @@ from urllib.parse import unquote, urlparse
 from loguru import logger
 
 from nanobot.config.paths import get_webui_dir
-from nanobot.session.automation_turns import is_automation_kind
+from nanobot.session.automation_turns import is_automation_history_message, is_automation_kind
-from nanobot.session.history_visibility import is_hidden_history_message
 from nanobot.session.manager import SessionManager
 from nanobot.webui.metadata import WEBUI_MESSAGE_SOURCE_METADATA_KEY, WEBUI_TURN_METADATA_KEY
 
@@ -783,7 +782,7 @@ def write_session_messages_as_transcript(
     target_chat_id = _chat_id_from_session_key(target_key)
     rows: list[dict[str, Any]] = []
     for msg in messages:
-        if is_hidden_history_message(msg):
+        if is_automation_history_message(msg):
             continue
         role = msg.get("role")
         content = msg.get("content")
@@ -855,28 +854,13 @@ def build_user_transcript_event(
     return event
 
 
-def _is_legacy_raw_subagent_result(message: dict[str, Any]) -> bool:
-    content = message.get("content")
-    if not isinstance(content, str):
-        return False
-    text = content.replace("\r\n", "\n").strip()
-    return (
-        text.startswith("[Subagent '")
-        and "\n\nTask:" in text
-        and "\n\nResult:" in text
-        and "Summarize this naturally" in text
-    )
-
-
 def _session_user_event(
     session_key: str,
     message: dict[str, Any],
 ) -> dict[str, Any] | None:
     if message.get("role") != "user":
         return None
-    if is_hidden_history_message(message):
+    if is_automation_history_message(message):
-        return None
-    if _is_legacy_raw_subagent_result(message):
         return None
     content = message.get("content")
     text = content if isinstance(content, str) else ""

tests/agent/test_runner_errors.py

@@ -103,48 +103,6 @@ async def test_llm_arrearage_error_surfaces_clear_message():
     assert result.final_content == _ARREARAGE_ERROR_MESSAGE
 
 
-@pytest.mark.asyncio
-@pytest.mark.parametrize(
-    ("finish_reason", "expected_stop_reason"),
-    [
-        ("refusal", "completed"),
-        ("content_filter", "completed"),
-        ("error", "error"),
-    ],
-)
-async def test_runner_ignores_tool_calls_when_finish_reason_blocks_execution(
-    finish_reason: str,
-    expected_stop_reason: str,
-):
-    """Provider/gateway-injected tool calls under terminal block reasons must not run."""
-    from nanobot.agent.runner import AgentRunSpec, AgentRunner
-
-    provider = MagicMock(spec=LLMProvider)
-    provider.chat_with_retry = AsyncMock(return_value=LLMResponse(
-        content="Request blocked by provider policy.",
-        finish_reason=finish_reason,
-        tool_calls=[ToolCallRequest(id="call_1", name="exec", arguments={"command": "echo nope"})],
-        usage={},
-    ))
-    tools = MagicMock()
-    tools.get_definitions.return_value = []
-    tools.execute = AsyncMock(return_value="should not run")
-
-    result = await AgentRunner(provider).run(AgentRunSpec(
-        initial_messages=[{"role": "user", "content": "run a command"}],
-        tools=tools,
-        model="test-model",
-        max_iterations=2,
-        max_tool_result_chars=_MAX_TOOL_RESULT_CHARS,
-    ))
-
-    tools.execute.assert_not_awaited()
-    assert result.stop_reason == expected_stop_reason
-    assert result.tools_used == []
-    assert result.final_content == "Request blocked by provider policy."
-    assert not any(msg.get("role") == "tool" for msg in result.messages)
-
-
 @pytest.mark.asyncio
 async def test_runner_tool_error_sets_final_content():
     from nanobot.agent.runner import AgentRunSpec, AgentRunner

tests/agent/test_runner_injections.py

@@ -465,68 +465,6 @@ async def test_loop_injected_followup_preserves_image_media(tmp_path):
     )
 
 
-@pytest.mark.asyncio
-async def test_subagent_pending_injection_is_hidden_history_and_not_merged(tmp_path):
-    from nanobot.agent.loop import AgentLoop
-    from nanobot.bus.events import InboundMessage
-    from nanobot.bus.queue import MessageBus
-    from nanobot.session.history_visibility import HIDDEN_HISTORY_META
-
-    bus = MessageBus()
-    provider = MagicMock()
-    provider.get_default_model.return_value = "test-model"
-    call_count = {"n": 0}
-
-    async def chat_with_retry(*, messages, **kwargs):
-        call_count["n"] += 1
-        if call_count["n"] == 1:
-            return LLMResponse(content="first answer", tool_calls=[], usage={})
-        return LLMResponse(content="second answer", tool_calls=[], usage={})
-
-    provider.chat_with_retry = chat_with_retry
-    loop = AgentLoop(bus=bus, provider=provider, workspace=tmp_path, model="test-model")
-    loop.tools.get_definitions = MagicMock(return_value=[])
-
-    payload = (
-        "[Subagent 'x' completed successfully]\n\n"
-        "Task: t\n\n"
-        "Result:\nr\n\n"
-        "Summarize this naturally for the user."
-    )
-    pending_queue = asyncio.Queue()
-    await pending_queue.put(InboundMessage(
-        channel="cli",
-        sender_id="user",
-        chat_id="c",
-        content="visible follow-up",
-    ))
-    await pending_queue.put(InboundMessage(
-        channel="system",
-        sender_id="subagent",
-        chat_id="cli:c",
-        content=payload,
-        metadata={"injected_event": "subagent_result", "subagent_task_id": "sub-1"},
-    ))
-
-    final_content, _, all_msgs, _, had_injections = await loop._run_agent_loop(
-        [{"role": "user", "content": "hello"}],
-        channel="cli",
-        chat_id="c",
-        pending_queue=pending_queue,
-    )
-
-    assert final_content == "second answer"
-    assert had_injections is True
-    assert call_count["n"] == 2
-    injected_users = [message for message in all_msgs if message.get("role") == "user"][-2:]
-    assert [message["content"] for message in injected_users] == ["visible follow-up", payload]
-    assert injected_users[1][HIDDEN_HISTORY_META] == {
-        "kind": "subagent_result",
-        "subagent_task_id": "sub-1",
-    }
-    assert injected_users[1]["injected_event"] == "subagent_result"
-
-
 @pytest.mark.asyncio
 async def test_runner_merges_multiple_injected_user_messages_without_losing_media():
     """Multiple injected follow-ups should not create lossy consecutive user messages."""

tests/agent/test_runner_tool_execution.py

@@ -465,3 +465,83 @@ async def test_runner_blocks_repeated_external_fetches():
         if msg.get("role") == "tool" and msg.get("tool_call_id") == "call_3"
     ][0]
     assert "repeated external lookup blocked" in blocked_tool_message["content"]
+
+
+@pytest.mark.asyncio
+async def test_runner_hints_repeated_tool_results():
+    provider = MagicMock()
+    captured_final_call: list[dict] = []
+    call_count = {"n": 0}
+
+    async def chat_with_retry(*, messages, **kwargs):
+        call_count["n"] += 1
+        if call_count["n"] <= 3:
+            return LLMResponse(
+                content="reading",
+                tool_calls=[ToolCallRequest(
+                    id=f"call_{call_count['n']}",
+                    name="grep",
+                    arguments={"pattern": "TODO", "path": "nanobot"},
+                )],
+                usage={},
+            )
+        captured_final_call[:] = messages
+        return LLMResponse(content="done", tool_calls=[], usage={})
+
+    provider.chat_with_retry = chat_with_retry
+    tools = MagicMock()
+    tools.get_definitions.return_value = []
+    tools.execute = AsyncMock(return_value="file content")
+
+    result = await AgentRunner(provider).run(AgentRunSpec(
+        initial_messages=[{"role": "user", "content": "review code"}],
+        tools=tools,
+        model="test-model",
+        max_iterations=4,
+        max_tool_result_chars=_MAX_TOOL_RESULT_CHARS,
+    ))
+
+    assert result.final_content == "done"
+    assert tools.execute.await_count == 3
+    hinted_tool_message = [
+        msg for msg in captured_final_call
+        if msg.get("role") == "tool" and msg.get("tool_call_id") == "call_3"
+    ][0]
+    assert "Repeated grep result" in hinted_tool_message["content"]
+
+
+@pytest.mark.asyncio
+async def test_runner_does_not_hint_different_tool_results():
+    provider = MagicMock()
+    call_count = {"n": 0}
+
+    async def chat_with_retry(*, messages, **kwargs):
+        call_count["n"] += 1
+        if call_count["n"] <= 3:
+            return LLMResponse(
+                content="reading",
+                tool_calls=[ToolCallRequest(
+                    id=f"call_{call_count['n']}",
+                    name="grep",
+                    arguments={"pattern": "TODO", "path": "nanobot"},
+                )],
+                usage={},
+            )
+        return LLMResponse(content="done", tool_calls=[], usage={})
+
+    provider.chat_with_retry = chat_with_retry
+    tools = MagicMock()
+    tools.get_definitions.return_value = []
+    tools.execute = AsyncMock(side_effect=["first result", "second result", "third result"])
+
+    result = await AgentRunner(provider).run(AgentRunSpec(
+        initial_messages=[{"role": "user", "content": "review code"}],
+        tools=tools,
+        model="test-model",
+        max_iterations=4,
+        max_tool_result_chars=_MAX_TOOL_RESULT_CHARS,
+    ))
+
+    assert result.final_content == "done"
+    assert tools.execute.await_count == 3
+    assert all("Repeated grep result" not in str(msg.get("content", "")) for msg in result.messages)

tests/channels/test_websocket_channel.py

@@ -3021,49 +3021,3 @@ def test_handle_webui_thread_get_does_not_backfill_trigger_internal_prompt(
     body = json.loads(resp.body.decode())
     assert [message["role"] for message in body["messages"]] == ["assistant"]
     assert [message["content"] for message in body["messages"]] == ["PR #4502 已经开始 review。"]
-
-
-def test_handle_webui_thread_get_does_not_backfill_hidden_subagent_result(
-    tmp_path,
-    monkeypatch,
-) -> None:
-    from urllib.parse import quote
-
-    from websockets.datastructures import Headers
-    from websockets.http11 import Request
-
-    from nanobot.session.history_visibility import HIDDEN_HISTORY_META
-    from nanobot.webui.transcript import append_transcript_object
-
-    monkeypatch.setattr("nanobot.config.paths.get_data_dir", lambda: tmp_path)
-    workspace = tmp_path / "workspace"
-    sessions = SessionManager(workspace)
-    key = "websocket:c-subagent"
-    session = sessions.get_or_create(key)
-    session.add_message(
-        "user",
-        "internal subagent result",
-        **{HIDDEN_HISTORY_META: {"kind": "subagent_result", "subagent_task_id": "sub-1"}},
-    )
-    session.add_message("assistant", "subagent summary")
-    sessions.save(session)
-    append_transcript_object(
-        key,
-        {"event": "message", "chat_id": "c-subagent", "text": "subagent summary"},
-    )
-
-    bus = MagicMock()
-    channel = WebSocketChannel(
-        {"enabled": True, "allowFrom": ["*"]},
-        bus,
-        gateway=_basic_handler(bus, session_manager=sessions, workspace_path=workspace),
-    )
-    channel.gateway.tokens.api_tokens["tok"] = time.monotonic() + 300.0
-    enc = quote(key, safe="")
-    req = Request(f"/api/sessions/{enc}/webui-thread", Headers([("Authorization", "Bearer tok")]))
-    resp = channel.gateway.http._handle_webui_thread_get(req, enc)
-
-    assert resp.status_code == 200
-    body = json.loads(resp.body.decode())
-    assert [message["role"] for message in body["messages"]] == ["assistant"]
-    assert [message["content"] for message in body["messages"]] == ["subagent summary"]

tests/cron/test_cron_service.py

@@ -699,36 +699,6 @@ async def test_external_update_preserves_run_history_records(tmp_path):
     fresh._save_store()
 
 
-def test_stale_instance_remove_preserves_external_add(tmp_path) -> None:
-    """A stopped instance must not save a stale snapshot over another instance's job."""
-    store_path = tmp_path / "cron" / "jobs.json"
-    schedule = CronSchedule(kind="every", every_ms=60_000)
-    service_a = CronService(store_path)
-    service_b = CronService(store_path)
-
-    first = service_a.add_job(
-        name="first",
-        schedule=schedule,
-        message="first",
-        **_bound_chat("first"),
-    )
-
-    # Prime service_b with a view that does not include later external changes.
-    assert [job.name for job in service_b.list_jobs(include_disabled=True)] == ["first"]
-
-    service_a.add_job(
-        name="second",
-        schedule=schedule,
-        message="second",
-        **_bound_chat("second"),
-    )
-
-    assert service_b.remove_job(first.id) == "removed"
-
-    reloaded = CronService(store_path)
-    assert [job.name for job in reloaded.list_jobs(include_disabled=True)] == ["second"]
-
-
 # ── timer race regression tests ──
 
 

tests/test_truncate_text_shadowing.py

@@ -1,4 +1,5 @@
 import inspect
+from types import SimpleNamespace
 
 
 def test_sanitize_persisted_blocks_truncate_text_shadowing_regression() -> None:
@@ -13,20 +14,16 @@ def test_sanitize_persisted_blocks_truncate_text_shadowing_regression() -> None:
     This test asserts the fixed API exists and truncation works without raising.
     """
 
-    from nanobot.session.turn_history import sanitize_persisted_blocks
+    from nanobot.agent.loop import AgentLoop
 
-    sig = inspect.signature(sanitize_persisted_blocks)
+    sig = inspect.signature(AgentLoop._sanitize_persisted_blocks)
     assert "should_truncate_text" in sig.parameters
     assert "truncate_text" not in sig.parameters
 
+    dummy = SimpleNamespace(max_tool_result_chars=5)
     content = [{"type": "text", "text": "0123456789"}]
 
-    out = sanitize_persisted_blocks(
+    out = AgentLoop._sanitize_persisted_blocks(dummy, content, should_truncate_text=True)
-        content,
-        max_tool_result_chars=5,
-        runtime_context_tag="[runtime]",
-        should_truncate_text=True,
-    )
     assert isinstance(out, list)
     assert out and out[0]["type"] == "text"
     assert isinstance(out[0]["text"], str)

tests/tools/test_exec_session_tools.py

@@ -5,7 +5,6 @@ import re
 import shlex
 import subprocess
 import sys
-import time
 
 from nanobot.agent.tools.exec_session import (
     ExecSessionManager,
@@ -77,23 +76,6 @@ def test_exec_returns_completed_session_output_when_yield_time_ms_is_used(tmp_pa
     assert "session_id:" not in result
 
 
-def test_exec_session_yield_returns_when_process_finishes_early(tmp_path):
-    async def run() -> tuple[str, float]:
-        manager = ExecSessionManager()
-        tool = ExecTool(working_dir=str(tmp_path), timeout=5, session_manager=manager)
-        command = _python_command("import time; time.sleep(0.1); print('done')")
-        started = time.monotonic()
-        result = await tool.execute(command=command, yield_time_ms=1200)
-        return result, time.monotonic() - started
-
-    result, elapsed = asyncio.run(run())
-
-    assert "done" in result
-    assert "Exit code: 0" in result
-    assert "session_id:" not in result
-    assert elapsed < 1.0
-
-
 def test_exec_session_accepts_max_output_tokens_alias(tmp_path):
     async def run() -> str:
         manager = ExecSessionManager()

tests/utils/test_repeated_tool_result_hint.py

@@ -0,0 +1,60 @@
+"""Tests for repeated tool-result hints."""
+
+from __future__ import annotations
+
+from nanobot.utils.runtime import (
+    repeated_external_lookup_error,
+    repeated_tool_result_hint,
+)
+
+
+def test_repeated_tool_result_hints_after_two_identical_results():
+    counts: dict[str, int] = {}
+
+    assert repeated_tool_result_hint("grep", "same result", counts) is None
+    assert repeated_tool_result_hint("grep", "same result", counts) is None
+    third = repeated_tool_result_hint("grep", "same result", counts)
+
+    assert third is not None
+    assert "Repeated grep result" in third
+
+
+def test_repeated_tool_result_ignores_different_results():
+    counts: dict[str, int] = {}
+
+    assert repeated_tool_result_hint("grep", "first", counts) is None
+    assert repeated_tool_result_hint("grep", "second", counts) is None
+    assert repeated_tool_result_hint("grep", "third", counts) is None
+
+
+def test_repeated_tool_result_is_per_tool():
+    counts: dict[str, int] = {}
+
+    repeated_tool_result_hint("grep", "same", counts)
+    repeated_tool_result_hint("grep", "same", counts)
+
+    assert repeated_tool_result_hint("read_file", "same", counts) is None
+
+
+def test_repeated_tool_result_handles_text_blocks():
+    counts: dict[str, int] = {}
+    result = [{"type": "text", "text": "same result"}]
+
+    repeated_tool_result_hint("mcp", result, counts)
+    repeated_tool_result_hint("mcp", result, counts)
+    third = repeated_tool_result_hint("mcp", result, counts)
+
+    assert third is not None
+    assert "Repeated mcp result" in third
+
+
+def test_repeated_external_lookup_still_blocks_after_two_attempts():
+    counts: dict[str, int] = {}
+    arguments = {"url": "https://example.com"}
+
+    repeated_external_lookup_error("web_fetch", arguments, counts)
+    repeated_external_lookup_error("web_fetch", arguments, counts)
+    third = repeated_external_lookup_error("web_fetch", arguments, counts)
+
+    assert third is not None
+    assert "repeated external lookup blocked" in third

tests/utils/test_webui_transcript.py

@@ -2,7 +2,6 @@
 
 from __future__ import annotations
 
-from nanobot.session.history_visibility import HIDDEN_HISTORY_META
 from nanobot.webui.transcript import (
     WEBUI_TRANSCRIPT_SCHEMA_VERSION,
     append_fork_marker,
@@ -699,47 +698,6 @@ def test_backfill_does_not_misalign_when_session_only_has_transcript_tail(
     ]
 
 
-def test_backfill_skips_internal_subagent_results(tmp_path, monkeypatch) -> None:
-    monkeypatch.setattr("nanobot.config.paths.get_data_dir", lambda: tmp_path)
-    key = "websocket:t-subagent"
-    for ev in (
-        {"event": "message", "chat_id": "t-subagent", "text": "summary one"},
-        {"event": "turn_end", "chat_id": "t-subagent"},
-        {"event": "message", "chat_id": "t-subagent", "text": "summary two"},
-        {"event": "turn_end", "chat_id": "t-subagent"},
-    ):
-        append_transcript_object(key, ev)
-
-    legacy_raw = (
-        "[Subagent 'legacy' completed successfully]\n\n"
-        "Task: t\n\n"
-        "Result:\nr\n\n"
-        "Summarize this naturally for the user."
-    )
-    out = build_webui_thread_response(
-        key,
-        session_messages=[
-            {"role": "user", "content": legacy_raw},
-            {"role": "assistant", "content": "summary one"},
-            {
-                "role": "user",
-                "content": "marked result",
-                HIDDEN_HISTORY_META: {
-                    "kind": "subagent_result",
-                    "subagent_task_id": "sub-1",
-                },
-            },
-            {"role": "assistant", "content": "summary two"},
-        ],
-    )
-
-    assert out is not None
-    assert [(message["role"], message["content"]) for message in out["messages"]] == [
-        ("assistant", "summary one"),
-        ("assistant", "summary two"),
-    ]
-
-
 def test_replay_infers_video_media_from_attachment_name() -> None:
     msgs = replay_transcript_to_ui_messages(
         [

tests/webui/test_session_list_index.py

@@ -7,7 +7,6 @@ from pathlib import Path
 import nanobot.webui.session_list_index as session_list_index
 from nanobot.cron.session_turns import CRON_HISTORY_META
 from nanobot.session.automation_turns import AUTOMATION_HISTORY_META
-from nanobot.session.history_visibility import HIDDEN_HISTORY_META
 from nanobot.session.manager import SessionManager
 
 
@@ -104,20 +103,6 @@ def test_webui_session_list_skips_trigger_internal_user_preview(tmp_path: Path)
     assert list_webui_sessions(manager)[0]["preview"] == "PR #4502 已经开始 review。"
 
 
-def test_webui_session_list_skips_hidden_history_user_preview(tmp_path: Path) -> None:
-    manager = SessionManager(tmp_path)
-    session = manager.get_or_create("websocket:hidden-preview")
-    session.add_message(
-        "user",
-        "internal subagent result",
-        **{HIDDEN_HISTORY_META: {"kind": "subagent_result", "subagent_task_id": "sub-1"}},
-    )
-    session.add_message("assistant", "subagent summary")
-    manager.save(session)
-
-    assert list_webui_sessions(manager)[0]["preview"] == "subagent summary"
-
-
 def test_webui_session_list_uses_webui_transcript_activity_for_sort(
     tmp_path: Path,
     monkeypatch,