refactor(session): extract turn history recovery

nanobot/agent/loop.py

@@ -55,13 +55,14 @@ from nanobot.security.workspace_access import (
     bind_workspace_scope,
     reset_workspace_scope,
 )
-from nanobot.session import turn_continuation
+from nanobot.session import turn_continuation, turn_history
 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,
@@ -70,8 +71,6 @@ 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 (
@@ -174,8 +173,8 @@ class AgentLoop:
         self._refresh_provider_snapshot()
         return LLMRuntime(self.provider, self.model)
 
-    _RUNTIME_CHECKPOINT_KEY = "runtime_checkpoint"
-    _PENDING_USER_TURN_KEY = "pending_user_turn"
+    _RUNTIME_CHECKPOINT_KEY = turn_history.RUNTIME_CHECKPOINT_KEY
+    _PENDING_USER_TURN_KEY = turn_history.PENDING_USER_TURN_KEY
 
     # Event-driven state transition table.
     # Handlers return an event string; the driver looks up the next state here.
@@ -795,7 +794,20 @@ class AgentLoop:
                     content, media = self._prepare_message_media(content, media)
                     media = media or None
                 user_content = self.context._build_user_content(content, media)
-                return {"role": "user", "content": user_content}
+                row: dict[str, Any] = {"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:
@@ -1640,38 +1652,13 @@ class AgentLoop:
         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(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
+        return turn_history.sanitize_persisted_blocks(
+            content,
+            max_tool_result_chars=self.max_tool_result_chars,
+            runtime_context_tag=ContextBuilder._RUNTIME_CONTEXT_TAG,
+            should_truncate_text=should_truncate_text,
+            drop_runtime=drop_runtime,
+        )
 
     def _save_turn(
         self,
@@ -1681,193 +1668,36 @@ class AgentLoop:
         *,
         turn_latency_ms: int | None = None,
     ) -> None:
-        """Save new-turn messages into session, truncating large tool results."""
-        from datetime import datetime
-
-        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) > 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()
+        turn_history.save_turn(
+            session,
+            messages,
+            skip,
+            max_tool_result_chars=self.max_tool_result_chars,
+            runtime_context_tag=ContextBuilder._RUNTIME_CONTEXT_TAG,
+            turn_latency_ms=turn_latency_ms,
+        )
 
     def _persist_subagent_followup(self, session: Session, msg: InboundMessage) -> 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
+        return turn_history.persist_subagent_followup(session, msg)
 
     def _set_runtime_checkpoint(self, session: Session, payload: dict[str, Any]) -> None:
-        """Persist the latest in-flight turn state into session metadata."""
-        session.metadata[self._RUNTIME_CHECKPOINT_KEY] = payload
+        turn_history.set_runtime_checkpoint(session, payload)
         self.sessions.save(session)
 
     def _mark_pending_user_turn(self, session: Session) -> None:
-        session.metadata[self._PENDING_USER_TURN_KEY] = True
+        turn_history.mark_pending_user_turn(session)
 
     def _clear_pending_user_turn(self, session: Session) -> None:
-        session.metadata.pop(self._PENDING_USER_TURN_KEY, None)
+        turn_history.clear_pending_user_turn(session)
 
     def _clear_runtime_checkpoint(self, session: Session) -> None:
-        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"),
-        )
+        turn_history.clear_runtime_checkpoint(session)
 
     def _restore_runtime_checkpoint(self, session: Session) -> bool:
-        """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
+        return turn_history.restore_runtime_checkpoint(session)
 
     def _restore_pending_user_turn(self, session: Session) -> bool:
-        """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
+        return turn_history.restore_pending_user_turn(session)
 
     async def process_direct(
         self,

nanobot/agent/runner.py

@@ -20,6 +20,7 @@ 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,
@@ -155,6 +156,8 @@ 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(

nanobot/agent/tools/exec_session.py

@@ -128,7 +128,15 @@ class _ExecSession:
     ) -> _SessionPoll:
         self.last_access = time.monotonic()
         if yield_time_ms > 0 and self.process.returncode is None:
-            await asyncio.sleep(min(yield_time_ms, MAX_YIELD_MS) / 1000)
+            wait_s = 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

@@ -0,0 +1,22 @@
+"""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,6 +15,7 @@ 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,
@@ -37,8 +38,8 @@ _SESSION_LIST_PREVIEW_MAX_RECORDS = 200
 _SESSION_LIST_PREVIEW_MAX_CHARS = 1_000_000
 _FORK_VOLATILE_METADATA_KEYS = {
     "goal_state",
-    "pending_user_turn",
-    "runtime_checkpoint",
+    PENDING_USER_TURN_KEY,
+    RUNTIME_CHECKPOINT_KEY,
     "thread_goal",
     "title",
     "title_user_edited",

nanobot/session/turn_history.py

@@ -0,0 +1,266 @@
+"""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_automation_history_message(message):
+        if is_hidden_history_message(message):
             continue
         role = message.get("role")
         content = message.get("content")

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.automation_turns import is_automation_history_message
+from nanobot.session.history_visibility import is_hidden_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_automation_history_message(message):
+        if is_hidden_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.automation_turns import is_automation_history_message
+from nanobot.session.history_visibility import is_hidden_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_automation_history_message(item):
+        if is_hidden_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_automation_history_message(item):
+    if is_hidden_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_automation_history_message(item):
+                    if is_hidden_history_message(item):
                         continue
                     text = _message_preview_text(item)
                     if not text:

nanobot/webui/transcript.py

@@ -17,7 +17,8 @@ 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_history_message, is_automation_kind
+from nanobot.session.automation_turns import 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
 
@@ -782,7 +783,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_automation_history_message(msg):
+        if is_hidden_history_message(msg):
             continue
         role = msg.get("role")
         content = msg.get("content")
@@ -854,13 +855,28 @@ 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_automation_history_message(message):
+    if is_hidden_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,6 +103,48 @@ 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,6 +465,68 @@ 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/channels/test_websocket_channel.py

@@ -3021,3 +3021,49 @@ 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,6 +699,36 @@ 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,5 +1,4 @@
 import inspect
-from types import SimpleNamespace
 
 
 def test_sanitize_persisted_blocks_truncate_text_shadowing_regression() -> None:
@@ -14,16 +13,20 @@ def test_sanitize_persisted_blocks_truncate_text_shadowing_regression() -> None:
     This test asserts the fixed API exists and truncation works without raising.
     """
 
-    from nanobot.agent.loop import AgentLoop
+    from nanobot.session.turn_history import sanitize_persisted_blocks
 
-    sig = inspect.signature(AgentLoop._sanitize_persisted_blocks)
+    sig = inspect.signature(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 = AgentLoop._sanitize_persisted_blocks(dummy, content, should_truncate_text=True)
+    out = sanitize_persisted_blocks(
+        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,6 +5,7 @@ import re
 import shlex
 import subprocess
 import sys
+import time
 
 from nanobot.agent.tools.exec_session import (
     ExecSessionManager,
@@ -76,6 +77,23 @@ 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_webui_transcript.py

@@ -2,6 +2,7 @@
 
 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,
@@ -698,6 +699,47 @@ 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,6 +7,7 @@ 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
 
 
@@ -103,6 +104,20 @@ 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,