fix(cli): preserve gateway log stream boundaries

docs/configuration.md

@@ -2268,12 +2268,16 @@ When a user is idle for longer than a configured threshold, nanobot **proactivel
 
 How it works:
 1. **Idle detection**: On each idle tick (~1 s), checks whether an idle-session scan is due. By default, the full scan runs at most once per minute.
-2. **Background compaction**: Older context is summarized while the most recent messages remain available.
+2. **Background compaction**: Idle sessions summarize the older live prefix via LLM and keep the most recent legal suffix (currently 8 messages).
-3. **Session preservation**: The complete session history remains stored for later inspection and reuse.
+3. **Summary injection**: When the user returns, the summary is injected as runtime context (one-shot, not persisted) alongside the retained recent suffix.
-4. **Restart-safe resume**: The compacted context remains available after a process restart.
+4. **Restart-safe resume**: The summary is also mirrored into session metadata so it can still be recovered after a process restart.
 
 > [!NOTE]
-> Auto compact shortens the context sent to the model without deleting the session's structured message history.
+> Mental model: "summarize older context, keep the freshest live turns, **and overwrite the session file with the compact form.**" It is not a full `session.clear()`, but it is a write — not a soft cursor move.
+>
+> Concretely, auto compact rewrites `sessions/<key>.jsonl` in place: older messages (including their structured `tool_calls` / `tool_call_id` / `reasoning_content`) are replaced by just the retained recent suffix (currently 8 messages), while the archived prefix is preserved only as a plain-text summary appended to `memory/history.jsonl` (or a `[RAW] ...` flattened dump if LLM summarization fails). The original structured JSON of those turns is no longer recoverable from the session file.
+>
+> This differs from the **token-driven soft consolidation** that fires when a prompt exceeds the context budget: that path only advances an internal `last_consolidated` cursor and leaves the session file untouched, so the raw tool-call trail stays on disk and can still be replayed or audited. If you rely on that trail for debugging or auditing, set `idleCompactAfterMinutes` to `0` and let only the token-driven path run.
 
 ## Timezone
 

nanobot/agent/context.py

@@ -30,7 +30,11 @@ from nanobot.security.workspace_access import WorkspaceScopeResolver
 from nanobot.session.keys import last_channel_from_metadata
 from nanobot.session.manager import Session
 from nanobot.session.summary import SessionSummary
-from nanobot.utils.helpers import detect_image_mime, load_bundled_template
+from nanobot.utils.helpers import (
+    detect_image_mime,
+    load_bundled_template,
+    truncate_text_to_tokens,
+)
 from nanobot.utils.prompt_templates import render_template
 
 
@@ -71,29 +75,14 @@ class PersistedPromptContextResolver:
         return channel, scope.project_path
 
 
-@dataclass(frozen=True, slots=True)
-class TranscriptInput:
-    """Raw turn inputs from which ``ContextBuilder`` assembles a transcript."""
-
-    history: list[dict[str, Any]]
-    current_message: str | None
-    media: Sequence[str] | None = None
-    current_role: str = "user"
-    session_summary: SessionSummary | None = None
-    runtime_context_blocks: Sequence[RuntimeContextBlock] | None = None
-
-    @property
-    def message_count(self) -> int:
-        """Number of boundary-preserving messages in the assembled transcript."""
-        return 1 + len(self.history) + (self.current_message is not None)
-
-
 class ContextBuilder:
     """Builds the context (system prompt + messages) for the agent."""
 
     BOOTSTRAP_FILES = ["AGENTS.md", "SOUL.md", "USER.md"]
     _SKIPPABLE_DEFAULTS = {"AGENTS.md", "USER.md"}
     _RUNTIME_CONTEXT_TAG = RUNTIME_CONTEXT_TAG
+    _MAX_RECENT_HISTORY = 50
+    _MAX_HISTORY_TOKENS = 8_000  # hard cap on recent history section size (tokens)
     _RUNTIME_CONTEXT_END = RUNTIME_CONTEXT_END
 
     def __init__(self, workspace: Path, timezone: str | None = None, disabled_skills: list[str] | None = None):
@@ -109,6 +98,9 @@ class ContextBuilder:
         session_summary: SessionSummary | None = None,
         workspace: Path | None = None,
         include_memory: bool = True,
+        include_memory_recent_history: bool = True,
+        session_key: str | None = None,
+        unified_session: bool = False,
     ) -> str:
         """Build the system prompt from identity, bootstrap files, memory, and skills."""
         root = workspace or self.workspace
@@ -146,6 +138,29 @@ class ContextBuilder:
         if skills_summary:
             parts.append(render_template("agent/skills_section.md", skills_summary=skills_summary))
 
+        if include_memory_recent_history:
+            entries = self.memory.read_recent_history_for_prompt(
+                since_cursor=self.memory.get_last_dream_cursor(),
+                session_key=session_key,
+                unified_session=unified_session,
+            )
+            if entries:
+                capped = entries[-self._MAX_RECENT_HISTORY:]
+                capped = self._without_duplicate_session_summary(
+                    capped,
+                    session_key=session_key,
+                    session_summary=session_summary,
+                )
+                if capped:
+                    history_text = "\n".join(
+                        f"- [{e['timestamp']}] {e['content']}" for e in capped
+                    )
+                    history_text = truncate_text_to_tokens(
+                        history_text,
+                        self._MAX_HISTORY_TOKENS,
+                    )
+                    parts.append("# Recent History\n\n" + history_text)
+
         if session_summary:
             parts.append(
                 "[Archived Context Summary]\n\n"
@@ -155,6 +170,25 @@ class ContextBuilder:
 
         return "\n\n---\n\n".join(parts)
 
+    @staticmethod
+    def _without_duplicate_session_summary(
+        entries: list[dict[str, Any]],
+        *,
+        session_key: str | None,
+        session_summary: SessionSummary | None,
+    ) -> list[dict[str, Any]]:
+        """Drop the history entry already represented by the session summary."""
+        if not session_summary:
+            return entries
+        for index in range(len(entries) - 1, -1, -1):
+            entry = entries[index]
+            if (
+                entry.get("session_key") == session_key
+                and entry.get("content") == session_summary["text"]
+            ):
+                return [*entries[:index], *entries[index + 1:]]
+        return entries
+
     def _get_identity(self, channel: str | None = None, workspace: Path | None = None) -> str:
         """Get the core identity section."""
         root = workspace or self.workspace
@@ -244,68 +278,46 @@ class ContextBuilder:
         runtime_context_blocks: Sequence[RuntimeContextBlock] | None = None,
         workspace: Path | None = None,
         include_memory: bool = True,
+        include_memory_recent_history: bool = True,
+        session_key: str | None = None,
+        unified_session: bool = False,
     ) -> list[dict[str, Any]]:
-        """Compatibility wrapper for callers that need merged adjacent roles."""
+        """Build the complete message list for an LLM call."""
-        messages = self.build_transcript(
-            TranscriptInput(
-                history=history,
-                current_message=current_message,
-                media=media,
-                current_role=current_role,
-                session_summary=session_summary,
-                runtime_context_blocks=runtime_context_blocks,
-            ),
-            channel=channel,
-            workspace=workspace,
-            include_memory=include_memory,
-        )
-        current = messages[-1]
-        if len(messages) < 2 or messages[-2].get("role") != current.get("role"):
-            return messages
-
-        merged = dict(messages[-2])
-        merged["content"] = self._merge_message_content(
-            merged.get("content"),
-            current.get("content"),
-        )
-        current_meta = current.get("_meta")
-        if current.get("role") == "user" and isinstance(current_meta, dict):
-            internal_meta = dict(merged.get("_meta") or {})
-            internal_meta.update(cast(dict[str, Any], current_meta))
-            merged["_meta"] = internal_meta
-        return [*messages[:-2], merged]
-
-    def build_transcript(
-        self,
-        transcript: TranscriptInput,
-        *,
-        channel: str | None = None,
-        workspace: Path | None = None,
-        include_memory: bool = True,
-    ) -> list[dict[str, Any]]:
-        """Build a model transcript while preserving the fresh-turn boundary."""
         root = workspace or self.workspace
         messages: list[dict[str, Any]] = [
             {
                 "role": "system",
                 "content": self.build_system_prompt(
                     channel=channel,
-                    session_summary=transcript.session_summary,
+                    session_summary=session_summary,
                     workspace=root,
                     include_memory=include_memory,
+                    include_memory_recent_history=include_memory_recent_history,
+                    session_key=session_key,
+                    unified_session=unified_session,
                 ),
             },
-            *transcript.history,
+            *history,
         ]
-        if transcript.current_message is None:
-            return messages
-
         current = self.build_current_message(
-            transcript.current_message,
+            current_message,
-            media=list(transcript.media) if transcript.media else None,
+            media=media,
-            current_role=transcript.current_role,
+            current_role=current_role,
-            runtime_context_blocks=transcript.runtime_context_blocks,
+            runtime_context_blocks=runtime_context_blocks,
         )
+        if messages[-1].get("role") == current_role:
+            last = dict(messages[-1])
+            last["content"] = self._merge_message_content(
+                last.get("content"),
+                current.get("content"),
+            )
+            current_meta = current.get("_meta")
+            if current_role == "user" and isinstance(current_meta, dict):
+                internal_meta = dict(last.get("_meta") or {})
+                internal_meta.update(cast(dict[str, Any], current_meta))
+                last["_meta"] = internal_meta
+            messages[-1] = last
+            return messages
         messages.append(current)
         return messages
 

nanobot/agent/context_governance.py

@@ -13,7 +13,6 @@ from typing import TYPE_CHECKING, Any, cast
 
 from loguru import logger
 
-from nanobot.providers.base import LLMUsage
 from nanobot.utils.helpers import (
     estimate_message_tokens,
     estimate_prompt_tokens_chain,
@@ -28,6 +27,12 @@ if TYPE_CHECKING:
     from nanobot.providers.base import LLMProvider
 
 SNIP_SAFETY_BUFFER = 1024
+MICROCOMPACT_MIN_CHARS = 500
+INFLIGHT_COMPACT_TARGET_RATIO = 0.85
+COMPACTABLE_TOOLS = frozenset({
+    "read_file", "exec", "grep", "find_files",
+    "web_search", "web_fetch", "list_dir", "list_exec_sessions",
+})
 # read_file is the recovery path for persisted results; exempting it prevents persist->read->persist loops.
 TOOL_RESULT_OFFLOAD_EXEMPT_TOOLS = frozenset({"read_file"})
 BACKFILL_CONTENT = "[Tool result unavailable — call was interrupted or lost]"
@@ -36,27 +41,6 @@ PLACEHOLDER_TEXTS = frozenset({
 })
 
 
-class ContextWindowExceededError(RuntimeError):
-    """Raised before a locally fitted request that still exceeds its budget."""
-
-    def __init__(
-        self,
-        *,
-        session_key: str | None,
-        estimated_tokens: int,
-        input_budget: int,
-        source: str,
-    ) -> None:
-        self.session_key = session_key
-        self.estimated_tokens = estimated_tokens
-        self.input_budget = input_budget
-        self.source = source
-        super().__init__(
-            "Model input still exceeds the local context budget after request fitting "
-            f"for {session_key or 'default'}: {estimated_tokens}/{input_budget} via {source}"
-        )
-
-
 def _tool_call_name_is_valid(tool_call: Any) -> bool:
     """Whether a persisted OpenAI-style tool_call carries a usable name.
 
@@ -83,6 +67,7 @@ class ContextGovernanceConfig:
     context_window_tokens: int | None = None
     context_block_limit: int | None = None
     max_tokens: int | None = None
+    inflight_start_index: int = 0
 
 
 class ContextGovernor:
@@ -92,85 +77,17 @@ class ContextGovernor:
         self,
         config: ContextGovernanceConfig,
         messages: list[dict[str, Any]],
+        compacted_tool_call_ids: set[str],
     ) -> list[dict[str, Any]]:
         updated = self.strip_placeholder_assistant_messages(messages)
         updated = self.strip_malformed_tool_calls(updated)
         updated = self.drop_orphan_tool_results(updated)
         updated = self.backfill_missing_tool_results(updated)
-        return self.apply_tool_result_budget(config, updated)
+        updated = self.apply_tool_result_budget(config, updated)
-
+        updated = self.compact_inflight_overflow(config, updated, compacted_tool_call_ids)
-    def fit_to_budget(
+        updated = self.snip_history(config, updated)
-        self,
-        config: ContextGovernanceConfig,
-        messages: list[dict[str, Any]],
-        *,
-        tool_definitions: list[dict[str, Any]] | None,
-    ) -> list[dict[str, Any]]:
-        """Fit a model-facing copy while keeping the source transcript intact."""
-        updated = self.snip_history(
-            config,
-            messages,
-            tool_definitions=tool_definitions,
-            force=True,
-        )
         updated = self.drop_orphan_tool_results(updated)
-        updated = self.backfill_missing_tool_results(updated)
+        return self.backfill_missing_tool_results(updated)
-        if not config.context_window_tokens:
-            return updated
-        budget = self.input_budget(config)
-        estimated, source = estimate_prompt_tokens_chain(
-            config.provider,
-            config.model,
-            updated,
-            tool_definitions,
-        )
-        if budget > 0 and estimated <= budget:
-            return updated
-        raise ContextWindowExceededError(
-            session_key=config.session_key,
-            estimated_tokens=estimated,
-            input_budget=budget,
-            source=source,
-        )
-
-    def fit_request(
-        self,
-        config: ContextGovernanceConfig,
-        messages: list[dict[str, Any]],
-        usage: LLMUsage | None,
-        *,
-        usage_matches_messages: bool,
-        tool_definitions: list[dict[str, Any]] | None,
-        request_context_tokens: int | None = None,
-    ) -> tuple[list[dict[str, Any]], bool]:
-        """Fit the request when its measured or estimated input is pressured."""
-        if not config.context_window_tokens:
-            return messages, False
-        budget = self.input_budget(config)
-        if (
-            request_context_tokens is None
-            and usage_matches_messages
-            and usage is not None
-            and usage.context_tokens is not None
-        ):
-            pressured = budget <= 0 or usage.context_tokens >= budget
-        else:
-            estimated, _ = estimate_prompt_tokens_chain(
-                config.provider,
-                config.model,
-                messages,
-                tool_definitions,
-            )
-            if request_context_tokens is not None:
-                estimated = max(estimated, request_context_tokens)
-            pressured = budget <= 0 or estimated >= budget
-        if not pressured:
-            return messages, False
-        return self.fit_to_budget(
-            config,
-            messages,
-            tool_definitions=tool_definitions,
-        ), True
 
     @staticmethod
     def input_budget(config: ContextGovernanceConfig) -> int:
@@ -409,13 +326,71 @@ class ContextGovernor:
                 updated[idx]["content"] = normalized
         return updated
 
+    def compact_inflight_overflow(
+        self,
+        config: ContextGovernanceConfig,
+        messages: list[dict[str, Any]],
+        compacted_tool_call_ids: set[str],
+    ) -> list[dict[str, Any]]:
+        """Compact in-flight tool results only when the request would overflow."""
+        budget = self.input_budget(config)
+        if budget <= 0:
+            return messages
+
+        tools = config.tools.get_definitions()
+        updated = self._apply_recorded_compactions(messages, compacted_tool_call_ids)
+        estimate, source = estimate_prompt_tokens_chain(
+            config.provider,
+            config.model,
+            updated,
+            tools,
+        )
+        if estimate <= budget:
+            return updated
+
+        target = int(budget * INFLIGHT_COMPACT_TARGET_RATIO)
+        candidates = self._inflight_compaction_candidates(
+            config,
+            updated,
+            compacted_tool_call_ids,
+        )
+        if not candidates:
+            return updated
+
+        for candidate_idx, (idx, tool_call_id) in enumerate(candidates):
+            is_newest_candidate = candidate_idx == len(candidates) - 1
+            if is_newest_candidate and estimate <= budget:
+                break
+            if tool_call_id in compacted_tool_call_ids:
+                continue
+            if updated is messages:
+                updated = [dict(m) for m in messages]
+            compacted_tool_call_ids.add(tool_call_id)
+            self._compact_tool_result_at(updated, idx)
+            estimate, source = estimate_prompt_tokens_chain(
+                config.provider,
+                config.model,
+                updated,
+                tools,
+            )
+            if estimate <= target:
+                break
+
+        logger.debug(
+            "In-flight context compaction for {}: prompt={} budget={} target={} via {}, ids={}",
+            config.session_key or "default",
+            estimate,
+            budget,
+            target,
+            source,
+            len(compacted_tool_call_ids),
+        )
+        return updated
+
     def snip_history(
         self,
         config: ContextGovernanceConfig,
         messages: list[dict[str, Any]],
-        *,
-        tool_definitions: list[dict[str, Any]] | None,
-        force: bool = False,
     ) -> list[dict[str, Any]]:
         if not messages or not config.context_window_tokens:
             return messages
@@ -424,13 +399,14 @@ class ContextGovernor:
         if budget <= 0:
             return messages
 
+        tools = config.tools.get_definitions()
         estimate, _ = estimate_prompt_tokens_chain(
             config.provider,
             config.model,
             messages,
-            tool_definitions,
+            tools,
         )
-        if not force and estimate <= budget:
+        if estimate <= budget:
             return messages
 
         system_messages = [dict(msg) for msg in messages if msg.get("role") == "system"]
@@ -443,7 +419,7 @@ class ContextGovernor:
             config.provider,
             config.model,
             system_messages,
-            tool_definitions,
+            tools,
         )
         remaining_budget = max(0, budget - max(system_tokens, fixed_tokens))
         kept: list[dict[str, Any]] = []
@@ -458,6 +434,16 @@ class ContextGovernor:
 
         return system_messages + self._legal_history_tail(kept, non_system)
 
+    @staticmethod
+    def _tool_result_compaction_message(message: dict[str, Any]) -> str:
+        name = message.get("name", "tool")
+        return (
+            f"Error: The previous {name} result was compacted to fit context because it was too "
+            "large. Do not repeat the same call unchanged. Retry with a narrower path, query, "
+            "range, or result limit, use another tool, or tell the user the task cannot fit in "
+            "the available context."
+        )
+
     def _legal_history_tail(
         self,
         kept: list[dict[str, Any]],
@@ -476,3 +462,50 @@ class ContextGovernor:
             if messages[idx].get("role") == "user":
                 return messages[idx:]
         return []
+
+    def _apply_recorded_compactions(
+        self,
+        messages: list[dict[str, Any]],
+        compacted_tool_call_ids: set[str],
+    ) -> list[dict[str, Any]]:
+        if not compacted_tool_call_ids:
+            return messages
+        updated = messages
+        for idx, msg in enumerate(messages):
+            if msg.get("role") != "tool":
+                continue
+            tool_call_id = msg.get("tool_call_id")
+            if not tool_call_id or str(tool_call_id) not in compacted_tool_call_ids:
+                continue
+            compaction_message = self._tool_result_compaction_message(msg)
+            if msg.get("content") == compaction_message:
+                continue
+            if updated is messages:
+                updated = [dict(m) for m in messages]
+            updated[idx]["content"] = compaction_message
+        return updated
+
+    def _inflight_compaction_candidates(
+        self,
+        config: ContextGovernanceConfig,
+        messages: list[dict[str, Any]],
+        compacted_tool_call_ids: set[str],
+    ) -> list[tuple[int, str]]:
+        compactable: list[tuple[int, str]] = []
+        for idx, msg in enumerate(messages):
+            if idx < config.inflight_start_index:
+                continue
+            if msg.get("role") != "tool" or msg.get("name") not in COMPACTABLE_TOOLS:
+                continue
+            tool_call_id = msg.get("tool_call_id")
+            if not tool_call_id or str(tool_call_id) in compacted_tool_call_ids:
+                continue
+            content = msg.get("content")
+            if not isinstance(content, str) or len(content) < MICROCOMPACT_MIN_CHARS:
+                continue
+            compactable.append((idx, str(tool_call_id)))
+
+        return compactable
+
+    def _compact_tool_result_at(self, messages: list[dict[str, Any]], idx: int) -> None:
+        messages[idx]["content"] = self._tool_result_compaction_message(messages[idx])

nanobot/agent/loop.py

@@ -14,7 +14,6 @@ from collections.abc import Coroutine, Iterable, Mapping
 from contextlib import AbstractContextManager, ExitStack, nullcontext, suppress
 from dataclasses import dataclass, field
 from enum import Enum, auto
-from functools import partial
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, Awaitable, Callable, TypeVar, cast
 
@@ -24,7 +23,7 @@ from nanobot.agent import context as agent_context
 from nanobot.agent import model_presets as preset_helpers
 from nanobot.agent.autocompact import AutoCompact
 from nanobot.agent.automation_turns import publish_next_deferred_turn
-from nanobot.agent.context import ContextBuilder, PersistedPromptContextResolver, TranscriptInput
+from nanobot.agent.context import ContextBuilder, PersistedPromptContextResolver
 from nanobot.agent.cron_turns import CronTurnCoordinator
 from nanobot.agent.hook import AgentHook, AgentTurnHookFactory
 from nanobot.agent.memory import Consolidator
@@ -136,7 +135,7 @@ class TurnContext:
     session: Session | None = None
 
     history: list[dict[str, Any]] = field(default_factory=list)
-    transcript_input: TranscriptInput | None = None
+    initial_messages: list[dict[str, Any]] = field(default_factory=list)
     provider_state: ProviderConversationState | None = field(default=None, repr=False)
     request_context: RequestContext | None = None
     runtime_context_blocks: list[RuntimeContextBlock] = field(default_factory=list)
@@ -444,6 +443,7 @@ class AgentLoop:
                 workspace_scopes=self.workspace_scopes,
                 unified_session=unified_session,
             ),
+            unified_session=unified_session,
         )
         self.auto_compact = AutoCompact(
             sessions=self.sessions,
@@ -723,15 +723,22 @@ class AgentLoop:
             return True
         return False
 
-    def _build_transcript_input(self, ctx: TurnContext) -> TranscriptInput:
+    def _build_initial_messages(self, ctx: TurnContext) -> list[dict[str, Any]]:
-        """Capture the persisted history and fresh input as separate transcript parts."""
+        """Build the initial message list for the LLM turn."""
         assert ctx.session is not None
-        return TranscriptInput(
+        scope = self.workspace_scopes.for_message(ctx.msg, ctx.session.metadata)
+        return self.context.build_messages(
             history=ctx.history,
             current_message=ctx.msg.content,
             media=ctx.msg.media if ctx.kind is TurnKind.USER and ctx.msg.media else None,
+            channel=ctx.delivery.route.channel,
             session_summary=ctx.pending_summary,
+            workspace=scope.project_path,
             runtime_context_blocks=ctx.runtime_context_blocks,
+            include_memory=ctx.session.policy.persist,
+            include_memory_recent_history=not ctx.ephemeral,
+            session_key=ctx.session.key,
+            unified_session=self._unified_session,
         )
 
     def _request_context_for_turn(self, ctx: TurnContext) -> RequestContext:
@@ -852,22 +859,6 @@ class AgentLoop:
             metadata={**metadata, "render_as": "text"},
         )
 
-    def _track_active_task(self, key: str, task: asyncio.Task[Any]) -> None:
-        """Track active session work until its task group becomes empty."""
-        tasks = self._active_tasks.setdefault(key, set())
-        tasks.add(task)
-        task.add_done_callback(partial(self._active_task_done, key, tasks))
-
-    def _active_task_done(
-        self,
-        key: str,
-        tasks: set[asyncio.Task[Any]],
-        task: asyncio.Task[Any],
-    ) -> None:
-        tasks.discard(task)
-        if not tasks and self._active_tasks.get(key) is tasks:
-            self._active_tasks.pop(key, None)
-
     async def _cancel_active_tasks(self, key: str) -> int:
         """Cancel and await all active work for *key*.
 
@@ -938,7 +929,7 @@ class AgentLoop:
 
     async def _run_agent_loop(
         self,
-        transcript_input: TranscriptInput,
+        initial_messages: list[dict[str, Any]],
         on_progress: Callable[..., Awaitable[None]] | None = None,
         on_stream: Callable[[str], Awaitable[None]] | None = None,
         on_stream_end: Callable[..., Awaitable[None]] | None = None,
@@ -1119,12 +1110,6 @@ class AgentLoop:
             message_metadata=request_metadata,
             session_metadata=session.metadata if session is not None else None,
         )
-        transcript_builder = partial(
-            self.context.build_transcript,
-            channel=request_ctx.channel,
-            workspace=effective_scope.project_path,
-            include_memory=session.policy.persist if session is not None else True,
-        )
         if request_context is None:
             request_ctx = dataclasses.replace(
                 request_ctx,
@@ -1171,13 +1156,11 @@ class AgentLoop:
                 run_extra_hooks_for_ephemeral=run_extra_hooks_for_ephemeral,
             ))
             result = await self.runner.run(AgentRunSpec(
-                initial_messages=None,
+                initial_messages=initial_messages,
                 tools=effective_tools,
                 runtime=runtime,
                 max_iterations=self.max_iterations,
                 max_tool_result_chars=self.max_tool_result_chars,
-                transcript_input=transcript_input,
-                transcript_builder=transcript_builder,
                 hook=hook,
                 concurrent_tools=True,
                 workspace=effective_scope.project_path,
@@ -1366,7 +1349,12 @@ class AgentLoop:
                 # Compute the effective session key before dispatching
                 # This ensures /stop command can find tasks correctly when unified session is enabled
                 task = asyncio.create_task(self._dispatch(msg))
-                self._track_active_task(effective_key, task)
+                active_tasks: set[asyncio.Task[Any]] = self._active_tasks.setdefault(
+                    effective_key,
+                    set(),
+                )
+                active_tasks.add(task)
+                task.add_done_callback(active_tasks.discard)
         finally:
             await self.aclose()
 
@@ -1890,13 +1878,6 @@ class AgentLoop:
                 session,
                 runtime=runtime,
             )
-            # Token consolidation may have committed a replacement checkpoint
-            # after the compact stage captured its summary for this request.
-            ctx.session, ctx.pending_summary = self.auto_compact.prepare_session(
-                session,
-                ctx.session_key,
-            )
-            session = ctx.require_session()
         is_subagent = ctx.kind is TurnKind.SYSTEM and ctx.msg.sender_id == "subagent"
 
         _hist_kwargs: dict[str, Any] = {
@@ -1987,7 +1968,7 @@ class AgentLoop:
             # Upgrade the replay-safe baseline to the resumable state before
             # prompt assembly and the first model checkpoint.
             self.sessions.save(session)
-        ctx.transcript_input = self._build_transcript_input(ctx)
+        ctx.initial_messages = self._build_initial_messages(ctx)
 
         if ctx.on_progress is None:
             ctx.on_progress = ctx.delivery.progress_callback()
@@ -1999,10 +1980,9 @@ class AgentLoop:
         if ctx.visible_run_started_at is None:
             ctx.visible_run_started_at = time.time()
         await ctx.delivery.running(started_at=ctx.visible_run_started_at)
-        assert ctx.transcript_input is not None
         with capture_message_deliveries() as message_sends:
             result = await self._run_agent_loop(
-                ctx.transcript_input,
+                ctx.initial_messages,
                 runtime=runtime,
                 on_progress=ctx.on_progress,
                 on_stream=ctx.on_stream,

nanobot/agent/memory.py

@@ -35,7 +35,6 @@ from nanobot.utils.helpers import (
     estimate_prompt_tokens_chain,
     strip_think,
     truncate_text,
-    truncate_text_to_tokens,
 )
 from nanobot.utils.prompt_templates import render_template
 from nanobot.utils.workspace_prompts import (
@@ -62,6 +61,12 @@ class MemoryStore:
     # Deliberately excludes memory/.dream_cursor so progress bookkeeping never
     # appears as a durable-memory edit in the audit record.
     _DREAM_CONTENT_PATHS = ("SOUL.md", "USER.md", "memory/MEMORY.md")
+    # Per-file cap when embedding current contents into the Dream prompt. The
+    # durable files are tiny in practice (~5 KB total), but a runaway file must
+    # not unbounded the prompt.
+    _DREAM_FILE_EMBED_CAP = 8000
+    _INTERNAL_HISTORY_SESSION_PREFIXES = ("cron:", "dream:")
+    _INTERNAL_HISTORY_SESSION_KEYS = {"heartbeat"}
     _LEGACY_ENTRY_START_RE = re.compile(r"^\[(\d{4}-\d{2}-\d{2}[^\]]*)\]\s*")
     _LEGACY_TIMESTAMP_RE = re.compile(r"^\[(\d{4}-\d{2}-\d{2} \d{2}:\d{2})\]\s*")
     _LEGACY_RAW_MESSAGE_RE = re.compile(
@@ -255,29 +260,6 @@ class MemoryStore:
 
     # -- history.jsonl — append-only, JSONL format ---------------------------
 
-    def _normalize_history_entry(
-        self,
-        entry: str,
-        *,
-        max_chars: int | None = None,
-    ) -> str:
-        """Return the exact bounded, model-safe text accepted by the journal."""
-        limit = max_chars if max_chars is not None else _HISTORY_ENTRY_HARD_CAP
-        raw = entry.rstrip()
-        content = strip_think(raw)
-        if len(content) > limit:
-            if not self._oversize_logged:
-                self._oversize_logged = True
-                logger.warning(
-                    "history entry exceeds {} chars ({}); truncating. "
-                    "Usually means a caller forgot its own cap; "
-                    "further occurrences suppressed.",
-                    limit,
-                    len(content),
-                )
-            content = truncate_text(content, limit)
-        return content
-
     def append_history(
         self,
         entry: str,
@@ -292,16 +274,27 @@ class MemoryStore:
         persisted. If the cleaned content is empty but the raw entry wasn't,
         the record is persisted with an empty string rather than falling back
         to the raw leak — otherwise `strip_think`'s guarantees would be
-        undone when Dream consumes the journal entry.
+        undone by history replay / consolidation downstream.
 
         A defensive cap (*max_chars*, default ``_HISTORY_ENTRY_HARD_CAP``) is
         applied as a final safety net: individual callers should cap their own
         content more tightly; this default only exists to catch unintentional
         large writes (e.g. an LLM echoing its input back as a "summary").
         """
+        limit = max_chars if max_chars is not None else _HISTORY_ENTRY_HARD_CAP
         ts = datetime.now().strftime("%Y-%m-%d %H:%M")
         raw = entry.rstrip()
-        content = self._normalize_history_entry(entry, max_chars=max_chars)
+        if len(raw) > limit:
+            if not self._oversize_logged:
+                self._oversize_logged = True
+                logger.warning(
+                    "history entry exceeds {} chars ({}); truncating. "
+                    "Usually means a caller forgot its own cap; "
+                    "further occurrences suppressed.",
+                    limit, len(raw),
+                )
+            raw = truncate_text(raw, limit)
+        content = strip_think(raw)
         # Cursor allocation and the append must be atomic: concurrent writers
         # could otherwise read the same current cursor and emit duplicates.
         with self._append_lock:
@@ -309,7 +302,7 @@ class MemoryStore:
             if raw and not content:
                 logger.debug(
                     "history entry {} stripped to empty (likely template leak); "
-                    "persisting empty content to avoid re-polluting Dream input",
+                    "persisting empty content to avoid re-polluting context",
                     cursor,
                 )
             record = {"cursor": cursor, "timestamp": ts, "content": content}
@@ -399,6 +392,36 @@ class MemoryStore:
         """Return history entries with a valid cursor > *since_cursor*."""
         return [e for e, c in self._iter_valid_entries() if c > since_cursor]
 
+    @classmethod
+    def _is_internal_history_session(cls, session_key: str | None) -> bool:
+        if not session_key:
+            return False
+        return (
+            session_key in cls._INTERNAL_HISTORY_SESSION_KEYS
+            or session_key.startswith(cls._INTERNAL_HISTORY_SESSION_PREFIXES)
+        )
+
+    def read_recent_history_for_prompt(
+        self,
+        since_cursor: int,
+        *,
+        session_key: str | None,
+        unified_session: bool = False,
+    ) -> list[dict[str, Any]]:
+        """Return unprocessed history entries safe to inject into a turn prompt."""
+        entries = self.read_unprocessed_history(since_cursor=since_cursor)
+        if session_key is None:
+            return entries
+        if not unified_session:
+            return [e for e in entries if e.get("session_key") == session_key]
+
+        return [
+            entry
+            for entry in entries
+            if (entry_session := entry.get("session_key")) == session_key
+            or not self._is_internal_history_session(entry_session)
+        ]
+
     def compact_history(self) -> None:
         """Drop oldest processed entries without discarding pending Dream input."""
         if self.max_history_entries <= 0:
@@ -545,7 +568,9 @@ class MemoryStore:
         Returns ``(prompt, last_cursor)`` or ``None`` if nothing to process.
 
         The current contents of the durable memory files (SOUL.md, USER.md,
-        memory/MEMORY.md) reach Dream through the normal agent system context.
+        memory/MEMORY.md) are embedded so the model edits the real files rather
+        than a stale mental model — eliminating a class of failed/out-of-bounds
+        edits that previously produced hallucinated audit records.
         """
         last_cursor = self.get_last_dream_cursor()
         entries = self.read_unprocessed_history(since_cursor=last_cursor)
@@ -558,9 +583,35 @@ class MemoryStore:
             for e in batch
         )
         template = self._dream_template()
-        prompt = f"{template}\n\n## Conversation History\n{history_text}"
+        files_section = self._render_current_memory_files()
+        prompt = (
+            f"{template}\n\n{files_section}\n\n"
+            f"## Conversation History\n{history_text}"
+        )
         return (prompt, batch[-1]["cursor"])
 
+    def _render_current_memory_files(self) -> str:
+        """Render the durable memory files' current contents for the Dream prompt.
+
+        Missing files render as ``(empty)``; oversized files are capped. The
+        section is the ground truth the model must edit against.
+        """
+        files = [
+            ("SOUL.md", self.soul_file),
+            ("USER.md", self.user_file),
+            ("memory/MEMORY.md", self.memory_file),
+        ]
+        blocks: list[str] = []
+        for label, path in files:
+            try:
+                content = path.read_text(encoding="utf-8") if path.exists() else ""
+            except OSError:
+                content = ""
+            if len(content) > self._DREAM_FILE_EMBED_CAP:
+                content = truncate_text(content, self._DREAM_FILE_EMBED_CAP) + "\n...[truncated]"
+            blocks.append(f"### {label}\n{content}" if content.strip() else f"### {label}\n(empty)")
+        return "## Current Memory Files\n" + "\n\n".join(blocks)
+
     def dream_content_diff(self) -> str:
         """Structured summary of uncommitted changes to the durable memory files.
 
@@ -667,28 +718,21 @@ class MemoryStore:
         *,
         max_chars: int | None = None,
         session_key: str | None = None,
-    ) -> str:
+    ) -> None:
-        """Persist and return a bounded raw checkpoint when summarization degrades."""
+        """Fallback: dump raw messages to history.jsonl without LLM summarization."""
-        checkpoint = self._build_raw_checkpoint(messages, max_chars=max_chars)
+        limit = max_chars if max_chars is not None else _RAW_ARCHIVE_MAX_CHARS
-        self.append_history(checkpoint, session_key=session_key)
+        formatted = truncate_text(
+            self._format_messages(public_history_messages(messages)),
+            limit,
+        )
+        self.append_history(
+            f"[RAW] {len(messages)} messages\n"
+            f"{formatted}",
+            session_key=session_key,
+        )
         logger.warning(
             "Memory consolidation degraded: raw-archived {} messages", len(messages)
         )
-        return checkpoint
-
-    def _build_raw_checkpoint(
-        self,
-        messages: list[dict[str, Any]],
-        *,
-        max_chars: int | None = None,
-    ) -> str:
-        """Build the same bounded checkpoint as :meth:`raw_archive` without writing it."""
-        limit = max_chars if max_chars is not None else _RAW_ARCHIVE_MAX_CHARS
-        checkpoint = (
-            f"[RAW] {len(messages)} messages\n"
-            f"{self._format_messages(public_history_messages(messages))}"
-        )
-        return self._normalize_history_entry(checkpoint, max_chars=limit)
 
     # ------------------------------------------------------------------
     # Dream helpers
@@ -743,10 +787,11 @@ class MemoryStore:
 # Memory ingestion and legacy context-pressure coordination
 # ---------------------------------------------------------------------------
 
-# Raw fallbacks use a tighter cap. Completed model summaries may scale with the
+# Individual history.jsonl writers cap their own payloads tightly; the
-# configured generation budget, while append_history() still enforces the
+# _HISTORY_ENTRY_HARD_CAP at append_history() is a belt-and-suspenders default
-# emergency hard cap against pathological provider output.
+# that catches any new caller that forgot to set its own cap.
 _RAW_ARCHIVE_MAX_CHARS = 16_000       # fallback dump (LLM failed)
+_ARCHIVE_SUMMARY_MAX_CHARS = 8_000    # LLM-produced consolidation summary
 _HISTORY_ENTRY_HARD_CAP = 64_000      # emergency cap in append_history
 
 
@@ -764,45 +809,13 @@ class MemoryArchiver:
         build_messages: Callable[..., list[dict[str, Any]]],
         get_tool_definitions: Callable[[], list[dict[str, Any]]],
         resolve_prompt_context: Callable[[Session], tuple[str | None, Path | None]] | None = None,
+        unified_session: bool = False,
     ) -> None:
         self.store = store
         self._build_messages = build_messages
         self._get_tool_definitions = get_tool_definitions
         self._resolve_prompt_context = resolve_prompt_context
-
+        self.unified_session = unified_session
-    def _raw_checkpoint(
-        self,
-        messages: list[dict[str, Any]],
-        *,
-        session_key: str,
-        previous_summary: str | None,
-        max_tokens: int,
-    ) -> str:
-        """Persist the failed chunk and return a bounded replacement checkpoint."""
-        raw = self.store.raw_archive(messages, session_key=session_key)
-        token_limit = max(1, max_tokens)
-        if not previous_summary:
-            return truncate_text_to_tokens(raw, token_limit)
-
-        combined = (
-            "[Previous archived context]\n"
-            f"{previous_summary}\n\n"
-            "[Newly archived raw context]\n"
-            f"{raw}"
-        )
-        bounded = truncate_text_to_tokens(combined, token_limit)
-        if bounded == combined:
-            return combined
-
-        # Keep evidence from both sides when their full concatenation cannot fit.
-        section_limit = max(1, (token_limit - 32) // 2)
-        return truncate_text_to_tokens(
-            "[Previous archived context]\n"
-            f"{truncate_text_to_tokens(previous_summary, section_limit)}\n\n"
-            "[Newly archived raw context]\n"
-            f"{truncate_text_to_tokens(raw, section_limit)}",
-            token_limit,
-        )
 
     async def archive(
         self,
@@ -812,53 +825,48 @@ class MemoryArchiver:
         session_key: str,
         request_messages: list[dict[str, Any]],
         request_tools: list[dict[str, Any]],
-        previous_summary: str | None = None,
     ) -> str | None:
         """Execute a prepared archive request and persist its result."""
         if not messages:
             return None
-
-        def raw_fallback() -> str:
-            return self._raw_checkpoint(
-                messages,
-                session_key=session_key,
-                previous_summary=previous_summary,
-                max_tokens=runtime.generation.max_tokens,
-            )
-
         try:
             with llm_usage_source("dream"):
                 response = await runtime.provider.chat_with_retry(
                     model=runtime.model,
                     messages=request_messages,
                     tools=request_tools,
+                    tool_choice="none",
                     temperature=runtime.generation.temperature,
                     max_tokens=runtime.generation.max_tokens,
                     reasoning_effort=runtime.generation.reasoning_effort,
                 )
         except Exception:
             logger.warning("Memory archive provider call failed, raw-dumping to history")
-            return raw_fallback()
+            self.store.raw_archive(messages, session_key=session_key)
+            return None
         if response.finish_reason in {"error", "length"}:
             logger.warning(
                 "Memory archive provider did not complete ({}), raw-dumping to history",
                 response.finish_reason,
             )
-            return raw_fallback()
+            self.store.raw_archive(messages, session_key=session_key)
+            return None
         if response.has_tool_calls is True:
             logger.warning("Memory archive provider returned tool calls, raw-dumping to history")
-            return raw_fallback()
+            self.store.raw_archive(messages, session_key=session_key)
+            return None
         summary = response.content
         if not summary or not summary.strip():
             logger.warning("Memory archive provider returned no summary, raw-dumping to history")
-            return raw_fallback()
+            self.store.raw_archive(messages, session_key=session_key)
-        summary = self.store._normalize_history_entry(summary)
+            return None
-        if not summary:
+        if summary.strip() == "(nothing)":
-            logger.warning("Memory archive provider summary was not safe to replay, raw-dumping")
-            return raw_fallback()
-        if summary == "(nothing)":
             return "(nothing)"
-        self.store.append_history(summary, session_key=session_key)
+        self.store.append_history(
+            summary,
+            max_chars=_ARCHIVE_SUMMARY_MAX_CHARS,
+            session_key=session_key,
+        )
         return summary
 
     async def archive_session(
@@ -873,26 +881,13 @@ class MemoryArchiver:
         messages = list(session.messages[session.last_archived:archive_end])
         if not messages:
             return None
-        session_summary = session_summary_from_metadata(
-            session.metadata,
-            fallback_last_active=session.updated_at,
-        )
-        previous_summary = session_summary["text"] if session_summary else None
-
-        def raw_fallback() -> str:
-            return self._raw_checkpoint(
-                messages,
-                session_key=session.key,
-                previous_summary=previous_summary,
-                max_tokens=runtime.generation.max_tokens,
-            )
-
         if input_token_budget <= 0:
             logger.debug(
                 "Memory archive has no safe input budget for {}; raw-dumping",
                 session.key,
             )
-            return raw_fallback()
+            self.store.raw_archive(messages, session_key=session.key)
+            return None
         prefix = Session(
             key=session.key,
             messages=list(session.messages[:archive_end]),
@@ -908,8 +903,13 @@ class MemoryArchiver:
                 "Memory archive cannot replay the full chunk for {}; raw-dumping",
                 session.key,
             )
-            return raw_fallback()
+            self.store.raw_archive(messages, session_key=session.key)
-        prompt = render_template("agent/consolidator_archive.md", strip=True)
+            return None
+        prompt = render_template(
+            "agent/consolidator_archive.md",
+            strip=True,
+            archive_count=len(archive_history),
+        )
         channel = session.key.split(":", 1)[0] if ":" in session.key else None
         workspace: Path | None = None
         if self._resolve_prompt_context is not None:
@@ -918,8 +918,13 @@ class MemoryArchiver:
             history=history,
             current_message=prompt,
             channel=channel,
-            session_summary=session_summary,
+            session_summary=session_summary_from_metadata(
+                session.metadata,
+                fallback_last_active=session.updated_at,
+            ),
             workspace=workspace,
+            session_key=session.key,
+            unified_session=self.unified_session,
         )
         tools = self._get_tool_definitions()
         estimated, source = estimate_prompt_tokens_chain(
@@ -936,14 +941,14 @@ class MemoryArchiver:
                 input_token_budget,
                 source,
             )
-            return raw_fallback()
+            self.store.raw_archive(messages, session_key=session.key)
+            return None
         return await self.archive(
             messages,
             runtime=runtime,
             session_key=session.key,
             request_messages=request_messages,
             request_tools=tools,
-            previous_summary=previous_summary,
         )
 
 
@@ -959,16 +964,20 @@ class Consolidator:
         build_messages: Callable[..., list[dict[str, Any]]],
         get_tool_definitions: Callable[[], list[dict[str, Any]]],
         resolve_prompt_context: Callable[[Session], tuple[str | None, Path | None]] | None = None,
+        unified_session: bool = False,
     ):
         self.store = store
         self.sessions = sessions
+        self.unified_session = unified_session
         self._build_messages = build_messages
         self._get_tool_definitions = get_tool_definitions
+        self._resolve_prompt_context = resolve_prompt_context
         self.archiver = MemoryArchiver(
             store=store,
             build_messages=build_messages,
             get_tool_definitions=get_tool_definitions,
             resolve_prompt_context=resolve_prompt_context,
+            unified_session=unified_session,
         )
         self._locks: weakref.WeakValueDictionary[str, asyncio.Lock] = (
             weakref.WeakValueDictionary()
@@ -1004,18 +1013,13 @@ class Consolidator:
             return []
         return session.get_history()
 
-    @staticmethod
+    def _persist_last_summary(self, session: Session, summary: str | None) -> None:
-    def _set_last_summary(
+        if summary and summary != "(nothing)":
-        session: Session,
-        summary: str,
-        *,
-        last_active: datetime | None = None,
-    ) -> None:
-        if summary != "(nothing)":
             session.metadata["_last_summary"] = {
                 "text": summary,
-                "last_active": (last_active or session.updated_at).isoformat(),
+                "last_active": session.updated_at.isoformat(),
             }
+            self.sessions.save(session)
 
     def estimate_session_prompt_tokens(
         self,
@@ -1035,6 +1039,8 @@ class Consolidator:
             current_message="[token-probe]",
             channel=channel,
             session_summary=summary,
+            session_key=session.key,
+            unified_session=self.unified_session,
         )
         return estimate_prompt_tokens_chain(
             runtime.provider,
@@ -1051,6 +1057,24 @@ class Consolidator:
             - self._SAFETY_BUFFER
         )
 
+    async def archive(
+        self,
+        messages: list[dict[str, Any]],
+        *,
+        runtime: LLMRuntime,
+        session_key: str,
+        request_messages: list[dict[str, Any]],
+        request_tools: list[dict[str, Any]],
+    ) -> str | None:
+        """Compatibility wrapper for the extracted MemoryArchiver."""
+        return await self.archiver.archive(
+            messages,
+            runtime=runtime,
+            session_key=session_key,
+            request_messages=request_messages,
+            request_tools=request_tools,
+        )
+
     async def archive_session(
         self,
         session: Session,
@@ -1077,23 +1101,26 @@ class Consolidator:
         The budget reserves space for completion tokens and a safety buffer
         so the LLM request never exceeds the context window.
         """
+        if runtime.context_window_tokens <= 0:
+            return
+
         lock = self.get_lock(session.key)
         async with lock:
             # Refresh session reference: AutoCompact may have replaced it.
             fresh = self.sessions.get_or_create(session.key)
             if fresh is not session:
                 session = fresh
-            if runtime.context_window_tokens <= 0:
-                return
             if not session.messages:
                 return
 
             budget = self._input_token_budget(runtime)
+            last_summary: str | None = None
             estimated, source = self.estimate_session_prompt_tokens(
                 session,
                 runtime=runtime,
             )
             if estimated <= 0:
+                self._persist_last_summary(session, last_summary)
                 return
             if estimated < budget:
                 unarchived_count = len(session.messages) - session.last_archived
@@ -1105,6 +1132,7 @@ class Consolidator:
                     source,
                     unarchived_count,
                 )
+                self._persist_last_summary(session, last_summary)
                 return
 
             end_idx = self.pick_consolidation_boundary(session)
@@ -1132,12 +1160,18 @@ class Consolidator:
                 archive_end=end_idx,
                 runtime=runtime,
             )
-            if summary is None:
+            # Advance either way: archive_session raw-archives on degradation,
-                return
+            # and replaying the same chunk would duplicate Memory material.
-            self._set_last_summary(session, summary)
+            if summary:
+                last_summary = summary
             session.last_archived = end_idx
             self.sessions.save(session)
 
+            # Persist the last summary to session metadata so it can be injected
+            # into the runtime context on the next prepare_session() call, aligning
+            # the summary injection strategy with AutoCompact._archive().
+            self._persist_last_summary(session, last_summary)
+
     async def compact_idle_session(
         self,
         session_key: str,
@@ -1175,10 +1209,12 @@ class Consolidator:
                 archive_end=archive_end,
                 runtime=runtime,
             )
-            if summary is None:
-                return None
 
-            self._set_last_summary(session, summary, last_active=last_active)
+            if summary and summary != "(nothing)":
+                session.metadata["_last_summary"] = {
+                    "text": summary,
+                    "last_active": last_active.isoformat(),
+                }
 
             # A turn can append while the provider call is in flight. Advance only
             # through the captured batch so new messages remain eligible next time.

nanobot/agent/runner.py

@@ -14,7 +14,6 @@ from typing import Any, cast
 
 from loguru import logger
 
-from nanobot.agent.context import TranscriptInput
 from nanobot.agent.context_governance import (
     ContextGovernanceConfig,
     ContextGovernor,
@@ -67,7 +66,6 @@ ContinuationCallback = Callable[[], str | None]
 RetryWaitCallback = Callable[[str], Awaitable[None]]
 CheckpointCallback = Callable[[dict[str, Any]], Awaitable[None]]
 InjectionCallback = Callable[..., Awaitable[Iterable[Any] | None]]
-TranscriptBuilder = Callable[[TranscriptInput], list[dict[str, Any]]]
 
 _DEFAULT_ERROR_MESSAGE = "Sorry, I encountered an error calling the AI model."
 _ARREARAGE_ERROR_MESSAGE = (
@@ -96,13 +94,11 @@ def _restore_outer_whitespace(content: str, original: str | None) -> str:
 class AgentRunSpec:
     """Configuration for a single agent execution."""
 
-    initial_messages: list[dict[str, Any]] | None
+    initial_messages: list[dict[str, Any]]
     tools: ToolRegistry
     runtime: LLMRuntime
     max_iterations: int
     max_tool_result_chars: int
-    transcript_input: TranscriptInput | None = None
-    transcript_builder: TranscriptBuilder | None = None
     hook: AgentHook | None = None
     error_message: str | None = _DEFAULT_ERROR_MESSAGE
     max_iterations_message: str | None = None
@@ -139,17 +135,6 @@ class AgentRunResult:
     provider_state: ProviderConversationState | None = field(default=None, repr=False)
 
 
-@dataclass(slots=True)
-class _ModelRequestState:
-    """Per-run state used to govern the next provider request."""
-
-    config: ContextGovernanceConfig
-    conversation: ProviderConversationStateController
-    usage: LLMUsage | None = None
-    messages: list[dict[str, Any]] | None = None
-    tool_definitions: list[dict[str, Any]] | None = None
-
-
 class AgentRunner:
     """Run a tool-capable LLM loop without product-layer concerns."""
 
@@ -425,7 +410,7 @@ class AgentRunner:
 
     async def run(self, spec: AgentRunSpec) -> AgentRunResult:
         hook = spec.hook or AgentHook()
-        messages = self._initial_transcript(spec)
+        messages = list(spec.initial_messages)
         context = AgentRunHookContext(messages=deepcopy(messages))
         llm_usage_source_token = bind_llm_usage_source(
             spec.llm_usage_source or source_from_session_key(spec.session_key)
@@ -477,19 +462,6 @@ class AgentRunner:
             finally:
                 reset_llm_usage_source(llm_usage_source_token)
 
-    @staticmethod
-    def _initial_transcript(spec: AgentRunSpec) -> list[dict[str, Any]]:
-        """Resolve exactly one supported source for the initial model transcript."""
-        if spec.transcript_input is not None:
-            if spec.initial_messages is not None:
-                raise ValueError("provide either transcript_input or initial_messages, not both")
-            if spec.transcript_builder is None:
-                raise ValueError("transcript_builder is required with transcript_input")
-            return list(spec.transcript_builder(spec.transcript_input))
-        if spec.initial_messages is None:
-            raise ValueError("initial_messages is required without transcript_input")
-        return list(spec.initial_messages)
-
     async def _run_core(
         self,
         spec: AgentRunSpec,
@@ -511,6 +483,7 @@ class AgentRunner:
         length_recovery_parts: list[str] = []
         had_injections = False
         injection_cycles = 0
+        compacted_tool_call_ids: set[str] = set()
         pending_stream_content: str | None = None
         conversation_state = ProviderConversationStateController(
             provider=spec.runtime.provider,
@@ -529,29 +502,39 @@ class AgentRunner:
             context_window_tokens=spec.runtime.context_window_tokens,
             context_block_limit=spec.context_block_limit,
             max_tokens=spec.runtime.generation.max_tokens,
-        )
+            inflight_start_index=len(spec.initial_messages),
-        request_state = _ModelRequestState(
-            config=governance_config,
-            conversation=conversation_state,
         )
 
         for iteration in range(spec.max_iterations):
+            # Keep the persisted conversation untouched. Context governance
+            # may repair or compact historical messages for the model, but
+            # those synthetic edits must not shift the append boundary used
+            # later when the caller saves only the new turn. A governance
+            # failure must stop the run instead of sending an ungoverned copy.
+            messages_for_model = self.context_governor.prepare_for_model(
+                governance_config,
+                messages,
+                compacted_tool_call_ids,
+            )
             context = AgentHookContext(
                 iteration=iteration,
                 messages=messages,
                 session_key=spec.session_key,
             )
             await hook.before_iteration(context)
+            provider_context = conversation_state.prepare_request(
+                messages,
+                context_window_tokens=spec.runtime.context_window_tokens,
+                model_messages=messages_for_model,
+            )
             response = await self._request_model(
                 spec,
-                messages,
+                messages_for_model,
                 hook,
                 context,
-                request_state=request_state,
+                conversation_state=conversation_state,
-                transcript=messages,
+                provider_context=provider_context,
             )
-            assert request_state.messages is not None
-            messages_for_model = request_state.messages
             conversation_state.observe_response(response, messages)
             context.response = response
             context.tool_calls = list(response.tool_calls)
@@ -563,7 +546,7 @@ class AgentRunner:
                 response.content,
             )
             response.content = cleaned_content
-            raw_usage = self._record_request_usage(spec, request_state, response)
+            raw_usage = self._usage_or_estimate(spec, messages_for_model, response)
             context.usage = raw_usage
             usage = self._merge_usage(usage, raw_usage)
             if reasoning_text and not context.streamed_reasoning:
@@ -637,6 +620,7 @@ class AgentRunner:
                     self.context_governor.prepare_for_model(
                         governance_config,
                         messages,
+                        compacted_tool_call_ids,
                     )
                     if response.provider_state is not None
                     else None
@@ -702,13 +686,14 @@ class AgentRunner:
                 )
                 if hook.wants_streaming():
                     await hook.on_stream_end(context, resuming=False)
+                retry_messages = self._finalization_retry_messages(messages_for_model)
                 response = await self._request_finalization_retry(
                     spec,
                     messages_for_model,
-                    request_state=request_state,
                     transcript=messages,
+                    conversation_state=conversation_state,
                 )
-                retry_usage = self._record_request_usage(spec, request_state, response)
+                retry_usage = self._usage_or_estimate(spec, retry_messages, response)
                 usage = self._merge_usage(usage, retry_usage)
                 raw_usage = self._merge_usage(raw_usage, retry_usage)
                 context.response = response
@@ -895,7 +880,7 @@ class AgentRunner:
                     hook,
                     messages,
                     usage,
-                    request_state=request_state,
+                    conversation_state,
                 )
             if terminal_content is None:
                 terminal_content = self._max_iterations_fallback(spec)
@@ -942,60 +927,6 @@ class AgentRunner:
         kwargs["reasoning_effort"] = generation.reasoning_effort
         return kwargs
 
-    def _prepare_model_request(
-        self,
-        state: _ModelRequestState,
-        messages: list[dict[str, Any]],
-        *,
-        tool_definitions: list[dict[str, Any]] | None,
-        transcript: list[dict[str, Any]] | None = None,
-    ) -> tuple[list[dict[str, Any]], ProviderCallContext | None]:
-        """Prepare, fit, and record the exact payload sent to a provider."""
-        prepared = self.context_governor.prepare_for_model(state.config, messages)
-        supplemental_messages = (
-            [prepared[-1]] if transcript is not None and tool_definitions is None else None
-        )
-        model_messages = None if supplemental_messages is not None else prepared
-        request_context_tokens = (
-            state.conversation.estimate_request_context_tokens(
-                transcript,
-                model_messages=model_messages,
-                supplemental_messages=supplemental_messages,
-                tool_definitions=tool_definitions,
-            )
-            if transcript is not None
-            else None
-        )
-        usage_matches_messages = (
-            state.messages is not None
-            and prepared == state.messages
-            and tool_definitions == state.tool_definitions
-        )
-        prepared, fitted = self.context_governor.fit_request(
-            state.config,
-            prepared,
-            state.usage,
-            usage_matches_messages=usage_matches_messages,
-            tool_definitions=tool_definitions,
-            request_context_tokens=request_context_tokens,
-        )
-        provider_context = (
-            state.conversation.prepare_request(
-                transcript,
-                context_window_tokens=state.config.context_window_tokens,
-                model_messages=model_messages,
-                supplemental_messages=supplemental_messages,
-                resume_state=not fitted,
-            )
-            if transcript is not None
-            else state.conversation.independent_request_context(
-                context_window_tokens=state.config.context_window_tokens,
-            )
-        )
-        state.messages = deepcopy(prepared)
-        state.tool_definitions = deepcopy(tool_definitions)
-        return prepared, provider_context
-
     async def _request_model(
         self,
         spec: AgentRunSpec,
@@ -1003,29 +934,21 @@ class AgentRunner:
         hook: AgentHook,
         context: AgentHookContext,
         *,
-        request_state: _ModelRequestState,
         malformed_retry: bool = False,
-        transcript: list[dict[str, Any]] | None,
+        conversation_state: ProviderConversationStateController,
+        provider_context: ProviderCallContext | None = None,
     ) -> LLMResponse:
         timeout_s = self._resolve_llm_timeout_s(spec)
-        tool_definitions = spec.tools.get_definitions()
-        messages, provider_context = self._prepare_model_request(
-            request_state,
-            messages,
-            tool_definitions=tool_definitions,
-            transcript=transcript,
-        )
 
         kwargs = self._build_request_kwargs(
             spec,
             messages,
-            tools=tool_definitions,
+            tools=spec.tools.get_definitions(),
         )
         wants_streaming = hook.wants_streaming()
 
         active_hosted_tools: dict[str, dict[str, Any]] = {}
         native_reasoning_open = False
-        native_reasoning_close_task: asyncio.Task[None] | None = None
         request_started_at = 0.0
         first_output_at: float | None = None
         generation_started_at: float | None = None
@@ -1049,29 +972,11 @@ class AgentRunner:
             generation_started_at = None
 
         async def _close_native_reasoning() -> None:
-            nonlocal native_reasoning_open, native_reasoning_close_task
+            nonlocal native_reasoning_open
-            if native_reasoning_close_task is None:
             if not native_reasoning_open:
                 return
             native_reasoning_open = False
-                native_reasoning_close_task = asyncio.create_task(
+            await hook.emit_reasoning_end()
-                    hook.emit_reasoning_end()
-                )
-
-            close_task = native_reasoning_close_task
-            cancellation: asyncio.CancelledError | None = None
-            while not close_task.done():
-                try:
-                    await asyncio.shield(close_task)
-                except asyncio.CancelledError as exc:
-                    cancellation = cancellation or exc
-            try:
-                close_task.result()
-            finally:
-                if native_reasoning_close_task is close_task:
-                    native_reasoning_close_task = None
-            if cancellation is not None:
-                raise cancellation
 
         async def _provider_tool_event(event: dict[str, Any]) -> None:
             if event.get("kind") != "hosted_tool":
@@ -1146,10 +1051,6 @@ class AgentRunner:
                 await coro if outer_timeout_s is None
                 else await asyncio.wait_for(coro, timeout=outer_timeout_s)
             )
-        except asyncio.CancelledError:
-            _pause_generation()
-            await _close_native_reasoning()
-            raise
         except asyncio.TimeoutError:
             if outer_timeout_s is None:
                 response = LLMResponse(
@@ -1197,9 +1098,11 @@ class AgentRunner:
             )
             return await self._request_model(
                 spec, retry_messages, hook, context,
-                request_state=request_state,
                 malformed_retry=True,
-                transcript=None,
+                conversation_state=conversation_state,
+                provider_context=conversation_state.independent_request_context(
+                    context_window_tokens=spec.runtime.context_window_tokens,
+                ),
             )
         if (
             all_dropped
@@ -1215,7 +1118,9 @@ class AgentRunner:
             return await self._request_no_tools(
                 spec,
                 fallback_messages,
-                request_state=request_state,
+                provider_context=conversation_state.independent_request_context(
+                    context_window_tokens=spec.runtime.context_window_tokens,
+                ),
             )
         return response
 
@@ -1283,17 +1188,21 @@ class AgentRunner:
         spec: AgentRunSpec,
         messages: list[dict[str, Any]],
         *,
-        request_state: _ModelRequestState,
         transcript: list[dict[str, Any]],
+        conversation_state: ProviderConversationStateController,
     ) -> LLMResponse:
         retry_messages = self._finalization_retry_messages(messages)
+        provider_context = conversation_state.prepare_request(
+            transcript,
+            context_window_tokens=spec.runtime.context_window_tokens,
+            supplemental_messages=[retry_messages[-1]],
+        )
         response = await self._request_no_tools(
             spec,
             retry_messages,
-            request_state=request_state,
+            provider_context=provider_context,
-            transcript=transcript,
         )
-        request_state.conversation.observe_response(
+        conversation_state.observe_response(
             response,
             transcript,
             adopt_candidate_state=False,
@@ -1312,15 +1221,16 @@ class AgentRunner:
         hook: AgentHook,
         messages: list[dict[str, Any]],
         usage: LLMUsage | None,
-        *,
+        conversation_state: ProviderConversationStateController,
-        request_state: _ModelRequestState,
     ) -> tuple[str | None, LLMUsage | None]:
         retry_messages = self._budget_exhausted_finalization_messages(messages)
         try:
             response = await self._request_no_tools(
                 spec,
                 retry_messages,
-                request_state=request_state,
+                provider_context=conversation_state.independent_request_context(
+                    context_window_tokens=spec.runtime.context_window_tokens,
+                ),
             )
         except Exception:
             logger.exception(
@@ -1329,7 +1239,7 @@ class AgentRunner:
             )
             return None, usage
 
-        raw_usage = self._record_request_usage(spec, request_state, response)
+        raw_usage = self._usage_or_estimate(spec, retry_messages, response)
         usage = self._merge_usage(usage, raw_usage)
         if response.finish_reason == "error" or response.has_tool_calls:
             logger.warning(
@@ -1358,15 +1268,8 @@ class AgentRunner:
         spec: AgentRunSpec,
         messages: list[dict[str, Any]],
         *,
-        request_state: _ModelRequestState,
+        provider_context: ProviderCallContext | None = None,
-        transcript: list[dict[str, Any]] | None = None,
     ) -> LLMResponse:
-        messages, provider_context = self._prepare_model_request(
-            request_state,
-            messages,
-            tool_definitions=None,
-            transcript=transcript,
-        )
         kwargs = self._build_request_kwargs(
             spec,
             messages,
@@ -1378,18 +1281,17 @@ class AgentRunner:
         )
         timeout_s = self._resolve_llm_timeout_s(spec)
         try:
-            response = (
+            return (
                 await coro
                 if timeout_s is None
                 else await asyncio.wait_for(coro, timeout=timeout_s)
             )
         except asyncio.TimeoutError:
-            response = LLMResponse(
+            return LLMResponse(
                 content=f"Error calling LLM: timed out after {timeout_s:g}s",
                 finish_reason="error",
                 error_kind="timeout",
             )
-        return response
 
     @staticmethod
     def _resolve_llm_timeout_s(spec: AgentRunSpec) -> float | None:
@@ -1431,53 +1333,33 @@ class AgentRunner:
         spec: AgentRunSpec,
         messages: list[dict[str, Any]],
         response: LLMResponse,
-        *,
-        tool_definitions: list[dict[str, Any]] | None,
     ) -> LLMUsage | None:
         usage = response.usage
         if response.finish_reason == "error":
             if usage is None or usage.total_tokens == 0:
                 usage = LLMUsage.empty_request()
         elif usage is None or usage.total_tokens == 0:
-            usage = self._estimate_response_usage(
+            usage = self._estimate_response_usage(spec, messages, response)
-                spec,
-                messages,
-                response,
-                tool_definitions=tool_definitions,
-            )
         return usage.with_timing(
             generation_ms=response.generation_ms,
             ttft_ms=response.ttft_ms,
         )
 
-    def _record_request_usage(
-        self,
-        spec: AgentRunSpec,
-        state: _ModelRequestState,
-        response: LLMResponse,
-    ) -> LLMUsage | None:
-        assert state.messages is not None
-        state.usage = self._usage_or_estimate(
-            spec,
-            state.messages,
-            response,
-            tool_definitions=state.tool_definitions,
-        )
-        return state.usage
-
     def _estimate_response_usage(
         self,
         spec: AgentRunSpec,
         messages: list[dict[str, Any]],
         response: LLMResponse,
-        *,
-        tool_definitions: list[dict[str, Any]] | None,
     ) -> LLMUsage:
+        try:
+            tools = spec.tools.get_definitions()
+        except Exception:
+            tools = None
         prompt_tokens, _ = estimate_prompt_tokens_chain(
             spec.runtime.provider,
             spec.runtime.model,
             messages,
-            tool_definitions,
+            tools,
         )
         assistant_message = build_assistant_message(
             response.content or "",

nanobot/agent/tools/execution.py

@@ -43,13 +43,6 @@ _WORKSPACE_VIOLATION_MARKERS: tuple[str, ...] = (
 )
 
 
-def _with_retry_hint(payload: str) -> str:
-    """Append the recovery hint exactly once."""
-    if payload.endswith(_RETRY_HINT):
-        return payload
-    return payload + _RETRY_HINT
-
-
 async def execute_tool_calls(
     tools: ToolRegistry,
     tool_calls: list[ToolCallRequest],
@@ -112,7 +105,7 @@ async def _execute_tool_call(
             "status": "error",
             "detail": "repeated external lookup blocked",
         }
-        return _with_retry_hint(lookup_error), event
+        return lookup_error + _RETRY_HINT, event
 
     prepare_call = cast(
         Callable[[str, Any], object] | None,
@@ -126,7 +119,6 @@ async def _execute_tool_call(
             if len(prepared_tuple) == 3:
                 tool, params, prep_error = cast(tuple[Any, Any, str | None], prepared_tuple)
     if prep_error:
-        payload = _with_retry_hint(prep_error)
         event = {
             "name": tool_call.name,
             "status": "error",
@@ -134,14 +126,14 @@ async def _execute_tool_call(
         }
         handled = _classify_violation(
             raw_text=prep_error,
-            soft_payload=payload,
+            soft_payload=prep_error + _RETRY_HINT,
             event=event,
             tool_call=tool_call,
             workspace_violation_counts=workspace_violation_counts,
         )
         if handled is not None:
             return handled
-        return payload, event
+        return prep_error + _RETRY_HINT, event
 
     await hook.before_execute_tool(context, tool_call, tool, params)
     try:
@@ -158,9 +150,10 @@ async def _execute_tool_call(
             "status": "error",
             "detail": str(exc),
         }
-        payload = _with_retry_hint(f"Error: {type(exc).__name__}: {exc}")
+        payload = f"Error: {type(exc).__name__}: {exc}"
         handled = _classify_violation(
             raw_text=str(exc),
+            # Preserve legacy exception payloads without the retry hint.
             soft_payload=payload,
             event=event,
             tool_call=tool_call,
@@ -172,7 +165,6 @@ async def _execute_tool_call(
 
     if is_tool_error_result(result):
         await hook.on_execute_tool_error(context, tool_call, tool, params, result)
-        payload = _with_retry_hint(result)
         event = {
             "name": tool_call.name,
             "status": "error",
@@ -180,14 +172,14 @@ async def _execute_tool_call(
         }
         handled = _classify_violation(
             raw_text=result,
-            soft_payload=payload,
+            soft_payload=result + _RETRY_HINT,
             event=event,
             tool_call=tool_call,
             workspace_violation_counts=workspace_violation_counts,
         )
         if handled is not None:
             return handled
-        return payload, event
+        return result + _RETRY_HINT, event
 
     await hook.after_execute_tool(context, tool_call, tool, params, result)
 

nanobot/agent/tools/filesystem.py

@@ -861,10 +861,8 @@ def _best_window(old_text: str, content: str) -> tuple[float, int, list[str], li
 @tool_parameters(
     tool_parameters_schema(
         path=StringSchema("The file path to edit"),
-        old_text=StringSchema("The text to find and replace; copy it from read_file."),
+        old_text=StringSchema("The text to find and replace"),
-        new_text=StringSchema(
+        new_text=StringSchema("The text to replace with"),
-            "The replacement text; must differ from old_text for an existing file."
-        ),
         replace_all=BooleanSchema(description="Replace all occurrences (default false)"),
         occurrence=IntegerSchema(
             description="Optional 1-based occurrence to replace when old_text appears multiple times.",
@@ -901,9 +899,15 @@ class EditFileTool(_FsTool):
     @property
     def description(self) -> str:
         return (
-            "Perform a small, exact replacement in one file. "
+            "Perform a small, exact replacement in one file by replacing "
-            "Prefer apply_patch for multi-file, structural, or generated edits. "
+            "old_text with new_text. When replacing text in an existing file, "
-            "occurrence, line_hint, and replace_all=true are mutually exclusive."
+            "old_text and new_text must be different. Use this for narrow text substitutions "
+            "with old_text copied from read_file. For multi-file, structural, "
+            "or generated code edits, prefer apply_patch. If old_text matches "
+            "multiple times, provide more context or set occurrence, line_hint, "
+            "replace_all, and expected_replacements. When editing from numbered "
+            "read_file output, set line_hint to the exact target line. "
+            "Shows closest-match diagnostics on failure."
         )
 
     @staticmethod

nanobot/agent/tools/session_messages.py

@@ -7,7 +7,7 @@ from __future__ import annotations
 import asyncio
 import json
 import time
-from collections import OrderedDict, deque
+from collections import deque
 from collections.abc import Callable
 from dataclasses import dataclass
 from typing import Any, Protocol
@@ -127,7 +127,7 @@ class SendSessionMessageTool(Tool):
         self._max_messages_per_minute = max_messages_per_minute
         self._schedule_later = schedule_later
         self._clock = clock or time.monotonic
-        self._sent_at: OrderedDict[str, deque[float]] = OrderedDict()
+        self._sent_at: dict[str, deque[float]] = {}
         self._pending_replies: dict[tuple[str, str], _PendingReply] = {}
         self._expiry_tasks: set[asyncio.Task[None]] = set()
         self._send_lock = asyncio.Lock()
@@ -240,11 +240,8 @@ class SendSessionMessageTool(Tool):
 
         async with self._send_lock:
             now = self._clock()
+            sent_at = self._sent_at.setdefault(source.session_key, deque())
             cutoff = now - _RATE_LIMIT_WINDOW_SECONDS
-            self._prune_expired_rate_limits(cutoff)
-            sent_at = self._sent_at.get(source.session_key)
-            if sent_at is None:
-                sent_at = deque[float]()
             while sent_at and sent_at[0] <= cutoff:
                 sent_at.popleft()
             if len(sent_at) >= self._max_messages_per_minute:
@@ -262,8 +259,6 @@ class SendSessionMessageTool(Tool):
                 input_role="user",
             ))
             sent_at.append(now)
-            self._sent_at[source.session_key] = sent_at
-            self._sent_at.move_to_end(source.session_key)
             self._cancel_pending_reply(reverse_wait_key)
             if timeout_seconds is not None:
                 self._cancel_pending_reply(wait_key)
@@ -276,14 +271,6 @@ class SendSessionMessageTool(Tool):
 
         return f"@{target.name}"
 
-    def _prune_expired_rate_limits(self, cutoff: float) -> None:
-        """Drop sources ordered by their most recent successful send."""
-        while self._sent_at:
-            _, sent_at = next(iter(self._sent_at.items()))
-            if sent_at[-1] > cutoff:
-                return
-            self._sent_at.popitem(last=False)
-
     @staticmethod
     def _validate_reply_timeout(
         expect_reply: bool,

nanobot/channels/dingtalk/runtime.py

@@ -182,12 +182,6 @@ class NanobotDingTalkHandler(_CallbackHandlerBase):
                 )
             )
 
-            if not self.channel._accepting_inbound_tasks:
-                self.channel.logger.debug(
-                    "Skipping DingTalk inbound dispatch during channel shutdown"
-                )
-                return AckMessage.STATUS_OK, "OK"
-
             self.channel.logger.info("Received message from {} ({}): {}", sender_name, sender_id, content)
 
             # Forward to Nanobot via _on_message (non-blocking).
@@ -202,7 +196,7 @@ class NanobotDingTalkHandler(_CallbackHandlerBase):
                 )
             )
             self.channel._background_tasks.add(task)
-            task.add_done_callback(self.channel._on_background_task_done)
+            task.add_done_callback(self.channel._background_tasks.discard)
 
             return AckMessage.STATUS_OK, "OK"
 
@@ -262,17 +256,6 @@ class DingTalkChannel(BaseChannel):
 
         # Hold references to background tasks to prevent GC
         self._background_tasks: set[asyncio.Task[None]] = set()
-        self._accepting_inbound_tasks = True
-
-    def _on_background_task_done(self, task: asyncio.Task[None]) -> None:
-        self._background_tasks.discard(task)
-        if task.cancelled():
-            return
-        exception = task.exception()
-        if exception is not None:
-            self.logger.opt(exception=exception).error(
-                "DingTalk inbound message task failed"
-            )
 
     async def start(self) -> None:
         """Start the DingTalk bot with Stream Mode."""
@@ -289,7 +272,6 @@ class DingTalkChannel(BaseChannel):
                 self.logger.error("client_id and client_secret not configured")
                 return
 
-            self._accepting_inbound_tasks = True
             self._running = True
             self._http = httpx.AsyncClient(
                 timeout=httpx.Timeout(10.0, connect=10.0, read=30.0, write=30.0, pool=10.0)
@@ -327,7 +309,6 @@ class DingTalkChannel(BaseChannel):
 
     async def stop(self) -> None:
         """Stop the DingTalk bot."""
-        self._accepting_inbound_tasks = False
         self._running = False
         await self._close_stream_client()
         start_task = self._start_task
@@ -345,11 +326,8 @@ class DingTalkChannel(BaseChannel):
             await self._http.aclose()
             self._http = None
         # Cancel outstanding background tasks
-        background_tasks = tuple(self._background_tasks)
+        for task in self._background_tasks:
-        for task in background_tasks:
             task.cancel()
-        if background_tasks:
-            await asyncio.gather(*background_tasks, return_exceptions=True)
         self._background_tasks.clear()
 
     async def _close_stream_client(self) -> None:

nanobot/channels/dingtalk/tests/test_dingtalk_channel.py

@@ -3,7 +3,7 @@ import json
 import zipfile
 from io import BytesIO
 from types import SimpleNamespace
-from unittest.mock import AsyncMock, MagicMock
+from unittest.mock import AsyncMock
 
 import httpx
 import pytest
@@ -402,61 +402,6 @@ async def test_handler_uses_voice_recognition_text_when_text_is_empty(monkeypatc
     assert msg.chat_id == "group:conv123"
 
 
-@pytest.mark.asyncio
-async def test_handler_retrieves_background_message_failure(monkeypatch) -> None:
-    bus = MessageBus()
-    channel = DingTalkChannel(
-        DingTalkConfig(client_id="app", client_secret="secret", allow_from=["user1"]),
-        bus,
-    )
-    handler = NanobotDingTalkHandler(channel)
-    failure = RuntimeError("inbound dispatch failed")
-    mock_logger = MagicMock()
-    channel.logger = mock_logger
-
-    class _FakeChatbotMessage:
-        text = SimpleNamespace(content="hello")
-        extensions = {}
-        sender_staff_id = "user1"
-        sender_id = "fallback-user"
-        sender_nick = "Alice"
-        message_type = "text"
-
-        @staticmethod
-        def from_dict(_data):
-            return _FakeChatbotMessage()
-
-    async def fail(*_args) -> None:
-        raise failure
-
-    monkeypatch.setattr(dingtalk_module, "ChatbotMessage", _FakeChatbotMessage)
-    monkeypatch.setattr(dingtalk_module, "AckMessage", SimpleNamespace(STATUS_OK="OK"))
-    monkeypatch.setattr(channel, "_on_message", fail)
-    event_loop = asyncio.get_running_loop()
-    previous_handler = event_loop.get_exception_handler()
-    loop_errors: list[dict[str, object]] = []
-    event_loop.set_exception_handler(lambda _loop, context: loop_errors.append(context))
-
-    try:
-        status, body = await handler.process(
-            SimpleNamespace(data={"conversationType": "1", "text": {"content": "hello"}})
-        )
-        for _ in range(10):
-            await asyncio.sleep(0)
-            if not channel._background_tasks:
-                break
-    finally:
-        event_loop.set_exception_handler(previous_handler)
-
-    assert (status, body) == ("OK", "OK")
-    assert not channel._background_tasks
-    assert not loop_errors
-    mock_logger.opt.assert_called_once_with(exception=failure)
-    mock_logger.opt.return_value.error.assert_called_once_with(
-        "DingTalk inbound message task failed"
-    )
-
-
 @pytest.mark.asyncio
 async def test_handler_processes_file_message(monkeypatch) -> None:
     """Test that file messages are handled and forwarded with downloaded path."""
@@ -506,72 +451,6 @@ async def test_handler_processes_file_message(monkeypatch) -> None:
     assert "/tmp/nanobot_dingtalk/user1/report.xlsx" in msg.content
 
 
-@pytest.mark.asyncio
-async def test_handler_does_not_spawn_message_task_after_stop_during_download(
-    monkeypatch,
-) -> None:
-    channel = DingTalkChannel(
-        DingTalkConfig(client_id="app", client_secret="secret", allow_from=["user1"]),
-        MessageBus(),
-    )
-    handler = NanobotDingTalkHandler(channel)
-    download_started = asyncio.Event()
-    release_download = asyncio.Event()
-    message_task_started = asyncio.Event()
-
-    class _FakeFileChatbotMessage:
-        text = None
-        extensions = {}
-        image_content = None
-        rich_text_content = None
-        sender_staff_id = "user1"
-        sender_id = "fallback-user"
-        sender_nick = "Alice"
-        message_type = "file"
-
-        @staticmethod
-        def from_dict(_data):
-            return _FakeFileChatbotMessage()
-
-    async def delayed_download(*_args):
-        download_started.set()
-        await release_download.wait()
-        return "/tmp/nanobot_dingtalk/user1/report.xlsx"
-
-    async def block_message(*_args) -> None:
-        message_task_started.set()
-        await asyncio.Future()
-
-    monkeypatch.setattr(dingtalk_module, "ChatbotMessage", _FakeFileChatbotMessage)
-    monkeypatch.setattr(dingtalk_module, "AckMessage", SimpleNamespace(STATUS_OK="OK"))
-    monkeypatch.setattr(channel, "_download_dingtalk_file", delayed_download)
-    monkeypatch.setattr(channel, "_on_message", block_message)
-
-    process_task = asyncio.create_task(handler.process(SimpleNamespace(data={
-        "conversationType": "1",
-        "content": {"downloadCode": "abc123", "fileName": "report.xlsx"},
-        "text": {"content": ""},
-    })))
-    await download_started.wait()
-
-    try:
-        await channel.stop()
-        release_download.set()
-        assert await process_task == ("OK", "OK")
-        await asyncio.sleep(0)
-
-        assert not message_task_started.is_set()
-        assert not channel._background_tasks
-    finally:
-        release_download.set()
-        if not process_task.done():
-            process_task.cancel()
-        pending = tuple(channel._background_tasks)
-        for task in pending:
-            task.cancel()
-        await asyncio.gather(process_task, *pending, return_exceptions=True)
-
-
 def _rich_text_message(rich_text_list):
     class _FakeRichTextChatbotMessage:
         text = None
@@ -771,41 +650,6 @@ async def test_stop_cancels_stream_client_after_sdk_swallows_first_cancel(monkey
     assert start_task.cancelled()
 
 
-@pytest.mark.asyncio
-async def test_stop_waits_for_background_message_tasks() -> None:
-    channel = DingTalkChannel(
-        DingTalkConfig(client_id="app", client_secret="secret", allow_from=["*"]),
-        MessageBus(),
-    )
-    mock_logger = MagicMock()
-    channel.logger = mock_logger
-    started = asyncio.Event()
-    cancelled = asyncio.Event()
-
-    async def wait_forever() -> None:
-        started.set()
-        try:
-            await asyncio.Future()
-        finally:
-            cancelled.set()
-
-    task = asyncio.create_task(wait_forever())
-    channel._background_tasks.add(task)
-    task.add_done_callback(channel._on_background_task_done)
-    await started.wait()
-
-    try:
-        await channel.stop()
-        assert task.done()
-        assert cancelled.is_set()
-        assert not channel._background_tasks
-        mock_logger.opt.assert_not_called()
-    finally:
-        if not task.done():
-            task.cancel()
-        await asyncio.gather(task, return_exceptions=True)
-
-
 @pytest.mark.asyncio
 async def test_download_dingtalk_file(tmp_path, monkeypatch) -> None:
     """Test the two-step file download flow (get URL then download content)."""

nanobot/channels/email/runtime.py

@@ -430,13 +430,7 @@ class EmailChannel(BaseChannel):
         skipped_uids: set[str],
         cycle_uids: set[str],
     ) -> list[dict[str, Any]] | None:
-        """Fetch messages by arbitrary IMAP search criteria.
+        """Fetch messages by arbitrary IMAP search criteria."""
-
-        Uses UID SEARCH so already-processed UIDs are recognized before any
-        FETCH at all, then fetches headers only to evaluate every filter — the
-        full body (and any attachments) is downloaded only for messages that
-        pass every check and are actually going to be delivered.
-        """
         mailbox = self.config.imap_mailbox or "INBOX"
 
         client = self._open_imap_client(mailbox=mailbox, missing_mailbox_ok=True)
@@ -444,30 +438,29 @@ class EmailChannel(BaseChannel):
             return messages
 
         try:
-            status, data = client.uid("SEARCH", None, *search_criteria)
+            status, data = client.search(None, *search_criteria)
-            if status != "OK" or not data or not data[0]:
+            if status != "OK" or not data:
                 return messages
 
-            uids = [raw.decode("ascii", errors="ignore") for raw in data[0].split()]
+            ids = data[0].split()
-            if limit > 0 and len(uids) > limit:
+            if limit > 0 and len(ids) > limit:
-                uids = uids[-limit:]
+                ids = ids[-limit:]
-
+            for imap_id in ids:
-            features: _ServerFeatures | None = None
+                status, fetched = client.fetch(imap_id, "(BODY.PEEK[] UID)")
-
-            for uid in uids:
-                if not uid or uid in cycle_uids:
-                    continue
-                if dedupe and uid in self._processed_uids:
-                    continue
-
-                status, fetched = client.uid("FETCH", uid, "(BODY.PEEK[HEADER])")
                 if status != "OK" or not fetched:
                     continue
-                header_bytes = self._extract_message_bytes(fetched)
+
-                if header_bytes is None:
+                raw_bytes = self._extract_message_bytes(fetched)
+                if raw_bytes is None:
                     continue
 
-                parsed = BytesParser(policy=policy.default).parsebytes(header_bytes)
+                uid = self._extract_uid(fetched)
+                if uid and uid in cycle_uids:
+                    continue
+                if dedupe and uid and uid in self._processed_uids:
+                    continue
+
+                parsed = BytesParser(policy=policy.default).parsebytes(raw_bytes)
                 sender = parseaddr(parsed.get("From", ""))[1].strip().lower()
                 if not sender:
                     continue
@@ -475,7 +468,8 @@ class EmailChannel(BaseChannel):
                     self.logger.info("From {} ignored: matches bot-owned address", sender)
                     self._remember_processed_uid(uid, dedupe, cycle_uids)
                     if mark_seen:
-                        features = self._mark_seen_uid(client, uid, features)
+                        client.store(imap_id, "+FLAGS", "\\Seen")
+                    if uid:
                         skipped_uids.add(uid)
                     continue
 
@@ -488,6 +482,7 @@ class EmailChannel(BaseChannel):
                         sender,
                     )
                     self._remember_processed_uid(uid, dedupe, cycle_uids)
+                    if uid:
                         skipped_uids.add(uid)
                     continue
                 if self.config.verify_dkim and not dkim_pass:
@@ -497,26 +492,18 @@ class EmailChannel(BaseChannel):
                         sender,
                     )
                     self._remember_processed_uid(uid, dedupe, cycle_uids)
+                    if uid:
                         skipped_uids.add(uid)
                     continue
 
                 if not self.is_allowed(sender):
                     self._remember_processed_uid(uid, dedupe, cycle_uids)
                     if mark_seen:
-                        features = self._mark_seen_uid(client, uid, features)
+                        client.store(imap_id, "+FLAGS", "\\Seen")
+                    if uid:
                         skipped_uids.add(uid)
                     continue
 
-                # Passed every filter — only now fetch the full message body
-                # (and any attachments) for the message we're actually delivering.
-                status, full_fetched = client.uid("FETCH", uid, "(BODY.PEEK[])")
-                if status != "OK" or not full_fetched:
-                    continue
-                raw_bytes = self._extract_message_bytes(full_fetched)
-                if raw_bytes is None:
-                    continue
-                parsed = BytesParser(policy=policy.default).parsebytes(raw_bytes)
-
                 subject = self._decode_header_value(parsed.get("Subject", ""))
                 date_value = parsed.get("Date", "")
                 message_id = parsed.get("Message-ID", "").strip()
@@ -569,19 +556,10 @@ class EmailChannel(BaseChannel):
                 self._remember_processed_uid(uid, dedupe, cycle_uids)
 
                 if mark_seen:
-                    features = self._mark_seen_uid(client, uid, features)
+                    client.store(imap_id, "+FLAGS", "\\Seen")
         finally:
             self._close_imap_client(client)
 
-    def _mark_seen_uid(
-        self, client: Any, uid: str, features: _ServerFeatures | None
-    ) -> _ServerFeatures:
-        """Mark a single UID \\Seen, reusing session-learned STORE support."""
-        if features is None:
-            features = self._server_features(client)
-        self._uid_store_flag(client, uid, "\\Seen", features)
-        return features
-
     def _open_imap_client(self, mailbox: str, *, missing_mailbox_ok: bool = False) -> Any | None:
         if self.config.imap_use_ssl:
             client: Any = imaplib.IMAP4_SSL(self.config.imap_host, self.config.imap_port)
@@ -736,14 +714,11 @@ class EmailChannel(BaseChannel):
         return data[0].split()[0]
 
     def _uid_store_deleted(self, client: Any, uid: str, features: _ServerFeatures) -> bool:
-        return self._uid_store_flag(client, uid, "\\Deleted", features)
-
-    def _uid_store_flag(self, client: Any, uid: str, flag: str, features: _ServerFeatures) -> bool:
         # Optimistic path: try UID STORE first because UID is stable and avoids
         # sequence-number lookup. If this fails once for the session, remember it
         # and use the sequence STORE fallback directly for remaining UIDs.
         if features.uid_store is not False:
-            status, _ = client.uid("STORE", uid, "+FLAGS", f"({flag})")
+            status, _ = client.uid("STORE", uid, "+FLAGS", "(\\Deleted)")
             if status == "OK":
                 features.uid_store = True
                 return True
@@ -753,12 +728,12 @@ class EmailChannel(BaseChannel):
         # unreliable: resolve the current sequence number from UID and use STORE.
         imap_id = self._lookup_imap_id_by_uid(client, uid)
         if not imap_id:
-            self.logger.warning("Could not locate UID {} to set flag {}", uid, flag)
+            self.logger.warning("Post-action skipped: UID {} not found", uid)
             return False
 
-        status, _ = client.store(imap_id, "+FLAGS", flag)
+        status, _ = client.store(imap_id, "+FLAGS", "\\Deleted")
         if status != "OK":
-            self.logger.warning("Failed to set flag {} on UID {}", flag, uid)
+            self.logger.warning("Post-action failed: could not mark UID {} as deleted", uid)
             return False
         return True
 
@@ -798,6 +773,16 @@ class EmailChannel(BaseChannel):
                     return bytes(fetched_item[1])
         return None
 
+    @staticmethod
+    def _extract_uid(fetched: list[Any]) -> str:
+        for item in fetched:
+            if isinstance(item, tuple) and item and isinstance(item[0], (bytes, bytearray)):
+                head = bytes(item[0]).decode("utf-8", errors="ignore")
+                m = re.search(r"UID\s+(\d+)", head)
+                if m:
+                    return m.group(1)
+        return ""
+
     @staticmethod
     def _decode_header_value(value: str) -> str:
         if not value:

nanobot/channels/email/tests/test_email_channel.py

@@ -53,7 +53,30 @@ def _make_raw_email(
 def test_fetch_new_messages_parses_unseen_and_marks_seen(monkeypatch) -> None:
     raw = _make_raw_email(subject="Invoice", body="Please pay")
 
-    fake = _make_fake_imap(raw, uid=b"123")
+    class FakeIMAP:
+        def __init__(self) -> None:
+            self.store_calls: list[tuple[bytes, str, str]] = []
+
+        def login(self, _user: str, _pw: str):
+            return "OK", [b"logged in"]
+
+        def select(self, _mailbox: str):
+            return "OK", [b"1"]
+
+        def search(self, *_args):
+            return "OK", [b"1"]
+
+        def fetch(self, _imap_id: bytes, _parts: str):
+            return "OK", [(b"1 (UID 123 BODY[] {200})", raw), b")"]
+
+        def store(self, imap_id: bytes, op: str, flags: str):
+            self.store_calls.append((imap_id, op, flags))
+            return "OK", [b""]
+
+        def logout(self):
+            return "BYE", [b""]
+
+    fake = FakeIMAP()
     monkeypatch.setattr("nanobot.channels.email.runtime.imaplib.IMAP4_SSL", lambda _h, _p: fake)
 
     channel = EmailChannel(_make_config(), MessageBus())
@@ -63,25 +86,38 @@ def test_fetch_new_messages_parses_unseen_and_marks_seen(monkeypatch) -> None:
     assert items[0]["sender"] == "alice@example.com"
     assert items[0]["subject"] == "Invoice"
     assert "Please pay" in items[0]["content"]
-    assert ("STORE", "123", "+FLAGS", "(\\Seen)") in fake.uid_calls
+    assert fake.store_calls == [(b"1", "+FLAGS", "\\Seen")]
-    assert [call for call in fake.uid_calls if call[0] == "FETCH"] == [
-        ("FETCH", "123", "(BODY.PEEK[HEADER])"),
-        ("FETCH", "123", "(BODY.PEEK[])"),
-    ]
     assert skipped_uids == set()
 
     # Same UID should be deduped in-process.
     items_again, skipped_again = channel._fetch_new_messages()
     assert items_again == []
     assert skipped_again == set()
-    assert len([call for call in fake.uid_calls if call[0] == "FETCH"]) == 2
 
 
 def test_fetch_new_messages_returns_accepted_and_skipped_uids(monkeypatch) -> None:
     raw = _make_raw_email(subject="Invoice", body="Please pay")
 
-    fake = _make_fake_imap(raw, uid=b"123")
+    class FakeIMAP:
-    monkeypatch.setattr("nanobot.channels.email.runtime.imaplib.IMAP4_SSL", lambda _h, _p: fake)
+        def login(self, _user: str, _pw: str):
+            return "OK", [b"logged in"]
+
+        def select(self, _mailbox: str):
+            return "OK", [b"1"]
+
+        def search(self, *_args):
+            return "OK", [b"1"]
+
+        def fetch(self, _imap_id: bytes, _parts: str):
+            return "OK", [(b"1 (UID 123 BODY[] {200})", raw), b")"]
+
+        def store(self, _imap_id: bytes, _op: str, _flags: str):
+            return "OK", [b""]
+
+        def logout(self):
+            return "BYE", [b""]
+
+    monkeypatch.setattr("nanobot.channels.email.runtime.imaplib.IMAP4_SSL", lambda _h, _p: FakeIMAP())
 
     channel = EmailChannel(_make_config(post_action="delete"), MessageBus())
     items, skipped_uids = channel._fetch_new_messages()
@@ -94,10 +130,26 @@ def test_fetch_new_messages_returns_accepted_and_skipped_uids(monkeypatch) -> No
 def test_fetch_new_messages_rejected_returns_skipped_uid(monkeypatch) -> None:
     raw = _make_raw_email(from_addr="Nanobot <bot@example.com>", subject="Loop test")
 
-    monkeypatch.setattr(
+    class FakeIMAP:
-        "nanobot.channels.email.runtime.imaplib.IMAP4_SSL",
+        def login(self, _user: str, _pw: str):
-        lambda _h, _p: _make_fake_imap(raw, uid=b"123"),
+            return "OK", [b"logged in"]
-    )
+
+        def select(self, _mailbox: str):
+            return "OK", [b"1"]
+
+        def search(self, *_args):
+            return "OK", [b"1"]
+
+        def fetch(self, _imap_id: bytes, _parts: str):
+            return "OK", [(b"1 (UID 123 BODY[] {200})", raw), b")"]
+
+        def store(self, _imap_id: bytes, _op: str, _flags: str):
+            return "OK", [b""]
+
+        def logout(self):
+            return "BYE", [b""]
+
+    monkeypatch.setattr("nanobot.channels.email.runtime.imaplib.IMAP4_SSL", lambda _h, _p: FakeIMAP())
 
     channel_skip = EmailChannel(
         _make_config(from_address="bot@example.com", post_action="delete", post_action_ignore_skipped=True),
@@ -493,7 +545,30 @@ async def test_start_keeps_post_actions_for_successful_emails_when_later_deliver
 def test_fetch_new_messages_skips_self_sent_email_and_marks_seen(monkeypatch) -> None:
     raw = _make_raw_email(from_addr="Nanobot <bot@example.com>", subject="Loop test")
 
-    fake = _make_fake_imap(raw, uid=b"123")
+    class FakeIMAP:
+        def __init__(self) -> None:
+            self.store_calls: list[tuple[bytes, str, str]] = []
+
+        def login(self, _user: str, _pw: str):
+            return "OK", [b"logged in"]
+
+        def select(self, _mailbox: str):
+            return "OK", [b"1"]
+
+        def search(self, *_args):
+            return "OK", [b"1"]
+
+        def fetch(self, _imap_id: bytes, _parts: str):
+            return "OK", [(b"1 (UID 123 BODY[] {200})", raw), b")"]
+
+        def store(self, imap_id: bytes, op: str, flags: str):
+            self.store_calls.append((imap_id, op, flags))
+            return "OK", [b""]
+
+        def logout(self):
+            return "BYE", [b""]
+
+    fake = FakeIMAP()
     monkeypatch.setattr("nanobot.channels.email.runtime.imaplib.IMAP4_SSL", lambda _h, _p: fake)
 
     channel = EmailChannel(_make_config(from_address="bot@example.com"), MessageBus())
@@ -501,7 +576,7 @@ def test_fetch_new_messages_skips_self_sent_email_and_marks_seen(monkeypatch) ->
 
     assert items == []
     assert skipped_uids == {"123"}
-    assert ("STORE", "123", "+FLAGS", "(\\Seen)") in fake.uid_calls
+    assert fake.store_calls == [(b"1", "+FLAGS", "\\Seen")]
 
     # Same UID should still be deduped after being ignored.
     items_again, skipped_again = channel._fetch_new_messages()
@@ -539,14 +614,37 @@ def test_fetch_new_messages_skips_self_sent_across_identity_sources(
     imap_username matches, and must be case-insensitive."""
     raw = _make_raw_email(from_addr=from_header, subject="Loop test")
 
-    fake = _make_fake_imap(raw, uid=b"123")
+    class FakeIMAP:
+        def __init__(self) -> None:
+            self.store_calls: list[tuple[bytes, str, str]] = []
+
+        def login(self, _user: str, _pw: str):
+            return "OK", [b"logged in"]
+
+        def select(self, _mailbox: str):
+            return "OK", [b"1"]
+
+        def search(self, *_args):
+            return "OK", [b"1"]
+
+        def fetch(self, _imap_id: bytes, _parts: str):
+            return "OK", [(b"1 (UID 123 BODY[] {200})", raw), b")"]
+
+        def store(self, imap_id: bytes, op: str, flags: str):
+            self.store_calls.append((imap_id, op, flags))
+            return "OK", [b""]
+
+        def logout(self):
+            return "BYE", [b""]
+
+    fake = FakeIMAP()
     monkeypatch.setattr("nanobot.channels.email.runtime.imaplib.IMAP4_SSL", lambda _h, _p: fake)
 
     channel = EmailChannel(_make_config(**config_override), MessageBus())
     items, _ = channel._fetch_new_messages()
 
     assert items == []
-    assert ("STORE", "123", "+FLAGS", "(\\Seen)") in fake.uid_calls
+    assert fake.store_calls == [(b"1", "+FLAGS", "\\Seen")]
 
 
 def test_fetch_new_messages_retries_once_when_imap_connection_goes_stale(monkeypatch) -> None:
@@ -564,16 +662,15 @@ def test_fetch_new_messages_retries_once_when_imap_connection_goes_stale(monkeyp
         def select(self, _mailbox: str):
             return "OK", [b"1"]
 
-        def uid(self, command: str, *args):
+        def search(self, *_args):
-            if command == "SEARCH":
             self.search_calls += 1
             if fail_once["pending"]:
                 fail_once["pending"] = False
                 raise imaplib.IMAP4.abort("socket error")
-                return "OK", [b"123"]
+            return "OK", [b"1"]
-            if command == "FETCH":
+
+        def fetch(self, _imap_id: bytes, _parts: str):
             return "OK", [(b"1 (UID 123 BODY[] {200})", raw), b")"]
-            return "OK", [b""]
 
         def store(self, imap_id: bytes, op: str, flags: str):
             self.store_calls.append((imap_id, op, flags))
@@ -603,7 +700,10 @@ def test_fetch_new_messages_retries_once_when_imap_connection_goes_stale(monkeyp
 def test_fetch_new_messages_keeps_messages_collected_before_stale_retry(monkeypatch) -> None:
     raw_first = _make_raw_email(subject="First", body="First body")
     raw_second = _make_raw_email(subject="Second", body="Second body")
-    mailbox_state = {"123": raw_first, "124": raw_second}
+    mailbox_state = {
+        b"1": {"uid": b"123", "raw": raw_first, "seen": False},
+        b"2": {"uid": b"124", "raw": raw_second, "seen": False},
+    }
     fail_once = {"pending": True}
 
     class FlakyIMAP:
@@ -613,18 +713,20 @@ def test_fetch_new_messages_keeps_messages_collected_before_stale_retry(monkeypa
         def select(self, _mailbox: str):
             return "OK", [b"2"]
 
-        def uid(self, command: str, *args):
+        def search(self, *_args):
-            if command == "SEARCH":
+            unseen_ids = [imap_id for imap_id, item in mailbox_state.items() if not item["seen"]]
-                keys = " ".join(sorted(mailbox_state.keys(), key=int))
+            return "OK", [b" ".join(unseen_ids)]
-                return "OK", [keys.encode()]
+
-            if command == "FETCH":
+        def fetch(self, imap_id: bytes, _parts: str):
-                uid = args[0]
+            if imap_id == b"2" and fail_once["pending"]:
-                if uid == "124" and fail_once["pending"]:
                 fail_once["pending"] = False
                 raise imaplib.IMAP4.abort("socket error")
-                raw = mailbox_state[uid]
+            item = mailbox_state[imap_id]
-                header = f"{uid} (UID {uid} BODY[] {{200}})".encode()
+            header = b"%s (UID %s BODY[] {200})" % (imap_id, item["uid"])
-                return "OK", [(header, raw), b")"]
+            return "OK", [(header, item["raw"]), b")"]
+
+        def store(self, imap_id: bytes, _op: str, _flags: str):
+            mailbox_state[imap_id]["seen"] = True
             return "OK", [b""]
 
         def logout(self):
@@ -942,13 +1044,12 @@ def test_fetch_messages_between_dates_uses_imap_since_before_without_mark_seen(m
         def select(self, _mailbox: str):
             return "OK", [b"1"]
 
-        def uid(self, command: str, *args):
+        def search(self, *_args):
-            if command == "SEARCH":
+            self.search_args = _args
-                self.search_args = args
+            return "OK", [b"5"]
-                return "OK", [b"999"]
+
-            if command == "FETCH":
+        def fetch(self, _imap_id: bytes, _parts: str):
             return "OK", [(b"5 (UID 999 BODY[] {200})", raw), b")"]
-            return "OK", [b""]
 
         def store(self, imap_id: bytes, op: str, flags: str):
             self.store_calls.append((imap_id, op, flags))
@@ -969,7 +1070,7 @@ def test_fetch_messages_between_dates_uses_imap_since_before_without_mark_seen(m
 
     assert len(items) == 1
     assert items[0]["subject"] == "Status"
-    # uid("SEARCH", None, "SINCE", "06-Feb-2026", "BEFORE", "07-Feb-2026")
+    # search(None, "SINCE", "06-Feb-2026", "BEFORE", "07-Feb-2026")
     assert fake.search_args is not None
     assert fake.search_args[1:] == ("SINCE", "06-Feb-2026", "BEFORE", "07-Feb-2026")
     assert fake.store_calls == []
@@ -979,12 +1080,11 @@ def test_fetch_messages_between_dates_uses_imap_since_before_without_mark_seen(m
 # Security: Anti-spoofing tests for Authentication-Results verification
 # ---------------------------------------------------------------------------
 
-def _make_fake_imap(raw: bytes, uid: bytes = b"500"):
+def _make_fake_imap(raw: bytes):
     """Return a FakeIMAP class pre-loaded with the given raw email."""
     class FakeIMAP:
         def __init__(self) -> None:
             self.store_calls: list[tuple[bytes, str, str]] = []
-            self.uid_calls: list[tuple] = []
 
         def login(self, _user: str, _pw: str):
             return "OK", [b"logged in"]
@@ -992,16 +1092,11 @@ def _make_fake_imap(raw: bytes, uid: bytes = b"500"):
         def select(self, _mailbox: str):
             return "OK", [b"1"]
 
-        def capability(self):
+        def search(self, *_args):
-            return "OK", [b"IMAP4rev1"]
+            return "OK", [b"1"]
 
-        def uid(self, command: str, *args):
+        def fetch(self, _imap_id: bytes, _parts: str):
-            self.uid_calls.append((command, *args))
+            return "OK", [(b"1 (UID 500 BODY[] {200})", raw), b")"]
-            if command == "SEARCH":
-                return "OK", [uid]
-            if command == "FETCH":
-                return "OK", [(b"1 (UID " + uid + b" BODY[] {200})", raw), b")"]
-            return "OK", [b""]
 
         def store(self, imap_id: bytes, op: str, flags: str):
             self.store_calls.append((imap_id, op, flags))
@@ -1197,10 +1292,7 @@ def test_fetch_new_messages_ignores_unauthorized_sender_before_attachments(monke
 
     assert channel._fetch_new_messages() == ([], {"500"})
     assert called["attachments"] is False
-    assert [call for call in fake.uid_calls if call[0] == "FETCH"] == [
+    assert fake.store_calls == [(b"1", "+FLAGS", "\\Seen")]
-        ("FETCH", "500", "(BODY.PEEK[HEADER])")
-    ]
-    assert ("STORE", "500", "+FLAGS", "(\\Seen)") in fake.uid_calls
 
 
 def test_extract_attachments_saves_pdf(tmp_path, monkeypatch) -> None:

nanobot/channels/telegram/runtime.py

@@ -897,68 +897,6 @@ class TelegramChannel(BaseChannel):
             self.logger.debug("sendRichMessage failed: {}", exc)
             return False
 
-    async def _try_edit_rich(self, chat_id: int, message_id: int, content: str) -> bool:
-        """Upgrade an existing message to rich in place via editMessageText (Bot API 10.1).
-
-        Editing in place keeps the message identity, so the streaming preview is
-        upgraded without the delete-and-resend pattern that caused flickering and
-        dropped line breaks (issue #4470).
-
-        Returns True when the rich edit is in place (including the ambiguous
-        "message is not modified" retry outcome after a response timeout).
-        Returns False only when the legacy HTML path should take over:
-        capability errors (server older than Bot API 10.1, which also trip the
-        rich latch) and content-shaped BadRequest rejections. Transport,
-        rate-limit, and unexpected errors propagate so the final-edit retry
-        contract is preserved — ChannelManager retries the buffered send
-        instead of an immediate legacy edit doubling connection demand.
-        """
-        if not self._app:
-            return False
-
-        payload: dict[str, Any] = {
-            "chat_id": chat_id,
-            "message_id": message_id,
-            "rich_message": {
-                "markdown": content,
-            },
-        }
-        try:
-            await self._call_with_retry(
-                self._app.bot.do_api_request,
-                "editMessageText",
-                api_kwargs=payload,
-            )
-            return True
-        except BadRequest as exc:
-            if self._is_not_modified_error(exc):
-                # Ambiguous success: the rich edit was applied server-side but
-                # its response timed out, so the retry hit "message is not
-                # modified". Treat it as done rather than letting the legacy
-                # edit overwrite the already-successful rich result.
-                self.logger.debug("Rich stream edit already applied for {}", chat_id)
-                return True
-            # Before Bot API 10.1, editMessageText ignores rich_message and
-            # reports the absent text argument instead.
-            pre_rich_edit_server = (
-                bool(content)
-                and str(exc).strip().lower() == "message text is empty"
-            )
-            if self._is_rich_capability_error(exc) or pre_rich_edit_server:
-                self.logger.debug("editMessageText rich_message not available, disabling")
-                self._rich_send_disabled = True
-                return False
-            # Content-shaped rejections (invalid markdown, unsupported media in
-            # the rich payload, …) fall back to the legacy HTML edit.
-            self.logger.debug("editMessageText rich_message rejected: {}", exc)
-            return False
-        except Exception:
-            # Transport, rate-limit, and unexpected errors propagate so the
-            # final-edit retry contract stays intact: ChannelManager retries
-            # the buffered send instead of this handler doubling connection
-            # demand with an immediate legacy edit.
-            raise
-
     async def send(self, msg: OutboundMessage) -> None:
         """Send a message through Telegram."""
         app = await self._wait_for_app()
@@ -1198,16 +1136,26 @@ class TelegramChannel(BaseChannel):
                 thread_kwargs["message_thread_id"] = message_thread_id
             raw_text = buf.text
 
-            # Try upgrading the streaming preview to rich in place (Bot API 10.1:
+            # Try sendRichMessage for final output (Bot API 10.1).
-            # editMessageText gained a rich_message parameter). Editing in place
+            # Skip when a streaming preview already exists to avoid the
-            # keeps the message identity, so there is no delete-and-resend and
+            # delete-and-resend pattern that causes flickering and drops
-            # none of the flickering / dropped line breaks from issue #4470.
+            # line breaks (issue #4470).
-            # The previous branch here was unreachable: it was guarded by
+            if not buf.message_id and self.config.rich_messages and not getattr(self, "_rich_send_disabled", False):
-            # ``not buf.message_id`` after an early return had already ensured
+                reply_params = None
-            # ``buf.message_id`` is set (issue #5516).
+                if reply_to_message_id := meta.get("message_id"):
-            if self.config.rich_messages and not getattr(self, "_rich_send_disabled", False):
+                    reply_params = {"message_id": int(reply_to_message_id), "allow_sending_without_reply": True}
-                rich_ok = await self._try_edit_rich(int_chat_id, buf.message_id, raw_text)
+                rich_ok = await self._try_send_rich(
+                    int_chat_id, raw_text, reply_params, thread_kwargs, None,
+                )
                 if rich_ok:
+                    # Delete the streaming preview message
+                    try:
+                        await self._call_with_retry(
+                            app.bot.delete_message,
+                            chat_id=int_chat_id, message_id=buf.message_id,
+                        )
+                    except Exception:
+                        pass  # Preview stays if delete fails
                     self._stream_bufs.pop(chat_id, None)
                     return
 

nanobot/channels/telegram/tests/test_telegram_channel.py

@@ -2735,130 +2735,3 @@ def test_markdown_to_html_code_block_same_line_no_newline() -> None:
 
     stripped = _strip_md_block(text)
     assert stripped == "Use <tag> here"
-
-
-@pytest.mark.asyncio
-async def test_send_delta_stream_end_upgrades_preview_to_rich_in_place() -> None:
-    """Rich messages finally work with streaming: the preview is upgraded via
-    editMessageText rich_message (in place), not delete-and-resend (issue #5516)."""
-    from telegram.error import BadRequest
-
-    channel = TelegramChannel(
-        TelegramConfig(enabled=True, token="123:abc", allow_from=["*"], rich_messages=True),
-        MessageBus(),
-    )
-    _install_ready_app(channel)
-    channel._app.bot.do_api_request = AsyncMock()
-    channel._app.bot.edit_message_text = AsyncMock(side_effect=BadRequest("should not be reached"))
-    channel._stream_bufs["123"] = _StreamBuf(text="**hello**", message_id=7, last_edit=0.0)
-
-    await channel.send_delta("123", "", stream_end=True)
-
-    # editMessageText with rich_message payload, in place (same message_id)
-    channel._app.bot.do_api_request.assert_awaited_once()
-    args, kwargs = channel._app.bot.do_api_request.await_args
-    assert args[0] == "editMessageText"
-    assert kwargs["api_kwargs"]["chat_id"] == 123
-    assert kwargs["api_kwargs"]["message_id"] == 7
-    assert kwargs["api_kwargs"]["rich_message"] == {"markdown": "**hello**"}
-    # No delete-and-resend, no legacy HTML edit
-    channel._app.bot.edit_message_text.assert_not_awaited()
-    assert "123" not in channel._stream_bufs
-
-
-@pytest.mark.asyncio
-async def test_send_delta_stream_end_rich_capability_error_latches_and_falls_back() -> None:
-    """On a pre-10.1 Bot API server the rich edit fails, the latch trips, and the
-    legacy HTML edit handles the final output."""
-    from telegram.error import BadRequest
-
-    channel = TelegramChannel(
-        TelegramConfig(enabled=True, token="123:abc", allow_from=["*"], rich_messages=True),
-        MessageBus(),
-    )
-    _install_ready_app(channel)
-    # Before Bot API 10.1, editMessageText ignores rich_message and requires text.
-    channel._app.bot.do_api_request = AsyncMock(
-        side_effect=BadRequest("Message text is empty")
-    )
-    channel._app.bot.edit_message_text = AsyncMock()
-    channel._stream_bufs["123"] = _StreamBuf(text="hello", message_id=7, last_edit=0.0)
-
-    await channel.send_delta("123", "", stream_end=True)
-
-    channel._app.bot.do_api_request.assert_awaited_once()
-    # Latch tripped: subsequent sends skip the rich path entirely
-    assert channel._rich_send_disabled is True
-    # Legacy HTML edit handled the final message
-    channel._app.bot.edit_message_text.assert_awaited_once()
-    assert "123" not in channel._stream_bufs
-
-
-@pytest.mark.asyncio
-async def test_send_delta_stream_end_rich_disabled_uses_legacy_html() -> None:
-    """rich_messages=False (the default) keeps the legacy HTML path untouched."""
-    channel = TelegramChannel(
-        TelegramConfig(enabled=True, token="123:abc", allow_from=["*"]),
-        MessageBus(),
-    )
-    _install_ready_app(channel)
-    channel._app.bot.do_api_request = AsyncMock()
-    channel._app.bot.edit_message_text = AsyncMock()
-    channel._stream_bufs["123"] = _StreamBuf(text="hello", message_id=7, last_edit=0.0)
-
-    await channel.send_delta("123", "", stream_end=True)
-
-    channel._app.bot.do_api_request.assert_not_called()
-    channel._app.bot.edit_message_text.assert_awaited_once()
-    assert "123" not in channel._stream_bufs
-
-
-@pytest.mark.asyncio
-async def test_send_delta_stream_end_rich_network_error_propagates_for_retry() -> None:
-    """A transport failure on the rich edit must propagate so ChannelManager
-    retries the buffered send — not fall through to an immediate legacy edit
-    that doubles connection demand during pool exhaustion."""
-    from telegram.error import NetworkError
-
-    channel = TelegramChannel(
-        TelegramConfig(enabled=True, token="123:abc", allow_from=["*"], rich_messages=True),
-        MessageBus(),
-    )
-    _install_ready_app(channel)
-    channel._app.bot.do_api_request = AsyncMock(side_effect=NetworkError("pool exhausted"))
-    channel._app.bot.edit_message_text = AsyncMock()
-    channel._stream_bufs["123"] = _StreamBuf(text="hello", message_id=7, last_edit=0.0)
-
-    with pytest.raises(NetworkError):
-        await channel.send_delta("123", "", stream_end=True)
-
-    # No legacy fallback edit: the buffered state stays for the manager retry.
-    channel._app.bot.edit_message_text.assert_not_awaited()
-    assert "123" in channel._stream_bufs
-
-
-@pytest.mark.asyncio
-async def test_send_delta_stream_end_rich_not_modified_after_timeout_is_success() -> None:
-    """Ambiguous success: the rich edit applied server-side but its response
-    timed out, so the retry hit "message is not modified". That is a completed
-    rich upgrade — the legacy edit must not overwrite it."""
-    from telegram.error import BadRequest, TimedOut
-
-    channel = TelegramChannel(
-        TelegramConfig(enabled=True, token="123:abc", allow_from=["*"], rich_messages=True),
-        MessageBus(),
-    )
-    _install_ready_app(channel)
-    # First attempt (inside _call_with_retry) times out, retry reports the
-    # edit as already applied.
-    channel._app.bot.do_api_request = AsyncMock(
-        side_effect=[TimedOut(), BadRequest("Message is not modified")]
-    )
-    channel._app.bot.edit_message_text = AsyncMock(side_effect=AssertionError("must not overwrite rich result"))
-    channel._stream_bufs["123"] = _StreamBuf(text="hello", message_id=7, last_edit=0.0)
-
-    await channel.send_delta("123", "", stream_end=True)
-
-    assert channel._app.bot.do_api_request.await_count == 2
-    channel._app.bot.edit_message_text.assert_not_awaited()
-    assert "123" not in channel._stream_bufs

nanobot/channels/websocket/runtime.py

@@ -3,7 +3,6 @@
 from __future__ import annotations
 
 import asyncio
-import errno
 import ipaddress
 import json
 import socket
@@ -539,32 +538,14 @@ class WebSocketChannel(BaseChannel):
     # -- Server lifecycle and connection ingress ---------------------------
 
     @staticmethod
-    def _socket_is_accepting(sock: socket.socket) -> bool:
+    def _listener_is_serving(server: Server) -> bool:
-        """Return whether a bound socket still advertises a listen capability.
-
-        ``SO_ACCEPTCONN`` is not portable: macOS/BSD raise ``OSError`` with
-        ``ENOPROTOOPT`` ("Protocol not available") for this option even on a
-        perfectly healthy listening socket. Treating that as "not serving"
-        makes the listener look permanently degraded, so the caller retries
-        forever and the channel never reaches a ready state. When the option
-        is unavailable we fall back to the file-descriptor liveness check.
-        """
-        if sock.fileno() < 0:
-            return False
-        try:
-            return bool(sock.getsockopt(socket.SOL_SOCKET, socket.SO_ACCEPTCONN))
-        except OSError as exc:
-            if exc.errno in (errno.ENOPROTOOPT, errno.EOPNOTSUPP):
-                return True
-            raise
-
-    @classmethod
-    def _listener_is_serving(cls, server: Server) -> bool:
         """Return whether every bound socket still has a live listen capability."""
         try:
             sockets = server.sockets
             return bool(sockets) and server.is_serving() and all(
-                cls._socket_is_accepting(sock) for sock in sockets
+                sock.fileno() >= 0
+                and bool(sock.getsockopt(socket.SOL_SOCKET, socket.SO_ACCEPTCONN))
+                for sock in sockets
             )
         except OSError:
             return False

nanobot/cron/service.py

@@ -25,7 +25,6 @@ from nanobot.cron.types import (
     CronSchedule,
     CronStore,
 )
-from nanobot.runtime_context import RUNTIME_CONTEXT_INPUT_META
 from nanobot.utils.run_records import (
     write_run_record as write_automation_run_record,
 )
@@ -116,21 +115,8 @@ def _disable_malformed_legacy_job(job: CronJob) -> None:
     logger.warning("Cron: disabled malformed legacy job '{}' ({}): {}", job.name, job.id, reason)
 
 
-def _persistable_origin_metadata(metadata: dict[str, Any]) -> dict[str, Any]:
-    """Return a detached JSON-safe routing snapshot for a cron payload."""
-    snapshot: dict[str, Any] = {}
-    for key, value in metadata.items():
-        if key == RUNTIME_CONTEXT_INPUT_META:
-            continue
-        try:
-            snapshot[key] = json.loads(json.dumps(value, ensure_ascii=False, allow_nan=False))
-        except (TypeError, ValueError, RecursionError):
-            continue
-    return snapshot
-
-
 def _normalize_agent_turn_job(job: CronJob) -> bool:
-    """Make routing metadata persistable and migrate legacy user cron payloads.
+    """Migrate legacy user cron payloads into session-bound payloads.
 
     Pre-bound user cron jobs stored their delivery target in ``channel``/``to``.
     Normal user-created legacy jobs always have those fields; if they are
@@ -138,12 +124,8 @@ def _normalize_agent_turn_job(job: CronJob) -> bool:
     a runtime legacy execution path.
     """
     payload = job.payload
-    origin_metadata = _persistable_origin_metadata(payload.origin_metadata)
-    changed = origin_metadata != payload.origin_metadata
-    payload.origin_metadata = origin_metadata
-
     if payload.kind != "agent_turn" or not _has_legacy_delivery_context(payload):
-        return changed
+        return False
 
     if not payload.channel or not payload.to:
         _disable_malformed_legacy_job(job)
@@ -153,7 +135,7 @@ def _normalize_agent_turn_job(job: CronJob) -> bool:
     payload.origin_channel = payload.origin_channel or payload.channel
     payload.origin_chat_id = payload.origin_chat_id or payload.to
     if not payload.origin_metadata:
-        payload.origin_metadata = _persistable_origin_metadata(payload.channel_meta or {})
+        payload.origin_metadata = dict(payload.channel_meta or {})
 
     payload.deliver = False
     payload.channel = None

nanobot/providers/base.py

@@ -1029,20 +1029,6 @@ class LLMProvider(ABC):
         # Unknown 429 defaults to WAIT+retry.
         return True
 
-    @staticmethod
-    def _content_as_blocks(content: Any) -> list[dict[str, Any]]:
-        """Convert message content to blocks so mixed user content can be merged."""
-        if isinstance(content, list):
-            return [
-                dict(cast(dict[str, Any], item))
-                if isinstance(item, dict)
-                else {"type": "text", "text": str(item)}
-                for item in cast(list[object], content)
-            ]
-        if content is None:
-            return []
-        return [{"type": "text", "text": str(content)}]
-
     @staticmethod
     def _enforce_role_alternation(messages: list[dict[str, Any]]) -> list[dict[str, Any]]:
         """Merge consecutive same-role messages and drop trailing assistant messages.
@@ -1077,13 +1063,6 @@ class LLMProvider(ABC):
                 curr_content = msg.get("content") or ""
                 if isinstance(prev_content, str) and isinstance(curr_content, str):
                     prev["content"] = (prev_content + "\n\n" + curr_content).strip()
-                elif role == "user":
-                    combined = dict(msg)
-                    combined["content"] = [
-                        *LLMProvider._content_as_blocks(prev_content),
-                        *LLMProvider._content_as_blocks(curr_content),
-                    ]
-                    merged[-1] = combined
                 else:
                     merged[-1] = dict(msg)
             else:

nanobot/providers/conversation_state.py

@@ -11,7 +11,6 @@ from nanobot.providers.base import (
     ProviderCallContext,
     ProviderConversationState,
 )
-from nanobot.utils.helpers import estimate_prompt_tokens_chain
 
 _PROVIDER_STATE_OUTPUT_META = "provider_state_output"
 _PROVIDER_STATE_BOUNDARY_META = "provider_state_boundary"
@@ -70,37 +69,6 @@ class ProviderConversationStateController:
             session_id=self._session_id,
         )
 
-    def estimate_request_context_tokens(
-        self,
-        messages: list[dict[str, Any]],
-        *,
-        model_messages: list[dict[str, Any]] | None = None,
-        supplemental_messages: list[dict[str, Any]] | None = None,
-        tool_definitions: list[dict[str, Any]] | None = None,
-    ) -> int | None:
-        """Estimate resumed state plus the pending delta for the next request."""
-        state = self.checkpoint(messages, model_messages=model_messages)
-        if state is None:
-            return None
-        context_tokens = state.payload.get("context_tokens")
-        if (
-            isinstance(context_tokens, bool)
-            or not isinstance(context_tokens, int)
-            or context_tokens < 0
-        ):
-            return None
-        pending_messages = [
-            *state.pending_messages,
-            *(supplemental_messages or []),
-        ]
-        delta_tokens, _ = estimate_prompt_tokens_chain(
-            self._provider,
-            self._model,
-            pending_messages,
-            tool_definitions,
-        )
-        return context_tokens + max(0, delta_tokens)
-
     def prepare_request(
         self,
         messages: list[dict[str, Any]],
@@ -108,20 +76,11 @@ class ProviderConversationStateController:
         context_window_tokens: int | None,
         model_messages: list[dict[str, Any]] | None = None,
         supplemental_messages: list[dict[str, Any]] | None = None,
-        resume_state: bool = True,
     ) -> ProviderCallContext | None:
-        """Build context for the next request and remember its durable delta.
+        """Build typed context for the next request and remember its durable delta."""
-
-        ``resume_state=False`` abandons opaque history when local request
-        fitting has produced a new independent model-facing context.
-        """
         independent_context = self.independent_request_context(
             context_window_tokens=context_window_tokens,
         )
-        if not resume_state:
-            self._state = None
-            self._request_messages = []
-            return independent_context
         if self._state is None:
             self._request_messages = []
             return independent_context

nanobot/session/turn_continuation.py

@@ -147,10 +147,10 @@ def prepare_save_boundary(ctx: TurnContext) -> None:
     if ctx.session is not None:
         clear_internal_continuation_state(ctx.session.metadata)
 
-    assert ctx.transcript_input is not None
     ctx.save_skip = _save_skip_for_turn(
         message_metadata=ctx.msg.metadata,
-        initial_message_count=ctx.transcript_input.message_count,
+        initial_message_count=len(ctx.initial_messages),
+        history_count=len(ctx.history),
         input_persisted_early=ctx.input_persisted_early,
     )
 
@@ -185,6 +185,7 @@ def _save_skip_for_turn(
     *,
     message_metadata: Mapping[str, Any] | None,
     initial_message_count: int,
+    history_count: int,
     input_persisted_early: bool,
 ) -> int:
     """Return the persisted-message append boundary for this turn."""
@@ -192,7 +193,10 @@ def _save_skip_for_turn(
         return initial_message_count
     if internal_continuation_inbound(message_metadata):
         return initial_message_count
-    if not input_persisted_early:
+    # build_messages may merge the current message into a same-role history tail.
+    # Runner-appended messages start at initial_message_count in either shape.
+    has_standalone_current = initial_message_count > 1 + history_count
+    if has_standalone_current and not input_persisted_early:
         return initial_message_count - 1
     return initial_message_count
 

nanobot/templates/agent/consolidator_archive.md

@@ -1,42 +1,27 @@
-Create a compact replacement checkpoint for this session.
+Create a memory overview for only the final {{ archive_count }} conversation messages immediately before this instruction. Earlier messages are context for resolving references; do not summarize them again.
 
-When `[Archived Context Summary]` appears in the system prompt, update that previous checkpoint to reflect the current conversation state.
+Use [skip] unless a fact meets all SNIP criteria:
+- Signal: would the user need to repeat this if forgotten?
+- Novel: not just a restatement of another fact in this same conversation chunk
+- Important: prevents rework or captures preferences / rules
+- Persistent: still relevant after 2 weeks
 
-## Merge rules
+Also preserve a compact working-state handoff even when it is not Persistent: the active objective, current status, completed steps, unresolved blockers, next action, and exact identifiers needed to continue without rework. Mark these facts [ephemeral].
 
-- Use the latest correction or decision as the current version of a fact, and merge duplicates.
+Format each fact as:
-- Preserve exact names, identifiers, paths, commands, decisions, results, and unresolved blockers when they are needed to continue the session.
+- [mark] fact content
-- Retain a fact already present in long-term memory when it is needed for session continuity.
 
-## What to retain
+Marks (choose the best match):
+- [permanent] Core preferences, personal traits, habits — never becomes stale
+- [durable] Technical discoveries, project knowledge, config details — valid for months
+- [ephemeral] Active task state, temporary decisions — may change in weeks
+- [correction] Correction to a previous memory — state what changed
+- [skip] Conversational filler, code/source facts derivable from the repo, or audit-only breadcrumbs
 
-Always retain a compact working-state handoff:
+Priority: user corrections and preferences > solutions > decisions > events > environment facts.
-- active objective
-- current status
-- completed results that constrain later work
-- unresolved blockers
-- next action
-- exact identifiers needed for that action
 
-Mark working-state facts `[ephemeral]`.
+Do not output facts already present in the system prompt's Recent History.
 
-For other facts, retain a candidate only when it meets all four SNIP criteria:
+Do not mark something [skip] merely because it might already exist in long-term memory.
-- Signal: remembering it saves the user from repeating it
-- Novel: it adds a distinct fact to this checkpoint
-- Important: losing it would cause rework or discard a preference or rule
-- Persistent: it is expected to remain useful for at least two weeks
 
-Assign each retained fact its best current mark:
+Return only formatted fact lines, or `(nothing)` if nothing noteworthy happened.
-- `[permanent]` for core preferences, personal traits, and habits that remain relevant indefinitely
-- `[durable]` for technical discoveries, project knowledge, and configuration that remains valid for months
-- `[ephemeral]` for active task state and temporary decisions that may change within weeks
-- `[correction]` for the current fact that supersedes conflicting earlier long-term memory
-
-When space is limited, prioritize user corrections and preferences, then solutions, decisions, events, and environment facts.
-
-## Output
-
-Return one concise retained fact per line in this form:
-- [mark] fact
-
-Use `(nothing)` when no fact qualifies and there is no active working state.

nanobot/templates/agent/dream.md

@@ -99,7 +99,7 @@ For [SKILL] entries:
 - Skills are instruction sets with concrete values, commands, and examples. MEMORY.md keeps strategic context and high-level facts only.
 
 ## Editing
-- Current contents of SOUL.md, USER.md, and memory/MEMORY.md are provided by the agent system context. Edit those files directly; do not rely on a remembered version of a file.
+- Current contents of SOUL.md, USER.md, and memory/MEMORY.md are embedded in this prompt under "Current Memory Files". Edit those files directly; do not rely on a remembered version of a file.
 - Batch changes into as few calls as possible. Surgical edits only.
 
 ## Verification

nanobot/templates/agent/tool_contract.md

@@ -40,7 +40,7 @@
   result with its original consumer or checker when one is available.
 - Use `apply_patch` as the default code editing tool, especially for multi-file changes, structural edits, generated code, moves, adds, or deletes.
 - Use `apply_patch dry_run=true` when the patch is uncertain and you want validation plus a change summary before writing.
-- Use `edit_file` only for small exact replacements in one file, with `old_text` copied from `read_file`.
+- Use `edit_file` only for small exact replacements in one file, with `old_text` copied from `read_file`; when editing a specific numbered line, pass that exact line as `line_hint`; add `occurrence` or `expected_replacements` when ambiguity matters.
 - Use `write_file` for new files or intentional full-file rewrites, not routine partial edits.
 - If `apply_patch` or `edit_file` fails, re-read with `force=true`, narrow the context, and try a smaller patch rather than switching to shell `sed` or `echo`.
 

tests/agent/test_attachment_references.py

@@ -5,7 +5,6 @@ from unittest.mock import AsyncMock, MagicMock
 
 import pytest
 
-from nanobot.agent.context import TranscriptInput
 from nanobot.agent.loop import AgentLoop, TurnContext, TurnKind
 from nanobot.agent.tools.context import RequestContext
 from nanobot.agent.tools.filesystem import ReadFileTool
@@ -149,10 +148,7 @@ async def test_pending_document_attachment_keeps_body_out_of_prompt(
 
     runtime = loop.llm_runtime()
     result = await loop._run_agent_loop(
-        TranscriptInput(
+        [{"role": "user", "content": "hello"}],
-            history=[{"role": "user", "content": "hello"}],
-            current_message=None,
-        ),
         runtime=runtime,
         request_context=RequestContext(channel="cli", chat_id="c", runtime=runtime),
         pending_queue=pending_queue,

tests/agent/test_auto_compact.py

@@ -1302,9 +1302,9 @@ class TestSummaryPersistence:
         assert "_last_summary" in reloaded.metadata
 
         # Simulate /new command
-        reloaded.clear()
+        session.clear()
-        loop.sessions.save(reloaded)
+        loop.sessions.save(session)
-        loop.sessions.invalidate(reloaded.key)
+        loop.sessions.invalidate(session.key)
 
         # After /new, metadata should no longer contain _last_summary
         fresh = loop.sessions.get_or_create("cli:test")

tests/agent/test_consolidator.py

@@ -1,4 +1,4 @@
-"""Tests for Memory checkpoint consolidation and history journaling."""
+"""Tests for the lightweight Consolidator — append-only to HISTORY.md."""
 
 from dataclasses import replace
 from unittest.mock import AsyncMock, MagicMock
@@ -6,7 +6,7 @@ from unittest.mock import AsyncMock, MagicMock
 import pytest
 
 from nanobot.agent.memory import (
-    _HISTORY_ENTRY_HARD_CAP,
+    _ARCHIVE_SUMMARY_MAX_CHARS,
     Consolidator,
     MemoryStore,
 )
@@ -26,8 +26,6 @@ from nanobot.session.manager import Session
 from nanobot.utils.llm_runtime import LLMRuntime
 from nanobot.utils.prompt_templates import render_template
 
-_ARCHIVE_PROMPT = render_template("agent/consolidator_archive.md", strip=True)
-
 
 @pytest.fixture
 def store(tmp_path):
@@ -100,15 +98,8 @@ def _build_test_messages(**kwargs):
     ]
 
 
-async def _archive(
+async def _archive(consolidator, messages, runtime, *, session_key="test:session"):
-    consolidator,
+    return await consolidator.archive(
-    messages,
-    runtime,
-    *,
-    session_key="test:session",
-    previous_summary=None,
-):
-    return await consolidator.archiver.archive(
         messages,
         runtime=runtime,
         session_key=session_key,
@@ -117,7 +108,6 @@ async def _archive(
             current_message="consolidate",
         ),
         request_tools=[],
-        previous_summary=previous_summary,
     )
 
 
@@ -211,9 +201,7 @@ class TestConsolidatorSummarize:
         mock_provider.chat_with_retry.side_effect = Exception("API error")
         messages = [{"role": "user", "content": "hello"}]
         result = await _archive(consolidator, messages, runtime)
-        assert result is not None
+        assert result is None  # no summary on raw dump fallback
-        assert "[RAW]" in result
-        assert "hello" in result
         entries = store.read_unprocessed_history(since_cursor=0)
         assert len(entries) == 1
         assert "[RAW]" in entries[0]["content"]
@@ -238,51 +226,23 @@ class TestConsolidatorSummarize:
         entries = store.read_unprocessed_history(since_cursor=0)
         assert entries[0]["session_key"] == "slack:chat-2"
 
-    async def test_raw_fallback_represents_previous_checkpoint_and_new_chunk(
-        self,
-        consolidator,
-        mock_provider,
-        runtime,
-    ):
-        runtime = replace(runtime, generation=GenerationSettings(max_tokens=96))
-        mock_provider.chat_with_retry.side_effect = RuntimeError("API error")
-
-        result = await _archive(
-            consolidator,
-            [{"role": "user", "content": "NEW_MARKER " + "new " * 200}],
-            runtime,
-            previous_summary="OLD_MARKER " + "old " * 200,
-        )
-
-        assert result is not None
-        assert "[Previous archived context]" in result
-        assert "OLD_MARKER" in result
-        assert "[Newly archived raw context]" in result
-        assert "NEW_MARKER" in result
-        assert "... (truncated)" in result
-
     async def test_summarize_skips_empty_messages(self, consolidator, runtime):
         result = await _archive(consolidator, [], runtime)
         assert result is None
 
 
 class TestConsolidatorPromptContract:
-    def test_archive_prompt_requests_a_cumulative_replacement_checkpoint(self):
+    def test_archive_prompt_preserves_working_state_with_memory_facts(self):
-        prompt = _ARCHIVE_PROMPT
+        prompt = render_template("agent/consolidator_archive.md", strip=True, archive_count=4)
 
-        for section in ("## Merge rules", "## What to retain", "## Output"):
-            assert section in prompt
-        assert "replacement checkpoint" in prompt
-        assert "[Archived Context Summary]" in prompt
-        assert "current conversation state" in prompt
         assert "SNIP" in prompt
-        for mark in ("[permanent]", "[durable]", "[ephemeral]", "[correction]"):
+        assert "final 4 conversation messages" in prompt
+        for mark in ("[permanent]", "[durable]", "[ephemeral]", "[correction]", "[skip]"):
             assert mark in prompt
         assert "working-state handoff" in prompt
-        assert "- [mark] fact" in prompt
+        assert "exact identifiers needed to continue without rework" in prompt
-        assert "[skip]" not in prompt
+        assert "Do not output facts already present in the system prompt's Recent History" in prompt
-        assert "(nothing)" in prompt
+        assert "Do not mark something [skip] merely because it might already exist" in prompt
-        assert "history.jsonl" not in prompt
 
 
 class TestConsolidatorArchiveErrorHandling:
@@ -312,8 +272,7 @@ class TestConsolidatorArchiveErrorHandling:
             {"role": "assistant", "content": "Done, fixed the race condition."},
         ]
         result = await _archive(consolidator, messages, runtime)
-        assert result is not None
+        assert result is None
-        assert "[RAW]" in result
         entries = store.read_unprocessed_history(since_cursor=0)
         assert len(entries) == 1
         assert "[RAW]" in entries[0]["content"]
@@ -477,9 +436,9 @@ class TestConsolidatorTokenBudget:
         assert [message["content"] for message in request["messages"][1:-1]] == [
             f"m{i}" for i in range(50)
         ]
-        assert request["messages"][-1]["content"] == _ARCHIVE_PROMPT
+        assert "final 50 conversation messages" in request["messages"][-1]["content"]
         assert request["tools"] == []
-        assert "tool_choice" not in request
+        assert request["tool_choice"] == "none"
         assert session.last_archived == 50
         assert session.provider_state == _provider_state()
 
@@ -501,7 +460,8 @@ class TestConsolidatorTokenBudget:
         consolidator.estimate_session_prompt_tokens = MagicMock(
             side_effect=[(1200, "tiktoken"), (400, "tiktoken")]
         )
-        consolidator.archive_session = AsyncMock(return_value="[RAW] checkpoint")
+        # LLM consolidation fails after raw_archive fires.
+        consolidator.archive_session = AsyncMock(return_value=None)
 
         await consolidator.maybe_consolidate_by_tokens(session, runtime=runtime)
 
@@ -531,7 +491,7 @@ class TestConsolidatorTokenBudget:
         consolidator.estimate_session_prompt_tokens = MagicMock(
             return_value=(1200, "tiktoken")
         )
-        consolidator.archive_session = AsyncMock(return_value="[RAW] checkpoint")
+        consolidator.archive_session = AsyncMock(return_value=None)
 
         await consolidator.maybe_consolidate_by_tokens(session, runtime=runtime)
 
@@ -653,62 +613,27 @@ class TestCompactIdleSession:
         assert reloaded.last_archived == 2
         assert [message["content"] for message in reloaded.get_history()] == ["hello", "hi"]
 
-    @pytest.mark.asyncio
-    async def test_idle_compaction_with_no_new_messages_is_noop(
-        self, real_consolidator, mock_provider, store, runtime
-    ):
-        sessions = real_consolidator.sessions
-        session = sessions.get_or_create("cli:archived-idle")
-        session.add_message("user", "already archived")
-        session.add_message("assistant", "old answer")
-        session.last_archived = 2
-        sessions.save(session)
-        sessions.invalidate("cli:archived-idle")
-
-        result = await real_consolidator.compact_idle_session(
-            "cli:archived-idle",
-            runtime=runtime,
-        )
-
-        assert result == ""
-        mock_provider.chat_with_retry.assert_not_awaited()
-        reloaded = sessions.get_or_create("cli:archived-idle")
-        assert reloaded.last_archived == 2
-        assert "_last_summary" not in reloaded.metadata
-        assert store.read_unprocessed_history(since_cursor=0) == []
-
     @pytest.mark.asyncio
     async def test_new_messages_advance_existing_archive_progress(
         self, real_consolidator, mock_provider, runtime
     ):
-        mock_provider.chat_with_retry.side_effect = [
+        mock_provider.chat_with_retry.return_value = MagicMock(
-            MagicMock(content="First replacement checkpoint.", finish_reason="stop"),
+            content="Summary.", finish_reason="stop"
-            MagicMock(content="Second replacement checkpoint.", finish_reason="stop"),
+        )
-        ]
         sessions = real_consolidator.sessions
         session = sessions.get_or_create("cli:incremental")
         session.add_message("user", "first user")
         session.add_message("assistant", "first assistant")
         sessions.save(session)
 
-        first = await real_consolidator.compact_idle_session(
+        await real_consolidator.compact_idle_session("cli:incremental", runtime=runtime)
-            "cli:incremental",
-            runtime=runtime,
-        )
         current = sessions.get_or_create("cli:incremental")
         current.add_message("user", "second user")
         current.add_message("assistant", "second assistant")
         sessions.save(current)
-        second = await real_consolidator.compact_idle_session(
+        await real_consolidator.compact_idle_session("cli:incremental", runtime=runtime)
-            "cli:incremental",
-            runtime=runtime,
-        )
 
-        assert first == "First replacement checkpoint."
-        assert second == "Second replacement checkpoint."
         assert mock_provider.chat_with_retry.await_count == 2
-        latest_build = real_consolidator.archiver._build_messages.call_args_list[-1].kwargs
-        assert latest_build["session_summary"]["text"] == "First replacement checkpoint."
         latest_messages = mock_provider.chat_with_retry.await_args_list[-1].kwargs["messages"]
         assert [message["content"] for message in latest_messages[1:5]] == [
             "first user",
@@ -716,91 +641,8 @@ class TestCompactIdleSession:
             "second user",
             "second assistant",
         ]
-        assert latest_messages[-1]["content"] == _ARCHIVE_PROMPT
+        assert "final 2 conversation messages" in latest_messages[-1]["content"]
-        sessions.invalidate("cli:incremental")
+        assert sessions.get_or_create("cli:incremental").last_archived == 4
-        reloaded = sessions.get_or_create("cli:incremental")
-        assert reloaded.last_archived == 4
-        assert reloaded.metadata["_last_summary"]["text"] == second
-
-    @pytest.mark.asyncio
-    async def test_raw_fallback_preserves_previous_checkpoint_and_new_chunk(
-        self,
-        real_consolidator,
-        mock_provider,
-        store,
-        runtime,
-    ):
-        mock_provider.chat_with_retry.side_effect = [
-            LLMResponse(content="Earlier durable checkpoint.", finish_reason="stop"),
-            RuntimeError("LLM unavailable"),
-        ]
-        sessions = real_consolidator.sessions
-        session = sessions.get_or_create("cli:cumulative-fallback")
-        session.add_message("user", "first user")
-        session.add_message("assistant", "first answer")
-        sessions.save(session)
-
-        await real_consolidator.compact_idle_session(
-            "cli:cumulative-fallback",
-            runtime=runtime,
-        )
-        current = sessions.get_or_create("cli:cumulative-fallback")
-        current.add_message("user", "second user")
-        current.add_message("assistant", "newest working state")
-        sessions.save(current)
-
-        fallback = await real_consolidator.compact_idle_session(
-            "cli:cumulative-fallback",
-            runtime=runtime,
-        )
-
-        assert fallback is not None
-        assert "[Previous archived context]" in fallback
-        assert "Earlier durable checkpoint." in fallback
-        assert "[Newly archived raw context]" in fallback
-        assert "newest working state" in fallback
-        entries = store.read_unprocessed_history(0)
-        assert entries[0]["content"] == "Earlier durable checkpoint."
-        assert entries[1]["content"].startswith("[RAW] 2 messages")
-        sessions.invalidate("cli:cumulative-fallback")
-        reloaded = sessions.get_or_create("cli:cumulative-fallback")
-        assert reloaded.metadata["_last_summary"]["text"] == fallback
-
-    @pytest.mark.asyncio
-    async def test_nothing_keeps_previous_replacement_checkpoint(
-        self,
-        real_consolidator,
-        mock_provider,
-        runtime,
-    ):
-        mock_provider.chat_with_retry.side_effect = [
-            LLMResponse(content="Existing checkpoint.", finish_reason="stop"),
-            LLMResponse(content="(nothing)", finish_reason="stop"),
-        ]
-        sessions = real_consolidator.sessions
-        session = sessions.get_or_create("cli:nothing-after-summary")
-        session.add_message("user", "important first turn")
-        session.add_message("assistant", "important result")
-        sessions.save(session)
-        await real_consolidator.compact_idle_session(
-            "cli:nothing-after-summary",
-            runtime=runtime,
-        )
-
-        current = sessions.get_or_create("cli:nothing-after-summary")
-        current.add_message("user", "thanks")
-        current.add_message("assistant", "you're welcome")
-        sessions.save(current)
-        result = await real_consolidator.compact_idle_session(
-            "cli:nothing-after-summary",
-            runtime=runtime,
-        )
-
-        assert result == "(nothing)"
-        sessions.invalidate("cli:nothing-after-summary")
-        reloaded = sessions.get_or_create("cli:nothing-after-summary")
-        assert reloaded.last_archived == 4
-        assert reloaded.metadata["_last_summary"]["text"] == "Existing checkpoint."
 
     @pytest.mark.asyncio
     async def test_concurrent_append_remains_unarchived(
@@ -950,16 +792,11 @@ class TestCompactIdleSession:
         result = await real_consolidator.compact_idle_session(
             "cli:nothing", runtime=runtime, max_suffix=4
         )
-        second = await real_consolidator.compact_idle_session(
-            "cli:nothing", runtime=runtime, max_suffix=4
-        )
         assert result == "(nothing)"
-        assert second == ""
 
         reloaded = sessions.get_or_create("cli:nothing")
         assert "_last_summary" not in reloaded.metadata
         assert real_consolidator.store.read_unprocessed_history(0) == []
-        mock_provider.chat_with_retry.assert_awaited_once()
 
     @pytest.mark.asyncio
     async def test_llm_failure_preserves_history_but_advances_replay_boundary(
@@ -976,8 +813,7 @@ class TestCompactIdleSession:
         result = await real_consolidator.compact_idle_session(
             "cli:fail", runtime=runtime, max_suffix=4
         )
-        assert result is not None
+        assert result is None
-        assert "[RAW]" in result
 
         # raw_archive should have been called (history.jsonl gets an entry)
         entries = store.read_unprocessed_history(since_cursor=0)
@@ -987,7 +823,6 @@ class TestCompactIdleSession:
         assert len(reloaded.messages) == 20
         assert reloaded.messages[0]["content"] == "u0"
         assert reloaded.last_archived == 20
-        assert reloaded.metadata["_last_summary"]["text"] == result
         assert [m["content"] for m in reloaded.get_history(max_messages=20)] == [
             "u6",
             "a6",
@@ -1028,10 +863,11 @@ class TestCompactIdleSession:
         archived_call = mock_provider.chat_with_retry.call_args
         sent_messages = archived_call.kwargs["messages"]
         sent_content = [message.get("content") for message in sent_messages]
-        # The replacement overview covers all model-visible conversation context.
+        # The ordinary replay prefix contributes recent context, while the
+        # temporary instruction limits the new overview to the unarchived tail.
         assert "u0" not in sent_content
         assert "u26" in sent_content
-        assert sent_messages[-1]["content"] == _ARCHIVE_PROMPT
+        assert "final 10 conversation messages" in sent_messages[-1]["content"]
 
     @pytest.mark.asyncio
     async def test_full_archive_keeps_extended_legal_replay_suffix(
@@ -1120,9 +956,9 @@ class TestCompactIdleSession:
             "user",
         ]
         assert sent_messages[2]["tool_calls"][0]["id"] == "call-1"
-        assert sent_messages[-1]["content"] == _ARCHIVE_PROMPT
+        assert "final 4 conversation messages" in sent_messages[-1]["content"]
         assert call["tools"] == tools
-        assert "tool_choice" not in call
+        assert call["tool_choice"] == "none"
 
         reloaded = sessions.get_or_create("cli:tool-history")
         assert len(reloaded.messages) == 4
@@ -1160,8 +996,7 @@ class TestCompactIdleSession:
             runtime=runtime,
         )
 
-        assert result is not None
+        assert result is None
-        assert "[RAW]" in result
         entries = store.read_unprocessed_history(since_cursor=0)
         assert len(entries) == 1
         assert entries[0]["content"].startswith("[RAW] ")
@@ -1191,8 +1026,7 @@ class TestCompactIdleSession:
             runtime=runtime,
         )
 
-        assert result is not None
+        assert result is None
-        assert "[RAW]" in result
         entries = store.read_unprocessed_history(since_cursor=0)
         assert len(entries) == 1
         assert entries[0]["content"].startswith("[RAW] ")
@@ -1218,8 +1052,7 @@ class TestCompactIdleSession:
             runtime=runtime,
         )
 
-        assert result is not None
+        assert result is None
-        assert "[RAW]" in result
         mock_provider.chat_with_retry.assert_not_awaited()
         entries = store.read_unprocessed_history(since_cursor=0)
         assert len(entries) == 1
@@ -1227,7 +1060,7 @@ class TestCompactIdleSession:
         assert sessions.get_or_create("sdk:oversized").last_archived == 1
 
     @pytest.mark.asyncio
-    async def test_archive_context_contains_only_model_visible_messages(
+    async def test_incremental_scope_counts_only_model_visible_messages(
         self,
         real_consolidator,
         mock_provider,
@@ -1260,7 +1093,7 @@ class TestCompactIdleSession:
             "new user",
             "new answer",
         ]
-        assert sent[-1]["content"] == _ARCHIVE_PROMPT
+        assert "final 2 conversation messages" in sent[-1]["content"]
 
     @pytest.mark.asyncio
     async def test_reuses_real_prefix_for_unified_session_workspace(
@@ -1293,6 +1126,8 @@ class TestCompactIdleSession:
             current_message="next project question",
             channel="websocket",
             workspace=project,
+            session_key=session.key,
+            unified_session=True,
         )
 
         await loop.consolidator.compact_idle_session(
@@ -1302,7 +1137,7 @@ class TestCompactIdleSession:
 
         sent_messages = runtime.provider.chat_with_retry.call_args.kwargs["messages"]
         assert sent_messages[:-1] == ordinary_messages[:-1]
-        assert sent_messages[-1]["content"] == _ARCHIVE_PROMPT
+        assert "final 2 conversation messages" in sent_messages[-1]["content"]
         system = sent_messages[0]["content"]
         assert "PROJECT_WORKSPACE_MARKER" in system
         assert "GLOBAL_WORKSPACE_MARKER" not in system
@@ -1472,21 +1307,6 @@ class TestRawArchiveTruncation:
         assert len(entries) == 1
         assert "hello" in entries[0]["content"]
 
-    def test_raw_archive_returns_the_sanitized_persisted_checkpoint(self, store):
-        messages = [
-            {
-                "role": "user",
-                "content": "<think>PRIVATE_REASONING</think>visible result",
-            }
-        ]
-
-        checkpoint = store.raw_archive(messages, session_key="cli:test")
-
-        persisted = store.read_unprocessed_history(since_cursor=0)[0]["content"]
-        assert checkpoint == persisted
-        assert "PRIVATE_REASONING" not in checkpoint
-        assert "visible result" in checkpoint
-
     def test_raw_archive_excludes_model_only_runtime_context(self, store):
         content, marker = append_runtime_context(
             "ship the feature",
@@ -1518,40 +1338,21 @@ class TestRawArchiveTruncation:
 
 
 class TestArchivePersistence:
-    async def test_archive_returns_the_sanitized_persisted_summary(
+    async def test_oversized_summary_is_capped_before_append(
-        self, consolidator, mock_provider, store, runtime
-    ):
-        mock_provider.chat_with_retry.return_value = MagicMock(
-            content="<think>PRIVATE_REASONING</think>safe summary",
-            finish_reason="stop",
-            has_tool_calls=False,
-        )
-
-        summary = await _archive(
-            consolidator,
-            [{"role": "user", "content": "hi"}],
-            runtime,
-        )
-
-        persisted = store.read_unprocessed_history(since_cursor=0)[0]["content"]
-        assert summary == persisted == "safe summary"
-
-    async def test_oversized_summary_uses_history_emergency_cap(
         self, consolidator, mock_provider, store, runtime
     ):
         """A pathologically large LLM summary must not land full-length in
         history.jsonl — that would re-open the #3412 bloat vector from the
         *success* path instead of the fallback path."""
         mock_provider.chat_with_retry.return_value = MagicMock(
-            content="S" * (_HISTORY_ENTRY_HARD_CAP * 2),
+            content="S" * (_ARCHIVE_SUMMARY_MAX_CHARS * 10),
             finish_reason="stop",
         )
-        summary = await _archive(
+        await _archive(
             consolidator,
             [{"role": "user", "content": "hi"}],
             runtime,
         )
 
         entry = store.read_unprocessed_history(since_cursor=0)[0]
-        assert len(entry["content"]) <= _HISTORY_ENTRY_HARD_CAP + 50
+        assert len(entry["content"]) <= _ARCHIVE_SUMMARY_MAX_CHARS + 50
-        assert summary == entry["content"]

tests/agent/test_context_builder.py

@@ -4,7 +4,7 @@ from pathlib import Path
 
 import pytest
 
-from nanobot.agent.context import ContextBuilder, TranscriptInput
+from nanobot.agent.context import ContextBuilder
 from nanobot.runtime_context import RuntimeContextBlock
 
 # ---------------------------------------------------------------------------
@@ -133,7 +133,10 @@ class TestLoadBootstrapFiles:
         (project / "SOUL.md").write_text("project soul collision", encoding="utf-8")
         (project / "USER.md").write_text("project user collision", encoding="utf-8")
 
-        result = ContextBuilder(agent_home).build_system_prompt(workspace=project)
+        result = ContextBuilder(agent_home).build_system_prompt(
+            workspace=project,
+            include_memory_recent_history=False,
+        )
 
         assert "selected project rules" in result
         assert "global project rules" not in result
@@ -149,7 +152,10 @@ class TestLoadBootstrapFiles:
         project.mkdir()
         (agent_home / "AGENTS.md").write_text("default workspace rules", encoding="utf-8")
 
-        result = ContextBuilder(agent_home).build_system_prompt(workspace=project)
+        result = ContextBuilder(agent_home).build_system_prompt(
+            workspace=project,
+            include_memory_recent_history=False,
+        )
 
         assert "default workspace rules" not in result
 
@@ -397,15 +403,6 @@ class TestBuildMessages:
         assert "user-only runtime context" not in messages[-1]["content"]
         assert "_meta" not in messages[-1]
 
-    def test_compatibility_builder_merges_system_role_without_history(self, tmp_path):
-        builder = _builder(tmp_path)
-
-        messages = builder.build_messages([], "system event", current_role="system")
-
-        assert len(messages) == 1
-        assert messages[0]["role"] == "system"
-        assert str(messages[0]["content"]).endswith("system event")
-
     def test_explicit_skill_reference_loads_full_instructions_for_this_turn(self, tmp_path):
         skill_dir = tmp_path / "skills" / "review"
         skill_dir.mkdir(parents=True)
@@ -475,20 +472,6 @@ class TestBuildMessages:
         assert "previous user message" in str(messages[1]["content"])
         assert "new message" in str(messages[1]["content"])
 
-    def test_structured_transcript_preserves_fresh_turn_boundary(self, tmp_path):
-        builder = _builder(tmp_path)
-        transcript = TranscriptInput(
-            history=[{"role": "user", "content": "previous user message"}],
-            current_message="new message",
-        )
-
-        messages = builder.build_transcript(transcript)
-
-        assert [message["role"] for message in messages] == ["system", "user", "user"]
-        assert messages[-2]["content"] == "previous user message"
-        assert messages[-1]["content"] == "new message"
-        assert transcript.message_count == 3
-
     def test_current_message_can_be_built_without_history_merge(self, tmp_path):
         builder = _builder(tmp_path)
         current = builder.build_current_message(

tests/agent/test_context_prompt_cache.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import datetime as datetime_module
+import re
 from datetime import datetime as real_datetime
 from importlib.resources import files as pkg_files
 from pathlib import Path
@@ -103,6 +104,173 @@ def test_provider_context_appended_after_user_content(tmp_path) -> None:
     assert user_pos < context_pos, "user content must precede provider context"
 
 
+def test_unprocessed_history_injected_into_system_prompt(tmp_path) -> None:
+    """Entries in history.jsonl not yet consumed by Dream appear with timestamps."""
+    workspace = _make_workspace(tmp_path)
+    builder = ContextBuilder(workspace)
+
+    builder.memory.append_history("User asked about weather in Tokyo")
+    builder.memory.append_history("Agent fetched forecast via web_search")
+
+    prompt = builder.build_system_prompt()
+    assert "# Recent History" in prompt
+    assert "User asked about weather in Tokyo" in prompt
+    assert "Agent fetched forecast via web_search" in prompt
+    assert re.search(r"\[\d{4}-\d{2}-\d{2} \d{2}:\d{2}\]", prompt)
+
+
+def test_recent_history_injection_is_session_scoped(tmp_path) -> None:
+    workspace = _make_workspace(tmp_path)
+    builder = ContextBuilder(workspace)
+
+    builder.memory.append_history("legacy entry without session")
+    builder.memory.append_history("telegram history", session_key="telegram:chat-1")
+    builder.memory.append_history("slack history", session_key="slack:chat-2")
+
+    prompt = builder.build_system_prompt(session_key="telegram:chat-1")
+
+    assert "# Recent History" in prompt
+    assert "telegram history" in prompt
+    assert "slack history" not in prompt
+    assert "legacy entry without session" not in prompt
+
+
+def test_session_summary_replaces_interleaved_recent_history_entry(tmp_path) -> None:
+    workspace = _make_workspace(tmp_path)
+    builder = ContextBuilder(workspace)
+    session_key = "unified:default"
+    overview = "CURRENT_SESSION_OVERVIEW_MARKER"
+
+    builder.memory.append_history("another session event", session_key=session_key)
+    builder.memory.append_history(overview, session_key=session_key)
+    latest_cursor = builder.memory.append_history(
+        "later telegram event",
+        session_key="telegram:chat-1",
+    )
+    summary = {"text": overview, "last_active": "2026-08-19T10:00:00"}
+
+    prompt = builder.build_system_prompt(
+        session_key=session_key,
+        session_summary=summary,
+        unified_session=True,
+    )
+
+    assert "# Recent History" in prompt
+    assert "another session event" in prompt
+    assert "later telegram event" in prompt
+    assert "[Archived Context Summary]" in prompt
+    assert prompt.count(overview) == 1
+
+    builder.memory.set_last_dream_cursor(latest_cursor)
+    processed_prompt = builder.build_system_prompt(
+        session_key=session_key,
+        session_summary=summary,
+        unified_session=True,
+    )
+    assert "# Recent History" not in processed_prompt
+    assert processed_prompt.count(overview) == 1
+
+
+def test_recent_history_injection_unified_excludes_cron_internals(tmp_path) -> None:
+    workspace = _make_workspace(tmp_path)
+    builder = ContextBuilder(workspace)
+
+    builder.memory.append_history("unified user history", session_key="unified:default")
+    builder.memory.append_history("channel user history", session_key="telegram:chat-1")
+    builder.memory.append_history("cron internal history", session_key="cron:job-1")
+
+    prompt = builder.build_system_prompt(
+        session_key="unified:default",
+        unified_session=True,
+    )
+
+    assert "unified user history" in prompt
+    assert "channel user history" in prompt
+    assert "cron internal history" not in prompt
+
+
+def test_cron_recent_history_can_see_own_history_and_unified_context(tmp_path) -> None:
+    workspace = _make_workspace(tmp_path)
+    builder = ContextBuilder(workspace)
+
+    builder.memory.append_history("unified user history", session_key="unified:default")
+    builder.memory.append_history("own cron history", session_key="cron:job-1")
+    builder.memory.append_history("other cron history", session_key="cron:job-2")
+
+    prompt = builder.build_system_prompt(
+        session_key="cron:job-1",
+        unified_session=True,
+    )
+
+    assert "unified user history" in prompt
+    assert "own cron history" in prompt
+    assert "other cron history" not in prompt
+
+
+def test_recent_history_capped_at_max(tmp_path) -> None:
+    """Only the most recent _MAX_RECENT_HISTORY entries are injected."""
+    workspace = _make_workspace(tmp_path)
+    builder = ContextBuilder(workspace)
+
+    for i in range(builder._MAX_RECENT_HISTORY + 20):
+        builder.memory.append_history(f"entry-{i}")
+
+    prompt = builder.build_system_prompt()
+    assert "entry-0" not in prompt
+    assert "entry-19" not in prompt
+    assert f"entry-{builder._MAX_RECENT_HISTORY + 19}" in prompt
+
+
+def test_recent_history_truncated_at_max_tokens(tmp_path) -> None:
+    """Recent History section must be truncated to _MAX_HISTORY_TOKENS."""
+    import tiktoken
+
+    workspace = _make_workspace(tmp_path)
+    builder = ContextBuilder(workspace)
+
+    big_entry = "word " * (builder._MAX_HISTORY_TOKENS + 5_000)
+    builder.memory.append_history(big_entry)
+
+    prompt = builder.build_system_prompt()
+    history_section = prompt.split("# Recent History\n\n", 1)
+    assert len(history_section) == 2
+
+    enc = tiktoken.get_encoding("cl100k_base")
+    assert len(enc.encode(history_section[1])) <= builder._MAX_HISTORY_TOKENS
+
+
+def test_no_recent_history_when_dream_has_processed_all(tmp_path) -> None:
+    """If Dream has consumed everything, no Recent History section should appear."""
+    workspace = _make_workspace(tmp_path)
+    builder = ContextBuilder(workspace)
+
+    cursor = builder.memory.append_history("already processed entry")
+    builder.memory.set_last_dream_cursor(cursor)
+
+    prompt = builder.build_system_prompt()
+    assert "# Recent History" not in prompt
+
+
+def test_partial_dream_processing_shows_only_remainder(tmp_path) -> None:
+    """When Dream has processed some entries, only the unprocessed ones appear."""
+    workspace = _make_workspace(tmp_path)
+    builder = ContextBuilder(workspace)
+
+    builder.memory.append_history("old conversation about Python")
+    c2 = builder.memory.append_history("old conversation about Rust")
+    builder.memory.append_history("recent question about Docker")
+    builder.memory.append_history("recent question about K8s")
+
+    builder.memory.set_last_dream_cursor(c2)
+
+    prompt = builder.build_system_prompt()
+    assert "# Recent History" in prompt
+    assert "old conversation about Python" not in prompt
+    assert "old conversation about Rust" not in prompt
+    assert "recent question about Docker" in prompt
+    assert "recent question about K8s" in prompt
+
+
 def test_execution_rules_in_system_prompt(tmp_path) -> None:
     """Execution rules should appear in the system prompt via the default templates."""
     from nanobot.utils.helpers import sync_workspace_templates

tests/agent/test_dream.py

@@ -62,14 +62,28 @@ class TestBuildDreamPrompt:
         prompt, _ = result
         assert "skill-creator" in prompt
 
-    def test_prompt_does_not_duplicate_current_memory_file_contents(self, store):
+    def test_prompt_embeds_current_memory_file_contents(self, store):
+        """Dream must see the real current file contents (Tier 4) so it edits the
+        files, not a stale mental model."""
         store.append_history("hello")
         result = store.build_dream_prompt()
         assert result is not None
         prompt, _ = result
-        assert "## Current Memory Files" not in prompt
+        assert "## Current Memory Files" in prompt
-        assert "Project X active" not in prompt
+        assert "### SOUL.md" in prompt
-        assert "Helpful" not in prompt
+        assert "### USER.md" in prompt
+        assert "### memory/MEMORY.md" in prompt
+        # Real current contents are embedded verbatim.
+        assert "Project X active" in prompt
+        assert "Helpful" in prompt
+
+    def test_prompt_renders_missing_files_as_empty(self, tmp_path):
+        store = MemoryStore(tmp_path)  # no durable files written
+        store.append_history("hello")
+        result = store.build_dream_prompt()
+        assert result is not None
+        prompt, _ = result
+        assert "(empty)" in prompt
 
     def test_workspace_dream_prompt_overrides_default(self, store):
         store.dream_prompt_file.parent.mkdir(parents=True)
@@ -412,7 +426,7 @@ class TestEphemeralDirect:
                 bus=bus,
                 provider=provider,
                 workspace=tmp_path,
-                context_window_tokens=32_000,
+                context_window_tokens=8000,
             )
 
         return loop, store
@@ -592,7 +606,7 @@ class TestEphemeralDirect:
             bus=MessageBus(),
             provider=provider,
             workspace=tmp_path,
-            context_window_tokens=32_000,
+            context_window_tokens=8000,
         )
 
         await loop.process_direct(
@@ -611,63 +625,6 @@ class TestEphemeralDirect:
         assert "entry-21" not in request_text
         assert "entry-60" not in request_text
 
-    async def test_dream_turn_injects_memory_files_once_and_persists_session(self, tmp_path):
-        """Dream gets durable files from system context without losing its session record."""
-        from unittest.mock import MagicMock
-
-        from nanobot.agent.loop import AgentLoop
-        from nanobot.bus.queue import MessageBus
-
-        markers = {
-            "SOUL.md": "DREAM_SOUL_MARKER",
-            "USER.md": "DREAM_USER_MARKER",
-            "memory/MEMORY.md": "DREAM_MEMORY_MARKER",
-        }
-        store = MemoryStore(tmp_path)
-        store.write_soul(markers["SOUL.md"])
-        store.write_user(markers["USER.md"])
-        store.write_memory(markers["memory/MEMORY.md"])
-        store.append_history("history-marker")
-        (tmp_path / "AGENTS.md").write_text("DREAM_AGENTS_MARKER", encoding="utf-8")
-
-        result = store.build_dream_prompt()
-        assert result is not None
-        prompt, _ = result
-
-        captured: dict[str, list[dict]] = {}
-        provider = MagicMock()
-        provider.get_default_model.return_value = "test-model"
-        provider.supports_tools = True
-        provider.generation = MagicMock(max_tokens=4096)
-
-        async def chat_with_retry(**kwargs):
-            captured["messages"] = kwargs["messages"]
-            return LLMResponse(content="done", finish_reason="stop")
-
-        provider.chat_with_retry = chat_with_retry
-        loop = AgentLoop(
-            bus=MessageBus(),
-            provider=provider,
-            workspace=tmp_path,
-            context_window_tokens=32_000,
-        )
-        session_key = "dream:single-memory-copy"
-
-        await loop.process_direct(
-            prompt,
-            session_key=session_key,
-            ephemeral=True,
-            tools=store.build_dream_tools(),
-        )
-
-        messages = captured["messages"]
-        system_prompt = str(messages[0]["content"])
-        request_text = "\n".join(str(message.get("content", "")) for message in messages)
-        for marker in [*markers.values(), "DREAM_AGENTS_MARKER"]:
-            assert marker in system_prompt
-            assert request_text.count(marker) == 1
-        assert loop.sessions._get_session_path(session_key).exists()
-
 
 class TestEphemeralHooks:
     """When ephemeral=True, extra hooks must not fire."""
@@ -709,7 +666,7 @@ class TestEphemeralHooks:
                 bus=bus,
                 provider=provider,
                 workspace=tmp_path,
-                context_window_tokens=32_000,
+                context_window_tokens=8000,
                 hooks=[spy],
             )
 

tests/agent/test_hook_composite.py

@@ -6,7 +6,6 @@ from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 
-from nanobot.agent.context import TranscriptInput
 from nanobot.agent.hook import (
     AgentHook,
     AgentHookContext,
@@ -460,7 +459,7 @@ async def test_agent_loop_extra_hook_receives_calls(tmp_path):
     loop.tools.get_definitions = MagicMock(return_value=[])
 
     result = await loop._run_agent_loop(
-        TranscriptInput(history=[{"role": "user", "content": "hi"}], current_message=None),
+        [{"role": "user", "content": "hi"}],
         runtime=loop.llm_runtime(),
     )
 
@@ -505,7 +504,7 @@ async def test_agent_loop_turn_hook_factories_receive_context(tmp_path):
 
     runtime = loop.llm_runtime()
     await loop._run_agent_loop(
-        TranscriptInput(history=[{"role": "user", "content": "hi"}], current_message=None),
+        [{"role": "user", "content": "hi"}],
         runtime=runtime,
         on_progress=on_progress,
         request_context=RequestContext(
@@ -552,7 +551,7 @@ async def test_agent_loop_extra_hook_error_isolation(tmp_path):
     loop.tools.get_definitions = MagicMock(return_value=[])
 
     result = await loop._run_agent_loop(
-        TranscriptInput(history=[{"role": "user", "content": "hi"}], current_message=None),
+        [{"role": "user", "content": "hi"}],
         runtime=loop.llm_runtime(),
     )
 
@@ -578,9 +577,7 @@ async def test_agent_loop_extra_hooks_do_not_swallow_loop_hook_errors(tmp_path):
 
     with pytest.raises(RuntimeError, match="progress failed"):
         await loop._run_agent_loop(
-            TranscriptInput(history=[], current_message=None),
+            [], runtime=loop.llm_runtime(), on_progress=bad_progress
-            runtime=loop.llm_runtime(),
-            on_progress=bad_progress,
         )
 
 
@@ -599,8 +596,7 @@ async def test_agent_loop_no_hooks_backward_compat(tmp_path):
     loop.max_iterations = 2
 
     result = await loop._run_agent_loop(
-        TranscriptInput(history=[], current_message=None),
+        [], runtime=loop.llm_runtime()
-        runtime=loop.llm_runtime(),
     )
     assert result.final_content == (
         "I reached the maximum number of tool call iterations (2) "

tests/agent/test_loop_consolidation_tokens.py

@@ -7,17 +7,11 @@ from nanobot.bus.queue import MessageBus
 from nanobot.providers.base import LLMResponse
 
 
-def _make_loop(
+def _make_loop(tmp_path, *, estimated_tokens: int, context_window_tokens: int) -> AgentLoop:
-    tmp_path,
-    *,
-    estimated_tokens: int,
-    context_window_tokens: int,
-    max_tokens: int = 0,
-) -> AgentLoop:
     from nanobot.providers.base import GenerationSettings
     provider = MagicMock()
     provider.get_default_model.return_value = "test-model"
-    provider.generation = GenerationSettings(max_tokens=max_tokens)
+    provider.generation = GenerationSettings(max_tokens=0)
     provider.estimate_prompt_tokens.return_value = (estimated_tokens, "test-counter")
     _response = LLMResponse(content="ok", tool_calls=[])
     provider.chat_with_retry = AsyncMock(return_value=_response)
@@ -29,9 +23,6 @@ def _make_loop(
         workspace=tmp_path,
         model="test-model",
         context_window_tokens=context_window_tokens,
-        # These tests isolate Memory consolidation; Runner request fitting is
-        # covered separately with realistic context windows.
-        context_block_limit=10_000,
     )
     loop.tools.get_definitions = MagicMock(return_value=[])
     loop.consolidator._SAFETY_BUFFER = 0
@@ -65,34 +56,6 @@ async def test_prompt_above_threshold_triggers_consolidation(tmp_path) -> None:
     assert loop.consolidator.archive_session.await_count >= 1
 
 
-@pytest.mark.asyncio
-async def test_token_consolidation_refreshes_summary_for_current_request(tmp_path) -> None:
-    loop = _make_loop(tmp_path, estimated_tokens=0, context_window_tokens=200)
-    loop.consolidator.archive_session = AsyncMock(  # type: ignore[method-assign]
-        return_value="FRESH_CHECKPOINT"
-    )
-    loop.consolidator.estimate_session_prompt_tokens = MagicMock(  # type: ignore[method-assign]
-        return_value=(1000, "test")
-    )
-    loop.schedule_background = lambda coro: coro.close()  # type: ignore[method-assign]
-
-    session = loop.sessions.get_or_create("cli:test")
-    session.messages = [
-        {"role": role, "content": f"{role[0]}{turn}"}
-        for turn in range(10)
-        for role in ("user", "assistant")
-    ]
-    loop.sessions.save(session)
-
-    await loop.process_direct("hello", session_key="cli:test")
-
-    request_messages = loop.provider.chat_with_retry.await_args.kwargs["messages"]
-    system_prompt = request_messages[0]["content"]
-    assert "FRESH_CHECKPOINT" in system_prompt
-    assert all(message.get("content") != "u0" for message in request_messages)
-    assert loop.sessions.get_or_create("cli:test").last_archived == 12
-
-
 @pytest.mark.asyncio
 async def test_prompt_above_threshold_uses_fixed_recent_tail(tmp_path) -> None:
     loop = _make_loop(tmp_path, estimated_tokens=1000, context_window_tokens=200)

tests/agent/test_loop_progress.py

@@ -6,7 +6,6 @@ from unittest.mock import AsyncMock, MagicMock
 
 import pytest
 
-from nanobot.agent.context import TranscriptInput
 from nanobot.agent.hooks import create_file_edit_activity_hook
 from nanobot.agent.loop import AgentLoop
 from nanobot.agent.tools.context import current_request_context
@@ -85,9 +84,7 @@ class TestToolEventProgress:
             progress.append((content, tool_hint, tool_events))
 
         result = await loop._run_agent_loop(
-            TranscriptInput(history=[], current_message=None),
+            [], runtime=loop.llm_runtime(), on_progress=on_progress
-            runtime=loop.llm_runtime(),
-            on_progress=on_progress,
         )
 
         assert result.final_content == "Done"
@@ -158,9 +155,7 @@ class TestToolEventProgress:
                 file_events.extend(file_edit_events)
 
         result = await loop._run_agent_loop(
-            TranscriptInput(history=[], current_message=None),
+            [], runtime=loop.llm_runtime(), on_progress=on_progress
-            runtime=loop.llm_runtime(),
-            on_progress=on_progress,
         )
 
         assert result.final_content == "Done"
@@ -230,9 +225,7 @@ class TestToolEventProgress:
         )
 
         result = await loop._run_agent_loop(
-            TranscriptInput(history=[], current_message=None),
+            [], runtime=loop.llm_runtime(), on_progress=on_progress
-            runtime=loop.llm_runtime(),
-            on_progress=on_progress,
         )
 
         assert result.final_content == "Done"
@@ -270,9 +263,7 @@ class TestToolEventProgress:
                 file_events.extend(file_edit_events)
 
         await loop._run_agent_loop(
-            TranscriptInput(history=[], current_message=None),
+            [], runtime=loop.llm_runtime(), on_progress=on_progress
-            runtime=loop.llm_runtime(),
-            on_progress=on_progress,
         )
 
         assert file_events == []
@@ -1028,7 +1019,7 @@ class TestToolEventProgress:
             progress.append((content, tool_hint, tool_events))
 
         result = await loop._run_agent_loop(
-            TranscriptInput(history=[], current_message=None),
+            [],
             runtime=loop.llm_runtime(),
             on_progress=on_progress,
             on_stream=on_stream,

tests/agent/test_loop_runner_integration.py

@@ -7,7 +7,6 @@ from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 
-from nanobot.agent.context import TranscriptInput
 from nanobot.agent.goal_permission import goal_mutation_allowed, goal_mutation_permission
 from nanobot.agent.tools.context import RequestContext
 from nanobot.bus.outbound_events import StreamedResponseEvent
@@ -56,7 +55,7 @@ async def test_ephemeral_runner_enters_and_restores_turn_scopes(tmp_path):
     loop.tools.get_definitions = MagicMock(return_value=[])
 
     await loop._run_agent_loop(
-        TranscriptInput(history=[], current_message=None),
+        [],
         runtime=loop.llm_runtime(),
         ephemeral=True,
         turn_scopes=[goal_mutation_permission(True)],
@@ -341,8 +340,7 @@ async def test_loop_max_iterations_message_stays_stable(tmp_path):
     loop.max_iterations = 2
 
     result = await loop._run_agent_loop(
-        TranscriptInput(history=[], current_message=None),
+        [], runtime=loop.llm_runtime()
-        runtime=loop.llm_runtime(),
     )
 
     assert result.final_content == (
@@ -364,7 +362,7 @@ async def test_loop_goal_turn_uses_standard_iteration_budget(tmp_path):
 
     runtime = loop.llm_runtime()
     result = await loop._run_agent_loop(
-        TranscriptInput(history=[], current_message=None),
+        [],
         runtime=runtime,
         request_context=RequestContext(
             channel="cli",
@@ -403,7 +401,7 @@ async def test_loop_stream_filter_handles_think_only_prefix_without_crashing(tmp
         endings.append(resuming)
 
     result = await loop._run_agent_loop(
-        TranscriptInput(history=[], current_message=None),
+        [],
         runtime=loop.llm_runtime(),
         on_stream=on_stream,
         on_stream_end=on_stream_end,
@@ -430,9 +428,7 @@ async def test_loop_stream_filter_hides_partial_trailing_think_prefix(tmp_path):
         deltas.append(delta)
 
     result = await loop._run_agent_loop(
-        TranscriptInput(history=[], current_message=None),
+        [], runtime=loop.llm_runtime(), on_stream=on_stream
-        runtime=loop.llm_runtime(),
-        on_stream=on_stream,
     )
 
     assert result.final_content == "Hello World"
@@ -455,9 +451,7 @@ async def test_loop_stream_filter_hides_complete_trailing_think_tag(tmp_path):
         deltas.append(delta)
 
     result = await loop._run_agent_loop(
-        TranscriptInput(history=[], current_message=None),
+        [], runtime=loop.llm_runtime(), on_stream=on_stream
-        runtime=loop.llm_runtime(),
-        on_stream=on_stream,
     )
 
     assert result.final_content == "Hello World"
@@ -478,8 +472,7 @@ async def test_loop_retries_think_only_final_response(tmp_path):
     loop.provider.chat_with_retry = chat_with_retry
 
     result = await loop._run_agent_loop(
-        TranscriptInput(history=[], current_message=None),
+        [], runtime=loop.llm_runtime()
-        runtime=loop.llm_runtime(),
     )
 
     assert result.final_content == "Recovered answer"

tests/agent/test_loop_save_turn.py

@@ -7,7 +7,7 @@ from unittest.mock import AsyncMock, MagicMock
 import pytest
 from loguru import logger
 
-from nanobot.agent.context import ContextBuilder, TranscriptInput
+from nanobot.agent.context import ContextBuilder
 from nanobot.agent.loop import AgentLoop
 from nanobot.agent.runner import AgentRunResult
 from nanobot.agent.tools.context import RequestContext, request_context
@@ -79,13 +79,6 @@ def _agent_run_result(
     )
 
 
-def _assembled_messages(
-    builder: ContextBuilder,
-    transcript_input: TranscriptInput,
-) -> list[dict]:
-    return builder.build_transcript(transcript_input, include_memory=False)
-
-
 def _mk_loop() -> AgentLoop:
     loop = AgentLoop.__new__(AgentLoop)
     from nanobot.config.schema import AgentDefaults
@@ -937,13 +930,10 @@ async def test_runtime_checkpoint_keeps_provider_state_out_of_public_metadata(
     session = loop.sessions.get_or_create("cli:private-checkpoint")
 
     await loop._run_agent_loop(
-        TranscriptInput(
+        [
-            history=[
             {"role": "system", "content": "system"},
             {"role": "user", "content": "question"},
         ],
-            current_message=None,
-        ),
         runtime=loop.llm_runtime(),
         session=session,
     )
@@ -1018,7 +1008,7 @@ async def test_subagent_followup_state_is_durable_before_prompt_assembly(
     loop = _make_full_loop(tmp_path)
     loop.consolidator.maybe_consolidate_by_tokens = AsyncMock(return_value=False)  # type: ignore[method-assign]
     loop.provider.can_resume_conversation_state.return_value = True
-    loop.context.build_system_prompt = MagicMock(  # type: ignore[method-assign]
+    loop._build_initial_messages = MagicMock(  # type: ignore[method-assign]
         side_effect=RuntimeError("prompt boom"),
     )
     session = loop.sessions.get_or_create("cli:subagent-prompt-crash")
@@ -1051,8 +1041,8 @@ async def test_subagent_redelivery_does_not_duplicate_staged_provider_input(
     loop = _make_full_loop(tmp_path)
     loop.consolidator.maybe_consolidate_by_tokens = AsyncMock(return_value=False)  # type: ignore[method-assign]
     loop.provider.can_resume_conversation_state.return_value = True
-    build_system_prompt = loop.context.build_system_prompt
+    build_initial_messages = loop._build_initial_messages
-    loop.context.build_system_prompt = MagicMock(  # type: ignore[method-assign]
+    loop._build_initial_messages = MagicMock(  # type: ignore[method-assign]
         side_effect=RuntimeError("prompt boom"),
     )
     session = loop.sessions.get_or_create("cli:subagent-redelivery")
@@ -1076,7 +1066,7 @@ async def test_subagent_redelivery_does_not_duplicate_staged_provider_input(
         message.get("content")
         for message in persisted.provider_state.pending_messages
     ].count("subagent result") == 1
-    loop.context.build_system_prompt = build_system_prompt  # type: ignore[method-assign]
+    loop._build_initial_messages = build_initial_messages  # type: ignore[method-assign]
     loop._run_agent_loop = AsyncMock(  # type: ignore[method-assign]
         side_effect=RuntimeError("provider boom"),
     )
@@ -1329,8 +1319,7 @@ async def test_internal_continuation_queues_turn_without_fake_user_history(
 
     calls: list[dict] = []
 
-    async def fake_run_agent_loop(transcript_input, *, metadata=None, **_kwargs):
+    async def fake_run_agent_loop(initial_messages, *, metadata=None, **_kwargs):
-        initial_messages = _assembled_messages(loop.context, transcript_input)
         calls.append({"initial_messages": initial_messages, "metadata": metadata})
         if len(calls) == 1:
             return _agent_run_result(
@@ -1398,9 +1387,8 @@ async def test_internal_continuation_preserves_streaming_route_metadata(
 
     calls = 0
 
-    async def fake_run_agent_loop(transcript_input, *, on_stream=None, on_stream_end=None, **_kwargs):
+    async def fake_run_agent_loop(initial_messages, *, on_stream=None, on_stream_end=None, **_kwargs):
         nonlocal calls
-        initial_messages = _assembled_messages(loop.context, transcript_input)
         calls += 1
         if calls == 1:
             return _agent_run_result(
@@ -1472,9 +1460,8 @@ async def test_websocket_internal_continuation_keeps_single_visible_run(
 
     calls = 0
 
-    async def fake_run_agent_loop(transcript_input, **_kwargs):
+    async def fake_run_agent_loop(initial_messages, **_kwargs):
         nonlocal calls
-        initial_messages = _assembled_messages(loop.context, transcript_input)
         calls += 1
         if calls == 1:
             return _agent_run_result(
@@ -1636,7 +1623,7 @@ async def test_run_agent_loop_continuation_reads_latest_goal_metadata(
 
     runtime = loop.llm_runtime()
     await loop._run_agent_loop(
-        TranscriptInput(history=[], current_message=None),
+        [],
         runtime=runtime,
         session=session,
         request_context=RequestContext(
@@ -1766,7 +1753,7 @@ async def test_stop_preserves_runtime_checkpoint_for_next_turn(tmp_path: Path) -
 
     checkpoint_saved = asyncio.Event()
 
-    async def interrupted_run_agent_loop(_transcript_input, *, session=None, **_kwargs):
+    async def interrupted_run_agent_loop(_initial_messages, *, session=None, **_kwargs):
         assert session is not None
         loop._set_runtime_checkpoint(
             session,
@@ -1826,8 +1813,7 @@ async def test_stop_preserves_runtime_checkpoint_for_next_turn(tmp_path: Path) -
     assert interrupted.metadata.get(AgentLoop._PENDING_USER_TURN_KEY) is True
     assert interrupted.metadata.get(AgentLoop._RUNTIME_CHECKPOINT_KEY) is not None
 
-    async def resumed_run_agent_loop(transcript_input, **_kwargs):
+    async def resumed_run_agent_loop(initial_messages, **_kwargs):
-        initial_messages = _assembled_messages(loop.context, transcript_input)
         return _agent_run_result(
             "next answer",
             [*initial_messages, {"role": "assistant", "content": "next answer"}],
@@ -1878,8 +1864,7 @@ async def test_system_subagent_followup_is_persisted_before_prompt_assembly(tmp_
     record_runtime = MagicMock(wraps=loop.runtime_event_publisher.record_turn_runtime)
     loop.runtime_event_publisher.record_turn_runtime = record_runtime
 
-    async def fake_run_agent_loop(transcript_input, **kwargs):
+    async def fake_run_agent_loop(initial_messages, **kwargs):
-        initial_messages = _assembled_messages(loop.context, transcript_input)
         seen["initial_messages"] = initial_messages
         seen["runtime"] = kwargs["runtime"]
         seen["request_context"] = kwargs["request_context"]
@@ -1955,8 +1940,7 @@ async def test_turn_usage_is_persisted_with_the_saved_session(tmp_path: Path) ->
     loop.consolidator.maybe_consolidate_by_tokens = AsyncMock(return_value=False)  # type: ignore[method-assign]
     turn_usage = LLMUsage.reported(input_tokens=64, output_tokens=9)
 
-    async def fake_run_agent_loop(transcript_input, **_kwargs):
+    async def fake_run_agent_loop(initial_messages, **_kwargs):
-        initial_messages = _assembled_messages(loop.context, transcript_input)
         return _agent_run_result(
             "done",
             [*initial_messages, {"role": "assistant", "content": "done"}],
@@ -1982,8 +1966,7 @@ async def test_system_subagent_followup_does_not_log_content(tmp_path: Path) ->
         return_value=False
     )
 
-    async def fake_run_agent_loop(transcript_input, **_kwargs):
+    async def fake_run_agent_loop(initial_messages, **_kwargs):
-        initial_messages = _assembled_messages(loop.context, transcript_input)
         return _agent_run_result(
             "done",
             [*initial_messages, {"role": "assistant", "content": "done"}],
@@ -2039,8 +2022,7 @@ async def test_system_subagent_followup_uses_common_turn_lifecycle(tmp_path: Pat
 
         setattr(loop, name, record)
 
-    async def fake_run_agent_loop(transcript_input, **_kwargs):
+    async def fake_run_agent_loop(initial_messages, **_kwargs):
-        initial_messages = _assembled_messages(loop.context, transcript_input)
         return _agent_run_result(
             "done",
             [*initial_messages, {"role": "assistant", "content": "done"}],
@@ -2083,8 +2065,7 @@ async def test_multiple_subagent_followups_all_persist_as_standalone_history(tmp
     loop = _make_full_loop(tmp_path)
     loop.consolidator.maybe_consolidate_by_tokens = AsyncMock(return_value=False)  # type: ignore[method-assign]
 
-    async def fake_run_agent_loop(transcript_input, **_kwargs):
+    async def fake_run_agent_loop(initial_messages, **_kwargs):
-        initial_messages = _assembled_messages(loop.context, transcript_input)
         return _agent_run_result(
             "ack",
             [*initial_messages, {"role": "assistant", "content": "ack"}],
@@ -2215,8 +2196,7 @@ async def test_system_subagent_followup_uses_thread_session_and_slack_metadata(t
 
     seen: dict[str, object] = {}
 
-    async def fake_run_agent_loop(transcript_input, **kwargs):
+    async def fake_run_agent_loop(initial_messages, **kwargs):
-        initial_messages = _assembled_messages(loop.context, transcript_input)
         seen["initial_messages"] = initial_messages
         seen["request_context"] = kwargs["request_context"]
         return _agent_run_result(
@@ -2272,11 +2252,8 @@ async def test_turn_after_unanswered_user_keeps_tool_call_pairing(tmp_path: Path
     session.add_message("user", "earlier question that never got an answer")
     loop.sessions.save(session)
 
-    async def fake_run_agent_loop(transcript_input, **_kwargs):
+    async def fake_run_agent_loop(initial_messages, **_kwargs):
-        initial_messages = _assembled_messages(loop.context, transcript_input)
+        assert [m["role"] for m in initial_messages] == ["system", "user"]
-        assert [m["role"] for m in initial_messages] == ["system", "user", "user"]
-        assert initial_messages[-2]["content"] == "earlier question that never got an answer"
-        assert initial_messages[-1]["content"] == "and another thing"
         return _agent_run_result(
             "done",
             [

tests/agent/test_loop_tool_context.py

@@ -5,7 +5,6 @@ from unittest.mock import AsyncMock, MagicMock
 
 import pytest
 
-from nanobot.agent.context import TranscriptInput
 from nanobot.agent.loop import AgentLoop
 from nanobot.agent.tools.context import (
     RequestContext,
@@ -134,7 +133,7 @@ async def test_loop_binds_request_context_for_tool_execution(tmp_path: Path) ->
     metadata = {"slack": {"thread_ts": "111.222", "channel_type": "channel"}}
     runtime = loop.llm_runtime()
     await loop._run_agent_loop(
-        TranscriptInput(history=[], current_message=None),
+        [],
         runtime=runtime,
         request_context=RequestContext(
             channel="slack",
@@ -235,7 +234,7 @@ async def test_agent_loop_restores_outer_request_context_after_runner_exception(
     try:
         with pytest.raises(RuntimeError, match="runner failed"):
             await loop._run_agent_loop(
-                TranscriptInput(history=[], current_message=None),
+                [],
                 runtime=runtime,
                 request_context=RequestContext(
                     channel="slack",

tests/agent/test_memory_store.py

@@ -113,6 +113,54 @@ class TestHistoryWithCursor:
         entries = store.read_unprocessed_history(since_cursor=0)
         assert len(entries) == 2
 
+    def test_prompt_history_filters_to_current_session(self, store):
+        store.append_history("legacy entry without session")
+        store.append_history("telegram entry", session_key="telegram:chat-1")
+        store.append_history("slack entry", session_key="slack:chat-2")
+
+        entries = store.read_recent_history_for_prompt(
+            since_cursor=0,
+            session_key="telegram:chat-1",
+        )
+
+        assert [e["content"] for e in entries] == ["telegram entry"]
+        assert [e["content"] for e in store.read_unprocessed_history(0)] == [
+            "legacy entry without session",
+            "telegram entry",
+            "slack entry",
+        ]
+
+    def test_unified_prompt_history_excludes_internal_cron_sessions(self, store):
+        store.append_history("legacy entry without session")
+        store.append_history("unified entry", session_key="unified:default")
+        store.append_history("telegram entry", session_key="telegram:chat-1")
+        store.append_history("cron internal entry", session_key="cron:job-1")
+
+        entries = store.read_recent_history_for_prompt(
+            since_cursor=0,
+            session_key="unified:default",
+            unified_session=True,
+        )
+
+        assert [e["content"] for e in entries] == [
+            "legacy entry without session",
+            "unified entry",
+            "telegram entry",
+        ]
+
+    def test_unified_cron_prompt_history_includes_own_cron_entry(self, store):
+        store.append_history("unified entry", session_key="unified:default")
+        store.append_history("other cron entry", session_key="cron:job-2")
+        store.append_history("own cron entry", session_key="cron:job-1")
+
+        entries = store.read_recent_history_for_prompt(
+            since_cursor=0,
+            session_key="cron:job-1",
+            unified_session=True,
+        )
+
+        assert [e["content"] for e in entries] == ["unified entry", "own cron entry"]
+
     def test_read_unprocessed_skips_entries_without_cursor(self, store):
         """Regression: entries missing the cursor key should be silently skipped."""
         store.history_file.write_text(

tests/agent/test_new_command_archival.py

@@ -8,10 +8,6 @@ from unittest.mock import AsyncMock, MagicMock
 
 import pytest
 
-from nanobot.utils.prompt_templates import render_template
-
-_ARCHIVE_PROMPT = render_template("agent/consolidator_archive.md", strip=True)
-
 
 class TestNewCommandArchival:
     """Test /new archival behavior with the structured archive flow."""
@@ -121,7 +117,7 @@ class TestNewCommandArchival:
         await loop.aclose()
         sent = loop.provider.chat_with_retry.call_args.kwargs["messages"]
         assert sent[1:-1] == ordinary_history
-        assert sent[-1]["content"] == _ARCHIVE_PROMPT
+        assert "final 2 conversation messages" in sent[-1]["content"]
 
     @pytest.mark.asyncio
     async def test_new_clears_session_and_responds(self, tmp_path: Path) -> None:

tests/agent/test_runner_core.py

@@ -10,8 +10,6 @@ from unittest.mock import AsyncMock, MagicMock, patch
 import pytest
 
 from agent.runner_helpers import make_run_spec
-from nanobot.agent.context import TranscriptInput
-from nanobot.agent.context_governance import ContextWindowExceededError
 from nanobot.config.schema import AgentDefaults
 from nanobot.providers.base import (
     LLMProvider,
@@ -36,35 +34,6 @@ def _make_usage_spec(provider, tools):
     )
 
 
-def test_initial_transcript_is_built_from_structured_turn_input() -> None:
-    from nanobot.agent.runner import AgentRunner
-
-    provider = MagicMock(spec=LLMProvider)
-    transcript_input = TranscriptInput(
-        history=[{"role": "user", "content": "earlier"}],
-        current_message="fresh",
-    )
-    expected = [
-        {"role": "system", "content": "system"},
-        {"role": "user", "content": "earlier"},
-        {"role": "user", "content": "fresh"},
-    ]
-    transcript_builder = MagicMock(return_value=expected)
-    spec = make_run_spec(
-        provider,
-        initial_messages=None,
-        transcript_input=transcript_input,
-        transcript_builder=transcript_builder,
-        tools=MagicMock(),
-        model="test-model",
-        max_iterations=1,
-        max_tool_result_chars=_MAX_TOOL_RESULT_CHARS,
-    )
-
-    assert AgentRunner._initial_transcript(spec) == expected
-    transcript_builder.assert_called_once_with(transcript_input)
-
-
 def test_usage_or_estimate_replaces_reported_zero_for_content(monkeypatch) -> None:
     from nanobot.agent.runner import AgentRunner
 
@@ -87,7 +56,6 @@ def test_usage_or_estimate_replaces_reported_zero_for_content(monkeypatch) -> No
         _make_usage_spec(provider, tools),
         [{"role": "user", "content": "hello"}],
         response,
-        tool_definitions=tools.get_definitions(),
     )
 
     assert usage == LLMUsage.estimated(input_tokens=12, output_tokens=7).with_timing(
@@ -132,7 +100,6 @@ def test_usage_or_estimate_counts_tool_call_output_for_reported_zero(monkeypatch
         _make_usage_spec(provider, tools),
         [{"role": "user", "content": "hello"}],
         response,
-        tool_definitions=tools.get_definitions(),
     )
 
     assert usage == LLMUsage.estimated(input_tokens=13, output_tokens=9)
@@ -165,7 +132,6 @@ def test_usage_or_estimate_counts_error_without_estimating_tokens(
         _make_usage_spec(provider, tools),
         [{"role": "user", "content": "hello"}],
         response,
-        tool_definitions=tools.get_definitions(),
     )
 
     assert usage is not None
@@ -201,7 +167,6 @@ def test_usage_or_estimate_trusts_positive_reported_total(monkeypatch) -> None:
         _make_usage_spec(provider, tools),
         [{"role": "user", "content": "hello"}],
         response,
-        tool_definitions=tools.get_definitions(),
     )
 
     assert usage is not None
@@ -371,12 +336,14 @@ async def test_runner_replays_provider_state_without_chat_projection_duplicates(
 
 
 @pytest.mark.asyncio
-async def test_runner_preserves_tool_result_before_rejecting_unfit_followup():
+async def test_runner_governs_tool_result_before_adding_it_to_provider_state():
     from nanobot.agent.runner import AgentRunner
 
     provider = MagicMock(spec=LLMProvider)
     provider.can_resume_conversation_state.return_value = True
+    provider.supports_native_compaction.return_value = False
     calls = 0
+    captured_context: ProviderCallContext | None = None
     checkpoints: list[dict] = []
     state = ProviderConversationState(
         kind="openai_responses",
@@ -387,7 +354,7 @@ async def test_runner_preserves_tool_result_before_rejecting_unfit_followup():
     )
 
     async def chat_with_retry(**kwargs):
-        nonlocal calls
+        nonlocal calls, captured_context
         calls += 1
         if calls == 1:
             return LLMResponse(
@@ -401,6 +368,7 @@ async def test_runner_preserves_tool_result_before_rejecting_unfit_followup():
                 ],
                 provider_state=state,
             )
+        captured_context = kwargs["provider_context"]
         return LLMResponse(content="done")
 
     provider.chat_with_retry = chat_with_retry
@@ -411,7 +379,6 @@ async def test_runner_preserves_tool_result_before_rejecting_unfit_followup():
     async def checkpoint(payload: dict) -> None:
         checkpoints.append(payload)
 
-    with pytest.raises(ContextWindowExceededError):
     await AgentRunner().run(make_run_spec(
         provider,
         initial_messages=[
@@ -428,19 +395,21 @@ async def test_runner_preserves_tool_result_before_rejecting_unfit_followup():
         checkpoint_callback=checkpoint,
     ))
 
-    assert calls == 1
+    assert captured_context is not None
+    assert captured_context.conversation_state is not None
+    pending = captured_context.conversation_state.pending_messages
+    assert len(pending) == 1
+    assert pending[0]["role"] == "tool"
+    assert "compacted to fit context" in pending[0]["content"]
+    assert pending[0]["content"] != "x" * 5_000
     completed_checkpoint = next(
         checkpoint
         for checkpoint in checkpoints
         if checkpoint["phase"] == "tools_completed"
     )
     checkpoint_pending = completed_checkpoint["provider_state"].pending_messages
-    assert checkpoint_pending == [{
+    assert "compacted to fit context" in checkpoint_pending[0]["content"]
-        "role": "tool",
+    assert checkpoint_pending[0]["content"] != "x" * 5_000
-        "tool_call_id": "call_1",
-        "name": "read_file",
-        "content": "x" * 5_000,
-    }]
 
 
 @pytest.mark.asyncio

tests/agent/test_runner_errors.py

@@ -52,31 +52,7 @@ async def test_runner_returns_tool_exception_to_model_for_recovery():
         {"name": "list_dir", "status": "error", "detail": "boom"}
     ]
     tool_message = next(message for message in result.messages if message.get("role") == "tool")
-    retry_hint = "[Analyze the error above and try a different approach.]"
     assert "Error: RuntimeError: boom" in tool_message["content"]
-    assert tool_message["content"].count(retry_hint) == 1
-
-
-@pytest.mark.asyncio
-async def test_tool_execution_does_not_duplicate_existing_retry_hint():
-    retry_hint = "\n\n[Analyze the error above and try a different approach.]"
-    tools = SimpleNamespace(
-        execute=AsyncMock(return_value=ToolResult.error("Error: boom" + retry_hint)),
-    )
-
-    results, events = await execute_tool_calls(
-        tools,
-        [ToolCallRequest(id="call_1", name="list_dir", arguments={})],
-        concurrent=False,
-        external_lookup_counts={},
-        workspace_violation_counts={},
-        hook=AgentHook(),
-        context=AgentHookContext(iteration=0, messages=[]),
-    )
-
-    assert results == ["Error: boom" + retry_hint]
-    assert results[0].count(retry_hint) == 1
-    assert events[0]["status"] == "error"
 
 
 @pytest.mark.asyncio

tests/agent/test_runner_governance.py

@@ -1,7 +1,8 @@
-"""Tests for AgentRunner context governance: repair and request fitting."""
+"""Tests for AgentRunner context governance: backfill, orphan cleanup, microcompact, snip_history."""
 
 from __future__ import annotations
 
+from types import SimpleNamespace
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
@@ -11,14 +12,11 @@ from nanobot.agent.context_governance import (
     BACKFILL_CONTENT,
     ContextGovernanceConfig,
     ContextGovernor,
-    ContextWindowExceededError,
 )
 from nanobot.agent.runner import AgentRunSpec
 from nanobot.config.schema import AgentDefaults
 from nanobot.providers.base import (
-    LLMProvider,
     LLMResponse,
-    LLMUsage,
     ProviderConversationState,
     ToolCallRequest,
 )
@@ -30,6 +28,8 @@ def _governance_config(
     provider,
     tools,
     spec: AgentRunSpec,
+    *,
+    inflight_start_index: int = 0,
 ) -> ContextGovernanceConfig:
     return ContextGovernanceConfig(
         provider=provider,
@@ -41,6 +41,7 @@ def _governance_config(
         context_window_tokens=spec.runtime.context_window_tokens,
         context_block_limit=spec.context_block_limit,
         max_tokens=spec.runtime.generation.max_tokens,
+        inflight_start_index=inflight_start_index,
     )
 
 
@@ -88,508 +89,6 @@ async def test_runner_propagates_context_governance_failure():
     provider.chat_with_retry.assert_not_awaited()
 
 
-@pytest.mark.asyncio
-async def test_runner_locally_fits_oversized_initial_transcript(monkeypatch):
-    from nanobot.agent.runner import AgentRunner
-
-    provider = MagicMock(spec=LLMProvider)
-    provider.chat_with_retry = AsyncMock(return_value=LLMResponse(content="done"))
-    tools = MagicMock()
-    tools.get_definitions.return_value = []
-    old_content = "x" * 20_000
-    monkeypatch.setattr(
-        "nanobot.agent.context_governance.estimate_prompt_tokens_chain",
-        lambda _provider, _model, messages, _tools: (
-            (600, "test-counter")
-            if any(message.get("content") == old_content for message in messages)
-            else (100, "test-counter")
-        ),
-    )
-
-    result = await AgentRunner().run(make_run_spec(
-        provider,
-        initial_messages=[
-            {"role": "system", "content": "system"},
-            {"role": "user", "content": "old question"},
-            {"role": "assistant", "content": old_content},
-            {"role": "user", "content": "continue"},
-        ],
-        tools=tools,
-        model="local-model",
-        context_window_tokens=2_000,
-        context_block_limit=500,
-        max_tokens=100,
-        max_iterations=1,
-        max_tool_result_chars=_MAX_TOOL_RESULT_CHARS,
-    ))
-
-    assert provider.chat_with_retry.await_args.kwargs["messages"] == [
-        {"role": "system", "content": "system"},
-        {"role": "user", "content": "continue"},
-    ]
-    assert any(message.get("content") == old_content for message in result.messages)
-
-
-@pytest.mark.asyncio
-async def test_runner_governs_messages_added_by_before_iteration_hook(monkeypatch):
-    from nanobot.agent.hook import AgentHook, AgentHookContext
-    from nanobot.agent.runner import AgentRunner
-
-    provider = MagicMock(spec=LLMProvider)
-    provider.chat_with_retry = AsyncMock(return_value=LLMResponse(content="unexpected"))
-    tools = MagicMock()
-    tools.get_definitions.return_value = []
-    oversized = "hook-added-oversized-message"
-
-    monkeypatch.setattr(
-        "nanobot.agent.context_governance.estimate_prompt_tokens_chain",
-        lambda _provider, _model, messages, _tools: (
-            (2_000, "test-counter")
-            if any(message.get("content") == oversized for message in messages)
-            else (100, "test-counter")
-        ),
-    )
-
-    class MutatingHook(AgentHook):
-        async def before_iteration(self, context: AgentHookContext) -> None:
-            context.messages.append({"role": "user", "content": oversized})
-
-    with pytest.raises(ContextWindowExceededError):
-        await AgentRunner().run(make_run_spec(
-            provider,
-            initial_messages=[{"role": "user", "content": "hello"}],
-            tools=tools,
-            model="local-model",
-            context_window_tokens=2_000,
-            context_block_limit=500,
-            max_iterations=1,
-            max_tool_result_chars=_MAX_TOOL_RESULT_CHARS,
-            hook=MutatingHook(),
-        ))
-
-    provider.chat_with_retry.assert_not_awaited()
-
-
-@pytest.mark.asyncio
-async def test_runner_drops_resumable_provider_state_when_request_is_fitted(monkeypatch):
-    from nanobot.agent.runner import AgentRunner
-
-    provider = MagicMock(spec=LLMProvider)
-    provider.can_resume_conversation_state.return_value = True
-    captured_contexts = []
-    old_content = "old-oversized-history"
-    candidate = ProviderConversationState(
-        kind="openai_responses",
-        provider="openai:test",
-        model="local-model",
-        version=1,
-        payload={"items": [{"type": "message", "content": "fresh state"}]},
-    )
-
-    async def chat_with_retry(*, provider_context=None, **_kwargs):
-        captured_contexts.append(provider_context)
-        return LLMResponse(
-            content="done",
-            usage=LLMUsage.reported(input_tokens=100, output_tokens=10),
-            provider_state=candidate,
-        )
-
-    provider.chat_with_retry = chat_with_retry
-    tools = MagicMock()
-    tools.get_definitions.return_value = []
-    monkeypatch.setattr(
-        "nanobot.agent.context_governance.estimate_prompt_tokens_chain",
-        lambda _provider, _model, messages, _tools: (
-            (600, "test-counter")
-            if any(message.get("content") == old_content for message in messages)
-            else (100, "test-counter")
-        ),
-    )
-    monkeypatch.setattr(
-        "nanobot.agent.context_governance.estimate_message_tokens",
-        lambda message: 450 if message.get("content") == old_content else 50,
-    )
-    saved_state = ProviderConversationState(
-        kind="openai_responses",
-        provider="openai:test",
-        model="local-model",
-        version=1,
-        payload={"items": [{"type": "message", "content": "stale state"}]},
-    )
-
-    result = await AgentRunner().run(make_run_spec(
-        provider,
-        initial_messages=[
-            {"role": "assistant", "content": old_content},
-            {"role": "user", "content": "continue"},
-        ],
-        tools=tools,
-        model="local-model",
-        context_window_tokens=2_000,
-        context_block_limit=500,
-        max_iterations=1,
-        max_tool_result_chars=_MAX_TOOL_RESULT_CHARS,
-        provider_state=saved_state,
-    ))
-
-    assert captured_contexts[0].conversation_state is None
-    assert result.provider_state is not None
-    assert result.provider_state.payload == candidate.payload
-
-
-@pytest.mark.asyncio
-async def test_runner_fits_each_malformed_retry_with_its_actual_tools(monkeypatch):
-    from nanobot.agent.runner import AgentRunner
-
-    provider = MagicMock(spec=LLMProvider)
-    calls: list[dict] = []
-    estimated_tools: list[object] = []
-    definitions = [{"type": "function", "function": {"name": "read_file"}}]
-
-    async def chat_with_retry(*, messages, tools=None, **_kwargs):
-        calls.append({"messages": [dict(message) for message in messages], "tools": tools})
-        if len(calls) < 3:
-            return LLMResponse(
-                content="bad tool request",
-                tool_calls=[ToolCallRequest(id=f"bad_{len(calls)}", name=None, arguments={})],
-                finish_reason="tool_calls",
-                usage=LLMUsage.reported(input_tokens=100, output_tokens=10),
-            )
-        return LLMResponse(
-            content="recovered",
-            usage=LLMUsage.reported(input_tokens=100, output_tokens=10),
-        )
-
-    def estimate(_provider, _model, messages, _tools):
-        estimated_tools.append(_tools)
-        user_count = sum(message.get("role") == "user" for message in messages)
-        return (600 if user_count > 1 else 100), "test-counter"
-
-    provider.chat_with_retry = chat_with_retry
-    tools = MagicMock()
-    tools.get_definitions.return_value = definitions
-    monkeypatch.setattr(
-        "nanobot.agent.context_governance.estimate_prompt_tokens_chain",
-        estimate,
-    )
-    monkeypatch.setattr(
-        "nanobot.agent.context_governance.estimate_message_tokens",
-        lambda _message: 300,
-    )
-
-    result = await AgentRunner().run(make_run_spec(
-        provider,
-        initial_messages=[{"role": "user", "content": "use a tool"}],
-        tools=tools,
-        model="local-model",
-        context_window_tokens=2_000,
-        context_block_limit=500,
-        max_iterations=1,
-        max_tool_result_chars=_MAX_TOOL_RESULT_CHARS,
-    ))
-
-    assert [call["tools"] for call in calls] == [definitions, definitions, None]
-    assert definitions in estimated_tools
-    assert None in estimated_tools
-    assert [len(call["messages"]) for call in calls] == [1, 1, 1]
-    assert result.final_content == "recovered"
-    assert result.messages == [
-        {"role": "user", "content": "use a tool"},
-        {"role": "assistant", "content": "recovered"},
-    ]
-
-
-@pytest.mark.asyncio
-async def test_runner_fits_empty_response_finalization_before_dispatch(monkeypatch):
-    from nanobot.agent.runner import AgentRunner
-
-    provider = MagicMock(spec=LLMProvider)
-    calls: list[dict] = []
-
-    async def chat_with_retry(*, messages, tools=None, **_kwargs):
-        calls.append({"messages": [dict(message) for message in messages], "tools": tools})
-        if len(calls) < 3:
-            return LLMResponse(
-                content=None,
-                usage=LLMUsage.reported(input_tokens=100, output_tokens=1),
-            )
-        return LLMResponse(
-            content="finalized",
-            usage=LLMUsage.reported(input_tokens=100, output_tokens=10),
-        )
-
-    def estimate(_provider, _model, messages, _tools):
-        contents = [str(message.get("content") or "") for message in messages]
-        has_original = "do task" in contents
-        has_finalization = any("conversation above" in content for content in contents)
-        return (600 if has_original and has_finalization else 100), "test-counter"
-
-    provider.chat_with_retry = chat_with_retry
-    tools = MagicMock()
-    tools.get_definitions.return_value = []
-    monkeypatch.setattr(
-        "nanobot.agent.context_governance.estimate_prompt_tokens_chain",
-        estimate,
-    )
-    monkeypatch.setattr(
-        "nanobot.agent.context_governance.estimate_message_tokens",
-        lambda _message: 300,
-    )
-
-    result = await AgentRunner().run(make_run_spec(
-        provider,
-        initial_messages=[{"role": "user", "content": "do task"}],
-        tools=tools,
-        model="local-model",
-        context_window_tokens=2_000,
-        context_block_limit=500,
-        max_iterations=3,
-        max_tool_result_chars=_MAX_TOOL_RESULT_CHARS,
-    ))
-
-    assert len(calls) == 3
-    assert calls[-1]["tools"] is None
-    assert all(message.get("content") != "do task" for message in calls[-1]["messages"])
-    assert result.final_content == "finalized"
-
-
-@pytest.mark.asyncio
-async def test_runner_fits_max_iteration_finalization_before_dispatch(monkeypatch):
-    from nanobot.agent.runner import AgentRunner
-
-    provider = MagicMock(spec=LLMProvider)
-    calls: list[dict] = []
-    oversized_result = "oversized-current-tool-result"
-
-    async def chat_with_retry(*, messages, tools=None, **_kwargs):
-        calls.append({"messages": [dict(message) for message in messages], "tools": tools})
-        if len(calls) == 1:
-            return LLMResponse(
-                content="working",
-                tool_calls=[ToolCallRequest(id="call_1", name="read_file", arguments={})],
-                finish_reason="tool_calls",
-                usage=LLMUsage.reported(input_tokens=100, output_tokens=10),
-            )
-        return LLMResponse(
-            content="safe summary",
-            usage=LLMUsage.reported(input_tokens=100, output_tokens=10),
-        )
-
-    def estimate(_provider, _model, messages, _tools):
-        has_oversized = any(
-            message.get("content") == oversized_result for message in messages
-        )
-        return (600 if has_oversized else 100), "test-counter"
-
-    provider.chat_with_retry = chat_with_retry
-    tools = MagicMock()
-    tools.get_definitions.return_value = []
-    tools.execute = AsyncMock(return_value=oversized_result)
-    monkeypatch.setattr(
-        "nanobot.agent.context_governance.estimate_prompt_tokens_chain",
-        estimate,
-    )
-    monkeypatch.setattr(
-        "nanobot.agent.context_governance.estimate_message_tokens",
-        lambda message: 600 if message.get("content") == oversized_result else 50,
-    )
-
-    result = await AgentRunner().run(make_run_spec(
-        provider,
-        initial_messages=[{"role": "user", "content": "inspect"}],
-        tools=tools,
-        model="local-model",
-        context_window_tokens=2_000,
-        context_block_limit=500,
-        max_iterations=1,
-        max_tool_result_chars=_MAX_TOOL_RESULT_CHARS,
-    ))
-
-    assert len(calls) == 2
-    assert calls[-1]["tools"] is None
-    assert all(
-        message.get("content") != oversized_result
-        for message in calls[-1]["messages"]
-    )
-    assert any(message.get("content") == oversized_result for message in result.messages)
-    assert result.final_content == "safe summary"
-
-
-@pytest.mark.parametrize(
-    ("input_tokens", "expected_fitted"),
-    [(500, True), (100, False)],
-)
-def test_matching_reported_provider_usage_avoids_local_estimate(
-    monkeypatch,
-    input_tokens,
-    expected_fitted,
-):
-    provider = MagicMock(spec=LLMProvider)
-    tools = MagicMock()
-    tools.get_definitions.return_value = []
-    spec = make_run_spec(
-        provider,
-        initial_messages=[{"role": "user", "content": "hello"}],
-        tools=tools,
-        model="local-model",
-        context_window_tokens=2_000,
-        context_block_limit=500,
-        max_iterations=1,
-        max_tool_result_chars=_MAX_TOOL_RESULT_CHARS,
-    )
-    monkeypatch.setattr(
-        "nanobot.agent.context_governance.estimate_prompt_tokens_chain",
-        lambda *_args, **_kwargs: (_ for _ in ()).throw(
-            AssertionError("matching provider usage must be authoritative")
-        ),
-    )
-
-    governor = ContextGovernor()
-    monkeypatch.setattr(governor, "fit_to_budget", lambda *_args, **_kwargs: [])
-    _messages, fitted = governor.fit_request(
-        _governance_config(provider, tools, spec),
-        spec.initial_messages,
-        LLMUsage.reported(input_tokens=input_tokens, output_tokens=10),
-        usage_matches_messages=True,
-        tool_definitions=tools.get_definitions(),
-    )
-
-    assert fitted is expected_fitted
-
-
-def test_changed_messages_use_local_estimate_after_reported_usage(monkeypatch):
-    provider = MagicMock(spec=LLMProvider)
-    tools = MagicMock()
-    tools.get_definitions.return_value = []
-    spec = make_run_spec(
-        provider,
-        initial_messages=[{"role": "user", "content": "new tool output"}],
-        tools=tools,
-        model="local-model",
-        context_window_tokens=2_000,
-        context_block_limit=500,
-        max_iterations=1,
-        max_tool_result_chars=_MAX_TOOL_RESULT_CHARS,
-    )
-    estimate = MagicMock(return_value=(600, "test-counter"))
-    monkeypatch.setattr(
-        "nanobot.agent.context_governance.estimate_prompt_tokens_chain",
-        estimate,
-    )
-
-    governor = ContextGovernor()
-    monkeypatch.setattr(governor, "fit_to_budget", lambda *_args, **_kwargs: [])
-    _messages, fitted = governor.fit_request(
-        _governance_config(provider, tools, spec),
-        spec.initial_messages,
-        LLMUsage.reported(input_tokens=900, output_tokens=10),
-        usage_matches_messages=False,
-        tool_definitions=tools.get_definitions(),
-    )
-
-    assert fitted is True
-    estimate.assert_called_once()
-
-
-@pytest.mark.asyncio
-async def test_runner_counts_resumed_provider_state_before_dispatch(monkeypatch):
-    from nanobot.agent.runner import AgentRunner
-
-    provider = MagicMock(spec=LLMProvider)
-    provider.can_resume_conversation_state.return_value = True
-    captured_contexts = []
-
-    async def chat_with_retry(*, provider_context=None, **_kwargs):
-        captured_contexts.append(provider_context)
-        return LLMResponse(
-            content="done",
-            usage=LLMUsage.reported(input_tokens=100, output_tokens=10),
-        )
-
-    provider.chat_with_retry = chat_with_retry
-    tools = MagicMock()
-    tools.get_definitions.return_value = []
-    current_message = {"role": "user", "content": "new delta"}
-    saved_state = ProviderConversationState(
-        kind="openai_responses",
-        provider="openai:test",
-        model="local-model",
-        version=1,
-        payload={
-            "items": [{"type": "reasoning", "encrypted_content": "opaque"}],
-            "context_tokens": 450,
-        },
-        pending_messages=[current_message],
-    )
-    monkeypatch.setattr(
-        "nanobot.agent.context_governance.estimate_prompt_tokens_chain",
-        lambda *_args, **_kwargs: (100, "test-counter"),
-    )
-    monkeypatch.setattr(
-        "nanobot.providers.conversation_state.estimate_prompt_tokens_chain",
-        lambda *_args, **_kwargs: (100, "test-counter"),
-    )
-
-    result = await AgentRunner().run(make_run_spec(
-        provider,
-        initial_messages=[current_message],
-        tools=tools,
-        model="local-model",
-        context_window_tokens=2_000,
-        context_block_limit=500,
-        max_iterations=1,
-        max_tool_result_chars=_MAX_TOOL_RESULT_CHARS,
-        provider_state=saved_state,
-    ))
-
-    assert captured_contexts[0].conversation_state is None
-    assert result.messages == [
-        current_message,
-        {"role": "assistant", "content": "done"},
-    ]
-
-
-@pytest.mark.asyncio
-@pytest.mark.parametrize(
-    ("context_block_limit", "expected_budget"),
-    [(500, 500), (None, 0)],
-)
-async def test_runner_refuses_locally_fitted_request_that_still_cannot_fit(
-    monkeypatch,
-    context_block_limit,
-    expected_budget,
-):
-    from nanobot.agent.runner import AgentRunner
-
-    provider = MagicMock(spec=LLMProvider)
-    provider.chat_with_retry = AsyncMock(return_value=LLMResponse(content="unexpected"))
-    tools = MagicMock()
-    tools.get_definitions.return_value = []
-    monkeypatch.setattr(
-        "nanobot.agent.context_governance.estimate_prompt_tokens_chain",
-        lambda *_args, **_kwargs: (2_000, "test-counter"),
-    )
-
-    with pytest.raises(ContextWindowExceededError) as exc_info:
-        await AgentRunner().run(make_run_spec(
-            provider,
-            initial_messages=[
-                {"role": "system", "content": "oversized system"},
-                {"role": "user", "content": "oversized user"},
-            ],
-            tools=tools,
-            model="local-model",
-            context_window_tokens=1_000,
-            context_block_limit=context_block_limit,
-            max_iterations=1,
-            max_tool_result_chars=_MAX_TOOL_RESULT_CHARS,
-        ))
-
-    assert exc_info.value.estimated_tokens == 2_000
-    assert exc_info.value.input_budget == expected_budget
-    provider.chat_with_retry.assert_not_awaited()
-
-
 def test_snip_history_drops_orphaned_tool_results_from_trimmed_slice(monkeypatch):
     provider = MagicMock()
     tools = MagicMock()
@@ -631,11 +130,7 @@ def test_snip_history_drops_orphaned_tool_results_from_trimmed_slice(monkeypatch
         lambda msg: token_sizes.get(str(msg.get("content")), 40),
     )
 
-    trimmed = ContextGovernor().snip_history(
+    trimmed = ContextGovernor().snip_history(_governance_config(provider, tools, spec), messages)
-        _governance_config(provider, tools, spec),
-        messages,
-        tool_definitions=tools.get_definitions(),
-    )
 
     # After the fix, the user message is recovered so the sequence is valid
     # for providers that require system → user (e.g. GLM error 1214).
@@ -687,11 +182,7 @@ def test_snip_history_reserves_budget_for_tool_definitions(monkeypatch):
         lambda msg: token_sizes.get(str(msg.get("content")), 40),
     )
 
-    trimmed = ContextGovernor().snip_history(
+    trimmed = ContextGovernor().snip_history(_governance_config(provider, tools, spec), messages)
-        _governance_config(provider, tools, spec),
-        messages,
-        tool_definitions=tools.get_definitions(),
-    )
 
     contents = [message.get("content") for message in trimmed]
     assert contents == ["system", "recent two"]
@@ -974,6 +465,260 @@ async def test_runner_backfill_only_mutates_model_context_not_returned_messages(
     ]
 
 
+# ---------------------------------------------------------------------------
+# Microcompact (stale tool result compaction)
+# ---------------------------------------------------------------------------
+
+
+def _microcompact_messages(*, total: int, tool_name: str, content: str) -> list[dict]:
+    messages: list[dict] = [{"role": "system", "content": "sys"}]
+    for i in range(total):
+        messages.append({
+            "role": "assistant",
+            "content": "",
+            "tool_calls": [{
+                "id": f"c{i}",
+                "type": "function",
+                "function": {"name": tool_name, "arguments": "{}"},
+            }],
+        })
+        messages.append({
+            "role": "tool",
+            "tool_call_id": f"c{i}",
+            "name": tool_name,
+            "content": content,
+        })
+    return messages
+
+
+def test_microcompact_skips_when_prompt_under_hard_budget(monkeypatch):
+    """Cache-friendly path: in-flight tool results stay stable while prompt fits."""
+    provider = MagicMock()
+    provider.generation = SimpleNamespace(max_tokens=0)
+    tools = MagicMock()
+    tools.get_definitions.return_value = []
+
+    total = 15
+    long_content = "x" * 600
+    messages = _microcompact_messages(total=total, tool_name="read_file", content=long_content)
+    spec = make_run_spec(provider,
+        initial_messages=messages,
+        tools=tools,
+        model="test-model",
+        max_iterations=1,
+        max_tool_result_chars=_MAX_TOOL_RESULT_CHARS,
+        max_tokens=0,
+        context_window_tokens=20_000,
+    )
+
+    monkeypatch.setattr(
+        "nanobot.agent.context_governance.estimate_prompt_tokens_chain",
+        lambda *_args, **_kwargs: (1000, "test"),
+    )
+
+    result = ContextGovernor().compact_inflight_overflow(
+        _governance_config(provider, tools, spec),
+        messages,
+        set(),
+    )
+
+    assert result is messages
+
+
+def test_microcompact_overflow_compacts_to_low_watermark(monkeypatch):
+    """Overflow path: compact in-flight stale results with headroom for later calls."""
+    provider = MagicMock()
+    provider.generation = SimpleNamespace(max_tokens=0)
+    tools = MagicMock()
+    tools.get_definitions.return_value = []
+
+    total = 18
+    long_content = "x" * 600
+    messages = _microcompact_messages(total=total, tool_name="read_file", content=long_content)
+    spec = make_run_spec(provider,
+        initial_messages=messages,
+        tools=tools,
+        model="test-model",
+        max_iterations=1,
+        max_tool_result_chars=_MAX_TOOL_RESULT_CHARS,
+        max_tokens=0,
+        context_window_tokens=2224,  # input budget 1200, low target 1020
+    )
+
+    def estimate(_provider, _model, msgs, _tools):
+        return sum(
+            100 if (content := msg.get("content")) == long_content
+            else 1 if isinstance(content, str) and "compacted to fit context" in content
+            else 0
+            for msg in msgs
+            if msg.get("role") == "tool"
+        ), "test"
+
+    monkeypatch.setattr("nanobot.agent.context_governance.estimate_prompt_tokens_chain", estimate)
+
+    result = ContextGovernor().compact_inflight_overflow(
+        _governance_config(provider, tools, spec),
+        messages,
+        set(),
+    )
+    tool_msgs = [m for m in result if m.get("role") == "tool"]
+    compacted = [m for m in tool_msgs if "compacted to fit context" in str(m.get("content", ""))]
+    preserved = [m for m in tool_msgs if m.get("content") == long_content]
+
+    assert len(compacted) == 8
+    assert len(preserved) == total - 8
+    assert [m["tool_call_id"] for m in compacted] == [f"c{i}" for i in range(8)]
+
+
+def test_microcompact_compacts_newest_when_it_alone_overflows(monkeypatch):
+    """An unfit newest result tells the model to retry narrowly or report the limit."""
+    provider = MagicMock()
+    provider.generation = SimpleNamespace(max_tokens=0)
+    tools = MagicMock()
+    tools.get_definitions.return_value = []
+
+    long_content = "x" * 600
+    messages = _microcompact_messages(total=1, tool_name="read_file", content=long_content)
+    spec = make_run_spec(provider,
+        initial_messages=messages,
+        tools=tools,
+        model="test-model",
+        max_iterations=1,
+        max_tool_result_chars=_MAX_TOOL_RESULT_CHARS,
+        max_tokens=0,
+        context_window_tokens=2000,
+        context_block_limit=500,
+    )
+
+    def estimate(_provider, _model, msgs, _tools):
+        return sum(
+            1000 if msg.get("content") == long_content else 1
+            for msg in msgs
+            if msg.get("role") == "tool"
+        ), "test"
+
+    monkeypatch.setattr("nanobot.agent.context_governance.estimate_prompt_tokens_chain", estimate)
+
+    compacted_tool_call_ids: set[str] = set()
+    result = ContextGovernor().compact_inflight_overflow(
+        _governance_config(provider, tools, spec),
+        messages,
+        compacted_tool_call_ids,
+    )
+
+    tool_msg = next(m for m in result if m.get("role") == "tool")
+    assert "compacted to fit context" in tool_msg["content"]
+    assert "Do not repeat the same call unchanged" in tool_msg["content"]
+    assert "Retry with a narrower path, query, range, or result limit" in tool_msg["content"]
+    assert "tell the user the task cannot fit" in tool_msg["content"]
+    assert compacted_tool_call_ids == {"c0"}
+
+
+def test_context_governor_keeps_compaction_boundary_stable(monkeypatch):
+    provider = MagicMock()
+    provider.generation = SimpleNamespace(max_tokens=0)
+    tools = MagicMock()
+    tools.get_definitions.return_value = []
+
+    total = 18
+    long_content = "x" * 600
+    messages = _microcompact_messages(total=total, tool_name="read_file", content=long_content)
+    spec = make_run_spec(provider,
+        initial_messages=messages,
+        tools=tools,
+        model="test-model",
+        max_iterations=1,
+        max_tool_result_chars=_MAX_TOOL_RESULT_CHARS,
+        max_tokens=0,
+        context_window_tokens=2224,
+    )
+
+    def estimate(_provider, _model, msgs, _tools):
+        return sum(
+            100 if msg.get("content") == long_content else 1
+            for msg in msgs
+            if msg.get("role") == "tool"
+        ), "test"
+
+    monkeypatch.setattr("nanobot.agent.context_governance.estimate_prompt_tokens_chain", estimate)
+
+    governor = ContextGovernor()
+    compacted_tool_call_ids: set[str] = set()
+    config = _governance_config(provider, tools, spec, inflight_start_index=0)
+    first = governor.compact_inflight_overflow(config, messages, compacted_tool_call_ids)
+    first_ids = set(compacted_tool_call_ids)
+
+    second = governor.compact_inflight_overflow(config, messages, compacted_tool_call_ids)
+
+    assert compacted_tool_call_ids == first_ids
+    assert [m.get("content") for m in second] == [m.get("content") for m in first]
+
+
+def test_microcompact_preserves_short_results(monkeypatch):
+    """Short tool results below the compaction threshold should not be replaced."""
+    provider = MagicMock()
+    provider.generation = SimpleNamespace(max_tokens=0)
+    tools = MagicMock()
+    tools.get_definitions.return_value = []
+
+    total = 15
+    messages = _microcompact_messages(total=total, tool_name="exec", content="short")
+    spec = make_run_spec(provider,
+        initial_messages=messages,
+        tools=tools,
+        model="test-model",
+        max_iterations=1,
+        max_tool_result_chars=_MAX_TOOL_RESULT_CHARS,
+        max_tokens=0,
+        context_window_tokens=2024,
+    )
+
+    monkeypatch.setattr(
+        "nanobot.agent.context_governance.estimate_prompt_tokens_chain",
+        lambda *_args, **_kwargs: (2000, "test"),
+    )
+
+    result = ContextGovernor().compact_inflight_overflow(
+        _governance_config(provider, tools, spec),
+        messages,
+        set(),
+    )
+    assert result is messages  # no copy needed — all stale results are short
+
+
+def test_microcompact_skips_non_compactable_tools(monkeypatch):
+    """Non-compactable tools (e.g. 'message') should never be replaced."""
+    provider = MagicMock()
+    provider.generation = SimpleNamespace(max_tokens=0)
+    tools = MagicMock()
+    tools.get_definitions.return_value = []
+
+    total = 15
+    long_content = "y" * 1000
+    messages = _microcompact_messages(total=total, tool_name="message", content=long_content)
+    spec = make_run_spec(provider,
+        initial_messages=messages,
+        tools=tools,
+        model="test-model",
+        max_iterations=1,
+        max_tool_result_chars=_MAX_TOOL_RESULT_CHARS,
+        max_tokens=0,
+        context_window_tokens=2024,
+    )
+
+    monkeypatch.setattr(
+        "nanobot.agent.context_governance.estimate_prompt_tokens_chain",
+        lambda *_args, **_kwargs: (2000, "test"),
+    )
+
+    result = ContextGovernor().compact_inflight_overflow(
+        _governance_config(provider, tools, spec),
+        messages,
+        set(),
+    )
+    assert result is messages  # no compactable tools found
+
+
 def test_governance_repairs_orphans_after_snip():
     """After snipping clips an assistant+tool_calls, orphan repair cleans up the tail."""
     # Simulate snipping that keeps only the tail: drop the assistant with
@@ -1073,11 +818,7 @@ def test_snip_history_preserves_user_message_after_truncation(monkeypatch):
         lambda msg: token_sizes.get(str(msg.get("content")), 100),
     )
 
-    trimmed = ContextGovernor().snip_history(
+    trimmed = ContextGovernor().snip_history(_governance_config(provider, tools, spec), messages)
-        _governance_config(provider, tools, spec),
-        messages,
-        tool_definitions=tools.get_definitions(),
-    )
 
     # The first non-system message MUST be user (not assistant).
     non_system = [m for m in trimmed if m.get("role") != "system"]
@@ -1122,11 +863,7 @@ def test_snip_history_no_user_at_all_falls_back_gracefully(monkeypatch):
         lambda msg: 100,
     )
 
-    trimmed = ContextGovernor().snip_history(
+    trimmed = ContextGovernor().snip_history(_governance_config(provider, tools, spec), messages)
-        _governance_config(provider, tools, spec),
-        messages,
-        tool_definitions=tools.get_definitions(),
-    )
 
     # Should not crash.  The result should still be a valid list.
     assert isinstance(trimmed, list)
@@ -1134,6 +871,7 @@ def test_snip_history_no_user_at_all_falls_back_gracefully(monkeypatch):
     assert any(m.get("role") == "system" for m in trimmed)
     # The _enforce_role_alternation safety net must be able to fix whatever
     # _snip_history returns here — verify it produces a valid sequence.
+    from nanobot.providers.base import LLMProvider
     fixed = LLMProvider._enforce_role_alternation(trimmed)
     non_system = [m for m in fixed if m["role"] != "system"]
     if non_system:

tests/agent/test_runner_injections.py

@@ -10,7 +10,6 @@ import pytest
 
 from agent.runner_helpers import make_run_spec
 from nanobot.agent.automation_turns import publish_next_deferred_turn
-from nanobot.agent.context import TranscriptInput
 from nanobot.agent.tools.context import RequestContext
 from nanobot.config.schema import AgentDefaults
 from nanobot.providers.base import LLMResponse, ToolCallRequest
@@ -618,7 +617,7 @@ async def test_loop_injected_followup_preserves_image_media(tmp_path):
 
     runtime = loop.llm_runtime()
     result = await loop._run_agent_loop(
-        TranscriptInput(history=[{"role": "user", "content": "hello"}], current_message=None),
+        [{"role": "user", "content": "hello"}],
         runtime=runtime,
         request_context=RequestContext(channel="cli", chat_id="c", runtime=runtime),
         pending_queue=pending_queue,
@@ -712,10 +711,7 @@ async def test_pending_injection_resolves_its_own_runtime_context(tmp_path):
 
     runtime = loop.llm_runtime()
     result = await loop._run_agent_loop(
-        TranscriptInput(
+        [{"role": "user", "content": "initial message from user A"}],
-            history=[{"role": "user", "content": "initial message from user A"}],
-            current_message=None,
-        ),
         runtime=runtime,
         session=session,
         request_context=RequestContext(
@@ -816,7 +812,7 @@ async def test_subagent_pending_injection_is_hidden_history_and_not_merged(tmp_p
 
     runtime = loop.llm_runtime()
     result = await loop._run_agent_loop(
-        TranscriptInput(history=[{"role": "user", "content": "hello"}], current_message=None),
+        [{"role": "user", "content": "hello"}],
         runtime=runtime,
         request_context=RequestContext(channel="cli", chat_id="c", runtime=runtime),
         pending_queue=pending_queue,
@@ -1480,7 +1476,7 @@ async def test_pending_queue_preserves_overflow_for_next_injection_cycle(tmp_pat
 
     runtime = loop.llm_runtime()
     result = await loop._run_agent_loop(
-        TranscriptInput(history=[{"role": "user", "content": "hello"}], current_message=None),
+        [{"role": "user", "content": "hello"}],
         runtime=runtime,
         request_context=RequestContext(channel="cli", chat_id="c", runtime=runtime),
         pending_queue=pending_queue,

tests/agent/test_runner_reasoning.py

@@ -9,7 +9,6 @@ channels, gated by ``context.streamed_reasoning`` rather than
 
 from __future__ import annotations
 
-import asyncio
 from typing import Any
 from unittest.mock import AsyncMock, MagicMock
 
@@ -83,18 +82,6 @@ class _LifecycleRecordingHook(AgentHook):
         self.events.append(f"hosted_tool:{event.get('phase')}")
 
 
-class _BlockingReasoningEndHook(_LifecycleRecordingHook):
-    def __init__(self) -> None:
-        super().__init__()
-        self.reasoning_end_started = asyncio.Event()
-        self.release_reasoning_end = asyncio.Event()
-
-    async def emit_reasoning_end(self) -> None:
-        self.reasoning_end_started.set()
-        await self.release_reasoning_end.wait()
-        await super().emit_reasoning_end()
-
-
 @pytest.mark.asyncio
 async def test_runner_preserves_reasoning_fields_in_assistant_history():
     """Reasoning fields ride along on the persisted assistant message so
@@ -567,86 +554,6 @@ async def test_runner_closes_native_reasoning_before_hosted_tool_event():
     ]
 
 
-@pytest.mark.asyncio
-async def test_runner_closes_native_reasoning_when_stream_is_cancelled():
-    from nanobot.agent.runner import AgentRunner
-
-    provider = MagicMock()
-    reasoning_started = asyncio.Event()
-    release_provider = asyncio.Event()
-
-    async def chat_stream_with_retry(
-        *, on_thinking_delta=None, **kwargs
-    ):
-        if on_thinking_delta:
-            await on_thinking_delta("inspect")
-        reasoning_started.set()
-        await release_provider.wait()
-        raise AssertionError("the cancelled provider call should not complete")
-
-    provider.chat_stream_with_retry = chat_stream_with_retry
-    tools = MagicMock()
-    tools.get_definitions.return_value = []
-    hook = _LifecycleRecordingHook()
-
-    task = asyncio.create_task(AgentRunner().run(make_run_spec(
-        provider,
-        initial_messages=[{"role": "user", "content": "inspect"}],
-        tools=tools,
-        model="test-model",
-        max_iterations=1,
-        max_tool_result_chars=_MAX_TOOL_RESULT_CHARS,
-        hook=hook,
-    )))
-    await reasoning_started.wait()
-
-    task.cancel()
-    with pytest.raises(asyncio.CancelledError):
-        await task
-
-    assert hook.events == ["reasoning:inspect", "reasoning_end"]
-
-
-@pytest.mark.asyncio
-async def test_runner_settles_native_reasoning_end_before_propagating_cancellation():
-    from nanobot.agent.runner import AgentRunner
-
-    provider = MagicMock()
-
-    async def chat_stream_with_retry(
-        *, on_content_delta=None, on_thinking_delta=None, **kwargs
-    ):
-        if on_thinking_delta:
-            await on_thinking_delta("inspect")
-        if on_content_delta:
-            await on_content_delta("done")
-        raise AssertionError("the cancelled provider call should not complete")
-
-    provider.chat_stream_with_retry = chat_stream_with_retry
-    tools = MagicMock()
-    tools.get_definitions.return_value = []
-    hook = _BlockingReasoningEndHook()
-
-    task = asyncio.create_task(AgentRunner().run(make_run_spec(
-        provider,
-        initial_messages=[{"role": "user", "content": "inspect"}],
-        tools=tools,
-        model="test-model",
-        max_iterations=1,
-        max_tool_result_chars=_MAX_TOOL_RESULT_CHARS,
-        hook=hook,
-    )))
-    await hook.reasoning_end_started.wait()
-
-    task.cancel()
-    await asyncio.sleep(0)
-    hook.release_reasoning_end.set()
-    with pytest.raises(asyncio.CancelledError):
-        await task
-
-    assert hook.events == ["reasoning:inspect", "reasoning_end"]
-
-
 @pytest.mark.asyncio
 async def test_runner_strips_thinking_tags_from_native_thinking_deltas():
     from nanobot.agent.runner import AgentRunner

tests/agent/test_session_model_runtime.py

@@ -114,7 +114,7 @@ async def test_removed_session_model_preset_falls_back_and_clears_metadata(tmp_p
         provider=base,
         workspace=tmp_path,
         model="base-model",
-        context_window_tokens=16_000,
+        context_window_tokens=8_000,
     )
     loop.schedule_background = lambda coro: coro.close()  # type: ignore[method-assign]
     session_key = "sdk:removed-preset"
@@ -196,7 +196,7 @@ async def test_sdk_custom_model_preset_metadata_does_not_select_runtime(
         provider=base,
         workspace=tmp_path,
         model="base-model",
-        context_window_tokens=16_000,
+        context_window_tokens=8_000,
     )
     loop.schedule_background = lambda coro: coro.close()  # type: ignore[method-assign]
     bot = Nanobot(loop)

tests/agent/test_task_cancel.py

@@ -42,65 +42,6 @@ def _make_loop(*, tools_config=None):
     return loop, bus
 
 
-class TestActiveTaskTracking:
-    @pytest.mark.asyncio
-    async def test_completed_task_removes_empty_session_group(self):
-        loop, _bus = _make_loop()
-        release = asyncio.Event()
-        task = asyncio.create_task(release.wait())
-
-        loop._track_active_task("test:c1", task)
-        release.set()
-        await task
-        await asyncio.sleep(0)
-
-        assert "test:c1" not in loop._active_tasks
-
-    @pytest.mark.asyncio
-    async def test_session_group_remains_until_last_task_completes(self):
-        loop, _bus = _make_loop()
-        releases = [asyncio.Event(), asyncio.Event()]
-        tasks = [asyncio.create_task(release.wait()) for release in releases]
-        for task in tasks:
-            loop._track_active_task("test:c1", task)
-
-        releases[0].set()
-        await tasks[0]
-        await asyncio.sleep(0)
-
-        assert loop._active_tasks["test:c1"] == {tasks[1]}
-
-        releases[1].set()
-        await tasks[1]
-        await asyncio.sleep(0)
-
-        assert "test:c1" not in loop._active_tasks
-
-    @pytest.mark.asyncio
-    async def test_old_callback_preserves_replacement_session_group(self):
-        loop, _bus = _make_loop()
-        old_release = asyncio.Event()
-        new_release = asyncio.Event()
-        old_task = asyncio.create_task(old_release.wait())
-        new_task = asyncio.create_task(new_release.wait())
-
-        loop._track_active_task("test:c1", old_task)
-        loop._active_tasks.pop("test:c1")
-        loop._track_active_task("test:c1", new_task)
-
-        old_release.set()
-        await old_task
-        await asyncio.sleep(0)
-
-        assert loop._active_tasks["test:c1"] == {new_task}
-
-        new_release.set()
-        await new_task
-        await asyncio.sleep(0)
-
-        assert "test:c1" not in loop._active_tasks
-
-
 class TestHandleStop:
     @pytest.mark.asyncio
     async def test_stop_no_active_task(self):

tests/agent/tools/test_subagent_tools.py

@@ -7,7 +7,6 @@ from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 
-from nanobot.agent.context import TranscriptInput
 from nanobot.agent.tools.context import RequestContext
 from nanobot.config.schema import AgentDefaults
 from nanobot.providers.base import GenerationSettings
@@ -569,10 +568,7 @@ async def test_agent_loop_syncs_updated_max_iterations_before_run(tmp_path):
     loop.runner.run = AsyncMock(side_effect=fake_run)
     loop.max_iterations = 55
 
-    await loop._run_agent_loop(
+    await loop._run_agent_loop([], runtime=loop.llm_runtime())
-        TranscriptInput(history=[], current_message=None),
-        runtime=loop.llm_runtime(),
-    )
 
     loop.runner.run.assert_awaited_once()
 
@@ -613,7 +609,7 @@ async def test_drain_pending_no_block_when_no_subagents(tmp_path):
 
     runtime = loop.llm_runtime()
     await loop._run_agent_loop(
-        TranscriptInput(history=[{"role": "user", "content": "test"}], current_message=None),
+        [{"role": "user", "content": "test"}],
         runtime=runtime,
         session=None,
         request_context=RequestContext(channel="test", chat_id="c1", runtime=runtime),
@@ -672,7 +668,7 @@ async def test_terminal_drain_timeout(tmp_path):
 
     runtime = loop.llm_runtime()
     await loop._run_agent_loop(
-        TranscriptInput(history=[{"role": "user", "content": "test"}], current_message=None),
+        [{"role": "user", "content": "test"}],
         runtime=runtime,
         session=session,
         request_context=RequestContext(
@@ -746,7 +742,7 @@ async def test_terminal_drain_reuses_one_timeout_budget(tmp_path):
     loop.subagents._running_tasks["sub-deadline-1"] = hang_task
 
     await loop._run_agent_loop(
-        TranscriptInput(history=[{"role": "user", "content": "test"}], current_message=None),
+        [{"role": "user", "content": "test"}],
         runtime=loop.llm_runtime(),
         session=session,
         pending_queue=pending_queue,

tests/channels/test_websocket_listener_health.py

@@ -1,117 +0,0 @@
-"""Regression tests for WebSocket listener health probing portability."""
-
-from __future__ import annotations
-
-import errno
-import socket
-from typing import Any
-
-import pytest
-
-from nanobot.channels.websocket.runtime import WebSocketChannel
-
-
-class _StubSocket:
-    """Minimal socket stand-in: real sockets forbid attribute patching."""
-
-    def __init__(self, *, fileno: int, error: OSError | None = None, value: int = 1):
-        self._fileno = fileno
-        self._error = error
-        self._value = value
-
-    def fileno(self) -> int:
-        return self._fileno
-
-    def getsockopt(self, *_args: Any, **_kwargs: Any) -> int:
-        if self._error is not None:
-            raise self._error
-        return self._value
-
-
-class _StubServer:
-    """Minimal server stand-in for the production listener-health boundary."""
-
-    def __init__(self, sock: _StubSocket, *, serving: bool = True):
-        self._sock = sock
-        self._serving = serving
-
-    @property
-    def sockets(self) -> tuple[_StubSocket, ...]:
-        return (self._sock,)
-
-    def is_serving(self) -> bool:
-        return self._serving
-
-
-@pytest.fixture
-def listening_socket() -> socket.socket:
-    sock = socket.socket()
-    sock.bind(("127.0.0.1", 0))
-    sock.listen(1)
-    yield sock
-    sock.close()
-
-
-def test_real_listening_socket_is_accepting(listening_socket: socket.socket) -> None:
-    """A genuinely listening socket must never be reported as degraded.
-
-    On macOS/BSD this exercises the ``ENOPROTOOPT`` fallback path; on Linux it
-    exercises the native ``SO_ACCEPTCONN`` path. Both must agree.
-    """
-    assert WebSocketChannel._socket_is_accepting(listening_socket) is True
-
-
-def test_closed_socket_is_not_accepting() -> None:
-    sock = socket.socket()
-    sock.bind(("127.0.0.1", 0))
-    sock.listen(1)
-    sock.close()
-
-    assert WebSocketChannel._socket_is_accepting(sock) is False
-
-
-@pytest.mark.parametrize(
-    "unsupported_errno",
-    [errno.ENOPROTOOPT, errno.EOPNOTSUPP],
-)
-def test_unsupported_sockopt_falls_back_to_fd_liveness(unsupported_errno: int) -> None:
-    """macOS/BSD reject ``SO_ACCEPTCONN`` even on healthy listeners.
-
-    Treating that rejection as "not serving" made the listener look permanently
-    degraded, so the channel retried forever and never became ready.
-    """
-    sock = _StubSocket(fileno=3, error=OSError(unsupported_errno, "Protocol not available"))
-
-    assert WebSocketChannel._socket_is_accepting(sock) is True
-
-
-def test_listener_health_uses_unsupported_sockopt_fallback() -> None:
-    """The fallback must be wired into the health check that controls readiness."""
-    sock = _StubSocket(fileno=3, error=OSError(errno.ENOPROTOOPT, "Protocol not available"))
-    server: Any = _StubServer(sock)
-
-    assert WebSocketChannel._listener_is_serving(server) is True
-
-
-def test_unexpected_oserror_propagates() -> None:
-    sock = _StubSocket(fileno=3, error=OSError(errno.EBADF, "Bad file descriptor"))
-
-    with pytest.raises(OSError) as excinfo:
-        WebSocketChannel._socket_is_accepting(sock)
-
-    assert excinfo.value.errno == errno.EBADF
-
-
-def test_listener_health_rejects_invalid_socket_state() -> None:
-    """``EINVAL`` can mean that a live socket isn't actually listening."""
-    sock = _StubSocket(fileno=3, error=OSError(errno.EINVAL, "Invalid argument"))
-    server: Any = _StubServer(sock)
-
-    assert WebSocketChannel._listener_is_serving(server) is False
-
-
-def test_unsupported_sockopt_still_rejects_dead_fd() -> None:
-    """The portability fallback must not mask an already-closed listener."""
-    sock = _StubSocket(fileno=-1, error=OSError(errno.ENOPROTOOPT, "Protocol not available"))
-
-    assert WebSocketChannel._socket_is_accepting(sock) is False

tests/cli/test_restart_command.py

@@ -11,7 +11,6 @@ from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 
-from nanobot.agent.context import TranscriptInput
 from nanobot.bus.events import InboundMessage
 from nanobot.providers.base import LLMResponse, LLMUsage
 
@@ -312,16 +311,10 @@ class TestRestartCommand:
             LLMResponse(content="second", usage=None),
         ])
 
-        first = await loop._run_agent_loop(
+        first = await loop._run_agent_loop([], runtime=loop.llm_runtime())
-            TranscriptInput(history=[], current_message=None),
-            runtime=loop.llm_runtime(),
-        )
         assert first.usage == LLMUsage.reported(input_tokens=9, output_tokens=4)
 
-        second = await loop._run_agent_loop(
+        second = await loop._run_agent_loop([], runtime=loop.llm_runtime())
-            TranscriptInput(history=[], current_message=None),
-            runtime=loop.llm_runtime(),
-        )
         assert second.usage == LLMUsage.estimated(input_tokens=123, output_tokens=7)
 
     @pytest.mark.asyncio

tests/cron/test_cron_service.py

@@ -7,7 +7,6 @@ import pytest
 
 from nanobot.cron.service import CronJobSkippedError, CronService
 from nanobot.cron.types import CronJob, CronPayload, CronSchedule
-from nanobot.runtime_context import RUNTIME_CONTEXT_INPUT_META
 
 
 async def _wait_until(predicate, *, timeout: float = 1.0, interval: float = 0.01) -> None:
@@ -293,12 +292,7 @@ def test_load_store_migrates_legacy_delivery_context(tmp_path) -> None:
                             "deliver": True,
                             "channel": "telegram",
                             "to": "user-1",
-                            "channelMeta": {
+                            "channelMeta": {"message_thread_id": 42},
-                                "message_thread_id": 42,
-                                RUNTIME_CONTEXT_INPUT_META: [
-                                    {"source": "webui_quote", "content": "stale quote"}
-                                ],
-                            },
                             "sessionKey": "telegram:user-1:topic:42",
                         },
                         "state": {},
@@ -417,39 +411,6 @@ def test_add_job_preserves_origin_delivery_context(tmp_path) -> None:
     assert reloaded.payload.origin_metadata == metadata
 
 
-@pytest.mark.asyncio
-async def test_start_heals_runtime_context_from_pending_external_add(tmp_path) -> None:
-    """Flattened runtime blocks from older action files must not be replayed."""
-    store_path = tmp_path / "cron" / "jobs.json"
-    external = CronService(store_path)
-    job = external.add_job(
-        name="quoted reminder",
-        schedule=CronSchedule(kind="every", every_ms=60_000),
-        message="remember this",
-        origin_metadata={"webui": True},
-        **_bound_chat("quoted"),
-    )
-
-    action_path = tmp_path / "cron" / "action.jsonl"
-    action = json.loads(action_path.read_text(encoding="utf-8"))
-    action["params"]["payload"]["origin_metadata"][RUNTIME_CONTEXT_INPUT_META] = [
-        {"source": "webui_quote", "content": "quoted reply"}
-    ]
-    action_path.write_text(json.dumps(action), encoding="utf-8")
-
-    owner = CronService(store_path)
-    await owner.start()
-    try:
-        loaded = owner.get_job(job.id)
-        assert loaded is not None
-        assert loaded.payload.origin_metadata == {"webui": True}
-
-        raw = json.loads(store_path.read_text(encoding="utf-8"))
-        assert raw["jobs"][0]["payload"]["originMetadata"] == {"webui": True}
-    finally:
-        owner.stop()
-
-
 @pytest.mark.asyncio
 async def test_channel_meta_and_session_key_survive_store_reload(tmp_path) -> None:
     store_path = tmp_path / "cron" / "jobs.json"

tests/providers/test_conversation_state.py

@@ -146,51 +146,6 @@ def test_controller_uses_governed_messages_for_provider_state_delta() -> None:
     assert governed_checkpoint.pending_messages[-1]["content"] == "compacted result"
 
 
-def test_controller_estimates_active_state_plus_pending_delta(monkeypatch) -> None:
-    provider = _provider()
-    current_message = {"role": "user", "content": "new delta"}
-    state = ProviderConversationState(
-        kind="openai_responses",
-        provider="openai:test",
-        model="gpt-5.6",
-        version=1,
-        payload={
-            "items": [{"type": "reasoning", "encrypted_content": "opaque"}],
-            "context_tokens": 450,
-        },
-        pending_messages=[current_message],
-    )
-    controller = ProviderConversationStateController(
-        provider=provider,
-        model="gpt-5.6",
-        messages=[current_message],
-        state=state,
-    )
-    seen = {}
-
-    def estimate(_provider, _model, messages, tools):
-        seen["messages"] = messages
-        seen["tools"] = tools
-        return 100, "test-counter"
-
-    monkeypatch.setattr(
-        "nanobot.providers.conversation_state.estimate_prompt_tokens_chain",
-        estimate,
-    )
-
-    tokens = controller.estimate_request_context_tokens(
-        [current_message],
-        model_messages=[current_message],
-        tool_definitions=[{"type": "web_search"}],
-    )
-
-    assert tokens == 550
-    assert seen == {
-        "messages": [current_message],
-        "tools": [{"type": "web_search"}],
-    }
-
-
 def test_transient_response_preserves_only_durable_request_messages() -> None:
     provider = _provider()
     current_message = {"role": "user", "content": "continue"}

tests/providers/test_enforce_role_alternation.py

@@ -112,49 +112,14 @@ class TestEnforceRoleAlternation:
         assert result[1]["content"] is None
         assert result[2]["role"] == "tool"
 
-    def test_consecutive_user_messages_preserve_text_before_multimodal_content(self):
+    def test_non_string_content_uses_latest(self):
-        image = {
-            "type": "image_url",
-            "image_url": {"url": "data:image/png;base64,aW1hZ2U="},
-        }
         msgs = [
-            {"role": "user", "content": "Earlier unanswered question"},
+            {"role": "user", "content": [{"type": "text", "text": "A"}]},
-            {
+            {"role": "user", "content": "B"},
-                "role": "user",
-                "content": [image, {"type": "text", "text": "The error is here"}],
-            },
         ]
         result = LLMProvider._enforce_role_alternation(msgs)
-        assert result == [{
+        assert len(result) == 1
-            "role": "user",
+        assert result[0]["content"] == "B"
-            "content": [
-                {"type": "text", "text": "Earlier unanswered question"},
-                image,
-                {"type": "text", "text": "The error is here"},
-            ],
-        }]
-
-    def test_consecutive_user_messages_preserve_multimodal_content_before_text(self):
-        image = {
-            "type": "image_url",
-            "image_url": {"url": "data:image/png;base64,aW1hZ2U="},
-        }
-        msgs = [
-            {
-                "role": "user",
-                "content": [image, {"type": "text", "text": "First question"}],
-            },
-            {"role": "user", "content": "Follow-up detail"},
-        ]
-        result = LLMProvider._enforce_role_alternation(msgs)
-        assert result == [{
-            "role": "user",
-            "content": [
-                image,
-                {"type": "text", "text": "First question"},
-                {"type": "text", "text": "Follow-up detail"},
-            ],
-        }]
 
     def test_original_messages_not_mutated(self):
         msgs = [

tests/session/test_turn_continuation.py

@@ -141,13 +141,14 @@ def test_internal_continuation_requires_budget_boundary_and_queue():
     )
 
 
-def test_save_skip_matches_prefix_when_current_message_was_persisted():
+def test_save_skip_matches_prefix_when_current_message_merged():
     skip = _save_skip_for_turn(
         message_metadata=None,
-        initial_message_count=3,  # [system, history user, current user]
+        initial_message_count=2,  # [system, merged user]
+        history_count=1,
         input_persisted_early=True,
     )
-    assert skip == 3
+    assert skip == 2
 
 
 def test_save_skip_unchanged_for_standalone_current_message():
@@ -155,10 +156,12 @@ def test_save_skip_unchanged_for_standalone_current_message():
     assert _save_skip_for_turn(
         message_metadata=None,
         initial_message_count=3,
+        history_count=1,
         input_persisted_early=True,
     ) == 3
     assert _save_skip_for_turn(
         message_metadata=None,
         initial_message_count=3,
+        history_count=1,
         input_persisted_early=False,
     ) == 2

tests/test_tool_contextvars.py

@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 import asyncio
-import json
 from unittest.mock import MagicMock
 
 import pytest
@@ -12,7 +11,6 @@ from nanobot.agent.tools.message import MessageTool
 from nanobot.agent.tools.spawn import SpawnTool
 from nanobot.cron.service import CronService
 from nanobot.providers.base import GenerationSettings, LLMProvider
-from nanobot.runtime_context import RUNTIME_CONTEXT_INPUT_META, RuntimeContextBlock
 from nanobot.session.keys import UNIFIED_SESSION_KEY
 from nanobot.utils.llm_runtime import LLMRuntime
 
@@ -301,41 +299,6 @@ async def test_webui_cron_tool_uses_origin_session_when_unified_enabled(tmp_path
     assert jobs[0].payload.origin_metadata == {"webui": True}
 
 
-@pytest.mark.asyncio
-async def test_cron_tool_snapshots_only_persistable_request_metadata(tmp_path) -> None:
-    """Live runtime context must not poison a persisted WebUI cron job."""
-    store_path = tmp_path / "jobs.json"
-    service = CronService(store_path)
-    tool = CronTool(service)
-    await service.start()
-    try:
-        with request_context(
-            RequestContext(
-                channel="websocket",
-                chat_id="chat-123",
-                metadata={
-                    "webui": True,
-                    RUNTIME_CONTEXT_INPUT_META: [
-                        RuntimeContextBlock(source="webui_quote", content="quoted reply")
-                    ],
-                    "opaque": object(),
-                },
-                session_key=UNIFIED_SESSION_KEY,
-            )
-        ):
-            result = await tool.execute(action="add", message="standup", every_seconds=300)
-
-        assert result.startswith("Created job")
-        jobs = service.list_jobs()
-        assert len(jobs) == 1
-        assert jobs[0].payload.origin_metadata == {"webui": True}
-
-        raw = json.loads(store_path.read_text(encoding="utf-8"))
-        assert raw["jobs"][0]["payload"]["originMetadata"] == {"webui": True}
-    finally:
-        service.stop()
-
-
 @pytest.mark.asyncio
 async def test_cron_tool_preserves_thread_scoped_session_key(tmp_path) -> None:
     """Channel-provided thread session keys should remain the cron owner."""

tests/tools/test_message_tool_suppress.py

@@ -6,7 +6,6 @@ from unittest.mock import AsyncMock, MagicMock
 
 import pytest
 
-from nanobot.agent.context import TranscriptInput
 from nanobot.agent.loop import AgentLoop
 from nanobot.agent.tools.message import MessageTool
 from nanobot.bus.events import InboundMessage, OutboundMessage
@@ -179,9 +178,7 @@ class TestMessageToolSuppressLogic:
             progress.append((content, tool_hint))
 
         result = await loop._run_agent_loop(
-            TranscriptInput(history=[], current_message=None),
+            [], runtime=loop.llm_runtime(), on_progress=on_progress
-            runtime=loop.llm_runtime(),
-            on_progress=on_progress,
         )
 
         assert result.final_content == "Done"

tests/tools/test_session_messages_tool.py

@@ -183,66 +183,6 @@ async def test_rate_limit_is_per_source_session_and_uses_a_rolling_minute(
     )
 
 
-@pytest.mark.asyncio
-async def test_rate_limit_releases_expired_source_state_and_keeps_recent_sources(
-    tmp_path: Path,
-) -> None:
-    sessions = SessionManager(tmp_path)
-    _persist(
-        sessions,
-        "websocket:a",
-        "websocket:b",
-        "websocket:c",
-        "websocket:target",
-    )
-    now = 0.0
-    tool = SendSessionMessageTool(
-        sessions=sessions,
-        bus=MessageBus(),
-        max_messages_per_minute=2,
-        clock=lambda: now,
-    )
-    target = _handle(sessions, "websocket:target").name
-
-    for source in ("websocket:a", "websocket:b"):
-        await tool.enqueue(
-            source_session_key=source,
-            target_handle=target,
-            content="initial",
-            expect_reply=False,
-        )
-    now = 30.0
-    await tool.enqueue(
-        source_session_key="websocket:a",
-        target_handle=target,
-        content="recent",
-        expect_reply=False,
-    )
-
-    now = 61.0
-    await tool.enqueue(
-        source_session_key="websocket:c",
-        target_handle=target,
-        content="trigger cleanup",
-        expect_reply=False,
-    )
-
-    assert set(tool._sent_at) == {"websocket:a", "websocket:c"}
-    await tool.enqueue(
-        source_session_key="websocket:a",
-        target_handle=target,
-        content="within rolling window",
-        expect_reply=False,
-    )
-    with pytest.raises(SessionMessageError, match="rate limit"):
-        await tool.enqueue(
-            source_session_key="websocket:a",
-            target_handle=target,
-            content="over limit",
-            expect_reply=False,
-        )
-
-
 @pytest.mark.asyncio
 async def test_reply_timeout_injects_a_user_input_back_into_the_source(
     tmp_path: Path,

tests/tools/test_tool_descriptions.py

@@ -9,9 +9,7 @@ from nanobot.agent.tools.shell import ExecTool
 
 def test_coding_tool_descriptions_steer_editing_priority() -> None:
     apply_patch = ApplyPatchTool().description.lower()
-    edit_tool = EditFileTool()
+    edit_file = EditFileTool().description.lower()
-    edit_file = edit_tool.description.lower()
-    edit_parameters = edit_tool.parameters["properties"]
     write_file = WriteFileTool().description.lower()
 
     assert "default tool for code edits" in apply_patch
@@ -20,10 +18,8 @@ def test_coding_tool_descriptions_steer_editing_priority() -> None:
     assert "edit_file only for small exact replacements" in apply_patch
 
     assert "small, exact replacement" in edit_file
+    assert "copied from read_file" in edit_file
     assert "prefer apply_patch" in edit_file
-    assert "occurrence, line_hint, and replace_all=true are mutually exclusive" in edit_file
-    assert "copy it from read_file" in edit_parameters["old_text"]["description"].lower()
-    assert "must differ from old_text" in edit_parameters["new_text"]["description"].lower()
 
     assert "replace an entire file" in write_file
     assert "prefer apply_patch" in write_file

tui/src/app.test.ts

@@ -205,7 +205,7 @@ describe("NanobotTui layout", () => {
       expect(occurrences(frame, "Ask nanobot anything")).toBe(1)
       expect(occurrences(frame, "Ready")).toBe(0)
       expect(occurrences(frame, "Getting ready…")).toBe(1)
-      expect(occurrences(frame, "default ▾")).toBe(1)
+      expect(occurrences(frame, "nanobot  ·  test/model")).toBe(1)
     }
 
     app.accept({ event: "attached", chat_id: "chat" })
@@ -253,23 +253,6 @@ describe("NanobotTui layout", () => {
     expect(sent).toEqual(["你好"])
   })
 
-  test("keeps input typed immediately after Enter in the next draft", async () => {
-    const sent: string[] = []
-    setup = await createRenderer({ width: 72, height: 20, screenMode: "alternate-screen" })
-    const app = mount(setup, sent)
-    app.accept({ event: "attached", chat_id: "chat" })
-    await Bun.sleep(1)
-    const composer = (app as unknown as { composer: TextareaRenderable }).composer
-
-    composer.setText("first")
-    setup.mockInput.pressEnter()
-    for (const key of "next") setup.mockInput.pressKey(key)
-    await waitUntil(() => sent.length > 0)
-
-    expect(sent).toEqual(["first"])
-    expect(composer.plainText).toBe("next")
-  })
-
   test("inserts newlines with Shift+Enter and the universal Ctrl+J fallback", async () => {
     const sent: string[] = []
     setup = await createRenderer({
@@ -1002,6 +985,7 @@ describe("NanobotTui layout", () => {
     const ui = app as unknown as {
       composer: TextareaRenderable
       sessionMenu: { visible: boolean }
+      titleText: { plainText: string }
       runtimeControls: { modelText: { plainText: string } }
     }
 
@@ -1015,7 +999,8 @@ describe("NanobotTui layout", () => {
       ui.composer.submit()
       await waitUntil(() => attached.length === 1)
       expect(attached).toEqual(["other"])
-      expect(ui.runtimeControls.modelText.plainText).toBe("Deep Research ▾")
+      expect(ui.titleText.plainText).toContain("Release checklist")
+      expect(ui.runtimeControls.modelText.plainText).toContain("Deep Research")
       expect(ui.runtimeControls.modelText.plainText).not.toContain("test/model")
 
       app.accept({ event: "attached", chat_id: "other" })
@@ -1024,7 +1009,8 @@ describe("NanobotTui layout", () => {
       ui.composer.submit()
       await waitUntil(() => newChats.length === 1)
       expect(newChats).toEqual(["new"])
-      expect(ui.runtimeControls.modelText.plainText).toBe("default ▾")
+      expect(ui.titleText.plainText).toContain("New chat")
+      expect(ui.runtimeControls.modelText.plainText).toContain("test/model")
     } finally {
       globalThis.fetch = original
     }
@@ -1196,8 +1182,7 @@ describe("NanobotTui layout", () => {
       model_preset: "Codex",
     })
     await setup.flush()
-    expect(ui.runtimeControls.modelText.plainText).toBe("Codex ▾")
+    expect(ui.runtimeControls.modelText.plainText).toContain("Codex  ·  openai/gpt-5.6")
-    expect(ui.runtimeControls.modelText.plainText).not.toContain("openai/gpt-5.6")
 
     app.accept({
       event: "runtime_model_updated",
@@ -1205,7 +1190,7 @@ describe("NanobotTui layout", () => {
       model_preset: "DeepSeek",
     })
     await setup.flush()
-    expect(ui.runtimeControls.modelText.plainText).toBe("Codex ▾")
+    expect(ui.runtimeControls.modelText.plainText).toContain("Codex  ·  openai/gpt-5.6")
     expect(ui.runtimeControls.modelText.plainText).not.toContain("DeepSeek")
   })
 
@@ -1227,8 +1212,8 @@ describe("NanobotTui layout", () => {
     })
     await setup.flush()
 
-    expect(ui.runtimeControls.modelText.plainText).toBe("default ▾")
+    expect(ui.runtimeControls.modelText.plainText).toContain("deepseek/deepseek-chat")
-    expect(ui.runtimeControls.modelText.plainText).not.toContain("deepseek/deepseek-chat")
+    expect(ui.runtimeControls.modelText.plainText).not.toContain("Codex")
   })
 
   test("refreshes the canonical preset after the model command completes", async () => {
@@ -1324,6 +1309,7 @@ describe("NanobotTui layout", () => {
         menuRoot: { getChildren(): unknown[] }
       }
       composer: TextareaRenderable
+      titleText: TextRenderable
       status: TextRenderable
       meta: TextRenderable
     }
@@ -1338,6 +1324,7 @@ describe("NanobotTui layout", () => {
       expect(ui.runtimeControls.modelText.selectable).toBe(false)
       expect(ui.runtimeControls.accessText.selectable).toBe(false)
       expect(ui.runtimeControls.contextText.selectable).toBe(false)
+      expect(ui.titleText.selectable).toBe(false)
       expect(ui.status.selectable).toBe(false)
       expect(ui.meta.selectable).toBe(false)
       app.accept({ event: "goal_status", chat_id: "chat", status: "running" })
@@ -1407,7 +1394,7 @@ describe("NanobotTui layout", () => {
     }
   })
 
-  test("switches sessions only through the sessions command", async () => {
+  test("opens and switches sessions from the clickable title", async () => {
     const original = globalThis.fetch
     globalThis.fetch = ((input: string | URL | Request) => {
       const url = String(input)
@@ -1433,18 +1420,14 @@ describe("NanobotTui layout", () => {
     const ui = app as unknown as {
       composer: TextareaRenderable
       sessionMenu: { visible: boolean; root: { getChildren(): unknown[] } }
-      title: { getChildren(): unknown[] }
+      titleText: TextRenderable
+      status: TextRenderable
     }
 
     try {
       await waitUntil(() => (app as unknown as { ready: boolean }).ready)
       await setup.renderOnce()
-      const titleItems = ui.title.getChildren() as TextRenderable[]
+      await setup.mockMouse.click(ui.titleText.x + 2, ui.titleText.y)
-      expect(titleItems.some((item) => item.id === "nanobot-tui-title-text")).toBe(false)
-      expect(ui.sessionMenu.visible).toBe(false)
-
-      ui.composer.setText("/sessions")
-      ui.composer.submit()
       await waitUntil(() => ui.sessionMenu.visible)
       await setup.flush()
       expect(ui.composer.placeholder).toBe("Search sessions")
@@ -1458,6 +1441,15 @@ describe("NanobotTui layout", () => {
       expect(attached).toEqual(["other"])
       expect(ui.sessionMenu.visible).toBe(false)
       expect(ui.composer.focused).toBe(true)
+      expect(ui.titleText.plainText).toContain("Release checklist")
+
+      app.accept({ event: "attached", chat_id: "other" })
+      await setup.mockMouse.click(ui.titleText.x + 2, ui.titleText.y)
+      await waitUntil(() => ui.sessionMenu.visible)
+      ui.composer.blur()
+      await setup.mockMouse.click(ui.status.x, ui.status.y)
+      expect(ui.sessionMenu.visible).toBe(false)
+      expect(ui.composer.focused).toBe(true)
     } finally {
       globalThis.fetch = original
     }
@@ -1723,7 +1715,7 @@ describe("NanobotTui layout", () => {
       await setup.flush()
       const frame = setup.captureCharFrame()
       expect(frame).toContain("Release checklist")
-      expect(occurrences(frame, "Current chat")).toBe(0)
+      expect(occurrences(frame, "Current chat")).toBe(1)
     } finally {
       globalThis.fetch = original
     }
@@ -1967,7 +1959,7 @@ describe("NanobotTui layout", () => {
       } else if (width >= 28 && height >= 9) {
         expect(occurrences(frame, "Enter now · Tab next")).toBe(1)
       }
-      expect(occurrences(frame, "default ▾")).toBe(height >= 14 ? 1 : 0)
+      expect(occurrences(frame, "nanobot  ·  test/model")).toBe(height >= 14 ? 1 : 0)
     }
   })
 
@@ -3130,7 +3122,7 @@ describe("NanobotTui with a Herdr pane title reporter", () => {
     await setup.flush()
     const activeFrame = setup.captureCharFrame()
     expect(activeFrame).toContain(">_  nanobot")
-    expect(activeFrame).toContain("default ▾")
+    expect(activeFrame).toContain("test/model")
     expect(occurrences(activeFrame, "› Ship the Herdr integration")).toBe(1)
     expect(occurrences(activeFrame, "app.ts")).toBe(1)
     expect(ui.composer.placeholder).toBe("Enter send now · Tab send next")

tui/src/app.ts

@@ -94,7 +94,7 @@ import {
   type FooterMode,
   type FooterHintTheme,
 } from "./footer-hints"
-import { configureOpenTuiEnvironment, createTuiHost, type TuiHost } from "./host"
+import { createTuiHost, type TuiHost } from "./host"
 
 interface AppOptions {
   wsUrl?: string
@@ -442,6 +442,7 @@ export class NanobotTui {
   private readonly client: ChatClient
   private readonly shell: BoxRenderable
   private readonly title: BoxRenderable
+  private readonly titleText: TextRenderable
   private readonly composerFrame: BoxRenderable
   private readonly composer: TextareaRenderable
   private composerSyntax: SyntaxStyle
@@ -648,6 +649,28 @@ export class NanobotTui {
       alignItems: "center",
       backgroundColor: RGBA.defaultBackground(),
     })
+    this.titleText = new TextRenderable(renderer, {
+      id: "nanobot-tui-title-text",
+      content: "nanobot",
+      height: 1,
+      flexShrink: 0,
+      truncate: true,
+      fg: this.palette.muted,
+      selectable: false,
+      onMouseOver: () => { this.titleText.fg = this.palette.accent },
+      onMouseOut: () => this.renderTitleColor(),
+      onMouseDown: (event) => {
+        if (event.button !== 0) return
+        event.preventDefault()
+        event.stopPropagation()
+        this.renderer.clearSelection()
+        if (this.sessionLoading || this.sessionMenu.visible) {
+          this.closeSessions()
+          return
+        }
+        void this.openSessions()
+      },
+    })
     this.runtimeControls = new RuntimeControls(
       renderer,
       runtimeControlsTheme(this.palette),
@@ -680,6 +703,7 @@ export class NanobotTui {
         },
       },
     )
+    this.title.add(this.titleText)
     this.title.add(this.runtimeControls.modelText)
     this.title.add(this.runtimeControls.accessText)
     this.title.add(this.runtimeControls.contextText)
@@ -733,10 +757,7 @@ export class NanobotTui {
       // IMEs may commit their final composed glyph after Enter. Matching the
       // OpenCode/OpenTUI integration, defer twice before reading plainText.
       onSubmit: () => this.deferSubmit(),
-      onPaste: (event) => {
+      onPaste: (event) => this.handlePaste(event),
-        this.flushSubmit()
-        if (!this.composer.isDestroyed) this.handlePaste(event)
-      },
     })
     this.status = new TextRenderable(renderer, {
       id: "nanobot-tui-status",
@@ -799,7 +820,6 @@ export class NanobotTui {
   }
 
   static async create(options: AppOptions): Promise<NanobotTui> {
-    configureOpenTuiEnvironment()
     const host = createTuiHost()
     const renderer = await createCliRenderer({
       targetFps: 30,
@@ -858,15 +878,12 @@ export class NanobotTui {
     if (this.submitPending) return
     this.submitPending = true
     const generation = ++this.submitGeneration
-    setTimeout(() => setTimeout(() => this.flushSubmit(generation), 0), 0)
+    setTimeout(() => setTimeout(() => {
-  }
+      if (generation !== this.submitGeneration) return
-
-  private flushSubmit(generation = this.submitGeneration): void {
-    if (!this.submitPending || generation !== this.submitGeneration) return
       this.submitPending = false
-    this.submitGeneration += 1
       if (this.composer.isDestroyed) return
       this.submit()
+    }, 0), 0)
   }
 
   private submit(): void {
@@ -1581,15 +1598,6 @@ export class NanobotTui {
   }
 
   private handleKey = (key: KeyEvent): void => {
-    // The app receives keypresses before the focused Textarea. Seal the pending
-    // submission first so this key is inserted into the next draft.
-    if (this.submitPending) {
-      this.flushSubmit()
-      if (this.quitting || this.composer.isDestroyed) {
-        key.preventDefault()
-        return
-      }
-    }
     if (this.diffViewer.visible) {
       if (key.ctrl && key.name === "c") {
         const selected = this.renderer.getSelection()?.getSelectedText()
@@ -1868,6 +1876,7 @@ export class NanobotTui {
     this.composer.syntaxStyle = this.composerSyntax
     this.syncComposerImageHighlights(this.composer.plainText)
     void this.renderer.idle().catch(() => {}).finally(() => previousComposerSyntax.destroy())
+    this.renderTitleColor()
     this.status.fg = this.palette.muted
     this.meta.fg = this.palette.faint
     this.updateMeta()
@@ -1947,15 +1956,24 @@ export class NanobotTui {
   }
 
   private updateTitle(): void {
+    const identity = this.sessionTitle.trim() || "nanobot"
+    this.titleText.maxWidth = Math.max(8, Math.floor(this.renderer.width * 0.38))
+    this.titleText.content = identity
     const context = this.contextTokens === null
       ? ""
-      : `     ~${formatTokenCount(this.contextTokens)}${this.contextWindowTokens
+      : `  ·  ~${formatTokenCount(this.contextTokens)}${this.contextWindowTokens
         ? `/${formatTokenCount(this.contextWindowTokens)}`
         : ""} ctx`
     this.runtimeControls.updateModel(this.modelName, this.modelPreset)
     this.runtimeControls.updateContext(context)
   }
 
+  private renderTitleColor(): void {
+    this.titleText.fg = this.sessionLoading || this.sessionMenu.visible
+      ? this.palette.accent
+      : this.palette.muted
+  }
+
   private resizeComposer(): void {
     const verticalPadding = this.renderer.height >= 12 ? 1 : 0
     const maxContentHeight = Math.max(1, Math.min(12, Math.floor(this.renderer.height / 3)))
@@ -2366,6 +2384,7 @@ export class NanobotTui {
     this.contextPanel.hide()
     this.clearComposer()
     this.sessionLoading = true
+    this.renderTitleColor()
     const loadId = ++this.sessionLoadId
     this.status.content = "Loading sessions…"
     try {
@@ -2392,6 +2411,7 @@ export class NanobotTui {
         this.defaultModelPreset,
       )
       this.startSessionRefresh()
+      this.renderTitleColor()
       this.sessionMenu.update(this.composer.plainText, limit)
       this.syncComposerPlaceholder()
       this.updateMeta()
@@ -2399,6 +2419,7 @@ export class NanobotTui {
     } catch (error) {
       if (loadId !== this.sessionLoadId) return
       this.sessionLoading = false
+      this.renderTitleColor()
       this.status.content = error instanceof Error ? error.message : String(error)
     }
   }
@@ -2557,6 +2578,7 @@ export class NanobotTui {
     this.sessionLoadId += 1
     this.sessionLoading = false
     this.hideSessionMenu()
+    this.renderTitleColor()
     this.clearComposer()
     this.syncComposerPlaceholder()
     this.composer.focus()

tui/src/host.test.ts

@@ -1,9 +1,6 @@
 import { describe, expect, test } from "bun:test"
 
-import {
+import { createTuiHost } from "./host"
-  configureOpenTuiEnvironment,
-  createTuiHost,
-} from "./host"
 
 async function settle(): Promise<void> {
   await Bun.sleep(0)
@@ -11,27 +8,6 @@ async function settle(): Promise<void> {
 }
 
 describe("TUI host integration", () => {
-  test("disables the explicit-width probe on Windows", () => {
-    const environment: Record<string, string | undefined> = {}
-
-    configureOpenTuiEnvironment(environment, "win32")
-
-    expect(environment.OPENTUI_FORCE_EXPLICIT_WIDTH).toBe("false")
-  })
-
-  test("preserves explicit probe choices and leaves other platforms unchanged", () => {
-    const overridden = {
-      OPENTUI_FORCE_EXPLICIT_WIDTH: "true",
-    }
-    const nonWindows: Record<string, string | undefined> = {}
-
-    configureOpenTuiEnvironment(overridden, "win32")
-    configureOpenTuiEnvironment(nonWindows, "linux")
-
-    expect(overridden.OPENTUI_FORCE_EXPLICIT_WIDTH).toBe("true")
-    expect(nonWindows.OPENTUI_FORCE_EXPLICIT_WIDTH).toBeUndefined()
-  })
-
   test("standalone terminals remain a no-op", async () => {
     const commands: string[][] = []
     const host = createTuiHost({}, async (command) => { commands.push([...command]) })

tui/src/host.ts

@@ -8,19 +8,6 @@ type CommandRunner = (command: readonly string[]) => Promise<void>
 
 const METADATA_SOURCE = "nanobot:tui:metadata"
 
-export function configureOpenTuiEnvironment(
-  environment: Environment = process.env,
-  platform = process.platform,
-): void {
-  if (platform !== "win32") return
-
-  // OpenTUI probes OSC 66 support on the main screen before its renderer is
-  // active. Some Windows terminal hosts do not restore the cursor around that
-  // probe, so shutdown resumes in terminal history instead of below the TUI.
-  // Keep an explicit user choice, but use the safe default on Windows.
-  environment.OPENTUI_FORCE_EXPLICIT_WIDTH ??= "false"
-}
-
 class StandaloneHost implements TuiHost {
   reportTitle(): void {}
   release(): void {}

tui/src/runtime-controls.ts

@@ -309,9 +309,12 @@ export class RuntimeControls {
   }
 
   private render(): void {
-    this.modelText.content = `${this.modelPreset} ▾`
+    const runtime = this.modelPreset !== "default"
+      ? [this.modelPreset, this.model].filter(Boolean).join("  ·  ")
+      : this.model
+    this.modelText.content = `  ·  ${runtime} ▾`
     const access = this.scope.access_mode === "full" ? "full access" : "workspace access"
-    this.accessText.content = `     ${access} ▾`
+    this.accessText.content = `  ·  ${access} ▾`
     this.renderColors()
   }
 

tui/src/transcript.ts

@@ -211,7 +211,7 @@ export class Transcript {
     const title = this.createText(`>_  nanobot  v${options.version}`, "text", true)
     const context = this.createText([
       "",
-      `${options.model}     ${options.access}`,
+      `${options.model}  ·  ${options.access}`,
       options.workspace,
     ].join("\n"), "muted")
     row.add(title)

webui/src/components/thread/ModelPresetBadge.tsx

@@ -6,7 +6,7 @@ import {
   type KeyboardEvent,
   type PointerEvent,
 } from "react";
-import { Check, SlidersHorizontal, Sparkles } from "lucide-react";
+import { Check, CircleHelp, SlidersHorizontal, Sparkles } from "lucide-react";
 import { useTranslation } from "react-i18next";
 
 import {
@@ -539,6 +539,7 @@ function PresetPill({
         "composer-model-badge composer-model-pill inline-flex h-full max-w-full min-w-0 shrink-0 items-center rounded-full border border-border/55 bg-card font-medium text-foreground/70",
         "w-fit",
         "transition-[color,background-color,border-color,transform] duration-150 ease-out group-focus-visible:ring-2 group-focus-visible:ring-ring/45",
+        needsSetup && "border-amber-500/35 bg-amber-50/70 text-amber-900 dark:bg-amber-500/10 dark:text-amber-200",
         isHero ? "gap-1.5 px-2.5 text-[12px]" : "gap-2 px-3 text-[12.5px]",
         offset !== undefined && "composer-model-pill-dock",
       )}
@@ -594,13 +595,13 @@ function PresetProviderIcon({
       data-testid={testId}
       className={cn(
         "grid shrink-0 place-items-center",
-        needsSetup && "text-muted-foreground",
+        needsSetup && "text-amber-800 dark:text-amber-200",
         isHero ? "h-4 w-4" : "h-[18px] w-[18px]",
       )}
       aria-hidden
     >
       {needsSetup ? (
-        <Sparkles className={cn(isHero ? "h-3 w-3" : "h-3.5 w-3.5")} strokeWidth={1.8} />
+        <CircleHelp className={cn(isHero ? "h-3 w-3" : "h-3.5 w-3.5")} strokeWidth={1.8} />
       ) : logoUrl ? (
         <img
           src={logoUrl}

webui/src/components/thread/ModelSetupDialog.tsx

@@ -1,123 +0,0 @@
-import { Check, Cloud, KeyRound, Laptop } from "lucide-react";
-import { useTranslation } from "react-i18next";
-
-import {
-  Dialog,
-  DialogContent,
-  DialogDescription,
-  DialogHeader,
-  DialogTitle,
-} from "@/components/ui/dialog";
-import { cn } from "@/lib/utils";
-
-export interface ModelSetupAvailability {
-  account: boolean;
-  apiKey: boolean;
-  local: boolean;
-}
-
-export type ModelSetupIntent = keyof ModelSetupAvailability;
-
-const SETUP_OPTIONS = [
-  {
-    intent: "account",
-    icon: Cloud,
-    titleKey: "thread.composer.modelSetup.account.title",
-    title: "Connect an account",
-    descriptionKey: "thread.composer.modelSetup.account.description",
-    description: "Use a supported AI subscription.",
-  },
-  {
-    intent: "apiKey",
-    icon: KeyRound,
-    titleKey: "thread.composer.modelSetup.apiKey.title",
-    title: "Use an API key",
-    descriptionKey: "thread.composer.modelSetup.apiKey.description",
-    description: "Bring a key from your preferred provider.",
-  },
-  {
-    intent: "local",
-    icon: Laptop,
-    titleKey: "thread.composer.modelSetup.local.title",
-    title: "Run locally",
-    descriptionKey: "thread.composer.modelSetup.local.description",
-    description: "Connect Ollama, LM Studio, or vLLM.",
-  },
-] as const;
-
-export function ModelSetupDialog({
-  availability,
-  open,
-  onOpenChange,
-  onReturnFocus,
-  onSelect,
-}: {
-  availability: ModelSetupAvailability;
-  open: boolean;
-  onOpenChange: (open: boolean) => void;
-  onReturnFocus: () => void;
-  onSelect: (intent: ModelSetupIntent) => void;
-}) {
-  const { t } = useTranslation();
-
-  return (
-    <Dialog open={open} onOpenChange={onOpenChange}>
-      <DialogContent
-        className="max-w-md gap-5 p-5 sm:p-6"
-        onCloseAutoFocus={(event) => {
-          event.preventDefault();
-          onReturnFocus();
-        }}
-      >
-        <DialogHeader className="pr-7">
-          <DialogTitle className="text-[18px] leading-6">
-            {t("thread.composer.modelSetup.title", { defaultValue: "Choose your AI" })}
-          </DialogTitle>
-          <DialogDescription className="leading-5">
-            {t("thread.composer.modelSetup.description", {
-              defaultValue: "Pick a starting point. You can change models at any time.",
-            })}
-          </DialogDescription>
-        </DialogHeader>
-
-        <div className="space-y-2">
-          {SETUP_OPTIONS.map((option) => {
-            const Icon = option.icon;
-            const ready = availability[option.intent];
-            return (
-              <button
-                key={option.intent}
-                type="button"
-                aria-label={t(option.titleKey, { defaultValue: option.title })}
-                onClick={() => onSelect(option.intent)}
-                className={cn(
-                  "group flex min-h-[68px] w-full items-center gap-3 rounded-control border border-border/55 bg-background px-3.5 py-3 text-left",
-                  "transition-[background-color,border-color,transform] duration-150 ease-out hover:border-border hover:bg-muted/45 active:scale-[0.99]",
-                  "focus-visible:outline-none focus-visible:ring-2 focus-visible:ring-ring/45",
-                )}
-              >
-                <span className="grid h-9 w-9 shrink-0 place-items-center rounded-full bg-muted/70 text-foreground/75 transition-colors group-hover:bg-background">
-                  <Icon className="h-[17px] w-[17px]" strokeWidth={1.8} aria-hidden />
-                </span>
-                <span className="min-w-0 flex-1">
-                  <span className="block text-[14px] font-semibold leading-5 text-foreground">
-                    {t(option.titleKey, { defaultValue: option.title })}
-                  </span>
-                  <span className="mt-0.5 block text-[12px] leading-[18px] text-muted-foreground">
-                    {t(option.descriptionKey, { defaultValue: option.description })}
-                  </span>
-                </span>
-                {ready ? (
-                  <span className="inline-flex shrink-0 items-center gap-1 rounded-full bg-emerald-500/10 px-2 py-1 text-[11px] font-medium text-emerald-700 dark:text-emerald-300">
-                    <Check className="h-3 w-3" strokeWidth={2.2} aria-hidden />
-                    {t("thread.composer.modelSetup.ready", { defaultValue: "Ready" })}
-                  </span>
-                ) : null}
-              </button>
-            );
-          })}
-        </div>
-      </DialogContent>
-    </Dialog>
-  );
-}

webui/src/components/thread/ThreadComposer.tsx

@@ -70,10 +70,6 @@ import {
   ModelPresetBadge,
   type ModelPresetOption,
 } from "@/components/thread/ModelPresetBadge";
-import {
-  ModelSetupDialog,
-  type ModelSetupAvailability,
-} from "@/components/thread/ModelSetupDialog";
 import {
   ACCEPT_ATTR,
   MAX_ATTACHMENTS_PER_MESSAGE,
@@ -302,7 +298,6 @@ interface ThreadComposerProps {
   modelProvider?: string | null;
   modelProviderLabel?: string | null;
   modelNeedsSetup?: boolean;
-  modelSetupAvailability?: ModelSetupAvailability;
   fallbackModelName?: string | null;
   onModelBadgeClick?: () => void;
   onManageModels?: () => void;
@@ -1002,7 +997,6 @@ export function ThreadComposer({
   modelProvider = null,
   modelProviderLabel = null,
   modelNeedsSetup = false,
-  modelSetupAvailability = { account: false, apiKey: false, local: false },
   fallbackModelName = null,
   onModelBadgeClick,
   onManageModels,
@@ -1042,7 +1036,6 @@ export function ThreadComposer({
   } | null>(null);
   const [inlineError, setInlineError] = useState<string | null>(null);
   const [sendPending, setSendPending] = useState(false);
-  const [modelSetupOpen, setModelSetupOpen] = useState(false);
   const interactionDisabled = !!disabled || sendPending;
   const [voiceErrorFading, setVoiceErrorFading] = useState(false);
   const [slashMenuDismissed, setSlashMenuDismissed] = useState(false);
@@ -2015,7 +2008,7 @@ export function ThreadComposer({
 
   const submit = useCallback(() => {
     if (modelNeedsSetup) {
-      setModelSetupOpen(true);
+      onModelBadgeClick?.();
       return;
     }
     if (!canSend) return;
@@ -2123,6 +2116,7 @@ export function ThreadComposer({
     isStreaming,
     maxTextBytes,
     modelNeedsSetup,
+    onModelBadgeClick,
     onSend,
     onStop,
     onQuotedContextChange,
@@ -2133,15 +2127,6 @@ export function ThreadComposer({
     value,
   ]);
 
-  const openModelSetup = useCallback(() => {
-    setModelSetupOpen(true);
-  }, []);
-
-  const continueModelSetup = useCallback(() => {
-    setModelSetupOpen(false);
-    onModelBadgeClick?.();
-  }, [onModelBadgeClick]);
-
   const onKeyDown = (e: ReactKeyboardEvent<HTMLTextAreaElement>) => {
     if (showCliAppMenu) {
       if (e.key === "ArrowDown") {
@@ -2563,7 +2548,7 @@ export function ThreadComposer({
                 needsSetup={modelNeedsSetup}
                 fallbackModelName={fallbackModelName}
                 isHero={isHero}
-                onClick={modelNeedsSetup ? openModelSetup : undefined}
+                onClick={modelNeedsSetup ? onModelBadgeClick : undefined}
               />
             ) : null}
             {!voiceRecorder.isRecording ? <ComposerContextBadge usage={contextUsage} /> : null}
@@ -2622,10 +2607,10 @@ export function ThreadComposer({
                 showStopButton
                   ? t("thread.composer.stop")
                   : modelNeedsSetup
-                    ? t("thread.composer.openModelSetup", { defaultValue: "Open AI setup" })
+                    ? t("thread.composer.configureModel", { defaultValue: "Configure model" })
                     : t("thread.composer.send")
               }
-              onClick={showStopButton ? handleStop : modelNeedsSetup ? openModelSetup : undefined}
+              onClick={showStopButton ? handleStop : modelNeedsSetup ? onModelBadgeClick : undefined}
               className={cn(
                 "thread-composer-action touch-target rounded-full transition-transform",
                 showStopButton
@@ -2671,13 +2656,6 @@ export function ThreadComposer({
           </div>
         ) : null}
       </div>
-      <ModelSetupDialog
-        availability={modelSetupAvailability}
-        open={modelSetupOpen}
-        onOpenChange={setModelSetupOpen}
-        onReturnFocus={() => textareaRef.current?.focus()}
-        onSelect={continueModelSetup}
-      />
     </form>
   );
 }

webui/src/components/thread/ThreadShell.tsx

@@ -14,7 +14,6 @@ import {
   type ComposerContextUsage,
 } from "@/components/thread/ThreadComposer";
 import type { ModelPresetOption } from "@/components/thread/ModelPresetBadge";
-import type { ModelSetupAvailability } from "@/components/thread/ModelSetupDialog";
 import { ThreadHeader } from "@/components/thread/ThreadHeader";
 import { StreamErrorNotice } from "@/components/thread/StreamErrorNotice";
 import { ThreadViewport, type ThreadViewportHandle } from "@/components/thread/ThreadViewport";
@@ -383,22 +382,6 @@ interface ModelBadgeInfo {
   needsSetup: boolean;
 }
 
-const LOCAL_MODEL_PROVIDERS = new Set(["atomic_chat", "lm_studio", "ollama", "vllm"]);
-
-function modelSetupAvailability(settings: SettingsPayload | null): ModelSetupAvailability {
-  const configured = settings?.providers.filter((provider) => provider.configured) ?? [];
-  const isLocal = (provider: SettingsPayload["providers"][number]) => {
-    if (LOCAL_MODEL_PROVIDERS.has(provider.name)) return true;
-    const apiBase = provider.api_base?.trim().toLowerCase() ?? "";
-    return apiBase.includes("localhost") || apiBase.includes("127.0.0.1") || apiBase.includes("[::1]");
-  };
-  return {
-    account: configured.some((provider) => provider.auth_type === "oauth"),
-    apiKey: configured.some((provider) => provider.auth_type !== "oauth" && !isLocal(provider)),
-    local: configured.some(isLocal),
-  };
-}
-
 function modelPresetForBadge(
   settings: SettingsPayload | null,
   scopedPreset: string | null,
@@ -978,9 +961,8 @@ export function ThreadShell({
     [activeModelPreset, modelName, settings],
   );
   const modelBadgeLabel = modelBadge.needsSetup
-    ? t("thread.composer.chooseAI", { defaultValue: "Choose your AI" })
+    ? t("thread.composer.modelNotConfigured", { defaultValue: "Model not configured" })
     : modelBadge.label;
-  const setupAvailability = useMemo(() => modelSetupAvailability(settings), [settings]);
   useEffect(() => {
     if (showHeroComposer && !wasShowingHeroComposerRef.current) {
       setHeroGreetingKey(randomHeroGreetingKey());
@@ -1535,7 +1517,6 @@ export function ThreadShell({
           modelProvider={modelBadge.provider}
           modelProviderLabel={modelBadge.providerLabel}
           modelNeedsSetup={modelBadge.needsSetup}
-          modelSetupAvailability={setupAvailability}
           fallbackModelName={fallbackModelName}
           onModelBadgeClick={modelBadge.needsSetup ? onOpenModelSettings : undefined}
           onManageModels={onOpenModelSettings}
@@ -1585,7 +1566,6 @@ export function ThreadShell({
           modelProvider={modelBadge.provider}
           modelProviderLabel={modelBadge.providerLabel}
           modelNeedsSetup={modelBadge.needsSetup}
-          modelSetupAvailability={setupAvailability}
           fallbackModelName={fallbackModelName}
           onModelBadgeClick={modelBadge.needsSetup ? onOpenModelSettings : undefined}
           onManageModels={onOpenModelSettings}

webui/src/i18n/locales/en/common.json

@@ -1193,25 +1193,8 @@
       "stop": "Stop response",
       "quotedContext": "Quoted context",
       "removeQuotedContext": "Remove quoted context",
-      "openModelSetup": "Open AI setup",
+      "modelNotConfigured": "Model not configured",
-      "chooseAI": "Choose your AI",
+      "configureModel": "Configure model",
-      "modelSetup": {
-        "title": "Choose your AI",
-        "description": "Pick a starting point. You can change models at any time.",
-        "ready": "Ready",
-        "account": {
-          "title": "Connect an account",
-          "description": "Use a supported AI subscription."
-        },
-        "apiKey": {
-          "title": "Use an API key",
-          "description": "Bring a key from your preferred provider."
-        },
-        "local": {
-          "title": "Run locally",
-          "description": "Connect Ollama, LM Studio, or vLLM."
-        }
-      },
       "switchModel": "Switch model for this chat",
       "manageModels": "Manage models",
       "context": {

webui/src/i18n/locales/es/common.json

@@ -1180,25 +1180,8 @@
       "stop": "Detener respuesta",
       "quotedContext": "Contexto citado",
       "removeQuotedContext": "Quitar contexto citado",
-      "openModelSetup": "Abrir configuración de IA",
+      "modelNotConfigured": "Modelo no configurado",
-      "chooseAI": "Elige tu IA",
+      "configureModel": "Configurar modelo",
-      "modelSetup": {
-        "title": "Elige tu IA",
-        "description": "Elige cómo empezar. Puedes cambiar de modelo en cualquier momento.",
-        "ready": "Listo",
-        "account": {
-          "title": "Conectar una cuenta",
-          "description": "Usa una suscripción de IA compatible."
-        },
-        "apiKey": {
-          "title": "Usar una clave API",
-          "description": "Usa una clave de tu proveedor preferido."
-        },
-        "local": {
-          "title": "Ejecutar localmente",
-          "description": "Conecta Ollama, LM Studio o vLLM."
-        }
-      },
       "switchModel": "Cambiar el modelo de este chat",
       "manageModels": "Gestionar modelos",
       "context": {

webui/src/i18n/locales/fr/common.json

@@ -1179,25 +1179,8 @@
       "stop": "Arrêter la réponse",
       "quotedContext": "Contexte cité",
       "removeQuotedContext": "Supprimer le contexte cité",
-      "openModelSetup": "Ouvrir la configuration de l’IA",
+      "modelNotConfigured": "Modèle non configuré",
-      "chooseAI": "Choisissez votre IA",
+      "configureModel": "Configurer le modèle",
-      "modelSetup": {
-        "title": "Choisissez votre IA",
-        "description": "Choisissez un point de départ. Vous pourrez changer de modèle à tout moment.",
-        "ready": "Prêt",
-        "account": {
-          "title": "Connecter un compte",
-          "description": "Utilisez un abonnement IA compatible."
-        },
-        "apiKey": {
-          "title": "Utiliser une clé API",
-          "description": "Utilisez la clé du fournisseur de votre choix."
-        },
-        "local": {
-          "title": "Exécuter localement",
-          "description": "Connectez Ollama, LM Studio ou vLLM."
-        }
-      },
       "switchModel": "Changer le modèle de cette conversation",
       "manageModels": "Gérer les modèles",
       "context": {

webui/src/i18n/locales/id/common.json

@@ -1179,25 +1179,8 @@
       "stop": "Hentikan respons",
       "quotedContext": "Konteks kutipan",
       "removeQuotedContext": "Hapus konteks kutipan",
-      "openModelSetup": "Buka penyiapan AI",
+      "modelNotConfigured": "Model belum dikonfigurasi",
-      "chooseAI": "Pilih AI Anda",
+      "configureModel": "Konfigurasi model",
-      "modelSetup": {
-        "title": "Pilih AI Anda",
-        "description": "Pilih cara memulai. Anda dapat mengganti model kapan saja.",
-        "ready": "Siap",
-        "account": {
-          "title": "Hubungkan akun",
-          "description": "Gunakan langganan AI yang didukung."
-        },
-        "apiKey": {
-          "title": "Gunakan kunci API",
-          "description": "Gunakan kunci dari penyedia pilihan Anda."
-        },
-        "local": {
-          "title": "Jalankan secara lokal",
-          "description": "Hubungkan Ollama, LM Studio, atau vLLM."
-        }
-      },
       "switchModel": "Ganti model untuk percakapan ini",
       "manageModels": "Kelola model",
       "context": {

webui/src/i18n/locales/ja/common.json

@@ -1179,25 +1179,8 @@
       "stop": "応答を停止",
       "quotedContext": "引用したコンテキスト",
       "removeQuotedContext": "引用したコンテキストを削除",
-      "openModelSetup": "AI 設定を開く",
+      "modelNotConfigured": "モデルが未設定です",
-      "chooseAI": "AI を選択",
+      "configureModel": "モデルを設定",
-      "modelSetup": {
-        "title": "AI を選択",
-        "description": "開始方法を選んでください。モデルはいつでも変更できます。",
-        "ready": "準備完了",
-        "account": {
-          "title": "アカウントを接続",
-          "description": "対応する AI サブスクリプションを使用します。"
-        },
-        "apiKey": {
-          "title": "API キーを使用",
-          "description": "お好みのプロバイダーのキーを使用します。"
-        },
-        "local": {
-          "title": "ローカルで実行",
-          "description": "Ollama、LM Studio、vLLM に接続します。"
-        }
-      },
       "switchModel": "この会話で使うモデルを切り替える",
       "manageModels": "モデルを管理",
       "context": {

webui/src/i18n/locales/ko/common.json

@@ -1179,25 +1179,8 @@
       "stop": "응답 중지",
       "quotedContext": "인용한 문맥",
       "removeQuotedContext": "인용한 문맥 제거",
-      "openModelSetup": "AI 설정 열기",
+      "modelNotConfigured": "모델이 설정되지 않음",
-      "chooseAI": "AI 선택",
+      "configureModel": "모델 설정",
-      "modelSetup": {
-        "title": "AI 선택",
-        "description": "시작 방법을 선택하세요. 모델은 언제든 변경할 수 있습니다.",
-        "ready": "준비됨",
-        "account": {
-          "title": "계정 연결",
-          "description": "지원되는 AI 구독을 사용합니다."
-        },
-        "apiKey": {
-          "title": "API 키 사용",
-          "description": "선호하는 제공업체의 키를 사용합니다."
-        },
-        "local": {
-          "title": "로컬에서 실행",
-          "description": "Ollama, LM Studio 또는 vLLM에 연결합니다."
-        }
-      },
       "switchModel": "이 대화에서 사용할 모델 전환",
       "manageModels": "모델 관리",
       "context": {

webui/src/i18n/locales/pt-BR/common.json

@@ -1193,25 +1193,8 @@
       "stop": "Parar resposta",
       "quotedContext": "Contexto citado",
       "removeQuotedContext": "Remover contexto citado",
-      "openModelSetup": "Abrir configuração de IA",
+      "modelNotConfigured": "Modelo não configurado",
-      "chooseAI": "Escolha sua IA",
+      "configureModel": "Configurar modelo",
-      "modelSetup": {
-        "title": "Escolha sua IA",
-        "description": "Escolha como começar. Você pode trocar de modelo a qualquer momento.",
-        "ready": "Pronto",
-        "account": {
-          "title": "Conectar uma conta",
-          "description": "Use uma assinatura de IA compatível."
-        },
-        "apiKey": {
-          "title": "Usar uma chave de API",
-          "description": "Use uma chave do seu provedor preferido."
-        },
-        "local": {
-          "title": "Executar localmente",
-          "description": "Conecte o Ollama, LM Studio ou vLLM."
-        }
-      },
       "switchModel": "Alternar o modelo desta conversa",
       "manageModels": "Gerenciar modelos",
       "context": {

webui/src/i18n/locales/vi/common.json

@@ -1179,25 +1179,8 @@
       "stop": "Dừng phản hồi",
       "quotedContext": "Ngữ cảnh được trích dẫn",
       "removeQuotedContext": "Xóa ngữ cảnh được trích dẫn",
-      "openModelSetup": "Mở thiết lập AI",
+      "modelNotConfigured": "Chưa cấu hình mô hình",
-      "chooseAI": "Chọn AI của bạn",
+      "configureModel": "Cấu hình mô hình",
-      "modelSetup": {
-        "title": "Chọn AI của bạn",
-        "description": "Chọn cách bắt đầu. Bạn có thể đổi mô hình bất cứ lúc nào.",
-        "ready": "Sẵn sàng",
-        "account": {
-          "title": "Kết nối tài khoản",
-          "description": "Dùng gói đăng ký AI được hỗ trợ."
-        },
-        "apiKey": {
-          "title": "Dùng khóa API",
-          "description": "Dùng khóa từ nhà cung cấp bạn chọn."
-        },
-        "local": {
-          "title": "Chạy cục bộ",
-          "description": "Kết nối Ollama, LM Studio hoặc vLLM."
-        }
-      },
       "switchModel": "Chuyển mô hình cho cuộc trò chuyện này",
       "manageModels": "Quản lý mô hình",
       "context": {

webui/src/i18n/locales/zh-CN/common.json

@@ -1192,25 +1192,8 @@
       "stop": "停止响应",
       "quotedContext": "引用内容",
       "removeQuotedContext": "移除引用内容",
-      "openModelSetup": "打开 AI 设置",
+      "modelNotConfigured": "模型未配置",
-      "chooseAI": "选择你的 AI",
+      "configureModel": "配置模型",
-      "modelSetup": {
-        "title": "选择你的 AI",
-        "description": "选择一种开始方式，之后可随时更换模型。",
-        "ready": "已就绪",
-        "account": {
-          "title": "连接账户",
-          "description": "使用支持的 AI 订阅。"
-        },
-        "apiKey": {
-          "title": "使用 API 密钥",
-          "description": "使用你偏好的服务商密钥。"
-        },
-        "local": {
-          "title": "在本地运行",
-          "description": "连接 Ollama、LM Studio 或 vLLM。"
-        }
-      },
       "switchModel": "切换本次对话所用模型",
       "manageModels": "管理模型预设",
       "context": {

webui/src/i18n/locales/zh-TW/common.json

@@ -1179,25 +1179,8 @@
       "stop": "停止回覆",
       "quotedContext": "引用內容",
       "removeQuotedContext": "移除引用內容",
-      "openModelSetup": "開啟 AI 設定",
+      "modelNotConfigured": "尚未設定模型",
-      "chooseAI": "選擇你的 AI",
+      "configureModel": "設定模型",
-      "modelSetup": {
-        "title": "選擇你的 AI",
-        "description": "選擇一種開始方式，之後可隨時更換模型。",
-        "ready": "已就緒",
-        "account": {
-          "title": "連結帳戶",
-          "description": "使用支援的 AI 訂閱。"
-        },
-        "apiKey": {
-          "title": "使用 API 金鑰",
-          "description": "使用你偏好的服務商金鑰。"
-        },
-        "local": {
-          "title": "在本機執行",
-          "description": "連結 Ollama、LM Studio 或 vLLM。"
-        }
-      },
       "switchModel": "切換此對話使用的模型",
       "manageModels": "管理模型預設",
       "context": {

webui/src/tests/thread-shell.test.tsx

@@ -681,7 +681,7 @@ describe("ThreadShell", () => {
     );
 
     expect(await screen.findByTitle("fast · gpt-5.5 · OpenAI Codex")).toBeInTheDocument();
-    expect(screen.queryByRole("button", { name: "Choose your AI" })).not.toBeInTheDocument();
+    expect(screen.queryByRole("button", { name: "Model not configured" })).not.toBeInTheDocument();
   });
 
   it("switches through every named preset while preserving call-order priority", async () => {
@@ -763,7 +763,7 @@ describe("ThreadShell", () => {
     );
 
     expect(await screen.findByTitle("fast · gpt-4 · Company Proxy")).toBeInTheDocument();
-    expect(screen.queryByRole("button", { name: "Choose your AI" })).not.toBeInTheDocument();
+    expect(screen.queryByRole("button", { name: "Model not configured" })).not.toBeInTheDocument();
   });
 
   it("shows the effective fallback model in the composer badge", async () => {
@@ -835,7 +835,7 @@ describe("ThreadShell", () => {
     expect(screen.getByText("Default")).toBeInTheDocument();
   });
 
-  it("opens first-run model setup without clearing the draft", async () => {
+  it("opens model settings from the unconfigured model badge", async () => {
     const client = makeClient();
     const settings = modelSettings("openai-codex/gpt-5.1-codex", "openai_codex");
     settings.agent.has_api_key = false;
@@ -844,20 +844,6 @@ describe("ThreadShell", () => {
         ? { ...provider, auth_type: "oauth", configured: false }
         : provider,
     );
-    settings.providers.push(
-      {
-        name: "xai_grok",
-        label: "xAI Grok",
-        auth_type: "oauth",
-        configured: true,
-      },
-      {
-        name: "ollama",
-        label: "Ollama",
-        configured: true,
-        api_base: "http://127.0.0.1:11434",
-      },
-    );
     const onOpenModelSettings = vi.fn();
 
     render(
@@ -874,31 +860,17 @@ describe("ThreadShell", () => {
       ),
     );
 
-    const badge = await screen.findByRole("button", { name: "Choose your AI" });
+    const badge = await screen.findByRole("button", { name: "Model not configured" });
-    const setupIcon = screen.getByTestId("composer-model-setup-icon");
+    expect(screen.getByTestId("composer-model-setup-icon")).toBeInTheDocument();
-    expect(setupIcon).toBeInTheDocument();
-    expect(setupIcon.parentElement).not.toHaveClass("border-amber-500/35");
     expect(screen.queryByTestId("composer-model-logo-openai_codex")).not.toBeInTheDocument();
     fireEvent.click(badge);
-    expect(await screen.findByRole("dialog", { name: "Choose your AI" })).toBeInTheDocument();
+    expect(onOpenModelSettings).toHaveBeenCalledTimes(1);
-    expect(screen.getAllByText("Ready")).toHaveLength(3);
-    expect(onOpenModelSettings).not.toHaveBeenCalled();
 
-    fireEvent.click(screen.getByRole("button", { name: "Close" }));
+    fireEvent.change(screen.getByRole("textbox", { name: "Message input" }), {
-
-    const input = screen.getByRole("textbox", { name: "Message input" });
-    fireEvent.change(input, {
       target: { value: "hello" },
     });
-    fireEvent.keyDown(input, { key: "Enter", code: "Enter" });
+    fireEvent.click(screen.getByRole("button", { name: "Configure model" }));
-
+    expect(onOpenModelSettings).toHaveBeenCalledTimes(2);
-    expect(await screen.findByRole("dialog", { name: "Choose your AI" })).toBeInTheDocument();
-    expect(input).toHaveValue("hello");
-    fireEvent.click(screen.getByRole("button", { name: "Use an API key" }));
-
-    expect(onOpenModelSettings).toHaveBeenCalledTimes(1);
-    expect(input).toHaveValue("hello");
-    await waitFor(() => expect(input).toHaveFocus());
     expect(client.sendMessage).not.toHaveBeenCalled();
   });