fix(agent): route direct subagent results in-turn

maintainer edit: remove defensive branches that normal session storage cannot produce and keep the idle-expiry helper direct.

docs/configuration.md

@@ -240,6 +240,7 @@ Tracing covers the providers that go through nanobot's OpenAI-compatible client
 > - **Xiaomi MiMo thinking mode**: MiMo models (e.g. `mimo-v2.5-pro`) default to enabled thinking. Use `agents.defaults.reasoningEffort: "none"` to disable it, or `"low"` / `"medium"` / `"high"` to keep it on. Omitting the field preserves the provider's per-model default.
 > - **Xiaomi MiMo Token Plan**: If you're on MiMo's token plan, set `"apiBase": "https://token-plan-sgp.xiaomimimo.com/v1"` in your xiaomi_mimo provider config.
 > - **Custom OpenAI-compatible providers**: Besides the built-in `custom` provider, any extra key under `providers` can define its own OpenAI-compatible endpoint. For example, `providers.companyProxy.apiBase` plus `modelPresets.primary.provider: "companyProxy"` creates a separate custom provider. Set `apiBase`; set `apiKey` only when the endpoint requires it. This named-custom path uses the OpenAI-compatible request format only. For Anthropic-compatible proxies, use `providers.anthropic.apiBase` with `provider: "anthropic"`.
+> - **Provider-scoped proxy**: `providers.<name>.proxy` routes only that provider through an HTTP proxy. It is supported for OpenAI-compatible providers and `openai_codex`. Native provider backends such as `anthropic`, `bedrock`, `azure_openai`, and `github_copilot` reject `proxy`.
 
 | Provider | Purpose | Get API Key |
 |----------|---------|-------------|
@@ -632,20 +633,37 @@ nanobot agent -m "Reply with one short sentence."
 <details>
 <summary><b>OpenAI Codex (OAuth)</b></summary>
 
-Codex uses OAuth instead of API keys. Requires a ChatGPT Plus or Pro account. No `providers.openaiCodex` block is needed in `config.json`; `nanobot provider login` stores the OAuth session outside config.
+Codex uses OAuth instead of API keys. Requires a ChatGPT Plus or Pro account. `nanobot provider login` stores the OAuth session outside config. A `providers.openai_codex` block is optional and is only needed for provider-specific settings such as a proxy.
 
 **1. Login:**
 ```bash
 nanobot provider login openai-codex
 ```
 
-**2. Set model** (merge into `~/.nanobot/config.json`):
+If the machine running nanobot cannot open a graphical browser, copy the printed URL into a real browser. For remote SSH login, open the URL locally, then paste the final `http://localhost:1455/auth/callback?...` redirect URL back into the terminal when prompted.
+
+**2. Optional proxy** (merge into `~/.nanobot/config.json` if Codex OAuth or Codex API traffic must use a proxy):
+
+```json
+{
+  "providers": {
+    "openai_codex": {
+      "proxy": "http://127.0.0.1:7890"
+    }
+  }
+}
+```
+
+The proxy applies to Codex OAuth token refresh, interactive token exchange, and Codex Responses API requests. It does not affect other providers; configure `proxy` separately on each supported provider that needs it.
+
+**3. Set model** (merge into `~/.nanobot/config.json`):
 ```json
 {
   "modelPresets": {
     "codex": {
       "provider": "openai_codex",
-      "model": "openai-codex/gpt-5.1-codex"
+      "model": "gpt-5.1-codex",
+      "reasoningEffort": "high"
     }
   },
   "agents": {
@@ -656,7 +674,9 @@ nanobot provider login openai-codex
 }
 ```
 
-**3. Chat:**
+Use `reasoningEffort` in the preset to send a Codex reasoning effort such as `"low"`, `"medium"`, `"high"`, or another value supported by the selected model. When `provider` is explicitly `openai_codex`, the model name does not need the `openai-codex/` prefix.
+
+**4. Chat:**
 ```bash
 nanobot agent -m "Hello!"
 
@@ -675,7 +695,17 @@ nanobot agent -c ~/.nanobot-telegram/config.json -w /tmp/nanobot-telegram-test -
 <details>
 <summary><b>GitHub Copilot (OAuth)</b></summary>
 
-GitHub Copilot uses OAuth instead of API keys. Requires a [GitHub account with a plan](https://github.com/features/copilot/plans) configured. No `providers.githubCopilot` block is needed in `config.json`; `nanobot provider login` stores the OAuth session outside config.
+GitHub Copilot uses OAuth instead of API keys. Requires a [GitHub account with a plan](https://github.com/features/copilot/plans) configured. No `providers.github_copilot` block is needed in `config.json`; `nanobot provider login` stores the OAuth session outside config.
+
+For GitHub Enterprise / Copilot for Business, set the endpoint overrides you need before login:
+```bash
+export NANOBOT_GITHUB_COPILOT_CLIENT_ID="your-enterprise-client-id"
+export NANOBOT_GITHUB_DEVICE_CODE_URL="https://ghe.example/login/device/code"
+export NANOBOT_GITHUB_ACCESS_TOKEN_URL="https://ghe.example/login/oauth/access_token"
+export NANOBOT_GITHUB_USER_URL="https://api.ghe.example/user"
+export NANOBOT_COPILOT_TOKEN_URL="https://api.ghe.example/copilot_internal/v2/token"
+export NANOBOT_COPILOT_BASE_URL="https://copilot-api.ghe.example"
+```
 
 **1. Login:**
 ```bash
@@ -1974,6 +2004,7 @@ The heartbeat job is backed by the same cron service as user-created reminders.
 | `gateway.heartbeat.enabled` | `true` | Register the built-in heartbeat cron job on gateway startup. |
 | `gateway.heartbeat.intervalS` | `1800` | Seconds between heartbeat checks. |
 | `gateway.heartbeat.keepRecentMessages` | `8` | Number of recent heartbeat-session messages to retain after each run. |
+| `gateway.restartMode` | `auto` | Restart strategy for `/restart`: `auto` uses `spawn` on Windows foreground runs and `exec` elsewhere. Use `exit` with Windows service wrappers such as WinSW or nssm so the service manager owns the restart. |
 
 
 ## Subagent Concurrency

docs/providers.md

@@ -61,9 +61,12 @@ These fields answer different questions:
 | `model` | `modelPresets.<name>.model` | The model ID expected by that provider or gateway. |
 | `apiKey` | `providers.<provider>.apiKey` | Credential for that provider. Use `${ENV_VAR}` for secrets. |
 | `apiBase` | `providers.<provider>.apiBase` | HTTP base URL of the provider endpoint. |
+| `proxy` | `providers.<provider>.proxy` | Optional HTTP proxy for this provider only. Supported for OpenAI-compatible providers and OpenAI Codex. |
 
 You usually omit `apiBase` for hosted built-in providers such as OpenRouter, Anthropic direct, OpenAI direct, Groq, or Bedrock because nanobot knows their default endpoints. Set `apiBase` for `custom`, local OpenAI-compatible servers, provider proxies, regional endpoints, or subscription endpoints. Include the API version path when the endpoint requires it, for example `https://api.example.com/v1` or `http://localhost:11434/v1`.
 
+Use `proxy` when one provider must send HTTP traffic through a proxy without changing process-wide `HTTP_PROXY` / `HTTPS_PROXY`. This is supported for providers that use nanobot's OpenAI-compatible client, including `openai`, `custom`, named custom providers, OpenRouter-style gateways, local OpenAI-compatible servers, and similar registry entries. It is also supported for `openai_codex`, including Codex OAuth token exchange/refresh and Codex Responses API requests. Native provider backends such as `anthropic`, `bedrock`, `azure_openai`, and `github_copilot` reject `proxy`; use their endpoint-specific configuration instead.
+
 ## Common Provider Patterns
 
 ### OpenRouter Gateway
@@ -422,6 +425,32 @@ nanobot provider login github-copilot
 
 Then explicitly select the provider and model in a preset. OAuth providers are not valid automatic fallbacks.
 
+For OpenAI Codex, add `providers.openai_codex.proxy` only when Codex OAuth/token refresh or Codex API requests must use a proxy:
+
+```json
+{
+  "providers": {
+    "openai_codex": {
+      "proxy": "http://127.0.0.1:7890"
+    }
+  },
+  "modelPresets": {
+    "codex": {
+      "provider": "openai_codex",
+      "model": "gpt-5.1-codex",
+      "reasoningEffort": "high"
+    }
+  },
+  "agents": {
+    "defaults": {
+      "modelPreset": "codex"
+    }
+  }
+}
+```
+
+If you run the login command on a remote/headless machine and open the authorization URL in a local browser, paste the final `http://localhost:1455/auth/callback?...` redirect URL back into the terminal when prompted. See [`configuration.md#providers`](./configuration.md#providers) for the full OAuth provider notes.
+
 ## Provider Resolution
 
 The recommended path is a named preset selected by `agents.defaults.modelPreset`. The effective model parameters come from:

nanobot/agent/autocompact.py

@@ -34,6 +34,26 @@ class AutoCompact:
             ts = datetime.fromisoformat(ts)
         return ((now or datetime.now()) - ts).total_seconds() >= self._ttl * 60
 
+    def _has_compactable_idle_tail(self, key: str) -> bool:
+        session = self.sessions.get_or_create(key)
+        tail = list(session.messages[session.last_consolidated:])
+        if not tail:
+            return False
+        probe = Session(
+            key=session.key,
+            messages=tail,
+            created_at=session.created_at,
+            updated_at=session.updated_at,
+            metadata={},
+            last_consolidated=0,
+        )
+        result = probe.retain_recent_legal_suffix(
+            self._RECENT_SUFFIX_MESSAGES,
+            extend_to_user=True,
+        )
+        messages_to_remove = result.dropped[result.already_consolidated_count:]
+        return bool(messages_to_remove)
+
     @staticmethod
     def _format_summary(text: str, last_active: datetime) -> str:
         return f"Previous conversation summary (last active {last_active.isoformat()}):\n{text}"
@@ -52,7 +72,8 @@ class AutoCompact:
                 continue
             if key in active_session_keys:
                 continue
-            if self._is_expired(info.get("updated_at"), now):
+            updated_at = info.get("updated_at")
+            if self._is_expired(updated_at, now) and self._has_compactable_idle_tail(key):
                 self._archiving.add(key)
                 schedule_background(self._archive(key))
 

nanobot/agent/context_governance.py

@@ -36,6 +36,23 @@ COMPACTABLE_TOOLS = frozenset({
 # 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]"
+PLACEHOLDER_TEXTS = frozenset({
+    "[Previous assistant message omitted.]",
+})
+
+
+def _tool_call_name_is_valid(tool_call: Any) -> bool:
+    """Whether a persisted OpenAI-style tool_call carries a usable name.
+
+    Mirrors ``ToolCallRequest.has_valid_name`` for the dict shape stored in
+    message history: a degenerate call with ``name=None`` / ``""`` cannot be
+    executed and is rejected by upstream APIs if replayed.
+    """
+    if not isinstance(tool_call, dict):
+        return False
+    fn = tool_call.get("function")
+    name = fn.get("name") if isinstance(fn, dict) else tool_call.get("name")
+    return isinstance(name, str) and bool(name)
 
 
 @dataclass(slots=True)
@@ -61,7 +78,9 @@ class ContextGovernor:
         messages: list[dict[str, Any]],
         compacted_tool_call_ids: set[str],
     ) -> list[dict[str, Any]]:
-        updated = self.drop_orphan_tool_results(messages)
+        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)
         updated = self.apply_tool_result_budget(config, updated)
         updated = self.compact_inflight_overflow(config, updated, compacted_tool_call_ids)
@@ -116,6 +135,99 @@ class ContextGovernor:
             return truncate_text(content, config.max_tool_result_chars)
         return content
 
+    @staticmethod
+    def strip_placeholder_assistant_messages(
+        messages: list[dict[str, Any]],
+    ) -> list[dict[str, Any]]:
+        """Remove assistant messages that are compaction placeholders.
+
+        Messages like ``[Previous assistant message omitted.]`` carry no useful
+        context for the model and can cause it to repeatedly attempt tool calls
+        that previously failed, producing malformed responses in a loop.
+        Consecutive same-role messages that result from removal are handled
+        downstream by the provider's merge-consecutive logic. Only the
+        model-facing copy is repaired; the persisted transcript is untouched
+        (a copy is returned, or the same list object when nothing changes).
+        """
+        updated: list[dict[str, Any]] | None = None
+        for idx, msg in enumerate(messages):
+            if msg.get("role") != "assistant":
+                if updated is not None:
+                    updated.append(msg)
+                continue
+            content = msg.get("content", "")
+            text = content if isinstance(content, str) else ""
+            is_placeholder = text.strip() in PLACEHOLDER_TEXTS
+            has_tool_calls = bool(msg.get("tool_calls"))
+            if is_placeholder and not has_tool_calls:
+                if updated is None:
+                    updated = list(messages[:idx])
+                logger.debug(
+                    "Stripping placeholder assistant message from history: {!r}",
+                    text[:60],
+                )
+                continue
+            if updated is not None:
+                updated.append(msg)
+        if updated is None:
+            return messages
+        return updated
+
+    @staticmethod
+    def strip_malformed_tool_calls(
+        messages: list[dict[str, Any]],
+    ) -> list[dict[str, Any]]:
+        """Drop persisted assistant tool_calls whose name is missing/non-string.
+
+        A degenerate tool call (``name=None`` or ``""``) that slipped into the
+        saved history before this guard existed gets replayed on every turn and
+        makes upstream APIs reject the whole request
+        (``messages.content.N.tool_use.name: Input should be a valid string``),
+        permanently wedging the session. Removing the bad call here lets the
+        existing orphan-result cleanup drop its now-dangling tool result, so a
+        polluted session self-heals on its next turn. The persisted transcript
+        is left untouched; only the model-facing copy is repaired (a copy is
+        returned, or the same list object when nothing changes).
+        """
+        updated: list[dict[str, Any]] | None = None
+        for idx, msg in enumerate(messages):
+            if msg.get("role") != "assistant":
+                if updated is not None:
+                    updated.append(msg)
+                continue
+            calls = msg.get("tool_calls")
+            if not calls:
+                if updated is not None:
+                    updated.append(msg)
+                continue
+            kept = [tc for tc in calls if _tool_call_name_is_valid(tc)]
+            if len(kept) == len(calls):
+                if updated is not None:
+                    updated.append(msg)
+                continue
+            if updated is None:
+                updated = [dict(m) for m in messages[:idx]]
+            logger.warning(
+                "Stripping {} malformed tool_call(s) with missing/non-string "
+                "name from assistant history before request",
+                len(calls) - len(kept),
+            )
+            repaired = dict(msg)
+            if kept:
+                repaired["tool_calls"] = kept
+            else:
+                repaired.pop("tool_calls", None)
+            # An assistant turn with neither content nor any valid tool call is
+            # itself invalid upstream; drop it entirely in that case.
+            has_content = bool(repaired.get("content"))
+            if not kept and not has_content:
+                continue
+            updated.append(repaired)
+
+        if updated is None:
+            return messages
+        return updated
+
     @staticmethod
     def drop_orphan_tool_results(
         messages: list[dict[str, Any]],

nanobot/agent/loop.py

@@ -57,7 +57,11 @@ from nanobot.session.goal_state import (
     sustained_goal_active,
 )
 from nanobot.session.keys import UNIFIED_SESSION_KEY, session_key_for_channel
-from nanobot.session.manager import Session, SessionManager
+from nanobot.session.manager import (
+    Session,
+    SessionManager,
+    replay_max_messages_for_context,
+)
 from nanobot.utils.document import extract_documents, reference_non_image_attachments
 from nanobot.utils.helpers import image_placeholder_text
 from nanobot.utils.helpers import truncate_text as truncate_text_fn
@@ -201,7 +205,6 @@ class AgentLoop:
         timezone: str | None = None,
         session_ttl_minutes: int = 0,
         consolidation_ratio: float = 0.5,
-        max_messages: int = 120,
         hooks: list[AgentHook] | None = None,
         unified_session: bool = False,
         disabled_skills: list[str] | None = None,
@@ -215,6 +218,7 @@ class AgentLoop:
         preset_snapshot_loader: preset_helpers.PresetSnapshotLoader | None = None,
         runtime_events: RuntimeEventBus | None = None,
         runtime_model_publisher: Callable[[str, str | None], None] | None = None,
+        restart_mode: str = "auto",
     ):
         from nanobot.config.schema import ToolsConfig
 
@@ -224,6 +228,7 @@ class AgentLoop:
         self.runtime_events = runtime_events or RuntimeEventBus()
         self.runtime_event_publisher = RuntimeEventPublisher(self.runtime_events)
         self.channels_config = channels_config
+        self.restart_mode = restart_mode
         self.provider = provider
         self._provider_snapshot_loader = provider_snapshot_loader
         self._preset_snapshot_loader = preset_snapshot_loader
@@ -292,7 +297,7 @@ class AgentLoop:
             llm_wall_timeout_for_session=lambda sk: runner_wall_llm_timeout_s(self.sessions, sk),
         )
         self._unified_session = unified_session
-        self._max_messages = max_messages if max_messages > 0 else 120
+        self._max_messages = replay_max_messages_for_context(self.context_window_tokens)
         self._running = False
         self._mcp_servers = mcp_servers or {}
         self._mcp_stacks: dict[str, AsyncExitStack] = {}
@@ -390,10 +395,10 @@ class AgentLoop:
             disabled_skills=defaults.disabled_skills,
             session_ttl_minutes=defaults.session_ttl_minutes,
             consolidation_ratio=defaults.consolidation_ratio,
-            max_messages=defaults.max_messages,
             tools_config=config.tools,
             model_presets=preset_helpers.configured_model_presets(config),
             model_preset=defaults.model_preset,
+            restart_mode=config.gateway.restart_mode,
             provider_snapshot_loader=provider_snapshot_loader,
             preset_snapshot_loader=preset_snapshot_loader,
             **extra,
@@ -421,6 +426,7 @@ class AgentLoop:
         self.runner.provider = provider
         self.subagents.set_provider(provider, model)
         self.consolidator.set_provider(provider, model, context_window_tokens)
+        self._sync_replay_max_messages()
         self._provider_signature = snapshot.signature
         if publish_update and self._runtime_model_publisher is not None:
             self._runtime_model_publisher(
@@ -434,6 +440,9 @@ class AgentLoop:
             )
         logger.info("Runtime model switched for next turn: {} -> {}", old_model, model)
 
+    def _sync_replay_max_messages(self) -> None:
+        self._max_messages = replay_max_messages_for_context(self.context_window_tokens)
+
     def _refresh_provider_snapshot(self) -> None:
         if self._provider_snapshot_loader is None:
             return
@@ -1843,13 +1852,17 @@ class AgentLoop:
         )
         # Share the dispatch lock so direct calls serialize with bus turns.
         lock = self._session_locks.setdefault(session_key, asyncio.Lock())
+        pending: asyncio.Queue[InboundMessage] = asyncio.Queue(maxsize=20)
         try:
             async with lock:
+                self._pending_queues[session_key] = pending
+                self.subagents.set_direct_result_queue(session_key, pending)
                 kwargs: dict[str, Any] = {
                     "session_key": session_key,
                     "on_progress": on_progress,
                     "on_stream": on_stream,
                     "on_stream_end": on_stream_end,
+                    "pending_queue": pending,
                     "ephemeral": ephemeral,
                 }
                 if _run_extra_hooks_for_ephemeral:
@@ -1863,5 +1876,13 @@ class AgentLoop:
                     **kwargs,
                 )
         finally:
+            self.subagents.clear_direct_result_queue(session_key, pending)
+            if self._pending_queues.get(session_key) is pending:
+                self._pending_queues.pop(session_key, None)
+            while True:
+                try:
+                    await self.bus.publish_inbound(pending.get_nowait())
+                except asyncio.QueueEmpty:
+                    break
             await self._runtime_events().run_status_changed(msg, session_key, "idle")
             self._runtime_events().clear_turn(session_key)

nanobot/agent/memory.py

@@ -33,7 +33,6 @@ if TYPE_CHECKING:
     from nanobot.providers.base import LLMProvider
     from nanobot.session.manager import SessionManager
 
-
 # ---------------------------------------------------------------------------
 # MemoryStore — pure file I/O layer
 # ---------------------------------------------------------------------------
@@ -1006,7 +1005,6 @@ class Consolidator:
 
             messages_to_summarize = list(session.messages[session.last_consolidated:])
             if not messages_to_summarize:
-                session.updated_at = datetime.now()
                 self.sessions.save(session)
                 return ""
 
@@ -1018,12 +1016,11 @@ class Consolidator:
                 metadata={},
                 last_consolidated=0,
             )
-            dropped, already_consolidated = probe.retain_recent_legal_suffix(max_suffix, extend_to_user=True)
+            result = probe.retain_recent_legal_suffix(max_suffix, extend_to_user=True)
             messages_to_keep = probe.messages
-            messages_to_remove = dropped[already_consolidated:]
+            messages_to_remove = result.dropped[result.already_consolidated_count:]
 
             if not messages_to_remove and not messages_to_keep:
-                session.updated_at = datetime.now()
                 self.sessions.save(session)
                 return ""
 
@@ -1046,7 +1043,6 @@ class Consolidator:
 
             session.messages = messages_to_keep
             session.last_consolidated = 0
-            session.updated_at = datetime.now()
             self.sessions.save(session)
 
             if messages_to_remove:

nanobot/agent/runner.py

@@ -389,7 +389,15 @@ class AgentRunner:
                     spec.session_key or "default",
                 )
                 try:
-                    messages_for_model = ContextGovernor.drop_orphan_tool_results(messages)
+                    messages_for_model = ContextGovernor.strip_placeholder_assistant_messages(
+                        messages
+                    )
+                    messages_for_model = ContextGovernor.strip_malformed_tool_calls(
+                        messages_for_model
+                    )
+                    messages_for_model = ContextGovernor.drop_orphan_tool_results(
+                        messages_for_model
+                    )
                     messages_for_model = ContextGovernor.backfill_missing_tool_results(
                         messages_for_model
                     )
@@ -725,6 +733,8 @@ class AgentRunner:
         messages: list[dict[str, Any]],
         hook: AgentHook,
         context: AgentHookContext,
+        *,
+        malformed_retry: bool = False,
     ):
         timeout_s: float | None = spec.llm_timeout_s
         if timeout_s is None:
@@ -867,8 +877,94 @@ class AgentRunner:
             )
         if progress_state and progress_state.get("reasoning_open"):
             await hook.emit_reasoning_end()
+        dropped, all_dropped, original_finish_reason = (
+            self._drop_malformed_tool_calls(response)
+        )
+        if (
+            all_dropped
+            and original_finish_reason in ("tool_calls", "function_call")
+            and not malformed_retry
+        ):
+            logger.warning(
+                "Retrying LLM request after all {} malformed tool call(s) were dropped",
+                dropped,
+            )
+            retry_messages = self._malformed_tool_call_retry_messages(
+                messages, response.content,
+            )
+            return await self._request_model(
+                spec, retry_messages, hook, context,
+                malformed_retry=True,
+            )
+        if (
+            all_dropped
+            and original_finish_reason in ("tool_calls", "function_call")
+            and malformed_retry
+        ):
+            logger.warning(
+                "Malformed tool calls persisted after retry; falling back to no-tools request",
+            )
+            fallback_messages = self._malformed_tool_call_retry_messages(
+                messages, response.content,
+            )
+            return await self._request_no_tools(spec, fallback_messages)
         return response
 
+    @staticmethod
+    def _drop_malformed_tool_calls(
+        response: LLMResponse,
+    ) -> tuple[int, bool, str | None]:
+        """Strip tool calls whose name is missing/non-string from the response.
+
+        Returns (dropped_count, all_dropped, original_finish_reason).
+
+        A degenerate call (name=None or "") cannot be executed, and if it were
+        persisted into the assistant message it would be replayed on every
+        subsequent turn, causing upstream validation errors
+        (``tool_use.name: Input should be a valid string``) that permanently
+        wedge the session. Dropping it here keeps it out of execution, the
+        assistant message, and the saved history in one place.
+        """
+        calls = getattr(response, "tool_calls", None)
+        if not calls:
+            return (0, False, getattr(response, "finish_reason", None))
+        valid = [tc for tc in calls if tc.has_valid_name()]
+        if len(valid) == len(calls):
+            return (0, False, getattr(response, "finish_reason", None))
+        dropped = len(calls) - len(valid)
+        original_finish_reason = getattr(response, "finish_reason", None)
+        logger.warning(
+            "Dropped {} malformed tool call(s) with missing/non-string name "
+            "from LLM response (finish_reason={!r})",
+            dropped,
+            original_finish_reason,
+        )
+        response.tool_calls = valid
+        if not valid:
+            response.finish_reason = "stop"
+        return (dropped, not valid, original_finish_reason)
+
+    @staticmethod
+    def _malformed_tool_call_retry_messages(
+        messages: list[dict[str, Any]],
+        assistant_text: str | None,
+    ) -> list[dict[str, Any]]:
+        retry_messages = list(messages)
+        note = (
+            "The previous model response attempted to call tools, but every tool call "
+            "was malformed: the tool_use blocks had missing or non-string tool names. "
+            "Do not answer with a promise to use tools. Either call the required tools again "
+            "using valid tool names from the provided tool list and JSON object inputs, or give "
+            "a final answer only if no tool is required."
+        )
+        if assistant_text:
+            note += (
+                f"\n\nPrevious assistant text before the malformed calls:\n"
+                f"{assistant_text}"
+            )
+        retry_messages.append({"role": "user", "content": note})
+        return retry_messages
+
     async def _request_finalization_retry(
         self,
         spec: AgentRunSpec,

nanobot/agent/subagent.py

@@ -118,6 +118,22 @@ class SubagentManager:
         self._running_tasks: dict[str, asyncio.Task[None]] = {}
         self._task_statuses: dict[str, SubagentStatus] = {}
         self._session_tasks: dict[str, set[str]] = {}  # session_key -> {task_id, ...}
+        self._direct_result_queues: dict[str, asyncio.Queue[InboundMessage]] = {}
+
+    def set_direct_result_queue(
+        self,
+        session_key: str,
+        queue: asyncio.Queue[InboundMessage],
+    ) -> None:
+        self._direct_result_queues[session_key] = queue
+
+    def clear_direct_result_queue(
+        self,
+        session_key: str,
+        queue: asyncio.Queue[InboundMessage],
+    ) -> None:
+        if self._direct_result_queues.get(session_key) is queue:
+            self._direct_result_queues.pop(session_key, None)
 
     def _subagent_tools_config(self) -> ToolsConfig:
         """Build a ToolsConfig scoped for subagent use."""
@@ -335,6 +351,10 @@ class SubagentManager:
             metadata=metadata,
         )
 
+        if queue := self._direct_result_queues.get(override):
+            await queue.put(msg)
+            logger.debug("Subagent [{}] queued result directly for {}", task_id, override)
+            return
         await self.bus.publish_inbound(msg)
         logger.debug("Subagent [{}] announced result to {}:{}", task_id, origin['channel'], origin['chat_id'])
 

nanobot/agent/tools/long_task.py

@@ -97,7 +97,8 @@ class _GoalToolsMixin(ContextAware):
             "Sustained objective for this chat thread. First read the built-in **long-goal** skill, "
             "especially its Start fast section, then call this promptly once the user's intent is clear. "
             "The goal must still be idempotent, self-contained, bounded, and explicit about done-ness; "
-            "do not delay this tool call to over-plan, research, or decide execution details.",
+            "do not delay this tool call to over-plan, research, or decide execution details. "
+            "Do not use this for a single current-turn answer, including one that uses spawn subagents.",
             max_length=12_000,
         ),
         ui_summary=StringSchema(
@@ -139,6 +140,8 @@ class LongTaskTool(Tool, _GoalToolsMixin):
     def description(self) -> str:
         return (
             "Mark this thread as a sustained long-running task. "
+            "Use only when the user wants work to persist across future turns or background check-ins; "
+            "do not use for a single current-turn answer, including one that uses spawn subagents. "
             "First read the built-in **long-goal** skill, especially its Start fast section; then call this "
             "as soon as the user's intent is clear. Write a good idempotent goal, but do not delay the tool "
             "call with long planning, research, or execution-detail thinking. "

nanobot/agent/tools/mcp.py

@@ -1,6 +1,7 @@
 """MCP client: connects to MCP servers and wraps their tools as native nanobot tools."""
 
 import asyncio
+import json
 import os
 import re
 import shutil
@@ -165,12 +166,31 @@ async def _probe_http_url(url: str, timeout: float = 3.0) -> bool:
         return False
 
 
+def _redact_url(url: str) -> str:
+    """Strip credentials and query/fragment before logging an MCP URL.
+
+    Server URLs may embed secrets (``https://user:token@host/sse`` or a
+    ``?token=`` query). Some deployments also put opaque tokens in the path, so
+    log only the origin and a path placeholder.
+    """
+    try:
+        parts = urllib.parse.urlsplit(url)
+        hostname = parts.hostname or ""
+        netloc = f"[{hostname}]" if ":" in hostname else hostname
+        if parts.port:
+            netloc = f"{netloc}:{parts.port}"
+        path = "/..." if parts.path and parts.path != "/" else parts.path
+        return urllib.parse.urlunsplit((parts.scheme, netloc, path, "", ""))
+    except Exception:
+        return "<redacted-url>"
+
+
 async def _validate_mcp_request_url(request: httpx.Request) -> None:
     """Validate each outgoing MCP HTTP request, including redirect targets."""
     ok, error = validate_url_target(str(request.url))
     if not ok:
         raise httpx.RequestError(
-            f"Blocked unsafe MCP URL {request.url} ({error})",
+            f"Blocked unsafe MCP URL {_redact_url(str(request.url))} ({error})",
             request=request,
         )
 
@@ -313,6 +333,52 @@ class _MCPWrapperBase(Tool):
         return True
 
 
+def _image_block_data_url(block: Any, types: Any) -> str | None:
+    """Return a base64 ``data:`` URL for an MCP image-bearing content block.
+
+    Handles ``ImageContent`` directly and ``EmbeddedResource`` wrapping a binary
+    blob with an ``image/*`` MIME type. Returns ``None`` for anything else.
+    ``getattr`` guards keep this safe when the installed/faked ``mcp`` SDK does
+    not expose a given type.
+    """
+    image_cls = getattr(types, "ImageContent", None)
+    if image_cls is not None and isinstance(block, image_cls):
+        mime = getattr(block, "mimeType", None) or "image/png"
+        return f"data:{mime};base64,{block.data}"
+
+    embedded_cls = getattr(types, "EmbeddedResource", None)
+    blob_cls = getattr(types, "BlobResourceContents", None)
+    if embedded_cls is not None and isinstance(block, embedded_cls):
+        resource = getattr(block, "resource", None)
+        if blob_cls is not None and isinstance(resource, blob_cls):
+            mime = getattr(resource, "mimeType", None) or ""
+            if isinstance(mime, str) and mime.startswith("image/"):
+                return f"data:{mime};base64,{resource.blob}"
+    return None
+
+
+def _mcp_image_tool_result(text_parts: list[str], artifacts: list[dict[str, Any]]) -> str:
+    """Build the compact tool result for an MCP call that returned image(s).
+
+    The base64 stays out of the model context entirely — only artifact paths and
+    metadata are returned, so the result is small and the channel can deliver the
+    saved file via the message tool.
+    """
+    payload: dict[str, Any] = {
+        "artifacts": artifacts,
+        "next_step": (
+            "These images were returned by an MCP tool and saved as local artifacts. "
+            "Call the message tool with the artifact 'path' values in the media "
+            "parameter to deliver the images to the user. Do not paste base64 or raw "
+            "paths into your reply unless the user asks for debug details."
+        ),
+    }
+    text = "\n".join(part for part in text_parts if part)
+    if text:
+        payload["text"] = text
+    return json.dumps(payload, ensure_ascii=False)
+
+
 class MCPToolWrapper(_MCPWrapperBase):
     """Wraps a single MCP server tool as a nanobot Tool."""
 
@@ -340,8 +406,6 @@ class MCPToolWrapper(_MCPWrapperBase):
         return self._parameters
 
     async def execute(self, **kwargs: Any) -> str:
-        from mcp import types
-
         retried_transient = False
         refreshed_session = False
         while True:
@@ -396,17 +460,63 @@ class MCPToolWrapper(_MCPWrapperBase):
                 )
                 return f"(MCP tool call failed: {type(exc).__name__})"
             else:
-                # Success — extract result
-                parts = []
-                for block in result.content:
-                    if isinstance(block, types.TextContent):
-                        parts.append(block.text)
-                    else:
-                        parts.append(str(block))
-                return "\n".join(parts) or "(no output)"
+                # Success — extract text and persist any image content as artifacts.
+                return self._render_call_result(result.content, kwargs)
 
         return "(MCP tool call failed)"  # Unreachable, but satisfies type checkers
 
+    def _render_call_result(self, content: Any, arguments: Mapping[str, Any]) -> str:
+        """Turn MCP content blocks into a tool result string.
+
+        Text is concatenated as before. Image blocks are decoded and saved as
+        local artifacts (mirroring the built-in image generation tool) so the
+        model can deliver them via the message tool instead of trying to forward
+        base64 — which would be truncated and bloat the context window.
+        """
+        from mcp import types
+
+        text_parts: list[str] = []
+        artifacts: list[dict[str, Any]] = []
+        for block in content:
+            if isinstance(block, types.TextContent):
+                text_parts.append(block.text)
+                continue
+            data_url = _image_block_data_url(block, types)
+            if data_url is not None:
+                stored = self._store_image_block(data_url, arguments)
+                if stored is not None:
+                    artifacts.append(stored)
+                else:
+                    text_parts.append("(MCP tool returned an image that could not be stored)")
+                continue
+            text_parts.append(str(block))
+
+        if artifacts:
+            return _mcp_image_tool_result(text_parts, artifacts)
+        return "\n".join(text_parts) or "(no output)"
+
+    def _store_image_block(
+        self, data_url: str, arguments: Mapping[str, Any]
+    ) -> dict[str, Any] | None:
+        """Persist one image data URL as an artifact; return its metadata or None."""
+        from nanobot.utils.artifacts import ArtifactError, store_generated_image_artifact
+
+        try:
+            return store_generated_image_artifact(
+                data_url,
+                prompt=str(arguments.get("prompt") or ""),
+                model=str(arguments.get("model") or ""),
+                save_dir="generated",
+                provider=f"mcp:{self._server_name}",
+            )
+        except (ArtifactError, OSError) as exc:
+            logger.warning(
+                "MCP tool '{}' returned an image that could not be stored: {}",
+                self._name,
+                exc,
+            )
+            return None
+
 
 class MCPResourceWrapper(_MCPWrapperBase):
     """Wraps an MCP resource URI as a read-only nanobot Tool."""
@@ -683,7 +793,7 @@ async def connect_mcp_servers(
                     logger.warning(
                         "MCP server '{}': blocked unsafe URL {} ({})",
                         name,
-                        cfg.url,
+                        _redact_url(cfg.url),
                         error,
                     )
                     await server_stack.aclose()
@@ -704,7 +814,7 @@ async def connect_mcp_servers(
                 read, write = await server_stack.enter_async_context(stdio_client(params))
             elif transport_type == "sse":
                 if not await _probe_http_url(cfg.url):
-                    logger.warning("MCP server '{}': {} unreachable, skipping", name, cfg.url)
+                    logger.warning("MCP server '{}': {} unreachable, skipping", name, _redact_url(cfg.url))
                     await server_stack.aclose()
                     return name, None
 
@@ -731,7 +841,7 @@ async def connect_mcp_servers(
                 )
             elif transport_type == "streamableHttp":
                 if not await _probe_http_url(cfg.url):
-                    logger.warning("MCP server '{}': {} unreachable, skipping", name, cfg.url)
+                    logger.warning("MCP server '{}': {} unreachable, skipping", name, _redact_url(cfg.url))
                     await server_stack.aclose()
                     return name, None
 

nanobot/agent/tools/self.py

@@ -438,6 +438,9 @@ class MyTool(Tool, ContextAware):
         setattr(self._runtime_state, key, value)
         if key == "model":
             self._runtime_state._active_preset = None
+        sync_replay = getattr(self._runtime_state, "_sync_replay_max_messages", None)
+        if key == "context_window_tokens" and callable(sync_replay):
+            sync_replay()
         if key == "max_iterations" and hasattr(self._runtime_state, "_sync_subagent_runtime_limits"):
             self._runtime_state._sync_subagent_runtime_limits()
         self._audit("modify", f"{key}: {old!r} -> {value!r}")

nanobot/agent/tools/spawn.py

@@ -63,6 +63,9 @@ class SpawnTool(Tool, ContextAware):
         return (
             "Spawn a subagent to handle a task in the background. "
             "Use this for complex or time-consuming tasks that can run independently. "
+            "For MapReduce-style work, spawn only independent map slices with clear "
+            "boundaries; keep reduction, conflict resolution, and final user-facing "
+            "synthesis in the main agent. "
             "The subagent will complete the task and report back when done. "
             "For deliverables or existing projects, inspect the workspace first "
             "and use a dedicated subdirectory when helpful."

nanobot/channels/manager.py

@@ -396,7 +396,7 @@ class ChannelManager:
     def _coalesce_stream_deltas(
         self, first_msg: OutboundMessage
     ) -> tuple[OutboundMessage, list[OutboundMessage]]:
-        """Merge consecutive _stream_delta messages for the same (channel, chat_id).
+        """Merge consecutive _stream_delta messages for the same (channel, chat_id, _stream_id).
 
         This reduces the number of API calls when the queue has accumulated multiple
         deltas, which happens when LLM generates faster than the channel can process.
@@ -404,7 +404,8 @@ class ChannelManager:
         Returns:
             tuple of (merged_message, list_of_non_matching_messages)
         """
-        target_key = (first_msg.channel, first_msg.chat_id)
+        first_metadata = first_msg.metadata or {}
+        target_key = (first_msg.channel, first_msg.chat_id, first_metadata.get("_stream_id"))
         combined_content = first_msg.content
         final_metadata = dict(first_msg.metadata or {})
         non_matching: list[OutboundMessage] = []
@@ -418,9 +419,14 @@ class ChannelManager:
                 break
 
             # Check if this message belongs to the same stream
-            same_target = (next_msg.channel, next_msg.chat_id) == target_key
-            is_delta = next_msg.metadata and next_msg.metadata.get("_stream_delta")
-            is_end = next_msg.metadata and next_msg.metadata.get("_stream_end")
+            next_metadata = next_msg.metadata or {}
+            same_target = (
+                next_msg.channel,
+                next_msg.chat_id,
+                next_metadata.get("_stream_id"),
+            ) == target_key
+            is_delta = next_metadata.get("_stream_delta")
+            is_end = next_metadata.get("_stream_end")
 
             if same_target and is_delta and not final_metadata.get("_stream_end"):
                 # Accumulate content

nanobot/channels/weixin.py

@@ -129,6 +129,13 @@ class WeixinConfig(Base):
     token: str = ""  # Manually set token, or obtained via QR login
     state_dir: str = ""  # Default: ~/.nanobot/weixin/
     poll_timeout: int = DEFAULT_LONG_POLL_TIMEOUT_S  # seconds for long-poll
+    # Default on: WeChat iLink has no native incremental delivery (send_delta is
+    # buffered and the final answer is still sent in one shot), so streaming has
+    # zero user-facing effect here — it only switches the LLM call to the
+    # streaming API. That avoids upstream Anthropic relays that drop tool_use
+    # id/name/input on the non-streaming Messages path (a common third-party
+    # relay bug). Set to false only if a relay's streaming/SSE path is broken.
+    streaming: bool = True
 
 
 class WeixinChannel(BaseChannel):
@@ -167,6 +174,10 @@ class WeixinChannel(BaseChannel):
         self._typing_tickets: dict[str, dict[str, Any]] = {}
         self._context_token_at: dict[str, float] = {}
         self._pending_tool_hints: dict[str, list[str]] = {}
+        # Buffers streamed content deltas per chat. WeChat iLink has no native
+        # incremental delivery, so when streaming is enabled we accumulate the
+        # deltas and flush the full reply in one shot at _stream_end.
+        self._stream_buffers: dict[str, list[str]] = {}
 
     # ------------------------------------------------------------------
     # State persistence
@@ -1223,14 +1234,38 @@ class WeixinChannel(BaseChannel):
     async def send_delta(
         self, chat_id: str, delta: str, metadata: dict[str, Any] | None = None
     ) -> None:
-        """Weixin iLink does not support native streaming deltas.
+        """Deliver a streamed reply to WeChat.
 
-        We only hook ``_stream_end`` so buffered tool hints are flushed even
-        when the final answer carries the ``_streamed`` flag and bypasses
-        :meth:`send`.
+        WeChat iLink has no native incremental delivery, and the manager
+        bypasses :meth:`send` for the ``_streamed`` final answer. So we
+        accumulate the content deltas here and flush the full reply as a
+        single message at ``_stream_end`` — otherwise a streamed reply would
+        never reach the user. Reasoning deltas are invisible in WeChat and are
+        dropped.
         """
-        if metadata and metadata.get("_stream_end"):
-            await self._flush_tool_hints(chat_id)
+        meta = metadata or {}
+        if meta.get("_reasoning_delta") or meta.get("_reasoning"):
+            return
+        is_end = meta.get("_stream_end")
+        # Accumulate intermediate deltas. The _stream_end message's own content
+        # (present when the manager coalesces deltas into the end message) is
+        # folded into `full` below instead of appended here, so a send retry
+        # recomputes the same `full` from an unchanged buffer rather than
+        # double-counting that delta.
+        if delta and not is_end:
+            self._stream_buffers.setdefault(chat_id, []).append(delta)
+        if not is_end:
+            return
+        full = ("".join(self._stream_buffers.get(chat_id, [])) + (delta or "")).strip()
+        await self._flush_tool_hints(chat_id)
+        if full:
+            # Send before clearing the buffer: if the send raises, the buffer is
+            # left intact so ChannelManager._send_with_retry can re-deliver the
+            # same _stream_end message instead of silently losing the reply.
+            await self.send(
+                OutboundMessage(channel=self.name, chat_id=chat_id, content=full)
+            )
+        self._stream_buffers.pop(chat_id, None)
 
     async def _start_typing(self, chat_id: str, context_token: str = "") -> None:
         """Start typing indicator immediately when a message is received."""

nanobot/channels/whatsapp.py

@@ -499,6 +499,30 @@ class WhatsAppChannel(BaseChannel):
                 self._self_jids.add(jid)
                 self._self_jids.add(_bare_jid(jid))
 
+    async def _send_read_receipt(self, client: Any, source: Any, message_id: str) -> None:
+        """Send a read receipt (blue double-check) for an incoming message.
+
+        Best-effort: any failure is logged at debug level and swallowed so it
+        never blocks message processing.
+        """
+        if not message_id:
+            return
+        try:
+            from neonize.utils.enum import ReceiptType
+
+            chat = _safe_attr(source, "Chat")
+            sender = _safe_attr(source, "Sender")
+            if chat is None or sender is None:
+                return
+            await client.mark_read(
+                message_id,
+                chat=chat,
+                sender=sender,
+                receipt=ReceiptType.READ,
+            )
+        except Exception as exc:  # noqa: BLE001 - read receipt is best-effort
+            self.logger.debug("Failed to send WhatsApp read receipt: {}", exc)
+
     async def _handle_neonize_message(self, client: Any, event: Any) -> None:
         info = _safe_attr(event, "Info")
         message = _safe_attr(event, "Message")
@@ -532,6 +556,9 @@ class WhatsAppChannel(BaseChannel):
             while len(self._processed_message_ids) > 1000:
                 self._processed_message_ids.popitem(last=False)
 
+        # Mark the incoming message as read (blue double-check). Best-effort.
+        await self._send_read_receipt(client, source, message_id)
+
         participant_jid = _normalize_jid(_safe_attr(source, "Sender"))
         sender_alt_jid = _normalize_jid(_safe_attr(source, "SenderAlt"))
         sender_candidates = [sender_alt_jid, participant_jid]

nanobot/cli/commands.py

@@ -1744,6 +1744,11 @@ _PROVIDER_DISPLAY: dict[str, str] = {
     "github_copilot": "GitHub Copilot",
 }
 
+_OAUTH_PROVIDER_DEFAULT_MODELS: dict[str, str] = {
+    "openai_codex": "openai-codex/gpt-5.4-mini",
+    "github_copilot": "github-copilot/gpt-5.4-mini",
+}
+
 
 def _register_login(name: str):
     """Register an OAuth login handler."""
@@ -1775,9 +1780,51 @@ def _resolve_oauth_provider(provider: str):
     return spec
 
 
+def _set_oauth_provider_as_main(
+    provider_name: str,
+    *,
+    model: str | None = None,
+    config_path: str | None = None,
+) -> None:
+    """Persist an OAuth provider as the active agent provider."""
+    from nanobot.config.loader import get_config_path, load_config, save_config, set_config_path
+
+    resolved_config_path = Path(config_path).expanduser().resolve() if config_path else None
+    if resolved_config_path is not None:
+        set_config_path(resolved_config_path)
+        console.print(f"[dim]Using config: {resolved_config_path}[/dim]")
+
+    config = load_config(resolved_config_path)
+    selected_model = (model or "").strip() or _OAUTH_PROVIDER_DEFAULT_MODELS[provider_name]
+    config.agents.defaults.model_preset = None
+    config.agents.defaults.provider = provider_name
+    config.agents.defaults.model = selected_model
+    save_config(config, resolved_config_path)
+
+    saved_path = resolved_config_path or get_config_path()
+    console.print(
+        f"[green]✓ Set {provider_name.replace('_', '-')} as the main provider[/green]  "
+        f"[dim]{selected_model}[/dim]"
+    )
+    console.print(f"[dim]Saved: {saved_path}[/dim]")
+
+
 @provider_app.command("login")
 def provider_login(
     provider: str = typer.Argument(..., help="OAuth provider (e.g. 'openai-codex', 'github-copilot')"),
+    set_main: bool = typer.Option(
+        False,
+        "--set-main",
+        "--main",
+        help="Set this OAuth provider as the active agent provider after login",
+    ),
+    model: str | None = typer.Option(
+        None,
+        "--model",
+        "-m",
+        help="Model to use when setting this provider as the active provider",
+    ),
+    config: str | None = typer.Option(None, "--config", "-c", help="Path to config file"),
 ):
     """Authenticate with an OAuth provider."""
     spec = _resolve_oauth_provider(provider)
@@ -1789,6 +1836,8 @@ def provider_login(
 
     console.print(f"{__logo__} OAuth Login - {spec.label}\n")
     handler()
+    if set_main or model:
+        _set_oauth_provider_as_main(spec.name, model=model, config_path=config)
 
 
 @provider_app.command("logout")
@@ -1812,14 +1861,23 @@ def _login_openai_codex() -> None:
     try:
         from oauth_cli_kit import get_token, login_oauth_interactive
 
+        from nanobot.config.loader import load_config, resolve_config_env_vars
+
+        proxy = None
+        try:
+            proxy = resolve_config_env_vars(load_config()).providers.openai_codex.proxy or None
+        except ValueError as e:
+            console.print(f"[red]{e}[/red]")
+            raise typer.Exit(1) from e
         token = None
         with suppress(Exception):
-            token = get_token()
+            token = get_token(proxy=proxy)
         if not (token and token.access):
             console.print("[cyan]Starting interactive OAuth login...[/cyan]\n")
             token = login_oauth_interactive(
                 print_fn=lambda s: console.print(s),
                 prompt_fn=lambda s: typer.prompt(s),
+                proxy=proxy,
             )
         if not (token and token.access):
             console.print("[red]✗ Authentication failed[/red]")

nanobot/command/builtin.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 
 import asyncio
 import os
+import subprocess
 import sys
 import time
 from contextlib import suppress
@@ -50,7 +51,7 @@ BUILTIN_COMMAND_SPECS: tuple[BuiltinCommandSpec, ...] = (
     BuiltinCommandSpec(
         "/restart",
         "Restart nanobot",
-        "Restart the bot process in place.",
+        "Restart the bot process.",
         "rotate-cw",
     ),
     BuiltinCommandSpec(
@@ -130,6 +131,15 @@ async def cmd_stop(ctx: CommandContext) -> OutboundMessage:
     loop = ctx.loop
     msg = ctx.msg
     total = await loop._cancel_active_tasks(ctx.key)
+    # Also drain pending queue to prevent mid-turn injection deadlock
+    pending = loop._pending_queues.pop(ctx.key, None)
+    if pending is not None:
+        while not pending.empty():
+            try:
+                pending.get_nowait()
+                total += 1
+            except Exception:
+                break
     content = f"Stopped {total} task(s)." if total else "No active task to stop."
     return OutboundMessage(
         channel=msg.channel, chat_id=msg.chat_id, content=content,
@@ -138,7 +148,7 @@ async def cmd_stop(ctx: CommandContext) -> OutboundMessage:
 
 
 async def cmd_restart(ctx: CommandContext) -> OutboundMessage:
-    """Restart the process in-place via os.execv."""
+    """Restart the process."""
     msg = ctx.msg
     set_restart_notice_to_env(
         channel=msg.channel,
@@ -148,7 +158,19 @@ async def cmd_restart(ctx: CommandContext) -> OutboundMessage:
 
     async def _do_restart():
         await asyncio.sleep(1)
-        os.execv(sys.executable, [sys.executable, "-m", "nanobot"] + sys.argv[1:])
+        argv = [sys.executable, "-m", "nanobot"] + sys.argv[1:]
+        mode = getattr(ctx.loop, "restart_mode", "auto") or "auto"
+        if mode == "auto":
+            mode = "spawn" if sys.platform == "win32" else "exec"
+        if mode == "exec":
+            os.execv(sys.executable, argv)
+            return
+        if mode == "spawn":
+            kwargs = {}
+            if sys.platform == "win32":
+                kwargs["creationflags"] = subprocess.CREATE_NEW_PROCESS_GROUP
+            subprocess.Popen(argv, **kwargs)
+        os._exit(0)
 
     asyncio.create_task(_do_restart())
     return OutboundMessage(

nanobot/config/loader.py

@@ -7,6 +7,7 @@ from pathlib import Path
 from typing import Any
 
 import pydantic
+from loguru import logger
 from pydantic import BaseModel
 
 from nanobot.config.schema import Config, _resolve_tool_config_refs
@@ -79,6 +80,10 @@ def save_config(config: Config, config_path: Path | None = None) -> None:
     path.parent.mkdir(parents=True, exist_ok=True)
 
     data = config.model_dump(mode="json", by_alias=True)
+    if config.providers.openai_codex.proxy is not None:
+        data.setdefault("providers", {})["openaiCodex"] = {
+            "proxy": config.providers.openai_codex.proxy,
+        }
 
     with open(path, "w", encoding="utf-8") as f:
         json.dump(data, f, indent=2, ensure_ascii=False)
@@ -152,6 +157,23 @@ def _env_replace(match: re.Match[str]) -> str:
 
 def _migrate_config(data: dict) -> dict:
     """Migrate old config formats to current."""
+    agents = data.get("agents", {})
+    defaults = agents.get("defaults", {}) if isinstance(agents, dict) else {}
+    if isinstance(defaults, dict):
+        had_legacy_max_messages = (
+            "maxMessages" in defaults or "max_messages" in defaults
+        )
+        defaults.pop("maxMessages", None)
+        defaults.pop("max_messages", None)
+        if had_legacy_max_messages:
+            # TODO(next version): Remove this legacy cleanup branch; the schema
+            # will silently ignore this field once the warning grace period ends.
+            logger.warning(
+                "agents.defaults.maxMessages/max_messages is legacy and ignored; "
+                "replay max messages is now an internal safety cap. Remove it from "
+                "config. This compatibility warning will be removed in the next version."
+            )
+
     # Move tools.exec.restrictToWorkspace → tools.restrictToWorkspace
     tools = data.get("tools", {})
     exec_cfg = tools.get("exec", {})

nanobot/config/schema.py

@@ -154,10 +154,6 @@ class AgentDefaults(Base):
         validation_alias=AliasChoices("idleCompactAfterMinutes", "sessionTtlMinutes"),
         serialization_alias="idleCompactAfterMinutes",
     )  # Auto-compact idle threshold in minutes (0 = disabled)
-    max_messages: int = Field(
-        default=120,
-        ge=0,
-    )  # Max messages to replay from session history (0 = use default 120, respects token budget)
     consolidation_ratio: float = Field(
         default=0.5,
         ge=0.1,
@@ -183,6 +179,7 @@ class ProviderConfig(Base):
     extra_headers: dict[str, str] | None = None  # Custom headers (e.g. APP-Code for AiHubMix)
     extra_body: dict[str, Any] | None = None  # Extra provider request fields; shape depends on provider/API surface
     extra_query: dict[str, str] | None = None  # Extra query params (e.g. api-version for Azure-style gateways)
+    proxy: str | None = None  # OpenAI-compatible/Codex HTTP proxy URL
     thinking_style: str | None = None  # Thinking/reasoning style for custom providers
 
     # Valid values mirror the keys of _THINKING_STYLE_MAP in
@@ -317,6 +314,7 @@ class GatewayConfig(Base):
 
     host: str = "127.0.0.1"  # Safer default: local-only bind.
     port: int = 18790
+    restart_mode: Literal["auto", "exec", "spawn", "exit"] = "auto"
     heartbeat: HeartbeatConfig = Field(default_factory=HeartbeatConfig)
 
 

nanobot/cron/service.py

@@ -357,6 +357,25 @@ class CronService:
 
         return self._store
 
+    def _require_store(self) -> CronStore:
+        """Return a usable store or raise a clear error.
+
+        ``_load_store`` deliberately returns ``None`` when the first load sees
+        a corrupt on-disk store and no previous in-memory snapshot exists.  The
+        public API requires a concrete store object before touching
+        ``store.jobs``; raising here keeps callers from seeing an accidental
+        ``AttributeError`` and, more importantly, prevents follow-up saves from
+        treating a corrupt store as an empty one.
+        """
+        store = self._load_store()
+        if store is None:
+            raise RuntimeError(
+                f"cron store at {self.store_path} could not be loaded and was preserved "
+                "as a .corrupt-<ts> backup; refusing to operate to avoid overwriting "
+                "scheduled jobs. Inspect the corrupt backup and restore jobs.json manually."
+            )
+        return store
+
     def _save_store(self) -> None:
         """Save jobs to disk."""
         if not self._store:
@@ -622,7 +641,7 @@ class CronService:
 
     def list_jobs(self, include_disabled: bool = False) -> list[CronJob]:
         """List all jobs."""
-        store = self._load_store()
+        store = self._require_store()
         jobs = store.jobs if include_disabled else [j for j in store.jobs if j.enabled]
         return sorted(jobs, key=lambda j: j.state.next_run_at_ms or float('inf'))
 
@@ -684,7 +703,7 @@ class CronService:
         _normalize_agent_turn_job(job)
         self._enforce_agent_binding(job)
         if self._running:
-            store = self._load_store()
+            store = self._require_store()
             store.jobs.append(job)
             self._save_store()
             self._arm_timer()
@@ -696,7 +715,7 @@ class CronService:
 
     def register_system_job(self, job: CronJob) -> CronJob:
         """Register an internal system job (idempotent on restart)."""
-        store = self._load_store()
+        store = self._require_store()
         now = _now_ms()
         job.state = CronJobState(next_run_at_ms=_compute_next_run(job.schedule, now))
         job.created_at_ms = now
@@ -710,7 +729,7 @@ class CronService:
 
     def remove_job(self, job_id: str) -> Literal["removed", "protected", "not_found"]:
         """Remove a job by ID, unless it is a protected system job."""
-        store = self._load_store()
+        store = self._require_store()
         job = next((j for j in store.jobs if j.id == job_id), None)
         if job is None:
             return "not_found"
@@ -735,7 +754,7 @@ class CronService:
 
     def enable_job(self, job_id: str, enabled: bool = True) -> CronJob | None:
         """Enable or disable a job."""
-        store = self._load_store()
+        store = self._require_store()
         for job in store.jobs:
             if job.id == job_id:
                 job.enabled = enabled
@@ -770,7 +789,7 @@ class CronService:
         For ``channel`` and ``to``, pass an explicit value (including ``None``)
         to update; omit (sentinel ``...``) to leave unchanged.
         """
-        store = self._load_store()
+        store = self._require_store()
         job = next((j for j in store.jobs if j.id == job_id), None)
         if job is None:
             return "not_found"
@@ -815,7 +834,7 @@ class CronService:
         was_running = self._running
         self._running = True
         try:
-            store = self._load_store()
+            store = self._require_store()
             for job in store.jobs:
                 if job.id == job_id:
                     if self._is_unbound_agent_job(job):
@@ -835,12 +854,12 @@ class CronService:
 
     def get_job(self, job_id: str) -> CronJob | None:
         """Get a job by ID."""
-        store = self._load_store()
+        store = self._require_store()
         return next((j for j in store.jobs if j.id == job_id), None)
 
     def status(self) -> dict:
         """Get service status."""
-        store = self._load_store()
+        store = self._require_store()
         return {
             "enabled": self._running,
             "jobs": len(store.jobs),

nanobot/providers/anthropic_provider.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 
 import asyncio
 import hashlib
+import json
 import re
 import secrets
 import string
@@ -275,7 +276,19 @@ class AnthropicProvider(LLMProvider):
             blocks.append({"type": "text", "text": content})
         elif isinstance(content, list):
             for item in content:
-                blocks.append(item if isinstance(item, dict) else {"type": "text", "text": str(item)})
+                if isinstance(item, dict):
+                    if not item.get("type"):
+                        # Anthropic requires every content block to declare a "type".
+                        # A tool that returned a bare dict lands here; coerce it to
+                        # a text block instead of emitting one that the API rejects.
+                        blocks.append({
+                            "type": "text",
+                            "text": AnthropicProvider._stringify_typeless_block(item),
+                        })
+                    else:
+                        blocks.append(item)
+                else:
+                    blocks.append({"type": "text", "text": str(item)})
 
         for tc in msg.get("tool_calls") or []:
             if not isinstance(tc, dict):
@@ -315,11 +328,18 @@ class AnthropicProvider(LLMProvider):
                 # A tool that returned a bare dict (or a list of dicts) lands
                 # here; coerce it to a text block instead of emitting a block
                 # the API rejects with "content.0.type: Field required".
-                result.append({"type": "text", "text": str(item)})
+                result.append({
+                    "type": "text",
+                    "text": AnthropicProvider._stringify_typeless_block(item),
+                })
                 continue
             result.append(item)
         return result or "(empty)"
 
+    @staticmethod
+    def _stringify_typeless_block(block: dict[str, Any]) -> str:
+        return json.dumps(block, ensure_ascii=False, sort_keys=True, default=str)
+
     @staticmethod
     def _convert_image_block(block: dict[str, Any]) -> dict[str, Any] | None:
         """Convert OpenAI image_url block to Anthropic image block."""

nanobot/providers/base.py

@@ -54,6 +54,18 @@ class ToolCallRequest:
     provider_specific_fields: dict[str, Any] | None = None
     function_provider_specific_fields: dict[str, Any] | None = None
 
+    def has_valid_name(self) -> bool:
+        """Whether this call carries a usable (non-empty string) tool name.
+
+        ToolCallRequest.name is typed ``str`` but not enforced at runtime: a
+        model/gateway can emit a degenerate call with ``name=None`` or ``""``.
+        Such a call cannot be executed and, if persisted and replayed, makes
+        upstream APIs reject the whole request (e.g. Anthropic-style
+        ``messages.content.N.tool_use.name: Input should be a valid string``),
+        which permanently wedges the session.
+        """
+        return isinstance(self.name, str) and bool(self.name)
+
     def to_openai_tool_call(self) -> dict[str, Any]:
         """Serialize to an OpenAI-style tool_call payload."""
         arguments = (

nanobot/providers/factory.py

@@ -58,6 +58,11 @@ def _make_provider_core(
     if spec and spec.is_transcription_only:
         raise ValueError(f"Provider '{provider_name}' only supports transcription.")
     backend = spec.backend if spec else "openai_compat"
+    if p and p.proxy and backend not in {"openai_compat", "openai_codex"}:
+        raise ValueError(
+            f"providers.{provider_name}.proxy is only supported for "
+            "OpenAI-compatible providers and OpenAI Codex."
+        )
 
     if backend == "azure_openai":
         if not p or not p.api_base:
@@ -79,7 +84,10 @@ def _make_provider_core(
     if backend == "openai_codex":
         from nanobot.providers.openai_codex_provider import OpenAICodexProvider
 
-        provider = OpenAICodexProvider(default_model=model)
+        provider = OpenAICodexProvider(
+            default_model=model,
+            proxy=getattr(p, "proxy", None) if p else None,
+        )
     elif backend == "azure_openai":
         from nanobot.providers.azure_openai_provider import AzureOpenAIProvider
 
@@ -124,6 +132,7 @@ def _make_provider_core(
             extra_body=p.extra_body if p else None,
             api_type=p.api_type if p and provider_name == "openai" else "auto",
             extra_query=p.extra_query if p else None,
+            proxy=p.proxy if p else None,
         )
 
     provider.generation = resolved.to_generation_settings()
@@ -218,6 +227,7 @@ def provider_signature(
             fallback.temperature,
             fallback.reasoning_effort,
             fallback.context_window_tokens,
+            getattr(fp, "proxy", None) if fp else None,
         )
 
     provider_name = config.get_provider_name(resolved.model, preset=resolved)
@@ -237,6 +247,7 @@ def provider_signature(
         resolved.temperature,
         resolved.reasoning_effort,
         resolved.context_window_tokens,
+        getattr(p, "proxy", None) if p else None,
         tuple(_fallback_signature(fallback) for fallback in fallback_presets),
     )
 

nanobot/providers/github_copilot_provider.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import os
 import time
 import webbrowser
 from collections.abc import Awaitable, Callable
@@ -29,6 +30,12 @@ _EXPIRY_SKEW_SECONDS = 60
 _LONG_LIVED_TOKEN_SECONDS = 315360000
 
 
+def _resolve(env_var: str, default: str) -> str:
+    """Allow GitHub Enterprise / Copilot for Business deployments to override defaults via env."""
+    value = os.environ.get(env_var)
+    return value.strip() if value and value.strip() else default
+
+
 def get_storage() -> FileTokenStorage:
     return FileTokenStorage(
         token_filename=TOKEN_FILENAME,
@@ -68,11 +75,16 @@ def login_github_copilot(
     printer = print_fn or print
     timeout = httpx.Timeout(20.0, connect=20.0)
 
+    client_id = _resolve("NANOBOT_GITHUB_COPILOT_CLIENT_ID", GITHUB_COPILOT_CLIENT_ID)
+    device_code_url = _resolve("NANOBOT_GITHUB_DEVICE_CODE_URL", DEFAULT_GITHUB_DEVICE_CODE_URL)
+    access_token_url = _resolve("NANOBOT_GITHUB_ACCESS_TOKEN_URL", DEFAULT_GITHUB_ACCESS_TOKEN_URL)
+    user_url = _resolve("NANOBOT_GITHUB_USER_URL", DEFAULT_GITHUB_USER_URL)
+
     with httpx.Client(timeout=timeout, follow_redirects=True, trust_env=True) as client:
         response = client.post(
-            DEFAULT_GITHUB_DEVICE_CODE_URL,
+            device_code_url,
             headers={"Accept": "application/json", "User-Agent": USER_AGENT},
-            data={"client_id": GITHUB_COPILOT_CLIENT_ID, "scope": GITHUB_COPILOT_SCOPE},
+            data={"client_id": client_id, "scope": GITHUB_COPILOT_SCOPE},
         )
         response.raise_for_status()
         payload = response.json()
@@ -96,10 +108,10 @@ def login_github_copilot(
         token_expires_in = _LONG_LIVED_TOKEN_SECONDS
         while time.time() < deadline:
             poll = client.post(
-                DEFAULT_GITHUB_ACCESS_TOKEN_URL,
+                access_token_url,
                 headers={"Accept": "application/json", "User-Agent": USER_AGENT},
                 data={
-                    "client_id": GITHUB_COPILOT_CLIENT_ID,
+                    "client_id": client_id,
                     "device_code": device_code,
                     "grant_type": "urn:ietf:params:oauth:grant-type:device_code",
                 },
@@ -132,7 +144,7 @@ def login_github_copilot(
             raise RuntimeError("GitHub device flow timed out.")
 
         user = client.get(
-            DEFAULT_GITHUB_USER_URL,
+            user_url,
             headers={
                 "Authorization": f"Bearer {access_token}",
                 "Accept": "application/vnd.github+json",
@@ -164,7 +176,7 @@ class GitHubCopilotProvider(OpenAICompatProvider):
         self._copilot_expires_at: float = 0.0
         super().__init__(
             api_key="no-key",
-            api_base=DEFAULT_COPILOT_BASE_URL,
+            api_base=_resolve("NANOBOT_COPILOT_BASE_URL", DEFAULT_COPILOT_BASE_URL),
             default_model=default_model,
             extra_headers={
                 "Editor-Version": EDITOR_VERSION,
@@ -186,7 +198,7 @@ class GitHubCopilotProvider(OpenAICompatProvider):
         timeout = httpx.Timeout(20.0, connect=20.0)
         async with httpx.AsyncClient(timeout=timeout, follow_redirects=True, trust_env=True) as client:
             response = await client.get(
-                DEFAULT_COPILOT_TOKEN_URL,
+                _resolve("NANOBOT_COPILOT_TOKEN_URL", DEFAULT_COPILOT_TOKEN_URL),
                 headers=_copilot_headers(github_token.access),
             )
             response.raise_for_status()

nanobot/providers/openai_codex_provider.py

@@ -33,9 +33,14 @@ class OpenAICodexProvider(LLMProvider):
 
     supports_progress_deltas = True
 
-    def __init__(self, default_model: str = "openai-codex/gpt-5.1-codex"):
+    def __init__(
+        self,
+        default_model: str = "openai-codex/gpt-5.1-codex",
+        proxy: str | None = None,
+    ):
         super().__init__(api_key=None, api_base=None)
         self.default_model = default_model
+        self.proxy = proxy or None
 
     async def _call_codex(
         self,
@@ -52,9 +57,6 @@ class OpenAICodexProvider(LLMProvider):
         model = model or self.default_model
         system_prompt, input_items = convert_messages(messages)
 
-        token = await asyncio.to_thread(get_codex_token)
-        headers = _build_headers(token.account_id, token.access)
-
         body: dict[str, Any] = {
             "model": _strip_model_prefix(model),
             "store": False,
@@ -74,9 +76,13 @@ class OpenAICodexProvider(LLMProvider):
             body["tools"] = convert_tools(tools)
 
         try:
+            token = await asyncio.to_thread(get_codex_token, proxy=self.proxy)
+            headers = _build_headers(token.account_id, token.access)
+
             try:
                 content, tool_calls, finish_reason, usage, reasoning_content = await _request_codex(
                     DEFAULT_CODEX_URL, headers, body, verify=True,
+                    proxy=self.proxy,
                     on_content_delta=on_content_delta,
                     on_thinking_delta=on_thinking_delta,
                     on_tool_call_delta=on_tool_call_delta,
@@ -87,6 +93,7 @@ class OpenAICodexProvider(LLMProvider):
                 logger.warning("SSL verification failed for Codex API; retrying with verify=False")
                 content, tool_calls, finish_reason, usage, reasoning_content = await _request_codex(
                     DEFAULT_CODEX_URL, headers, body, verify=False,
+                    proxy=self.proxy,
                     on_content_delta=on_content_delta,
                     on_thinking_delta=on_thinking_delta,
                     on_tool_call_delta=on_tool_call_delta,
@@ -199,12 +206,17 @@ async def _request_codex(
     headers: dict[str, str],
     body: dict[str, Any],
     verify: bool,
+    proxy: str | None = None,
     on_content_delta: Callable[[str], Awaitable[None]] | None = None,
     on_thinking_delta: Callable[[str], Awaitable[None]] | None = None,
     on_tool_call_delta: Callable[[dict[str, Any]], Awaitable[None]] | None = None,
 ) -> tuple[str, list[ToolCallRequest], str, dict[str, int], str | None]:
     idle_timeout_s = resolve_stream_idle_timeout_s()
-    async with httpx.AsyncClient(timeout=idle_timeout_s, verify=verify) as client:
+    client_kwargs: dict[str, Any] = {"timeout": idle_timeout_s, "verify": verify}
+    if proxy:
+        client_kwargs["proxy"] = proxy
+        client_kwargs["trust_env"] = False
+    async with httpx.AsyncClient(**client_kwargs) as client:
         async with client.stream("POST", url, headers=headers, json=body) as response:
             if response.status_code != 200:
                 text = await response.aread()

nanobot/providers/openai_compat_provider.py

@@ -358,6 +358,7 @@ class OpenAICompatProvider(LLMProvider):
         extra_body: dict[str, Any] | None = None,
         api_type: str = "auto",
         extra_query: dict[str, str] | None = None,
+        proxy: str | None = None,
     ):
         super().__init__(api_key, api_base)
         self.default_model = default_model
@@ -366,6 +367,7 @@ class OpenAICompatProvider(LLMProvider):
         self._extra_body = extra_body or {}
         self._api_type = api_type if spec and spec.name == "openai" else "auto"
         self._extra_query = extra_query or {}
+        self._proxy = proxy or None
 
         if api_key and spec and spec.env_key:
             self._setup_env(api_key, api_base)
@@ -396,7 +398,14 @@ class OpenAICompatProvider(LLMProvider):
 
         timeout_s = _openai_compat_timeout_s()
         http_client: httpx.AsyncClient | None = None
-        if self._is_local:
+        if self._proxy:
+            http_client = httpx.AsyncClient(
+                timeout=timeout_s,
+                proxy=self._proxy,
+                trust_env=False,
+                follow_redirects=True,
+            )
+        elif self._is_local:
             # Local model servers (Ollama, llama.cpp, vLLM) often close idle
             # HTTP connections before the client-side keepalive expires. When
             # two LLM calls happen seconds apart (e.g. heartbeat _decide then
@@ -1131,14 +1140,21 @@ class OpenAICompatProvider(LLMProvider):
                 if reasoning_content is None:
                     reasoning_content = m.get("reasoning_content")
 
+            # Deduplicate tool call IDs (same pattern as streaming path)
+            # Some providers reuse the same ID for parallel tool calls.
+            _seen_tc_ids: set[str] = set()
             parsed_tool_calls = []
             for tc in raw_tool_calls:
                 tc_map = self._maybe_mapping(tc) or {}
                 fn = self._maybe_mapping(tc_map.get("function")) or {}
                 args = parse_tool_arguments(fn.get("arguments", {}))
                 ec, prov, fn_prov = _extract_tc_extras(tc)
+                raw_id = str(tc_map.get("id") or _short_tool_id())
+                if not raw_id or raw_id in _seen_tc_ids:
+                    raw_id = _short_tool_id()
+                _seen_tc_ids.add(raw_id)
                 parsed_tool_calls.append(ToolCallRequest(
-                    id=str(tc_map.get("id") or _short_tool_id()),
+                    id=raw_id,
                     name=str(fn.get("name") or ""),
                     arguments=args,
                     extra_content=ec,

nanobot/session/manager.py

@@ -1,5 +1,6 @@
 """Session management for conversation history."""
 
+import base64
 import json
 import os
 import re
@@ -26,6 +27,8 @@ from nanobot.utils.helpers import (
 from nanobot.utils.subagent_channel_display import scrub_subagent_announce_body
 
 FILE_MAX_MESSAGES = 2000
+MIN_REPLAY_MAX_MESSAGES = 120
+REPLAY_TOKENS_PER_MESSAGE = 100
 _MESSAGE_TIME_PREFIX_RE = re.compile(r"^\[Message Time: [^\]]+\]\n?")
 _LOCAL_IMAGE_BREADCRUMB_RE = re.compile(r"^\[image: (?:/|~)[^\]]+\]\s*$")
 _TOOL_CALL_ECHO_RE = re.compile(r'^\s*(?:generate_image|message)\([^)]*\)\s*$')
@@ -42,6 +45,15 @@ _FORK_VOLATILE_METADATA_KEYS = {
 }
 
 
+def replay_max_messages_for_context(context_window_tokens: int | None) -> int:
+    if not context_window_tokens or context_window_tokens <= 0:
+        return FILE_MAX_MESSAGES
+    return min(
+        FILE_MAX_MESSAGES,
+        max(MIN_REPLAY_MAX_MESSAGES, context_window_tokens // REPLAY_TOKENS_PER_MESSAGE),
+    )
+
+
 def _sanitize_assistant_replay_text(content: str) -> str:
     """Remove internal replay artifacts that the model may have copied before.
 
@@ -98,6 +110,12 @@ def _metadata_title(metadata: Any) -> str:
     return strip_think(title)
 
 
+@dataclass
+class RetentionResult:
+    dropped: list[dict]
+    already_consolidated_count: int
+
+
 @dataclass
 class Session:
     """A conversation session."""
@@ -131,7 +149,7 @@ class Session:
 
     def get_history(
         self,
-        max_messages: int = 120,
+        max_messages: int = FILE_MAX_MESSAGES,
         *,
         max_tokens: int = 0,
         extend_to_user: bool = False,
@@ -142,7 +160,7 @@ class Session:
         token budget from the tail (``max_tokens``) when provided.
         """
         unconsolidated = self.messages[self.last_consolidated:]
-        max_messages = max_messages if max_messages > 0 else 120
+        max_messages = max_messages if max_messages > 0 else FILE_MAX_MESSAGES
         start_idx = recent_message_start_index(
             unconsolidated,
             max_messages,
@@ -277,22 +295,26 @@ class Session:
         max_messages: int,
         *,
         extend_to_user: bool = False,
-    ) -> tuple[list[dict], int]:
+    ) -> RetentionResult:
         """Keep a legal recent suffix, optionally extending it back to a user turn.
 
-        Returns ``(dropped, already_consolidated_count)`` where *dropped* is
-        the list of removed messages (in original order) and
-        *already_consolidated_count* is how many of those were inside the
-        pre-existing ``last_consolidated`` prefix and therefore do not need
-        raw archiving.
+        Returns a RetentionResult with dropped messages and how many of those
+        were in the already-consolidated prefix. This method mutates
+        self.messages and self.last_consolidated in place.
         """
         if max_messages <= 0:
             dropped = list(self.messages)
             lc = self.last_consolidated
             self.clear()
-            return dropped, min(lc, len(dropped))
+            return RetentionResult(
+                dropped=dropped,
+                already_consolidated_count=min(lc, len(dropped)),
+            )
         if len(self.messages) <= max_messages:
-            return [], 0
+            return RetentionResult(
+                dropped=[],
+                already_consolidated_count=0,
+            )
 
         original = list(self.messages)
         before_lc = self.last_consolidated
@@ -358,7 +380,10 @@ class Session:
         self.messages = retained
         self.last_consolidated = new_lc
         self.updated_at = datetime.now()
-        return dropped, already_consolidated
+        return RetentionResult(
+            dropped=dropped,
+            already_consolidated_count=already_consolidated,
+        )
 
     def enforce_file_cap(
         self,
@@ -369,17 +394,17 @@ class Session:
         if limit <= 0 or len(self.messages) <= limit:
             return
 
-        dropped, already_consolidated = self.retain_recent_legal_suffix(limit)
-        if not dropped:
+        result = self.retain_recent_legal_suffix(limit)
+        if not result.dropped:
             return
 
-        archive_chunk = dropped[already_consolidated:]
+        archive_chunk = result.dropped[result.already_consolidated_count:]
         if archive_chunk and on_archive:
             on_archive(archive_chunk)
         logger.info(
             "Session file cap hit for {}: dropped {}, raw-archived {}, kept {}",
             self.key,
-            len(dropped),
+            len(result.dropped),
             len(archive_chunk),
             len(self.messages),
         )
@@ -403,14 +428,53 @@ class SessionManager:
         """Public helper used by HTTP handlers to map an arbitrary key to a stable filename stem."""
         return safe_filename(key.replace(":", "_"))
 
+    @staticmethod
+    def _storage_key(key: str) -> str:
+        """Collision-resistant encoding for internal session storage filenames."""
+        return base64.urlsafe_b64encode(key.encode()).decode().rstrip("=")
+
+    @staticmethod
+    def _decode_storage_key(stem: str) -> str | None:
+        """Reverse _storage_key(): decode a base64url (no-padding) stem back to the original key."""
+        try:
+            # Restore padding stripped by rstrip("=")
+            padding = 4 - len(stem) % 4
+            if padding != 4:
+                stem += "=" * padding
+            return base64.urlsafe_b64decode(stem).decode("utf-8")
+        except Exception:
+            return None
+
     def _get_session_path(self, key: str) -> Path:
-        """Get the file path for a session."""
-        return self.sessions_dir / f"{self.safe_key(key)}.jsonl"
+        """Get the collision-resistant workspace path for a session."""
+        return self.sessions_dir / f"{self._storage_key(key)}.jsonl"
+
+    def _get_legacy_lossy_path(self, key: str) -> Path:
+        """Previous workspace session path using lossy ':' to '_' replacement."""
+        return self.sessions_dir / f"{safe_filename(key.replace(':', '_'))}.jsonl"
 
     def _get_legacy_session_path(self, key: str) -> Path:
         """Legacy global session path (~/.nanobot/sessions/)."""
         return self.legacy_sessions_dir / f"{self.safe_key(key)}.jsonl"
 
+    @staticmethod
+    def _stored_key_for_path(path: Path) -> str | None:
+        """Read the stored session key from a JSONL metadata row, if present."""
+        try:
+            with open(path, encoding="utf-8") as f:
+                for line in f:
+                    line = line.strip()
+                    if not line:
+                        continue
+                    data = json.loads(line)
+                    if data.get("_type") == "metadata":
+                        stored_key = data.get("key")
+                        return stored_key if isinstance(stored_key, str) else None
+                    return None
+        except Exception:
+            return None
+        return None
+
     def get_or_create(self, key: str) -> Session:
         """
         Get an existing session or create a new one.
@@ -435,13 +499,28 @@ class SessionManager:
         """Load a session from disk."""
         path = self._get_session_path(key)
         if not path.exists():
-            legacy_path = self._get_legacy_session_path(key)
-            if legacy_path.exists():
+            fallback_paths = [
+                (self._get_legacy_lossy_path(key), "legacy lossy path"),
+                (self._get_legacy_session_path(key), "legacy path"),
+            ]
+            for fallback_path, description in fallback_paths:
+                if not fallback_path.exists():
+                    continue
+                stored_key = self._stored_key_for_path(fallback_path)
+                if stored_key and stored_key != key:
+                    logger.info(
+                        "Skipping migration for {} from {} because it belongs to {}",
+                        key,
+                        description,
+                        stored_key,
+                    )
+                    continue
                 try:
-                    shutil.move(str(legacy_path), str(path))
-                    logger.info("Migrated session {} from legacy path", key)
+                    shutil.move(str(fallback_path), str(path))
+                    logger.info("Migrated session {} from {}", key, description)
                 except Exception:
                     logger.exception("Failed to migrate session {}", key)
+                break
 
         if not path.exists():
             return None
@@ -484,9 +563,10 @@ class SessionManager:
                 logger.info("Recovered session {} from corrupt file ({} messages)", key, len(repaired.messages))
             return repaired
 
-    def _repair(self, key: str) -> Session | None:
+    def _repair(self, key: str, *, path: Path | None = None) -> Session | None:
         """Attempt to recover a session from a corrupt JSONL file."""
-        path = self._get_session_path(key)
+        if path is None:
+            path = self._get_session_path(key)
         if not path.exists():
             return None
 
@@ -623,7 +703,11 @@ class SessionManager:
 
         Returns True if at least one JSONL file was found and unlinked.
         """
-        paths = [self._get_session_path(key), self._get_legacy_session_path(key)]
+        paths = [
+            self._get_session_path(key),
+            self._get_legacy_lossy_path(key),
+            self._get_legacy_session_path(key),
+        ]
         self.invalidate(key)
         deleted = False
         for path in paths:
@@ -784,7 +868,8 @@ class SessionManager:
         sessions = []
 
         for path in self.sessions_dir.glob("*.jsonl"):
-            fallback_key = path.stem.replace("_", ":", 1)
+            decoded = self._decode_storage_key(path.stem)
+            fallback_key = decoded or path.stem.replace("_", ":", 1)
             try:
                 # Read the metadata line and a small preview for session lists.
                 with open(path, encoding="utf-8") as f:
@@ -792,7 +877,7 @@ class SessionManager:
                     if first_line:
                         data = json.loads(first_line)
                         if data.get("_type") == "metadata":
-                            key = data.get("key") or path.stem.replace("_", ":", 1)
+                            key = data.get("key") or fallback_key
                             metadata = data.get("metadata", {})
                             title = _metadata_title(metadata)
                             preview = ""
@@ -833,7 +918,7 @@ class SessionManager:
                                 }
                             )
             except Exception:
-                repaired = self._repair(fallback_key)
+                repaired = self._repair(fallback_key, path=path)
                 if repaired is not None:
                     sessions.append(
                         {

nanobot/templates/agent/subagent_announce.md

@@ -5,4 +5,7 @@ Task: {{ task }}
 Result:
 {{ result }}
 
-Summarize this naturally for the user. Keep it brief (1-2 sentences). Do not mention technical details like "subagent" or task IDs.
+Use this result as evidence for the current turn. For MapReduce-style work,
+preserve any Summary / Evidence / Open issues structure when reducing multiple
+results. Mention gaps or failures if they affect the answer; avoid exposing
+internal task IDs unless they are needed for clarity.

nanobot/templates/agent/subagent_system.md

@@ -4,6 +4,15 @@
 
 You are a subagent spawned by the main agent to complete a specific task.
 Stay focused on the assigned task. Your final response will be reported back to the main agent.
+If this task is one slice of a larger MapReduce-style effort, treat yourself as
+the map step: do only the assigned slice, avoid cross-slice coordination, and
+leave reduction or final synthesis to the main agent.
+
+For MapReduce-style slices, end with a compact, mergeable result:
+
+- Summary: what you found or changed
+- Evidence: relevant files, commands, URLs, or observations
+- Open issues: blockers, failures, or "none"
 
 {% include 'agent/_snippets/untrusted_content.md' %}
 

nanobot/utils/file_edit_events.py

@@ -529,6 +529,9 @@ class StreamingFileEditTracker:
         """Keep final start/end events keyed to any earlier streamed placeholder."""
         used_canonicals: set[str] = set()
         for tool_call in final_tool_calls:
+            name = getattr(tool_call, "name", None)
+            if not is_file_edit_tool(name):
+                continue
             canonical = self.canonical_call_id_for(tool_call)
             if canonical and canonical not in used_canonicals:
                 try:

nanobot/utils/tool_hints.py

@@ -35,10 +35,15 @@ def format_tool_hints(tool_calls: list, max_length: int = 40) -> str:
 
     formatted = []
     for tc in tool_calls:
-        fmt = _TOOL_FORMATS.get(tc.name)
+        name = getattr(tc, "name", None)
+        if not isinstance(name, str) or not name:
+            # Degenerate/malformed tool call (e.g. a model emits name=None);
+            # skip it instead of raising AttributeError on the whole turn.
+            continue
+        fmt = _TOOL_FORMATS.get(name)
         if fmt:
             formatted.append(_fmt_known(tc, fmt, max_length))
-        elif tc.name.startswith("mcp_"):
+        elif name.startswith("mcp_"):
             formatted.append(_fmt_mcp(tc, max_length))
         else:
             formatted.append(_fmt_fallback(tc, max_length))

nanobot/webui/session_list_index.py

@@ -26,10 +26,11 @@ from nanobot.session.manager import (
     _metadata_title,
 )
 
-_INDEX_VERSION = 1
+_INDEX_VERSION = 2
 _INDEX_FILENAME = ".webui_session_index.json"
 _WEBUI_ACTIVITY_MTIME_NS = "webui_activity_mtime_ns"
 _WEBUI_ACTIVITY_SIZE = "webui_activity_size"
+_VISIBLE_TRANSCRIPT_ROLES = {"user", "assistant"}
 
 
 def list_webui_sessions(session_manager: SessionManager) -> list[dict[str, Any]]:
@@ -214,14 +215,45 @@ def _latest_updated_at(stored: str | None, activity: str | None) -> str | None:
     return stored
 
 
+def _visible_message_timestamp(item: dict[str, Any]) -> str | None:
+    if item.get(CRON_HISTORY_META) is True:
+        return None
+    if item.get("role") not in _VISIBLE_TRANSCRIPT_ROLES:
+        return None
+    timestamp = item.get("timestamp")
+    return timestamp if isinstance(timestamp, str) else None
+
+
+def _last_visible_message_at(messages: list[dict[str, Any]]) -> str | None:
+    latest: str | None = None
+    for item in messages:
+        timestamp = _visible_message_timestamp(item)
+        if timestamp is not None:
+            latest = _latest_updated_at(latest, timestamp)
+    return latest
+
+
+def _visible_activity_updated_at(
+    stored: str | None,
+    visible_message_at: str | None,
+    webui_activity: str | None,
+) -> str | None:
+    return _latest_updated_at(visible_message_at, webui_activity) or stored
+
+
 def _indexed_row_for_session(session: Session, path: Path) -> dict[str, Any]:
     signature = _file_signature(path)
     activity_signature = _webui_activity_signature(session.key)
     activity_updated_at = _webui_activity_updated_at(activity_signature)
+    visible_message_at = _last_visible_message_at(session.messages)
     return {
         "key": session.key,
         "created_at": session.created_at.isoformat(),
-        "updated_at": _latest_updated_at(session.updated_at.isoformat(), activity_updated_at),
+        "updated_at": _visible_activity_updated_at(
+            session.updated_at.isoformat(),
+            visible_message_at,
+            activity_updated_at,
+        ),
         "title": _metadata_title(session.metadata),
         "preview": _preview_from_messages(session.messages),
         "file": path.name,
@@ -232,7 +264,8 @@ def _indexed_row_for_session(session: Session, path: Path) -> dict[str, Any]:
 
 
 def _scan_session_row(session_manager: SessionManager, path: Path) -> dict[str, Any] | None:
-    fallback_key = path.stem.replace("_", ":", 1)
+    storage_key = SessionManager._decode_storage_key(path.stem)
+    fallback_key = storage_key or path.stem.replace("_", ":", 1)
     try:
         with open(path, encoding="utf-8") as f:
             first_line = f.readline().strip()
@@ -243,31 +276,37 @@ def _scan_session_row(session_manager: SessionManager, path: Path) -> dict[str,
                 return None
             preview = ""
             fallback_preview = ""
+            visible_message_at = None
+            preview_done = False
             scanned_records = 0
             scanned_chars = 0
             for line in f:
                 if not line.strip():
                     continue
-                scanned_records += 1
-                scanned_chars += len(line)
-                if (
-                    scanned_records > _SESSION_LIST_PREVIEW_MAX_RECORDS
-                    or scanned_chars > _SESSION_LIST_PREVIEW_MAX_CHARS
-                ):
-                    break
                 item = json.loads(line)
-                if item.get("_type") == "metadata":
-                    continue
-                if item.get(CRON_HISTORY_META) is True:
-                    continue
-                text = _message_preview_text(item)
-                if not text:
-                    continue
-                if item.get("role") == "user":
-                    preview = text
-                    break
-                if not fallback_preview and item.get("role") == "assistant":
-                    fallback_preview = text
+                timestamp = _visible_message_timestamp(item)
+                if timestamp is not None:
+                    visible_message_at = _latest_updated_at(visible_message_at, timestamp)
+                if not preview_done:
+                    scanned_records += 1
+                    scanned_chars += len(line)
+                    if (
+                        scanned_records > _SESSION_LIST_PREVIEW_MAX_RECORDS
+                        or scanned_chars > _SESSION_LIST_PREVIEW_MAX_CHARS
+                    ):
+                        preview_done = True
+                        continue
+                    if item.get(CRON_HISTORY_META) is True:
+                        continue
+                    text = _message_preview_text(item)
+                    if not text:
+                        continue
+                    if item.get("role") == "user":
+                        preview = text
+                        preview_done = True
+                        continue
+                    if not fallback_preview and item.get("role") == "assistant":
+                        fallback_preview = text
             signature = _file_signature(path)
             created_at_s = data.get("created_at")
             updated_at_s = data.get("updated_at")
@@ -281,7 +320,11 @@ def _scan_session_row(session_manager: SessionManager, path: Path) -> dict[str,
             return {
                 "key": key,
                 "created_at": created_at_s,
-                "updated_at": _latest_updated_at(updated_at_s, activity_updated_at),
+                "updated_at": _visible_activity_updated_at(
+                    updated_at_s,
+                    visible_message_at,
+                    activity_updated_at,
+                ),
                 "title": _metadata_title(data.get("metadata", {})),
                 "preview": preview or fallback_preview,
                 "file": path.name,

nanobot/webui/settings_api.py

@@ -22,7 +22,7 @@ from nanobot.audio.transcription_registry import (
     resolve_transcription_provider,
     transcription_provider_names,
 )
-from nanobot.config.loader import get_config_path, load_config, save_config
+from nanobot.config.loader import get_config_path, load_config, resolve_config_env_vars, save_config
 from nanobot.config.schema import ModelPresetConfig, ProviderConfig
 from nanobot.providers.image_generation import (
     get_image_gen_provider,
@@ -1166,14 +1166,19 @@ def login_oauth_provider(query: QueryParams) -> dict[str, Any]:
         except ImportError:
             raise WebUISettingsError("oauth_cli_kit is not installed", status=500) from None
 
+        try:
+            proxy = resolve_config_env_vars(load_config()).providers.openai_codex.proxy or None
+        except ValueError as e:
+            raise WebUISettingsError(str(e), status=400) from e
         token = None
         with suppress(Exception):
-            token = get_token()
+            token = get_token(proxy=proxy)
         if not (token and token.access):
             messages: list[str] = []
             token = login_oauth_interactive(
                 print_fn=lambda message: messages.append(str(message)),
                 prompt_fn=lambda _prompt: "",
+                proxy=proxy,
             )
         if not (token and token.access):
             raise WebUISettingsError("OAuth login failed", status=401)

pyproject.toml

@@ -31,7 +31,7 @@ dependencies = [
     "websocket-client>=1.9.0,<2.0.0",
     "httpx>=0.28.0,<1.0.0",
     "ddgs>=9.5.5,<10.0.0",
-    "oauth-cli-kit>=0.1.3,<1.0.0",
+    "oauth-cli-kit>=0.1.6,<1.0.0",
     "loguru>=0.7.3,<1.0.0",
     "readability-lxml>=0.8.4,<1.0.0",
     "lxml-html-clean>=0.4.0,<1.0.0",

scripts/install.sh

@@ -269,7 +269,15 @@ if [ "${NANOBOT_SKIP_WIZARD:-}" = "1" ]; then
   exit 0
 fi
 
-info "Starting setup wizard..."
-run_nanobot onboard --wizard
+if [ -t 0 ]; then
+  info "Starting setup wizard..."
+  run_nanobot onboard --wizard
+elif : 2>/dev/null < /dev/tty; then
+  info "Starting setup wizard..."
+  run_nanobot onboard --wizard < /dev/tty
+else
+  info "Skipping setup wizard because no interactive terminal is available."
+  info "Run this later: $(nanobot_try_command) onboard --wizard"
+fi
 
 info "Done. Try: $(nanobot_try_command) agent -m \"Hello!\""

tests/agent/conftest.py

@@ -38,7 +38,6 @@ def make_loop(
     model: str = "test-model",
     context_window_tokens: int = 128_000,
     session_ttl_minutes: int = 0,
-    max_messages: int = 120,
     unified_session: bool = False,
     mcp_servers: dict | None = None,
     tools_config=None,
@@ -64,7 +63,6 @@ def make_loop(
         model=model,
         context_window_tokens=context_window_tokens,
         session_ttl_minutes=session_ttl_minutes,
-        max_messages=max_messages,
         unified_session=unified_session,
     )
     if mcp_servers is not None:
@@ -79,8 +77,8 @@ def make_loop(
     if patch_deps:
         with patch("nanobot.agent.loop.ContextBuilder"), \
              patch("nanobot.agent.loop.SessionManager"), \
-             patch("nanobot.agent.loop.SubagentManager") as MockSubMgr:
-            MockSubMgr.return_value.cancel_by_session = AsyncMock(return_value=0)
+             patch("nanobot.agent.loop.SubagentManager") as mock_sub_mgr:
+            mock_sub_mgr.return_value.cancel_by_session = AsyncMock(return_value=0)
             return AgentLoop(**kwargs)
     return AgentLoop(**kwargs)
 

tests/agent/test_auto_compact.py

@@ -91,7 +91,6 @@ def _make_fake_compact(
 
         tail = list(session.messages[session.last_consolidated:])
         if not tail:
-            session.updated_at = datetime.now()
             loop.sessions.save(session)
             return ""
 
@@ -103,15 +102,14 @@ def _make_fake_compact(
             metadata={},
             last_consolidated=0,
         )
-        dropped, already_consolidated = probe.retain_recent_legal_suffix(
+        result = probe.retain_recent_legal_suffix(
             max_suffix,
             extend_to_user=True,
         )
         kept = probe.messages
-        archive_msgs = dropped[already_consolidated:]
+        archive_msgs = result.dropped[result.already_consolidated_count:]
 
         if not archive_msgs and not kept:
-            session.updated_at = datetime.now()
             loop.sessions.save(session)
             return ""
 
@@ -132,7 +130,6 @@ def _make_fake_compact(
 
         session.messages = kept
         session.last_consolidated = 0
-        session.updated_at = datetime.now()
         loop.sessions.save(session)
         return s
 
@@ -1021,27 +1018,28 @@ class TestProactiveAutoCompact:
         await self._run_check_expired(loop)
         assert _fake_compact.state["count"] == 1
 
-        # Second tick: should NOT re-schedule (updated_at is fresh after clear)
+        # Second tick: should NOT re-schedule because the session has no removable tail.
         await self._run_check_expired(loop)
         assert _fake_compact.state["count"] == 1  # Still 1, not re-scheduled
         await loop.close_mcp()
 
     @pytest.mark.asyncio
-    async def test_empty_skip_refreshes_updated_at_prevents_reschedule(self, tmp_path):
-        """Empty session skip refreshes updated_at, preventing immediate re-scheduling."""
+    async def test_empty_session_does_not_schedule_idle_compact(self, tmp_path):
+        """Empty expired sessions have no removable tail and should not schedule."""
         loop = _make_loop(tmp_path, session_ttl_minutes=15)
         session = loop.sessions.get_or_create("cli:test")
         session.updated_at = datetime.now() - timedelta(minutes=20)
         loop.sessions.save(session)
 
-        loop.consolidator.compact_idle_session = _make_fake_compact(loop)
+        _fake_compact = _make_fake_compact(loop)
+        loop.consolidator.compact_idle_session = _fake_compact
 
-        # First tick: skips (no messages), refreshes updated_at
         await self._run_check_expired(loop)
+        assert _fake_compact.state["count"] == 0
         assert "cli:test" not in loop.auto_compact._summaries
 
-        # Second tick: should NOT re-schedule because updated_at is fresh
         await self._run_check_expired(loop)
+        assert _fake_compact.state["count"] == 0
         assert "cli:test" not in loop.auto_compact._summaries
         await loop.close_mcp()
 

tests/agent/test_autocompact_unit.py

@@ -200,8 +200,11 @@ class TestCheckExpired:
         """Expired session should trigger schedule_background."""
         ac = _make_autocompact(ttl=15)
         mock_sm = MagicMock(spec=SessionManager)
-        old_ts = (datetime.now() - timedelta(minutes=20)).isoformat()
-        mock_sm.list_sessions.return_value = [{"key": "cli:old", "updated_at": old_ts}]
+        old_dt = datetime.now() - timedelta(minutes=20)
+        session = _make_session("cli:old", updated_at=old_dt)
+        _add_turns(session, 5)
+        mock_sm.list_sessions.return_value = [{"key": "cli:old", "updated_at": old_dt.isoformat()}]
+        mock_sm.get_or_create.return_value = session
         ac.sessions = mock_sm
 
         scheduled = []
@@ -273,6 +276,24 @@ class TestCheckExpired:
         scheduler.assert_not_called()
         assert "dream:20260602-155256" not in ac._archiving
 
+    def test_already_trimmed_session_skips(self):
+        """Expired session with no removable tail should not be re-scheduled."""
+        ac = _make_autocompact(ttl=15)
+        mock_sm = MagicMock(spec=SessionManager)
+        last_active = datetime(2026, 1, 1, 10, 0, 0)
+        session = _make_session("cli:done", updated_at=last_active)
+        _add_turns(session, 2)
+        mock_sm.list_sessions.return_value = [
+            {"key": "cli:done", "updated_at": last_active.isoformat()},
+        ]
+        mock_sm.get_or_create.return_value = session
+        ac.sessions = mock_sm
+
+        scheduler = MagicMock()
+        ac.check_expired(scheduler)
+
+        scheduler.assert_not_called()
+
 
 # ---------------------------------------------------------------------------
 # _archive

tests/agent/test_consolidator.py

@@ -430,9 +430,11 @@ class TestCompactIdleSession:
         )
         sessions = real_consolidator.sessions
         session = sessions.get_or_create("cli:test")
+        old_ts = session.updated_at
         for i in range(20):
             session.add_message("user", f"user msg {i}")
             session.add_message("assistant", f"assistant msg {i}")
+        session.updated_at = old_ts
         sessions.save(session)
 
         result = await real_consolidator.compact_idle_session("cli:test", max_suffix=8)
@@ -445,6 +447,7 @@ class TestCompactIdleSession:
         assert meta is not None
         assert meta["text"] == "Summary of old conversation."
         assert "last_active" in meta
+        assert reloaded.updated_at == old_ts
 
     @pytest.mark.asyncio
     async def test_summarizes_retained_suffix_not_just_dropped_prefix(
@@ -518,8 +521,10 @@ class TestCompactIdleSession:
         assert entries[0]["session_key"] == "cli:test"
 
     @pytest.mark.asyncio
-    async def test_empty_session_refreshes_timestamp(self, real_consolidator):
-        """Empty session with old updated_at → refreshed after call, returns ''."""
+    async def test_empty_session_does_not_refresh_timestamp(
+        self, real_consolidator
+    ):
+        """Empty session with old updated_at does not look active after compaction."""
         from datetime import datetime, timedelta
 
         sessions = real_consolidator.sessions
@@ -532,7 +537,8 @@ class TestCompactIdleSession:
         assert result == ""
 
         reloaded = sessions.get_or_create("cli:empty")
-        assert reloaded.updated_at > old_ts
+        assert reloaded.updated_at == old_ts
+        assert reloaded.metadata == {}
 
     @pytest.mark.asyncio
     async def test_nothing_summary_not_stored(self, real_consolidator, mock_provider):

tests/agent/test_dream_session.py

@@ -24,9 +24,14 @@ class TestDreamSessionKey:
 
 class TestPruneDreamSessions:
     def test_keeps_n_most_recent(self, tmp_path):
+        import os
+        import time
+
         sessions_dir = tmp_path / "sessions"
         sessions_dir.mkdir()
 
+        base_time = time.time() - 100
+
         for i in range(15):
             key = f"dream:20260528-{100000 + i:06d}"
             safe_key = key.replace(":", "_")
@@ -37,6 +42,7 @@ class TestPruneDreamSessions:
                 f'"updated_at": "2026-05-28T10:00:{i:02d}"}}\n',
                 encoding="utf-8",
             )
+            os.utime(path, (base_time + i, base_time + i))
 
         normal_path = sessions_dir / "telegram_123.jsonl"
         normal_path.write_text('{"_type": "metadata"}\n', encoding="utf-8")

tests/agent/test_gemini_thought_signature.py

@@ -11,7 +11,6 @@ from unittest.mock import patch
 from nanobot.providers.base import ToolCallRequest
 from nanobot.providers.openai_compat_provider import OpenAICompatProvider
 
-
 GEMINI_EXTRA = {"google": {"thought_signature": "sig-abc-123"}}
 
 
@@ -125,6 +124,47 @@ def test_parse_dict_preserves_extra_content() -> None:
     assert payload["extra_content"] == GEMINI_EXTRA
 
 
+def test_parse_dict_deduplicates_duplicate_tool_call_ids() -> None:
+    with patch("nanobot.providers.openai_compat_provider.AsyncOpenAI"):
+        provider = OpenAICompatProvider()
+
+    response_dict = {
+        "choices": [
+            {
+                "message": {
+                    "content": None,
+                    "tool_calls": [{
+                        "id": "call_same",
+                        "type": "function",
+                        "function": {"name": "read_file", "arguments": '{"path":"a.txt"}'},
+                    }],
+                },
+                "finish_reason": "tool_calls",
+            },
+            {
+                "message": {
+                    "content": None,
+                    "tool_calls": [{
+                        "id": "call_same",
+                        "type": "function",
+                        "function": {"name": "read_file", "arguments": '{"path":"b.txt"}'},
+                    }],
+                },
+                "finish_reason": "tool_calls",
+            },
+        ],
+    }
+
+    result = provider._parse(response_dict)
+
+    ids = [tc.id for tc in result.tool_calls]
+    assert len(ids) == 2
+    assert ids[0] == "call_same"
+    assert ids[1] != "call_same"
+    assert len(set(ids)) == 2
+    assert [tc.arguments for tc in result.tool_calls] == [{"path": "a.txt"}, {"path": "b.txt"}]
+
+
 # ── _parse_chunks: streaming round-trip ───────────────────────────────
 
 def test_parse_chunks_sdk_preserves_extra_content() -> None:

tests/agent/test_max_messages_config.py

@@ -1,4 +1,4 @@
-"""Tests for max_messages config wiring into session history replay."""
+"""Tests for the internal max_messages replay cap."""
 
 from __future__ import annotations
 
@@ -11,20 +11,27 @@ from nanobot.agent.loop import AgentLoop
 from nanobot.bus.events import InboundMessage
 from nanobot.bus.queue import MessageBus
 from nanobot.providers.base import LLMResponse
-from nanobot.session.manager import Session
-
-DEFAULT_MAX_MESSAGES = 120
+from nanobot.providers.factory import ProviderSnapshot
+from nanobot.session.manager import (
+    FILE_MAX_MESSAGES,
+    Session,
+    replay_max_messages_for_context,
+)
 
 
-def _make_loop(tmp_path: Path, max_messages: int = DEFAULT_MAX_MESSAGES) -> AgentLoop:
+def _make_loop(
+    tmp_path: Path,
+    context_window_tokens: int = 200_000,
+) -> AgentLoop:
     provider = MagicMock()
     provider.get_default_model.return_value = "test-model"
+    provider.generation.max_tokens = 4096
     return AgentLoop(
         bus=MessageBus(),
         provider=provider,
         workspace=tmp_path,
         model="test-model",
-        max_messages=max_messages,
+        context_window_tokens=context_window_tokens,
     )
 
 
@@ -51,24 +58,44 @@ def _tool_round(call_id: str) -> list[dict]:
 
 
 class TestMaxMessagesInit:
-    """Verify AgentLoop stores the config value correctly."""
+    """Verify AgentLoop derives the internal replay cap correctly."""
 
-    def test_default_is_builtin_limit(self, tmp_path: Path) -> None:
+    def test_context_formula(self) -> None:
+        assert replay_max_messages_for_context(8_000) == 120
+        assert replay_max_messages_for_context(32_768) == 327
+        assert replay_max_messages_for_context(200_000) == FILE_MAX_MESSAGES
+
+    def test_default_for_200k_context_reaches_file_cap(self, tmp_path: Path) -> None:
         loop = _make_loop(tmp_path)
-        assert loop._max_messages == DEFAULT_MAX_MESSAGES
+        assert loop._max_messages == FILE_MAX_MESSAGES
 
-    def test_positive_value_stored(self, tmp_path: Path) -> None:
-        loop = _make_loop(tmp_path, max_messages=25)
-        assert loop._max_messages == 25
+    def test_default_scales_with_context_window(self, tmp_path: Path) -> None:
+        loop = _make_loop(tmp_path, context_window_tokens=32_768)
+        assert loop._max_messages == 327
 
-    def test_zero_uses_builtin_limit(self, tmp_path: Path) -> None:
-        loop = _make_loop(tmp_path, max_messages=0)
-        assert loop._max_messages == DEFAULT_MAX_MESSAGES
+    def test_provider_refresh_resyncs_context_derived_limit(self, tmp_path: Path) -> None:
+        old_provider = MagicMock()
+        old_provider.get_default_model.return_value = "old-model"
+        old_provider.generation.max_tokens = 4096
+        new_provider = MagicMock()
+        new_provider.generation.max_tokens = 4096
+        loop = AgentLoop(
+            bus=MessageBus(),
+            provider=old_provider,
+            workspace=tmp_path,
+            model="old-model",
+            context_window_tokens=32_768,
+            provider_snapshot_loader=lambda: ProviderSnapshot(
+                provider=new_provider,
+                model="new-model",
+                context_window_tokens=200_000,
+                signature=("new-model",),
+            ),
+        )
 
-    def test_negative_treated_as_builtin_limit(self, tmp_path: Path) -> None:
-        """Negative values should not produce negative slicing."""
-        loop = _make_loop(tmp_path, max_messages=-5)
-        assert loop._max_messages == DEFAULT_MAX_MESSAGES
+        assert loop._max_messages == 327
+        loop._refresh_provider_snapshot()
+        assert loop._max_messages == FILE_MAX_MESSAGES
 
 
 class TestGetHistoryWithMaxMessages:
@@ -77,7 +104,7 @@ class TestGetHistoryWithMaxMessages:
     def test_default_uses_builtin_limit(self) -> None:
         session = _populated_session(80)
         history = session.get_history()
-        assert len(history) <= DEFAULT_MAX_MESSAGES
+        assert len(history) <= FILE_MAX_MESSAGES
 
     def test_explicit_max_messages_limits_output(self) -> None:
         session = _populated_session(40)  # 80 messages total
@@ -93,7 +120,7 @@ class TestGetHistoryWithMaxMessages:
     def test_max_messages_zero_uses_builtin_limit(self) -> None:
         session = _populated_session(80)  # 160 messages total
         history = session.get_history(max_messages=0)
-        assert len(history) <= DEFAULT_MAX_MESSAGES
+        assert len(history) <= FILE_MAX_MESSAGES
 
     def test_small_session_unaffected(self) -> None:
         """When session has fewer messages than max_messages, all are returned."""
@@ -103,12 +130,13 @@ class TestGetHistoryWithMaxMessages:
 
 
 class TestMaxMessagesIntegration:
-    """Verify the config flows from AgentLoop into get_history calls."""
+    """Verify AgentLoop passes the replay cap into get_history calls."""
 
     @pytest.mark.asyncio
-    async def test_process_message_passes_config_to_history_call(self, tmp_path: Path) -> None:
+    async def test_process_message_passes_limit_to_history_call(self, tmp_path: Path) -> None:
         """The real message path should pass max_messages into session history replay."""
-        loop = _make_loop(tmp_path, max_messages=25)
+        loop = _make_loop(tmp_path)
+        loop._max_messages = 25
         loop.provider.chat_with_retry = AsyncMock(
             return_value=LLMResponse(content="ok", tool_calls=[], usage={})
         )
@@ -127,8 +155,11 @@ class TestMaxMessagesIntegration:
         assert mock_hist.call_args.kwargs["extend_to_user"] is False
 
     @pytest.mark.asyncio
-    async def test_zero_config_passes_builtin_limit_to_history_call(self, tmp_path: Path) -> None:
-        loop = _make_loop(tmp_path, max_messages=0)
+    async def test_default_limit_passes_context_derived_limit_to_history_call(
+        self,
+        tmp_path: Path,
+    ) -> None:
+        loop = _make_loop(tmp_path)
         loop.provider.chat_with_retry = AsyncMock(
             return_value=LLMResponse(content="ok", tool_calls=[], usage={})
         )
@@ -142,7 +173,7 @@ class TestMaxMessagesIntegration:
             )
 
         assert result is not None
-        assert mock_hist.call_args.kwargs["max_messages"] == DEFAULT_MAX_MESSAGES
+        assert mock_hist.call_args.kwargs["max_messages"] == FILE_MAX_MESSAGES
         assert mock_hist.call_args.kwargs["extend_to_user"] is False
 
     @pytest.mark.asyncio
@@ -151,7 +182,8 @@ class TestMaxMessagesIntegration:
         tmp_path: Path,
     ) -> None:
         """A live user turn should not extend history to an older long tool turn."""
-        loop = _make_loop(tmp_path, max_messages=6)
+        loop = _make_loop(tmp_path)
+        loop._max_messages = 6
         loop.provider.chat_with_retry = AsyncMock(
             return_value=LLMResponse(content="ok", tool_calls=[], usage={})
         )
@@ -182,31 +214,3 @@ class TestMaxMessagesIntegration:
         sent_text = "\n".join(str(message.get("content")) for message in sent_messages)
         assert "new question" in sent_text
         assert "long older turn" not in sent_text
-
-
-class TestSchemaConfig:
-    """Verify the config schema accepts max_messages."""
-
-    def test_schema_default(self) -> None:
-        from nanobot.config.schema import AgentDefaults
-
-        defaults = AgentDefaults()
-        assert defaults.max_messages == DEFAULT_MAX_MESSAGES
-
-    def test_schema_accepts_zero_as_builtin_limit(self) -> None:
-        from nanobot.config.schema import AgentDefaults
-
-        defaults = AgentDefaults(max_messages=0)
-        assert defaults.max_messages == 0
-
-    def test_schema_accepts_positive(self) -> None:
-        from nanobot.config.schema import AgentDefaults
-
-        defaults = AgentDefaults(max_messages=25)
-        assert defaults.max_messages == 25
-
-    def test_schema_rejects_negative(self) -> None:
-        from nanobot.config.schema import AgentDefaults
-
-        with pytest.raises(Exception):  # Pydantic validation error
-            AgentDefaults(max_messages=-1)

tests/agent/test_runner_governance.py

@@ -15,7 +15,7 @@ from nanobot.agent.context_governance import (
 )
 from nanobot.agent.runner import AgentRunSpec
 from nanobot.config.schema import AgentDefaults
-from nanobot.providers.base import LLMResponse
+from nanobot.providers.base import LLMResponse, ToolCallRequest
 
 _MAX_TOOL_RESULT_CHARS = AgentDefaults().max_tool_result_chars
 
@@ -877,3 +877,162 @@ def test_snip_history_no_user_at_all_falls_back_gracefully(monkeypatch):
         assert non_system[0]["role"] in ("user", "tool"), (
             f"Safety net should ensure first non-system is user/tool, got {non_system[0]['role']}"
         )
+
+
+# ---------------------------------------------------------------------------
+# Malformed tool_call name guard (missing/non-string name wedges the session
+# upstream: messages.content.N.tool_use.name: Input should be a valid string)
+# ---------------------------------------------------------------------------
+
+
+def test_drop_malformed_tool_calls_trims_response():
+    """LLM response tool_calls with a missing/empty name are dropped in place."""
+    from nanobot.agent.runner import AgentRunner
+
+    response = LLMResponse(
+        content=None,
+        tool_calls=[
+            ToolCallRequest(id="1", name=None, arguments={}),
+            ToolCallRequest(id="2", name="", arguments={}),
+            ToolCallRequest(id="3", name="read_file", arguments={}),
+        ],
+        finish_reason="tool_calls",
+    )
+    dropped, all_dropped, orig = AgentRunner._drop_malformed_tool_calls(response)
+    assert [tc.name for tc in response.tool_calls] == ["read_file"]
+    assert response.finish_reason == "tool_calls"
+    assert response.should_execute_tools is True
+    assert dropped == 2
+    assert all_dropped is False
+    assert orig == "tool_calls"
+
+
+def test_drop_malformed_tool_calls_all_bad_disables_execution():
+    """If every tool call is malformed, execution is disabled (no empty exec)."""
+    from nanobot.agent.runner import AgentRunner
+
+    response = LLMResponse(
+        content="some text",
+        tool_calls=[ToolCallRequest(id="1", name=None, arguments={})],
+        finish_reason="tool_calls",
+    )
+    dropped, all_dropped, orig = AgentRunner._drop_malformed_tool_calls(response)
+    assert response.tool_calls == []
+    assert response.finish_reason == "stop"
+    assert response.should_execute_tools is False
+    assert dropped == 1
+    assert all_dropped is True
+    assert orig == "tool_calls"
+
+
+def test_drop_malformed_returns_tuple_no_calls():
+    """No tool calls returns (0, False, current_finish_reason)."""
+    from nanobot.agent.runner import AgentRunner
+
+    response = LLMResponse(content="hi", finish_reason="stop")
+    dropped, all_dropped, orig = AgentRunner._drop_malformed_tool_calls(response)
+    assert dropped == 0
+    assert all_dropped is False
+    assert orig == "stop"
+
+
+def test_strip_malformed_tool_calls_keeps_valid_calls_in_history():
+    """A mixed assistant turn keeps only its valid tool_calls."""
+    messages = [
+        {"role": "user", "content": "hi"},
+        {
+            "role": "assistant",
+            "content": "",
+            "tool_calls": [
+                {"id": "bad", "type": "function", "function": {"name": None, "arguments": "{}"}},
+                {"id": "ok", "type": "function", "function": {"name": "exec", "arguments": "{}"}},
+            ],
+        },
+        {"role": "tool", "tool_call_id": "ok", "name": "exec", "content": "done"},
+    ]
+    result = ContextGovernor.strip_malformed_tool_calls(messages)
+    assert result is not messages  # copied, original untouched
+    assert len(messages[1]["tool_calls"]) == 2  # original preserved
+    kept = result[1]["tool_calls"]
+    assert [tc["function"]["name"] for tc in kept] == ["exec"]
+
+
+def test_strip_malformed_tool_calls_drops_empty_assistant_turn():
+    """An assistant turn that is only a malformed call is removed entirely;
+    the existing orphan-result cleanup then drops its dangling tool result,
+    so a polluted session self-heals."""
+    messages = [
+        {"role": "user", "content": "hi"},
+        {
+            "role": "assistant",
+            "content": None,
+            "tool_calls": [
+                {"id": "bad", "type": "function", "function": {"name": None, "arguments": "{}"}},
+            ],
+        },
+        {"role": "tool", "tool_call_id": "bad", "name": "", "content": "r"},
+    ]
+    stripped = ContextGovernor.strip_malformed_tool_calls(messages)
+    assert [m["role"] for m in stripped] == ["user", "tool"]
+    healed = ContextGovernor.drop_orphan_tool_results(stripped)
+    assert [m["role"] for m in healed] == ["user"]
+
+
+def test_strip_malformed_tool_calls_noop_when_clean():
+    """Clean history is returned unchanged (same object)."""
+    messages = [
+        {"role": "user", "content": "hi"},
+        {
+            "role": "assistant",
+            "content": "",
+            "tool_calls": [
+                {"id": "ok", "type": "function", "function": {"name": "exec", "arguments": "{}"}},
+            ],
+        },
+        {"role": "tool", "tool_call_id": "ok", "name": "exec", "content": "done"},
+    ]
+    assert ContextGovernor.strip_malformed_tool_calls(messages) is messages
+
+
+def test_strip_placeholder_assistant_messages_removes_omitted():
+    """Placeholder assistant messages are removed; real messages kept."""
+    messages = [
+        {"role": "user", "content": "hi"},
+        {"role": "assistant", "content": "real response"},
+        {"role": "user", "content": "ok"},
+        {"role": "assistant", "content": "[Previous assistant message omitted.]"},
+        {"role": "user", "content": "?"},
+        {"role": "assistant", "content": "[Previous assistant message omitted.]"},
+        {"role": "user", "content": "hello"},
+    ]
+    result = ContextGovernor.strip_placeholder_assistant_messages(messages)
+    assert [m["role"] for m in result] == [
+        "user", "assistant", "user", "user", "user",
+    ]
+    assert result[1]["content"] == "real response"
+
+
+def test_strip_placeholder_noop_when_clean():
+    """Clean history is returned unchanged (same object)."""
+    messages = [
+        {"role": "user", "content": "hi"},
+        {"role": "assistant", "content": "hello back"},
+    ]
+    assert ContextGovernor.strip_placeholder_assistant_messages(messages) is messages
+
+
+def test_strip_placeholder_keeps_assistant_with_tool_calls():
+    """A placeholder assistant that also carries tool_calls is kept."""
+    messages = [
+        {"role": "user", "content": "hi"},
+        {
+            "role": "assistant",
+            "content": "[Previous assistant message omitted.]",
+            "tool_calls": [
+                {"id": "1", "type": "function", "function": {"name": "exec", "arguments": "{}"}},
+            ],
+        },
+        {"role": "tool", "tool_call_id": "1", "name": "exec", "content": "done"},
+    ]
+    result = ContextGovernor.strip_placeholder_assistant_messages(messages)
+    assert result is messages

tests/agent/test_session_collision.py

@@ -0,0 +1,170 @@
+"""Regression tests for collision-resistant session filenames."""
+
+import json
+from datetime import datetime
+from pathlib import Path
+
+from nanobot.session.manager import Session, SessionManager
+from nanobot.utils.helpers import safe_filename
+
+
+def _manager(tmp_path: Path, monkeypatch) -> SessionManager:
+    monkeypatch.setattr(
+        "nanobot.session.manager.get_legacy_sessions_dir",
+        lambda: tmp_path / "legacy_sessions",
+    )
+    return SessionManager(tmp_path / "workspace")
+
+
+def _write_session_file(path: Path, key: str, content: str) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    metadata = {
+        "_type": "metadata",
+        "key": key,
+        "created_at": datetime(2025, 1, 1).isoformat(),
+        "updated_at": datetime(2025, 1, 1).isoformat(),
+        "metadata": {"source": "test"},
+        "last_consolidated": 0,
+    }
+    message = {"role": "user", "content": content}
+    path.write_text(
+        json.dumps(metadata) + "\n" + json.dumps(message) + "\n",
+        encoding="utf-8",
+    )
+
+
+def test_distinct_keys_have_distinct_filenames(tmp_path: Path, monkeypatch) -> None:
+    sm = _manager(tmp_path, monkeypatch)
+
+    first = sm._get_session_path("telegram:a_b")
+    second = sm._get_session_path("telegram:a:b")
+
+    assert first.name != second.name
+    assert sm.safe_key("telegram:a_b") == sm.safe_key("telegram:a:b")
+    assert sm._storage_key("telegram:a_b") != sm._storage_key("telegram:a:b")
+
+
+def test_save_uses_new_path_not_lossy(tmp_path: Path, monkeypatch) -> None:
+    sm = _manager(tmp_path, monkeypatch)
+    key = "telegram:a:b"
+    session = Session(key=key)
+    session.add_message("user", "first")
+    sm.save(session)
+
+    new_path = sm._get_session_path(key)
+    lossy_path = sm._get_legacy_lossy_path(key)
+    _write_session_file(lossy_path, key, "stale lossy content")
+    stale_lossy = lossy_path.read_text(encoding="utf-8")
+
+    session.add_message("assistant", "latest content")
+    sm.save(session)
+
+    assert new_path.exists()
+    assert lossy_path.exists()
+    assert "latest content" in new_path.read_text(encoding="utf-8")
+    assert lossy_path.read_text(encoding="utf-8") == stale_lossy
+
+
+def test_load_falls_back_to_lossy_path(tmp_path: Path, monkeypatch) -> None:
+    sm = _manager(tmp_path, monkeypatch)
+    key = "telegram:legacy:lossy"
+    lossy_path = sm._get_legacy_lossy_path(key)
+    _write_session_file(lossy_path, key, "loaded from lossy")
+
+    session = sm._load(key)
+
+    assert session is not None
+    assert session.metadata == {"source": "test"}
+    assert session.messages[0]["content"] == "loaded from lossy"
+
+
+def test_load_migrates_lossy_to_new_path(tmp_path: Path, monkeypatch) -> None:
+    sm = _manager(tmp_path, monkeypatch)
+    key = "telegram:migrate:lossy"
+    new_path = sm._get_session_path(key)
+    lossy_path = sm._get_legacy_lossy_path(key)
+    _write_session_file(lossy_path, key, "migrate me")
+
+    session = sm._load(key)
+
+    assert session is not None
+    assert session.messages[0]["content"] == "migrate me"
+    assert new_path.exists()
+    assert not lossy_path.exists()
+
+
+def test_load_does_not_migrate_lossy_path_for_different_stored_key(
+    tmp_path: Path,
+    monkeypatch,
+) -> None:
+    sm = _manager(tmp_path, monkeypatch)
+    first_key = "telegram:a_b"
+    second_key = "telegram:a:b"
+    lossy_path = sm._get_legacy_lossy_path(first_key)
+    assert lossy_path == sm._get_legacy_lossy_path(second_key)
+    _write_session_file(lossy_path, first_key, "belongs to first")
+
+    loaded_second = sm._load(second_key)
+
+    assert loaded_second is None
+    assert lossy_path.exists()
+    assert not sm._get_session_path(second_key).exists()
+
+    loaded_first = sm._load(first_key)
+
+    assert loaded_first is not None
+    assert loaded_first.messages[0]["content"] == "belongs to first"
+    assert sm._get_session_path(first_key).exists()
+    assert not lossy_path.exists()
+
+
+def test_safe_key_is_lossy() -> None:
+    assert SessionManager.safe_key("telegram:a_b") == SessionManager.safe_key("telegram:a:b")
+
+
+def test_storage_key_is_collision_resistant() -> None:
+    encoded = {
+        SessionManager._storage_key("a:b"),
+        SessionManager._storage_key("a_b"),
+        SessionManager._storage_key("a:b:c"),
+    }
+
+    assert len(encoded) == 3
+    assert SessionManager._storage_key("telegram:a_b") != SessionManager._storage_key("telegram:a:b")
+
+
+def test_lossy_path_helper_returns_expected_path(tmp_path: Path, monkeypatch) -> None:
+    sm = _manager(tmp_path, monkeypatch)
+    key = "telegram:a:b"
+    expected = sm.sessions_dir / f"{safe_filename(key.replace(':', '_'))}.jsonl"
+
+    assert sm._get_legacy_lossy_path(key) == expected
+
+
+def test_storage_paths_are_distinct_when_keys_collide_under_safe_key(
+    tmp_path: Path,
+    monkeypatch,
+) -> None:
+    sm = _manager(tmp_path, monkeypatch)
+    first = Session(key="telegram:a_b")
+    first.add_message("user", "underscore history")
+    second = Session(key="telegram:a:b")
+    second.add_message("user", "colon history")
+
+    sm.save(first)
+    sm.save(second)
+
+    assert sm.safe_key(first.key) == sm.safe_key(second.key)
+    assert sm._get_session_path(first.key).exists()
+    assert sm._get_session_path(second.key).exists()
+    assert sm._get_session_path(first.key) != sm._get_session_path(second.key)
+
+    sm.invalidate(first.key)
+    sm.invalidate(second.key)
+    loaded_first = sm._load(first.key)
+    loaded_second = sm._load(second.key)
+
+    assert loaded_first is not None
+    assert loaded_second is not None
+    assert loaded_first.messages[0]["content"] == "underscore history"
+    assert loaded_second.messages[0]["content"] == "colon history"

tests/agent/test_session_delete.py

@@ -58,11 +58,11 @@ def test_read_session_file_missing(tmp_path: Path) -> None:
     assert sm.read_session_file("nope:none") is None
 
 
-def test_safe_key_matches_internal_path(tmp_path: Path) -> None:
+def test_storage_key_matches_internal_path(tmp_path: Path) -> None:
     sm = SessionManager(tmp_path)
     key = "telegram:abc/def"
     expected = sm._get_session_path(key).name
-    assert SessionManager.safe_key(key) + ".jsonl" == expected
+    assert SessionManager._storage_key(key) + ".jsonl" == expected
 
 
 def _write_legacy_session(legacy_dir: Path, key: str, roles: list[str]) -> Path:

tests/agent/test_session_manager_history.py

@@ -685,12 +685,12 @@ def test_retain_recent_legal_suffix_returns_dropped_messages():
     for i in range(10):
         session.messages.append({"role": "user", "content": f"msg{i}"})
 
-    dropped, already_cons = session.retain_recent_legal_suffix(4)
+    result = session.retain_recent_legal_suffix(4)
 
-    assert len(dropped) == 6
-    assert [m["content"] for m in dropped] == [f"msg{i}" for i in range(6)]
+    assert len(result.dropped) == 6
+    assert [m["content"] for m in result.dropped] == [f"msg{i}" for i in range(6)]
     assert len(session.messages) == 4
-    assert already_cons == 0
+    assert result.already_consolidated_count == 0
 
 
 def test_retain_recent_legal_suffix_returns_empty_when_no_drop():
@@ -699,10 +699,10 @@ def test_retain_recent_legal_suffix_returns_empty_when_no_drop():
     for i in range(3):
         session.messages.append({"role": "user", "content": f"msg{i}"})
 
-    dropped, already_cons = session.retain_recent_legal_suffix(4)
+    result = session.retain_recent_legal_suffix(4)
 
-    assert dropped == []
-    assert already_cons == 0
+    assert result.dropped == []
+    assert result.already_consolidated_count == 0
     assert len(session.messages) == 3
 
 
@@ -713,10 +713,10 @@ def test_retain_recent_legal_suffix_returns_all_on_zero():
         session.messages.append({"role": "user", "content": f"msg{i}"})
     session.last_consolidated = 3
 
-    dropped, already_cons = session.retain_recent_legal_suffix(0)
+    result = session.retain_recent_legal_suffix(0)
 
-    assert len(dropped) == 5
-    assert already_cons == 3
+    assert len(result.dropped) == 5
+    assert result.already_consolidated_count == 3
     assert session.messages == []
 
 
@@ -820,11 +820,11 @@ def test_retain_recent_legal_suffix_last_consolidated_correct_in_else_branch():
         session.messages.append({"role": "assistant", "content": f"a{i}"})
     session.last_consolidated = 12  # u0..u9, a0, a1 consolidated
 
-    dropped, already_cons = session.retain_recent_legal_suffix(4)
+    result = session.retain_recent_legal_suffix(4)
 
     # Retained messages start from latest user (u9) + max_messages forward
     # so retained = [u9, a0..a9][:4] → but these are from original indices 9..12
     # Of those, indices 9,10,11 are < 12 (before_lc), so new_lc = 3
     assert session.last_consolidated == 3
     # already_cons should count dropped messages with original index < 12
-    assert already_cons == 9
+    assert result.already_consolidated_count == 9

tests/agent/test_tool_hint.py

@@ -1,7 +1,7 @@
 """Tests for tool hint formatting (nanobot.utils.tool_hints)."""
 
-from nanobot.utils.tool_hints import format_tool_hints
 from nanobot.providers.base import ToolCallRequest
+from nanobot.utils.tool_hints import format_tool_hints
 
 
 def _tc(name: str, args) -> ToolCallRequest:
@@ -306,3 +306,22 @@ class TestToolHintMaxLength:
         short = _hint([_tc("list_dir", {"path": long_path})], max_length=40)
         long = _hint([_tc("list_dir", {"path": long_path})], max_length=120)
         assert len(long) > len(short)
+
+
+class TestToolHintMalformedCalls:
+    """Malformed tool calls must not crash hint formatting (see HKUDS/nanobot)."""
+
+    def test_none_name_is_skipped(self):
+        """A tool call with name=None should be skipped, not raise AttributeError."""
+        result = _hint([_tc(None, None)])
+        assert result == ""
+
+    def test_empty_name_is_skipped(self):
+        """A tool call with an empty name should be skipped."""
+        result = _hint([_tc("", {"path": "foo.txt"})])
+        assert result == ""
+
+    def test_none_name_mixed_with_valid_call(self):
+        """A degenerate call must not suppress hints for the valid calls beside it."""
+        result = _hint([_tc(None, None), _tc("read_file", {"path": "foo.txt"})])
+        assert result == "read foo.txt"

tests/agent/tools/test_self_tool.py

@@ -236,10 +236,12 @@ class TestModifyRestricted:
 
     @pytest.mark.asyncio
     async def test_modify_context_window_valid(self):
-        tool = _make_tool()
+        loop = _make_mock_loop(_sync_replay_max_messages=MagicMock())
+        tool = _make_tool(runtime_state=loop)
         result = await tool.execute(action="set", key="context_window_tokens", value=131072)
         assert "Set context_window_tokens" in result
-        assert tool._runtime_state.context_window_tokens == 131072
+        assert loop.context_window_tokens == 131072
+        loop._sync_replay_max_messages.assert_called_once_with()
 
     @pytest.mark.asyncio
     async def test_modify_none_value_for_restricted_int(self):

tests/agent/tools/test_subagent_tools.py

@@ -7,6 +7,7 @@ from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 
+from nanobot.bus.events import OutboundMessage
 from nanobot.config.schema import AgentDefaults
 
 _MAX_TOOL_RESULT_CHARS = AgentDefaults().max_tool_result_chars
@@ -482,3 +483,49 @@ async def test_drain_pending_timeout(tmp_path):
         await hang_task
     except asyncio.CancelledError:
         pass
+
+
+@pytest.mark.asyncio
+async def test_process_direct_routes_subagent_results_to_pending_queue(tmp_path):
+    """Single-message CLI mode should consume subagent announcements mid-turn."""
+    from nanobot.agent.loop import AgentLoop
+    from nanobot.bus.events import InboundMessage
+    from nanobot.bus.queue import MessageBus
+
+    loop = AgentLoop(
+        bus=MessageBus(),
+        provider=MagicMock(),
+        workspace=tmp_path,
+        model="test-model",
+    )
+    loop._connect_mcp = AsyncMock()  # type: ignore[method-assign]
+
+    async def fake_process_message(msg, **kwargs):
+        pending_queue = kwargs["pending_queue"]
+        await loop.bus.publish_inbound(InboundMessage(
+            channel="other",
+            sender_id="u",
+            chat_id="room",
+            content="unrelated",
+        ))
+        await loop.subagents._announce_result(
+            "sub-1",
+            "label",
+            "task",
+            "subagent result",
+            {"channel": "cli", "chat_id": "direct", "session_key": "cli:direct"},
+            "ok",
+        )
+        routed = await asyncio.wait_for(pending_queue.get(), timeout=1)
+        assert "subagent result" in routed.content
+        assert routed.metadata["subagent_task_id"] == "sub-1"
+        return OutboundMessage(channel="cli", chat_id="direct", content="done")
+
+    loop._process_message = fake_process_message  # type: ignore[method-assign]
+
+    response = await loop.process_direct("start", session_key="cli:direct")
+
+    assert response is not None
+    assert response.content == "done"
+    unrelated = await asyncio.wait_for(loop.bus.consume_inbound(), timeout=1)
+    assert unrelated.content == "unrelated"

tests/channels/test_channel_manager_delta_coalescing.py

@@ -142,6 +142,31 @@ class TestDeltaCoalescing:
         assert pending[0].chat_id == "chat2"
         assert pending[0].content == "World"
 
+    @pytest.mark.asyncio
+    async def test_deltas_different_stream_ids_not_coalesced(self, manager, bus):
+        """Deltas for the same chat but different streams should not be merged."""
+        await bus.publish_outbound(OutboundMessage(
+            channel="mock",
+            chat_id="chat1",
+            content="A1",
+            metadata={"_stream_delta": True, "_stream_id": "stream-a"},
+        ))
+        await bus.publish_outbound(OutboundMessage(
+            channel="mock",
+            chat_id="chat1",
+            content="B1",
+            metadata={"_stream_delta": True, "_stream_id": "stream-b"},
+        ))
+
+        first_msg = await bus.consume_outbound()
+        merged, pending = manager._coalesce_stream_deltas(first_msg)
+
+        assert merged.content == "A1"
+        assert merged.metadata.get("_stream_id") == "stream-a"
+        assert len(pending) == 1
+        assert pending[0].content == "B1"
+        assert pending[0].metadata.get("_stream_id") == "stream-b"
+
     @pytest.mark.asyncio
     async def test_stream_end_terminates_coalescing(self, manager, bus):
         """_stream_end should stop coalescing and be included in final message."""

tests/channels/test_websocket_reconnect_idle.py

@@ -0,0 +1,59 @@
+"""Test websocket subscribe hydration only replays known active turns."""
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from nanobot.channels.websocket import WebSocketChannel
+
+
+@pytest.mark.asyncio
+async def test_hydrate_after_subscribe_is_quiet_when_no_turn_active():
+    """Subscribe hydration must not inject an idle event into normal message order."""
+    channel = WebSocketChannel.__new__(WebSocketChannel)
+    channel.gateway = MagicMock()
+    channel.gateway.session_manager = MagicMock()
+    channel.gateway.session_manager.read_session_file = MagicMock(return_value={})
+
+    sent_events = []
+
+    async def mock_send_goal_state(chat_id, blob):
+        sent_events.append(("goal_state", chat_id, blob))
+
+    async def mock_send_goal_status(chat_id, status, **kwargs):
+        sent_events.append(("goal_status", chat_id, status, kwargs))
+
+    channel.send_goal_state = mock_send_goal_state
+    channel.send_goal_status = mock_send_goal_status
+
+    with patch("nanobot.channels.websocket.websocket_turn_wall_started_at", return_value=None):
+        await channel._hydrate_after_subscribe("test-chat")
+
+    assert sent_events == []
+
+
+@pytest.mark.asyncio
+async def test_hydrate_after_subscribe_pushes_running_when_turn_active():
+    """Reconnecting client should receive running status when turn is active."""
+    channel = WebSocketChannel.__new__(WebSocketChannel)
+    channel.gateway = MagicMock()
+    channel.gateway.session_manager = MagicMock()
+    channel.gateway.session_manager.read_session_file = MagicMock(return_value={})
+
+    sent_events = []
+
+    async def mock_send_goal_state(chat_id, blob):
+        sent_events.append(("goal_state", chat_id, blob))
+
+    async def mock_send_goal_status(chat_id, status, **kwargs):
+        sent_events.append(("goal_status", chat_id, status, kwargs))
+
+    channel.send_goal_state = mock_send_goal_state
+    channel.send_goal_status = mock_send_goal_status
+
+    with patch("nanobot.channels.websocket.websocket_turn_wall_started_at", return_value=1234567890.0):
+        await channel._hydrate_after_subscribe("test-chat")
+
+    running_events = [e for e in sent_events if e[0] == "goal_status" and e[2] == "running"]
+    assert len(running_events) == 1
+    assert running_events[0][3]["started_at"] == 1234567890.0

tests/channels/test_weixin_channel.py

@@ -1764,6 +1764,44 @@ async def test_buffer_flushed_on_stream_end() -> None:
     assert "wx-user" not in channel._pending_tool_hints
 
 
+@pytest.mark.asyncio
+async def test_stream_end_flushes_buffered_answer() -> None:
+    channel, _bus = _make_channel()
+    channel._client = object()
+    channel._token = "token"
+    channel._context_tokens["wx-user"] = "ctx-1"
+    channel._context_token_at["wx-user"] = time.time()
+    channel._send_text = AsyncMock()
+
+    await channel.send_delta("wx-user", "hello ", {"_stream_delta": True})
+    await channel.send_delta("wx-user", "world", {"_stream_end": True})
+
+    channel._send_text.assert_awaited_once_with("wx-user", "hello world", "ctx-1")
+    assert "wx-user" not in channel._stream_buffers
+
+
+@pytest.mark.asyncio
+async def test_stream_end_send_failure_keeps_buffer_for_retry() -> None:
+    channel, _bus = _make_channel()
+    channel._client = object()
+    channel._token = "token"
+    channel._context_tokens["wx-user"] = "ctx-1"
+    channel._context_token_at["wx-user"] = time.time()
+    channel._send_text = AsyncMock(side_effect=RuntimeError("temporary send failure"))
+
+    await channel.send_delta("wx-user", "hello ", {"_stream_delta": True})
+    with pytest.raises(RuntimeError):
+        await channel.send_delta("wx-user", "world", {"_stream_end": True})
+
+    assert channel._stream_buffers["wx-user"] == ["hello "]
+
+    channel._send_text = AsyncMock()
+    await channel.send_delta("wx-user", "world", {"_stream_end": True})
+
+    channel._send_text.assert_awaited_once_with("wx-user", "hello world", "ctx-1")
+    assert "wx-user" not in channel._stream_buffers
+
+
 @pytest.mark.asyncio
 async def test_stop_clears_buffer() -> None:
     channel, _bus = _make_channel()

tests/channels/test_whatsapp_channel.py

@@ -1,6 +1,8 @@
 from __future__ import annotations
 
 import asyncio
+import sys
+import types
 from types import SimpleNamespace
 from unittest.mock import AsyncMock, MagicMock
 
@@ -86,6 +88,23 @@ def _patch_neonize_api(monkeypatch) -> None:
     )
 
 
+def _patch_receipt_type(monkeypatch):
+    neonize = types.ModuleType("neonize")
+    utils = types.ModuleType("neonize.utils")
+    enum = types.ModuleType("neonize.utils.enum")
+
+    class ReceiptType:
+        READ = "read"
+
+    enum.ReceiptType = ReceiptType
+    neonize.utils = utils
+    utils.enum = enum
+    monkeypatch.setitem(sys.modules, "neonize", neonize)
+    monkeypatch.setitem(sys.modules, "neonize.utils", utils)
+    monkeypatch.setitem(sys.modules, "neonize.utils.enum", enum)
+    return ReceiptType
+
+
 class _FakeLoginClient:
     def __init__(self) -> None:
         self.handlers = {}
@@ -301,6 +320,60 @@ async def test_group_sender_id_uses_participant_not_group_jid() -> None:
     assert kwargs["metadata"]["participant"] == "SENDERLID@lid"
 
 
+@pytest.mark.asyncio
+async def test_read_receipt_is_requested_once_after_dedup() -> None:
+    ch = _make_channel()
+    ch._send_read_receipt = AsyncMock()
+    ch._handle_message = AsyncMock()
+    client = SimpleNamespace(download_any=AsyncMock())
+    event = _event(
+        message=_Proto(conversation="hi"),
+        sender=_jid("15551234567", "s.whatsapp.net"),
+    )
+
+    await ch._handle_neonize_message(client, event)
+    await ch._handle_neonize_message(client, event)
+
+    ch._send_read_receipt.assert_awaited_once_with(
+        client,
+        event.Info.MessageSource,
+        "m1",
+    )
+    ch._handle_message.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_send_read_receipt_uses_mark_read_and_swallows_failures(monkeypatch) -> None:
+    receipt_type = _patch_receipt_type(monkeypatch)
+    ch = _make_channel()
+    source = _event(
+        message=_Proto(conversation="hi"),
+        sender=_jid("15551234567", "s.whatsapp.net"),
+    ).Info.MessageSource
+    client = SimpleNamespace(
+        mark_read=AsyncMock(),
+        download_any=AsyncMock(),
+    )
+
+    await ch._send_read_receipt(client, source, "m1")
+
+    client.mark_read.assert_awaited_once_with(
+        "m1",
+        chat=source.Chat,
+        sender=source.Sender,
+        receipt=receipt_type.READ,
+    )
+
+    failing_client = SimpleNamespace(
+        mark_read=AsyncMock(side_effect=RuntimeError("boom")),
+        download_any=AsyncMock(),
+    )
+
+    await ch._send_read_receipt(failing_client, source, "m2")
+
+    failing_client.mark_read.assert_awaited_once()
+
+
 @pytest.mark.asyncio
 async def test_lid_to_phone_cache_resolves_lid_only_messages() -> None:
     ch = _make_channel()

tests/cli/test_commands.py

@@ -5,6 +5,7 @@ import shutil
 import signal
 from contextlib import suppress
 from pathlib import Path
+from types import SimpleNamespace
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
@@ -19,7 +20,7 @@ from nanobot.cron.service import CronJobSkippedError
 from nanobot.cron.session_turns import CRON_DEFER_UNTIL_IDLE_META, CRON_TRIGGER_META
 from nanobot.cron.types import CronJob, CronPayload
 from nanobot.cron.webui_metadata import cron_proactive_delivery_metadata
-from nanobot.providers.factory import ProviderSnapshot, make_provider
+from nanobot.providers.factory import ProviderSnapshot, make_provider, provider_signature
 from nanobot.providers.openai_codex_provider import _strip_model_prefix
 from nanobot.providers.registry import find_by_name
 from nanobot.webui.metadata import (
@@ -434,6 +435,154 @@ def test_provider_login_rejects_unknown_provider():
     assert "Unknown OAuth provider" in result.stdout
 
 
+def test_provider_login_can_set_openai_codex_as_main_provider(tmp_path):
+    config_path = tmp_path / "config.json"
+    called = False
+    original = cli_commands._LOGIN_HANDLERS["openai_codex"]
+
+    def fake_login() -> None:
+        nonlocal called
+        called = True
+
+    cli_commands._LOGIN_HANDLERS["openai_codex"] = fake_login
+    try:
+        result = runner.invoke(
+            app,
+            [
+                "provider",
+                "login",
+                "openai-codex",
+                "--set-main",
+                "--config",
+                str(config_path),
+            ],
+        )
+    finally:
+        cli_commands._LOGIN_HANDLERS["openai_codex"] = original
+
+    assert result.exit_code == 0
+    assert called is True
+    assert "Set openai-codex as the main provider" in result.stdout
+
+    saved = Config.model_validate(json.loads(config_path.read_text(encoding="utf-8")))
+    assert saved.agents.defaults.provider == "openai_codex"
+    assert saved.agents.defaults.model == "openai-codex/gpt-5.4-mini"
+    assert saved.agents.defaults.model_preset is None
+    assert make_provider(saved).__class__.__name__ == "OpenAICodexProvider"
+
+
+def test_provider_login_can_set_github_copilot_as_main_provider(tmp_path):
+    config_path = tmp_path / "config.json"
+    original = cli_commands._LOGIN_HANDLERS["github_copilot"]
+    cli_commands._LOGIN_HANDLERS["github_copilot"] = lambda: None
+    try:
+        result = runner.invoke(
+            app,
+            [
+                "provider",
+                "login",
+                "github-copilot",
+                "--set-main",
+                "--config",
+                str(config_path),
+            ],
+        )
+    finally:
+        cli_commands._LOGIN_HANDLERS["github_copilot"] = original
+
+    assert result.exit_code == 0
+    assert "Set github-copilot as the main provider" in result.stdout
+
+    saved = Config.model_validate(json.loads(config_path.read_text(encoding="utf-8")))
+    assert saved.agents.defaults.provider == "github_copilot"
+    assert saved.agents.defaults.model == "github-copilot/gpt-5.4-mini"
+    assert saved.agents.defaults.model_preset is None
+    assert make_provider(saved).__class__.__name__ == "GitHubCopilotProvider"
+
+
+def test_provider_login_model_implies_set_main_provider(tmp_path):
+    config_path = tmp_path / "config.json"
+    original = cli_commands._LOGIN_HANDLERS["github_copilot"]
+    cli_commands._LOGIN_HANDLERS["github_copilot"] = lambda: None
+    try:
+        result = runner.invoke(
+            app,
+            [
+                "provider",
+                "login",
+                "github-copilot",
+                "--model",
+                "github-copilot/gpt-5.4-mini",
+                "--config",
+                str(config_path),
+            ],
+        )
+    finally:
+        cli_commands._LOGIN_HANDLERS["github_copilot"] = original
+
+    assert result.exit_code == 0
+    assert "Set github-copilot as the main provider" in result.stdout
+
+    saved = Config.model_validate(json.loads(config_path.read_text(encoding="utf-8")))
+    assert saved.agents.defaults.provider == "github_copilot"
+    assert saved.agents.defaults.model == "github-copilot/gpt-5.4-mini"
+    assert make_provider(saved).__class__.__name__ == "GitHubCopilotProvider"
+
+
+def test_provider_login_openai_codex_passes_configured_proxy(monkeypatch):
+    proxy = "http://127.0.0.1:23458"
+    monkeypatch.setattr(
+        "nanobot.config.loader.load_config",
+        lambda: Config.model_validate({"providers": {"openaiCodex": {"proxy": proxy}}}),
+    )
+
+    import oauth_cli_kit
+
+    def fake_get_token(**_kwargs):
+        raise RuntimeError("no-token")
+
+    monkeypatch.setattr(oauth_cli_kit, "get_token", fake_get_token)
+
+    captured: dict[str, str | None] = {}
+
+    def fake_login(*, print_fn, prompt_fn, proxy=None):
+        captured["proxy"] = proxy
+        return SimpleNamespace(access="access-token", account_id="acct-test")
+
+    monkeypatch.setattr(oauth_cli_kit, "login_oauth_interactive", fake_login)
+
+    result = runner.invoke(app, ["provider", "login", "openai-codex"])
+
+    assert result.exit_code == 0
+    assert captured["proxy"] == proxy
+
+
+def test_provider_login_openai_codex_resolves_proxy_env_ref(monkeypatch):
+    proxy = "http://127.0.0.1:23458"
+    monkeypatch.setenv("CODEX_PROXY_FOR_TEST", proxy)
+    monkeypatch.setattr(
+        "nanobot.config.loader.load_config",
+        lambda: Config.model_validate(
+            {"providers": {"openaiCodex": {"proxy": "${CODEX_PROXY_FOR_TEST}"}}}
+        ),
+    )
+
+    import oauth_cli_kit
+
+    captured: dict[str, str | None] = {}
+
+    def fake_get_token(*, proxy=None):
+        captured["proxy"] = proxy
+        return SimpleNamespace(access="access-token", account_id="acct-test")
+
+    monkeypatch.setattr(oauth_cli_kit, "get_token", fake_get_token)
+
+    result = runner.invoke(app, ["provider", "login", "openai-codex"])
+
+    assert result.exit_code == 0
+    assert captured["proxy"] == proxy
+
+
 def test_config_matches_explicit_ollama_prefix_without_api_key():
     config = Config()
     config.agents.defaults.model = "ollama/llama3.2"
@@ -685,6 +834,54 @@ def test_make_provider_uses_github_copilot_backend():
     assert provider.__class__.__name__ == "GitHubCopilotProvider"
 
 
+def test_openai_codex_proxy_config_affects_provider_and_signature():
+    def config_with_proxy(proxy: str) -> Config:
+        return Config.model_validate(
+            {
+                "agents": {
+                    "defaults": {
+                        "provider": "openai-codex",
+                        "model": "openai-codex/gpt-5.5",
+                    }
+                },
+                "providers": {"openaiCodex": {"proxy": proxy}},
+            }
+        )
+
+    proxy = "http://127.0.0.1:23458"
+    config = config_with_proxy(proxy)
+
+    provider = make_provider(config)
+
+    assert provider.__class__.__name__ == "OpenAICodexProvider"
+    assert provider.proxy == proxy
+    assert provider_signature(config) != provider_signature(
+        config_with_proxy("http://127.0.0.1:23459")
+    )
+
+
+def test_provider_proxy_rejects_unsupported_backend():
+    config = Config.model_validate(
+        {
+            "agents": {
+                "defaults": {
+                    "provider": "anthropic",
+                    "model": "anthropic/claude-opus-4-5",
+                }
+            },
+            "providers": {
+                "anthropic": {
+                    "apiKey": "sk-test",
+                    "proxy": "http://127.0.0.1:23458",
+                }
+            },
+        }
+    )
+
+    with pytest.raises(ValueError, match=r"providers\.anthropic\.proxy"):
+        make_provider(config)
+
+
 def test_github_copilot_provider_strips_prefixed_model_name():
     from nanobot.providers.github_copilot_provider import GitHubCopilotProvider
 
@@ -752,7 +949,7 @@ def test_make_provider_passes_extra_headers_to_custom_provider():
                         "x-session-affinity": "sticky-session",
                     },
                 }
-            },
+            }
         }
     )
 

tests/cli/test_restart_command.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 
 import asyncio
 import os
+import sys
 import time
 from types import SimpleNamespace
 from unittest.mock import AsyncMock, MagicMock, patch
@@ -44,7 +45,8 @@ class TestRestartCommand:
             RESTART_STARTED_AT_ENV,
         )
 
-        loop, bus = _make_loop()
+        loop, _bus = _make_loop()
+        loop.restart_mode = "exec"
         msg = InboundMessage(channel="cli", sender_id="user", chat_id="direct", content="/restart")
         ctx = CommandContext(msg=msg, session=None, key=msg.session_key, raw="/restart", loop=loop)
 
@@ -76,10 +78,75 @@ class TestRestartCommand:
             await scheduled[0]
             mock_execv.assert_called_once()
 
+    @pytest.mark.asyncio
+    async def test_restart_windows_auto_spawns_and_exits(self):
+        from nanobot.command.builtin import cmd_restart
+        from nanobot.command.router import CommandContext
+
+        loop, _bus = _make_loop()
+        msg = InboundMessage(channel="cli", sender_id="user", chat_id="direct", content="/restart")
+        ctx = CommandContext(msg=msg, session=None, key=msg.session_key, raw="/restart", loop=loop)
+
+        async def _fast_sleep(_delay: float) -> None:
+            return None
+
+        scheduled: list[asyncio.Task] = []
+        fake_asyncio = SimpleNamespace(
+            sleep=_fast_sleep,
+            create_task=lambda coro: scheduled.append(asyncio.create_task(coro)) or scheduled[-1],
+        )
+
+        with patch("nanobot.command.builtin.asyncio", new=fake_asyncio), \
+             patch("nanobot.command.builtin.sys.platform", "win32"), \
+             patch("nanobot.command.builtin.subprocess.CREATE_NEW_PROCESS_GROUP", 512, create=True), \
+             patch("nanobot.command.builtin.subprocess.Popen") as mock_popen, \
+             patch("nanobot.command.builtin.os._exit") as mock_exit, \
+             patch("nanobot.command.builtin.os.execv") as mock_execv:
+            await cmd_restart(ctx)
+            await scheduled[0]
+
+        mock_popen.assert_called_once_with(
+            [sys.executable, "-m", "nanobot"] + sys.argv[1:],
+            creationflags=512,
+        )
+        mock_exit.assert_called_once_with(0)
+        mock_execv.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_restart_exit_mode_does_not_spawn(self):
+        from nanobot.command.builtin import cmd_restart
+        from nanobot.command.router import CommandContext
+
+        loop, _bus = _make_loop()
+        loop.restart_mode = "exit"
+        msg = InboundMessage(channel="cli", sender_id="user", chat_id="direct", content="/restart")
+        ctx = CommandContext(msg=msg, session=None, key=msg.session_key, raw="/restart", loop=loop)
+
+        async def _fast_sleep(_delay: float) -> None:
+            return None
+
+        scheduled: list[asyncio.Task] = []
+        fake_asyncio = SimpleNamespace(
+            sleep=_fast_sleep,
+            create_task=lambda coro: scheduled.append(asyncio.create_task(coro)) or scheduled[-1],
+        )
+
+        with patch("nanobot.command.builtin.asyncio", new=fake_asyncio), \
+             patch("nanobot.command.builtin.subprocess.Popen") as mock_popen, \
+             patch("nanobot.command.builtin.os._exit") as mock_exit, \
+             patch("nanobot.command.builtin.os.execv") as mock_execv:
+            await cmd_restart(ctx)
+            await scheduled[0]
+
+        mock_exit.assert_called_once_with(0)
+        mock_popen.assert_not_called()
+        mock_execv.assert_not_called()
+
     @pytest.mark.asyncio
     async def test_restart_intercepted_in_run_loop(self):
         """Verify /restart is handled at the run-loop level, not inside _dispatch."""
         loop, bus = _make_loop()
+        loop.restart_mode = "exec"
         msg = InboundMessage(channel="telegram", sender_id="u1", chat_id="c1", content="/restart")
 
         async def _fast_sleep(_delay: float) -> None:

tests/command/test_stop_pending_queue.py

@@ -0,0 +1,81 @@
+"""Test cmd_stop drains pending queue to prevent mid-turn injection deadlock."""
+
+import asyncio
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from nanobot.bus.events import OutboundMessage
+from nanobot.command.builtin import cmd_stop
+from nanobot.command.router import CommandContext
+
+
+@pytest.mark.asyncio
+async def test_cmd_stop_drains_pending_queue():
+    """cmd_stop should drain pending queue in addition to cancelling active tasks."""
+    mock_loop = MagicMock()
+    mock_loop._cancel_active_tasks = AsyncMock(return_value=1)
+    mock_loop._pending_queues = {}
+
+    pending = asyncio.Queue()
+    await pending.put("msg1")
+    await pending.put("msg2")
+    mock_loop._pending_queues["test-session"] = pending
+
+    ctx = CommandContext(
+        msg=MagicMock(channel="websocket", chat_id="test-chat", metadata={}),
+        session=None,
+        key="test-session",
+        raw="/stop",
+        loop=mock_loop,
+    )
+
+    result = await cmd_stop(ctx)
+
+    assert isinstance(result, OutboundMessage)
+    assert "Stopped 3 task(s)" in result.content  # 1 cancelled + 2 drained
+    assert "test-session" not in mock_loop._pending_queues
+
+
+@pytest.mark.asyncio
+async def test_cmd_stop_with_empty_pending_queue():
+    """cmd_stop should work correctly when pending queue is empty."""
+    mock_loop = MagicMock()
+    mock_loop._cancel_active_tasks = AsyncMock(return_value=2)
+    mock_loop._pending_queues = {}
+
+    pending = asyncio.Queue()
+    mock_loop._pending_queues["test-session"] = pending
+
+    ctx = CommandContext(
+        msg=MagicMock(channel="websocket", chat_id="test-chat", metadata={}),
+        session=None,
+        key="test-session",
+        raw="/stop",
+        loop=mock_loop,
+    )
+
+    result = await cmd_stop(ctx)
+
+    assert "Stopped 2 task(s)" in result.content
+    assert "test-session" not in mock_loop._pending_queues
+
+
+@pytest.mark.asyncio
+async def test_cmd_stop_no_pending_queue():
+    """cmd_stop should work when no pending queue exists."""
+    mock_loop = MagicMock()
+    mock_loop._cancel_active_tasks = AsyncMock(return_value=0)
+    mock_loop._pending_queues = {}
+
+    ctx = CommandContext(
+        msg=MagicMock(channel="websocket", chat_id="test-chat", metadata={}),
+        session=None,
+        key="test-session",
+        raw="/stop",
+        loop=mock_loop,
+    )
+
+    result = await cmd_stop(ctx)
+
+    assert "No active task to stop" in result.content

tests/config/test_config_migration.py

@@ -2,6 +2,8 @@ import json
 import socket
 from unittest.mock import patch
 
+import pytest
+
 from nanobot.config.loader import load_config, save_config
 from nanobot.security.network import validate_url_target
 
@@ -93,6 +95,41 @@ def test_onboard_does_not_crash_with_legacy_memory_window(tmp_path, monkeypatch)
     assert result.exit_code == 0
 
 
+@pytest.mark.parametrize("field_name", ["maxMessages", "max_messages"])
+def test_load_config_warns_and_ignores_legacy_max_messages(tmp_path, field_name) -> None:
+    config_path = tmp_path / "config.json"
+    config_path.write_text(
+        json.dumps({"agents": {"defaults": {field_name: 25, "maxTokens": 1234}}}),
+        encoding="utf-8",
+    )
+
+    with patch("nanobot.config.loader.logger.warning") as warning:
+        config = load_config(config_path)
+
+    assert config.agents.defaults.max_tokens == 1234
+    assert not hasattr(config.agents.defaults, "max_messages")
+    warning.assert_called_once()
+    message = warning.call_args.args[0]
+    assert "legacy and ignored" in message
+    assert "next version" in message
+
+
+def test_save_config_drops_legacy_max_messages(tmp_path) -> None:
+    config_path = tmp_path / "config.json"
+    config_path.write_text(
+        json.dumps({"agents": {"defaults": {"maxMessages": 25}}}),
+        encoding="utf-8",
+    )
+
+    with patch("nanobot.config.loader.logger.warning"):
+        config = load_config(config_path)
+    save_config(config, config_path)
+    saved = json.loads(config_path.read_text(encoding="utf-8"))
+
+    assert "maxMessages" not in saved["agents"]["defaults"]
+    assert "max_messages" not in saved["agents"]["defaults"]
+
+
 def test_onboard_refresh_backfills_missing_channel_fields(tmp_path, monkeypatch) -> None:
     from types import SimpleNamespace
 

tests/config/test_env_interpolation.py

@@ -8,6 +8,7 @@ from nanobot.config.loader import (
     resolve_config_env_vars,
     save_config,
 )
+from nanobot.config.schema import Config
 
 
 class TestResolveEnvVars:
@@ -127,6 +128,31 @@ class TestResolveConfig:
         assert "githubCopilot" not in saved["providers"]
         assert saved["providers"]["groq"]["apiKey"] == "groq-secret"
 
+    def test_save_preserves_openai_codex_proxy_config(self, tmp_path):
+        config_path = tmp_path / "config.json"
+        proxy = "http://127.0.0.1:23458"
+        config = Config.model_validate(
+            {
+                "providers": {
+                    "openaiCodex": {
+                        "apiKey": "codex-secret",
+                        "proxy": proxy,
+                    },
+                    "groq": {"apiKey": "groq-secret"},
+                }
+            }
+        )
+
+        save_config(config, config_path)
+
+        saved = json.loads(config_path.read_text(encoding="utf-8"))
+        assert saved["providers"]["openaiCodex"] == {"proxy": proxy}
+        assert saved["providers"]["groq"]["apiKey"] == "groq-secret"
+
+        reloaded = load_config(config_path)
+        assert reloaded.providers.openai_codex.proxy == proxy
+        assert reloaded.providers.openai_codex.api_key is None
+
     def test_preserves_excluded_fields_when_no_env_refs(self, tmp_path):
         """Regression: fields with ``exclude=True`` (e.g. ProviderConfig.openai_codex)
         must survive ``resolve_config_env_vars`` when the config has no

tests/config/test_gateway_config.py

@@ -0,0 +1,15 @@
+import pytest
+
+from nanobot.config.schema import Config, GatewayConfig
+
+
+def test_gateway_restart_mode_accepts_camel_alias():
+    config = Config.model_validate({"gateway": {"restartMode": "exit"}})
+
+    assert config.gateway.restart_mode == "exit"
+    assert config.model_dump(by_alias=True)["gateway"]["restartMode"] == "exit"
+
+
+def test_gateway_restart_mode_rejects_unknown_value():
+    with pytest.raises(ValueError):
+        GatewayConfig(restart_mode="service")

tests/cron/test_cron_persistence.py

@@ -10,11 +10,12 @@ from __future__ import annotations
 
 import json
 from pathlib import Path
+from typing import Callable
 
 import pytest
 
 from nanobot.cron.service import CronService
-from nanobot.cron.types import CronSchedule
+from nanobot.cron.types import CronJob, CronPayload, CronSchedule
 
 
 def _seeded_store(tmp_path: Path) -> tuple[CronService, Path]:
@@ -41,6 +42,29 @@ def _seeded_store(tmp_path: Path) -> tuple[CronService, Path]:
     return service, store_path
 
 
+def _corrupt_store(tmp_path: Path) -> Path:
+    store_path = tmp_path / "cron" / "jobs.json"
+    store_path.parent.mkdir(parents=True)
+    store_path.write_text("{not valid json", encoding="utf-8")
+    return store_path
+
+
+def _assert_single_corrupt_backup(store_path: Path) -> None:
+    assert not store_path.exists()
+    backups = list(store_path.parent.glob("jobs.json.corrupt-*"))
+    assert len(backups) == 1
+    assert backups[0].read_text(encoding="utf-8") == "{not valid json"
+
+
+def _system_job(job_id: str = "dream") -> CronJob:
+    return CronJob(
+        id=job_id,
+        name="Dream",
+        schedule=CronSchedule(kind="cron", expr="0 */2 * * *", tz="UTC"),
+        payload=CronPayload(kind="system_event"),
+    )
+
+
 def test_save_store_is_atomic(tmp_path: Path) -> None:
     """``_save_store`` must use temp-file + rename so an interrupted write
     cannot leave the destination truncated or invalid."""
@@ -148,6 +172,126 @@ def test_load_store_falls_back_to_in_memory_on_corruption_after_start(
     assert result.jobs[0].name == "Daily Loving Message"
 
 
+@pytest.mark.parametrize(
+    ("api_name", "call"),
+    [
+        ("list_jobs", lambda service: service.list_jobs()),
+        ("get_job", lambda service: service.get_job("missing")),
+        ("status", lambda service: service.status()),
+        ("remove_job", lambda service: service.remove_job("missing")),
+        ("enable_job", lambda service: service.enable_job("missing", enabled=False)),
+        ("update_job", lambda service: service.update_job("missing", name="new name")),
+        ("register_system_job", lambda service: service.register_system_job(_system_job())),
+    ],
+)
+def test_public_apis_raise_clear_error_for_unavailable_corrupt_store(
+    tmp_path: Path,
+    api_name: str,
+    call: Callable[[CronService], object],
+) -> None:
+    """Public APIs should report the corrupt store explicitly instead of
+    leaking ``AttributeError`` when the first load cannot produce a store."""
+    store_path = _corrupt_store(tmp_path)
+    service = CronService(store_path)
+
+    with pytest.raises(RuntimeError, match="corrupt.*restore jobs.json") as exc_info:
+        call(service)
+
+    assert api_name
+    assert str(store_path) in str(exc_info.value)
+    _assert_single_corrupt_backup(store_path)
+
+
+@pytest.mark.asyncio
+async def test_run_job_raises_clear_error_and_restores_running_state_for_corrupt_store(
+    tmp_path: Path,
+) -> None:
+    store_path = _corrupt_store(tmp_path)
+    service = CronService(store_path)
+
+    with pytest.raises(RuntimeError, match="corrupt.*restore jobs.json"):
+        await service.run_job("missing")
+
+    assert service._running is False
+    _assert_single_corrupt_backup(store_path)
+
+
+@pytest.mark.asyncio
+async def test_run_job_preserves_running_state_when_corrupt_store_unavailable(
+    tmp_path: Path,
+) -> None:
+    store_path = _corrupt_store(tmp_path)
+    service = CronService(store_path)
+    service._running = True
+    service._arm_timer = lambda: None
+
+    with pytest.raises(RuntimeError, match="corrupt.*restore jobs.json"):
+        await service.run_job("missing")
+
+    assert service._running is True
+    service.stop()
+
+
+def test_running_add_job_raises_clear_error_for_unavailable_corrupt_store(
+    tmp_path: Path,
+) -> None:
+    store_path = _corrupt_store(tmp_path)
+    service = CronService(store_path)
+    service._running = True
+
+    with pytest.raises(RuntimeError, match="corrupt.*restore jobs.json"):
+        service.add_job(
+            name="running add",
+            schedule=CronSchedule(kind="every", every_ms=60_000),
+            message="hello",
+            session_key="websocket:chat-1",
+            origin_channel="websocket",
+            origin_chat_id="chat-1",
+        )
+
+    _assert_single_corrupt_backup(store_path)
+
+
+def test_stopped_add_job_still_appends_action_without_loading_corrupt_store(
+    tmp_path: Path,
+) -> None:
+    """The stopped-service add path is an action-log write and must not start
+    requiring a readable store."""
+    store_path = _corrupt_store(tmp_path)
+    service = CronService(store_path)
+
+    job = service.add_job(
+        name="offline add",
+        schedule=CronSchedule(kind="every", every_ms=60_000),
+        message="hello",
+        session_key="websocket:chat-1",
+        origin_channel="websocket",
+        origin_chat_id="chat-1",
+    )
+
+    assert job.name == "offline add"
+    assert store_path.exists()
+    assert store_path.read_text(encoding="utf-8") == "{not valid json"
+    assert list(store_path.parent.glob("jobs.json.corrupt-*")) == []
+    actions = (store_path.parent / "action.jsonl").read_text(encoding="utf-8").splitlines()
+    assert len(actions) == 1
+    assert json.loads(actions[0])["action"] == "add"
+
+
+def test_public_api_uses_in_memory_snapshot_when_disk_becomes_corrupt(
+    tmp_path: Path,
+) -> None:
+    service, store_path = _seeded_store(tmp_path)
+    service._load_store()
+    assert service._store is not None
+    store_path.write_text("{not valid json", encoding="utf-8")
+
+    jobs = service.list_jobs(include_disabled=True)
+
+    assert len(jobs) == 1
+    assert jobs[0].name == "Daily Loving Message"
+
+
 def test_full_round_trip_survives_repeated_save_load(tmp_path: Path) -> None:
     """Sanity check: jobs survive add → save → reload across fresh
     ``CronService`` instances pointing at the same store."""

tests/providers/test_anthropic_tool_result.py

@@ -70,7 +70,7 @@ def test_convert_user_content_coerces_typeless_dict():
         {"foo": "bar"},
         {"type": "text", "text": "ok"},
     ])
-    assert result[0] == {"type": "text", "text": str({"foo": "bar"})}
+    assert result[0] == {"type": "text", "text": '{"foo": "bar"}'}
     assert result[1] == {"type": "text", "text": "ok"}
 
 
@@ -81,7 +81,16 @@ def test_convert_user_content_coerces_mixed_typeless():
         {"key": "val"},
     ])
     assert result[0] == {"type": "text", "text": "42"}
-    assert result[1] == {"type": "text", "text": str({"key": "val"})}
+    assert result[1] == {"type": "text", "text": '{"key": "val"}'}
+
+
+def test_assistant_blocks_coerce_typeless_dict_to_json_text():
+    blocks = AnthropicProvider._assistant_blocks({
+        "role": "assistant",
+        "content": [{"answer": "ok", "count": 2}],
+    })
+
+    assert blocks == [{"type": "text", "text": '{"answer": "ok", "count": 2}'}]
 
 
 def test_convert_assistant_message_repairs_history_tool_arguments():

tests/providers/test_github_copilot_enterprise.py

@@ -0,0 +1,143 @@
+"""Regression tests for GitHub Enterprise / Copilot for Business endpoint overrides (#4220)."""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+
+import pytest
+
+from nanobot.providers import github_copilot_provider as gc
+
+
+def test_resolve_falls_back_to_default_without_env(monkeypatch):
+    monkeypatch.delenv("NANOBOT_COPILOT_BASE_URL", raising=False)
+    assert gc._resolve("NANOBOT_COPILOT_BASE_URL", gc.DEFAULT_COPILOT_BASE_URL) == (
+        gc.DEFAULT_COPILOT_BASE_URL
+    )
+
+
+def test_resolve_uses_env_override_and_strips(monkeypatch):
+    monkeypatch.setenv("NANOBOT_COPILOT_TOKEN_URL", "  https://api.acme.ghe.com/copilot_internal/v2/token  ")
+    assert gc._resolve("NANOBOT_COPILOT_TOKEN_URL", gc.DEFAULT_COPILOT_TOKEN_URL) == (
+        "https://api.acme.ghe.com/copilot_internal/v2/token"
+    )
+
+
+def test_blank_env_override_falls_back_to_default(monkeypatch):
+    monkeypatch.setenv("NANOBOT_COPILOT_BASE_URL", "   ")
+    assert gc._resolve("NANOBOT_COPILOT_BASE_URL", gc.DEFAULT_COPILOT_BASE_URL) == (
+        gc.DEFAULT_COPILOT_BASE_URL
+    )
+
+
+def test_provider_api_base_honors_env_override(monkeypatch):
+    monkeypatch.setenv("NANOBOT_COPILOT_BASE_URL", "https://copilot-api.acme.ghe.com")
+    provider = gc.GitHubCopilotProvider()
+    assert provider.api_base == "https://copilot-api.acme.ghe.com"
+
+
+def test_login_uses_enterprise_endpoint_overrides(monkeypatch):
+    monkeypatch.setenv("NANOBOT_GITHUB_COPILOT_CLIENT_ID", "enterprise-client-id")
+    monkeypatch.setenv("NANOBOT_GITHUB_DEVICE_CODE_URL", "https://ghe.example/login/device/code")
+    monkeypatch.setenv(
+        "NANOBOT_GITHUB_ACCESS_TOKEN_URL",
+        "https://ghe.example/login/oauth/access_token",
+    )
+    monkeypatch.setenv("NANOBOT_GITHUB_USER_URL", "https://api.ghe.example/user")
+    monkeypatch.setattr(gc.webbrowser, "open", lambda _url: None)
+
+    calls = []
+    saved = []
+
+    class FakeResponse:
+        def __init__(self, payload):
+            self._payload = payload
+
+        def raise_for_status(self):
+            pass
+
+        def json(self):
+            return self._payload
+
+    class FakeClient:
+        def __init__(self, *args, **kwargs):
+            pass
+
+        def __enter__(self):
+            return self
+
+        def __exit__(self, *args):
+            return False
+
+        def post(self, url, *, headers, data):
+            calls.append(("post", url, data))
+            if url.endswith("/device/code"):
+                return FakeResponse(
+                    {
+                        "device_code": "device-code",
+                        "user_code": "user-code",
+                        "verification_uri": "https://ghe.example/device",
+                        "interval": 1,
+                        "expires_in": 60,
+                    }
+                )
+            return FakeResponse({"access_token": "github-token", "expires_in": 3600})
+
+        def get(self, url, *, headers):
+            calls.append(("get", url, headers))
+            return FakeResponse({"login": "enterprise-user"})
+
+    monkeypatch.setattr(gc.httpx, "Client", FakeClient)
+    monkeypatch.setattr(gc, "get_storage", lambda: SimpleNamespace(save=saved.append))
+
+    token = gc.login_github_copilot(print_fn=lambda _message: None)
+
+    assert token.access == "github-token"
+    assert saved[0].account_id == "enterprise-user"
+    assert calls[0] == (
+        "post",
+        "https://ghe.example/login/device/code",
+        {"client_id": "enterprise-client-id", "scope": gc.GITHUB_COPILOT_SCOPE},
+    )
+    assert calls[1][0:2] == ("post", "https://ghe.example/login/oauth/access_token")
+    assert calls[1][2]["client_id"] == "enterprise-client-id"
+    assert calls[2][0:2] == ("get", "https://api.ghe.example/user")
+
+
+@pytest.mark.asyncio
+async def test_copilot_token_exchange_uses_enterprise_endpoint_override(monkeypatch):
+    monkeypatch.setenv(
+        "NANOBOT_COPILOT_TOKEN_URL",
+        "https://api.ghe.example/copilot_internal/v2/token",
+    )
+    monkeypatch.setattr(gc, "_load_github_token", lambda: SimpleNamespace(access="github-token"))
+
+    calls = []
+
+    class FakeResponse:
+        def raise_for_status(self):
+            pass
+
+        def json(self):
+            return {"token": "copilot-token", "refresh_in": 120}
+
+    class FakeAsyncClient:
+        def __init__(self, *args, **kwargs):
+            pass
+
+        async def __aenter__(self):
+            return self
+
+        async def __aexit__(self, *args):
+            return False
+
+        async def get(self, url, *, headers):
+            calls.append((url, headers))
+            return FakeResponse()
+
+    monkeypatch.setattr(gc.httpx, "AsyncClient", FakeAsyncClient)
+
+    provider = gc.GitHubCopilotProvider()
+
+    assert await provider._get_copilot_access_token() == "copilot-token"
+    assert calls[0][0] == "https://api.ghe.example/copilot_internal/v2/token"

tests/providers/test_openai_codex_provider.py

@@ -21,9 +21,12 @@ from nanobot.providers.openai_codex_provider import (
 
 
 def _mock_codex_token(monkeypatch: pytest.MonkeyPatch) -> None:
+    def fake_token(**_kwargs):
+        return SimpleNamespace(account_id="acct", access="token")
+
     monkeypatch.setattr(
         "nanobot.providers.openai_codex_provider.get_codex_token",
-        lambda: SimpleNamespace(account_id="acct", access="token"),
+        fake_token,
     )
 
 
@@ -77,7 +80,12 @@ async def test_codex_request_non_200_populates_http_metadata(monkeypatch) -> Non
             request=request,
         )
 
-    def fake_client(*, timeout: int, verify: bool) -> httpx.AsyncClient:
+    def fake_client(
+        *,
+        timeout: int,
+        verify: bool,
+        **_kwargs: object,
+    ) -> httpx.AsyncClient:
         assert timeout == 90
         assert verify is True
         return original_client(transport=httpx.MockTransport(handler), timeout=timeout)
@@ -106,7 +114,12 @@ async def test_codex_request_honors_stream_idle_timeout_env(monkeypatch) -> None
     def handler(request: httpx.Request) -> httpx.Response:
         return httpx.Response(200, request=request)
 
-    def fake_client(*, timeout: int, verify: bool) -> httpx.AsyncClient:
+    def fake_client(
+        *,
+        timeout: int,
+        verify: bool,
+        **_kwargs: object,
+    ) -> httpx.AsyncClient:
         seen["timeout"] = timeout
         return original_client(transport=httpx.MockTransport(handler), timeout=timeout)
 
@@ -117,6 +130,39 @@ async def test_codex_request_honors_stream_idle_timeout_env(monkeypatch) -> None
     assert seen["timeout"] == 5
 
 
+@pytest.mark.asyncio
+async def test_codex_request_uses_configured_proxy(monkeypatch) -> None:
+    original_client = httpx.AsyncClient
+    seen: dict[str, object] = {}
+    proxy = "http://127.0.0.1:23458"
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        return httpx.Response(200, request=request)
+
+    def fake_client(
+        *,
+        timeout: int,
+        verify: bool,
+        proxy: str | None = None,
+        trust_env: bool = True,
+    ) -> httpx.AsyncClient:
+        seen["proxy"] = proxy
+        seen["trust_env"] = trust_env
+        return original_client(transport=httpx.MockTransport(handler), timeout=timeout)
+
+    monkeypatch.setattr("nanobot.providers.openai_codex_provider.httpx.AsyncClient", fake_client)
+
+    await _request_codex(
+        "https://codex.example/responses",
+        {},
+        {"input": []},
+        verify=True,
+        proxy=proxy,
+    )
+
+    assert seen == {"proxy": proxy, "trust_env": False}
+
+
 @pytest.mark.asyncio
 async def test_codex_prompt_cache_key_uses_stable_conversation_prefix(monkeypatch) -> None:
     bodies: list[dict] = []
@@ -128,11 +174,12 @@ async def test_codex_prompt_cache_key_uses_stable_conversation_prefix(monkeypatc
         headers,
         body,
         verify,
+        proxy=None,
         on_content_delta=None,
         on_thinking_delta=None,
         on_tool_call_delta=None,
     ):
-        _ = on_thinking_delta, on_tool_call_delta
+        _ = proxy, on_thinking_delta, on_tool_call_delta
         bodies.append(body)
         return "ok", [], "stop", {}, None
 
@@ -186,6 +233,40 @@ async def test_codex_timeout_error_is_typed_and_retryable(monkeypatch) -> None:
     assert response.error_should_retry is True
 
 
+@pytest.mark.asyncio
+async def test_codex_provider_passes_proxy_to_oauth_and_response_request(monkeypatch) -> None:
+    proxy = "http://127.0.0.1:23458"
+    seen: dict[str, object] = {}
+
+    def fake_token(*, proxy=None):
+        seen["token_proxy"] = proxy
+        return SimpleNamespace(account_id="acct", access="token")
+
+    async def fake_request(
+        url,
+        headers,
+        body,
+        verify,
+        proxy=None,
+        on_content_delta=None,
+        on_thinking_delta=None,
+        on_tool_call_delta=None,
+    ):
+        _ = url, headers, body, verify, on_content_delta, on_thinking_delta, on_tool_call_delta
+        seen["request_proxy"] = proxy
+        return "ok", [], "stop", {}, None
+
+    monkeypatch.setattr("nanobot.providers.openai_codex_provider.get_codex_token", fake_token)
+    monkeypatch.setattr("nanobot.providers.openai_codex_provider._request_codex", fake_request)
+
+    provider = OpenAICodexProvider(proxy=proxy)
+    response = await provider.chat([{"role": "user", "content": "hello"}])
+
+    assert response.content == "ok"
+    assert seen["token_proxy"] == proxy
+    assert seen["request_proxy"] == proxy
+
+
 @pytest.mark.asyncio
 async def test_codex_timeout_error_writes_diagnostic_log(monkeypatch) -> None:
     log_capture = _capture_codex_warnings(monkeypatch)
@@ -409,9 +490,12 @@ def test_codex_reasoning_options_request_summary_without_forcing_effort() -> Non
 
 @pytest.mark.asyncio
 async def test_codex_stream_surfaces_reasoning_summary(monkeypatch) -> None:
+    def fake_token(**_kwargs):
+        return SimpleNamespace(account_id="acct", access="token")
+
     monkeypatch.setattr(
         "nanobot.providers.openai_codex_provider.get_codex_token",
-        lambda: SimpleNamespace(account_id="acct", access="token"),
+        fake_token,
     )
 
     async def fake_request(
@@ -419,11 +503,12 @@ async def test_codex_stream_surfaces_reasoning_summary(monkeypatch) -> None:
         headers,
         body,
         verify,
+        proxy=None,
         on_content_delta=None,
         on_thinking_delta=None,
         on_tool_call_delta=None,
     ):
-        _ = url, headers, verify, on_tool_call_delta
+        _ = url, headers, verify, proxy, on_tool_call_delta
         assert body["reasoning"] == {"summary": "auto", "effort": "medium"}
         if on_content_delta:
             await on_content_delta("answer")

tests/providers/test_proxy_env.py

@@ -4,6 +4,7 @@ from unittest.mock import MagicMock
 
 import httpx
 
+import nanobot.providers.openai_compat_provider as openai_compat_provider
 from nanobot.providers.openai_compat_provider import OpenAICompatProvider
 
 
@@ -54,3 +55,32 @@ class TestCloudEndpointProxyEnabled:
         client = provider._client._client
         # trust_env should be True so httpx reads HTTP_PROXY etc.
         assert client._trust_env is True
+
+    async def test_explicit_provider_proxy_overrides_env(self, monkeypatch):
+        spec = _make_spec(is_local=False)
+        spec.env_key = ""
+        spec.default_api_base = "https://api.openai.com/v1"
+        proxy = "http://127.0.0.1:23458"
+        monkeypatch.delenv("NANOBOT_OPENAI_COMPAT_TIMEOUT_S", raising=False)
+
+        http_client = MagicMock()
+        async_client = MagicMock(return_value=http_client)
+        openai_client = MagicMock(return_value=object())
+        monkeypatch.setattr(httpx, "AsyncClient", async_client)
+        monkeypatch.setattr(openai_compat_provider, "AsyncOpenAI", openai_client)
+
+        provider = OpenAICompatProvider(
+            api_key="test",
+            api_base=None,
+            spec=spec,
+            proxy=proxy,
+        )
+        provider._build_client()
+
+        async_client.assert_called_once_with(
+            timeout=120.0,
+            proxy=proxy,
+            trust_env=False,
+            follow_redirects=True,
+        )
+        assert openai_client.call_args.kwargs["http_client"] is http_client

tests/session/test_session_list_repair_legacy.py

@@ -0,0 +1,40 @@
+"""Reproduction test: list_sessions drops corrupt legacy-stem sessions during repair."""
+import json
+from datetime import datetime
+from pathlib import Path
+
+from nanobot.session.manager import SessionManager
+
+
+def test_list_sessions_repairs_corrupt_legacy_stem(tmp_path: Path, monkeypatch) -> None:
+    monkeypatch.setattr(
+        "nanobot.session.manager.get_legacy_sessions_dir",
+        lambda: tmp_path / "legacy_sessions",
+    )
+    manager = SessionManager(tmp_path / "workspace")
+
+    # Simulate a legacy lossy-path filename (telegram_12345.jsonl) with a corrupt
+    # first line that triggers the repair branch in list_sessions.
+    legacy_stem = "telegram_12345"
+    corrupt_path = manager.sessions_dir / f"{legacy_stem}.jsonl"
+    corrupt_path.parent.mkdir(parents=True, exist_ok=True)
+    metadata = json.dumps({
+        "_type": "metadata",
+        "key": "telegram:12345",
+        "created_at": datetime(2025, 1, 1).isoformat(),
+        "updated_at": datetime(2025, 1, 1).isoformat(),
+    })
+    # Corrupt line followed by valid message
+    corrupt_path.write_text(
+        metadata + "\n{INVALID JSON LINE\n"
+        + json.dumps({"role": "user", "content": "recoverable message"}) + "\n",
+        encoding="utf-8",
+    )
+
+    sessions = manager.list_sessions()
+
+    # BUG: repair fails because _repair re-encodes the fallback_key via
+    # _get_session_path, producing a base64 stem that doesn't match the
+    # actual legacy filename. The session is silently dropped.
+    assert len(sessions) == 1, f"Expected 1 session, got {len(sessions)}"
+    assert sessions[0]["key"] == "telegram:12345"

tests/tools/test_mcp_tool.py

@@ -1,8 +1,10 @@
 from __future__ import annotations
 
 import asyncio
+import json
 import sys
 from contextlib import asynccontextmanager
+from pathlib import Path
 from types import ModuleType, SimpleNamespace
 
 import httpx
@@ -36,6 +38,12 @@ class _FakeBlobResourceContents:
         self.blob = blob
 
 
+class _FakeImageContent:
+    def __init__(self, data: str, mime_type: str = "image/png") -> None:
+        self.data = data
+        self.mimeType = mime_type
+
+
 @pytest.fixture
 def fake_mcp_runtime() -> dict[str, object | None]:
     return {"session": None}
@@ -50,6 +58,7 @@ def _fake_mcp_module(
         TextContent=_FakeTextContent,
         TextResourceContents=_FakeTextResourceContents,
         BlobResourceContents=_FakeBlobResourceContents,
+        ImageContent=_FakeImageContent,
     )
 
     class _FakeStdioServerParameters:
@@ -295,6 +304,60 @@ async def test_execute_returns_text_blocks() -> None:
     assert result == "hello\n42"
 
 
+# Smallest valid 1x1 PNG, base64 without the data: prefix.
+_PNG_B64 = (
+    "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8"
+    "/x8AAwMCAO+/p9sAAAAASUVORK5CYII="
+)
+
+
+@pytest.mark.asyncio
+async def test_execute_persists_image_block_as_artifact(tmp_path: Path) -> None:
+    from nanobot.config.loader import set_config_path
+
+    set_config_path(tmp_path / "config.json")
+
+    async def call_tool(_name: str, arguments: dict) -> object:
+        return SimpleNamespace(
+            content=[
+                _FakeTextContent("here you go"),
+                _FakeImageContent(_PNG_B64, "image/png"),
+            ]
+        )
+
+    wrapper = _make_wrapper(SimpleNamespace(call_tool=call_tool))
+
+    result = await wrapper.execute(prompt="a cat", model="sdxl")
+
+    payload = json.loads(result)
+    assert payload["text"] == "here you go"
+    assert len(payload["artifacts"]) == 1
+    artifact = payload["artifacts"][0]
+    assert artifact["mime"] == "image/png"
+    assert artifact["prompt"] == "a cat"
+    assert artifact["provider"] == "mcp:test"
+    assert Path(artifact["path"]).is_file()
+    # The base64 payload must NOT leak into the model-facing result.
+    assert _PNG_B64 not in result
+    assert "message tool" in payload["next_step"]
+
+
+@pytest.mark.asyncio
+async def test_execute_notes_unstorable_image_block(tmp_path: Path) -> None:
+    from nanobot.config.loader import set_config_path
+
+    set_config_path(tmp_path / "config.json")
+
+    async def call_tool(_name: str, arguments: dict) -> object:
+        return SimpleNamespace(content=[_FakeImageContent("not-valid-base64!!", "image/png")])
+
+    wrapper = _make_wrapper(SimpleNamespace(call_tool=call_tool))
+
+    result = await wrapper.execute()
+
+    assert result == "(MCP tool returned an image that could not be stored)"
+
+
 @pytest.mark.asyncio
 async def test_execute_returns_timeout_message() -> None:
     async def call_tool(_name: str, arguments: dict) -> object:
@@ -1177,3 +1240,18 @@ async def test_connect_mcp_servers_enabled_tools_matches_sanitized_name(
         await stack.aclose()
 
     assert registry.tool_names == ["mcp_test_My_Tool"]
+
+
+@pytest.mark.parametrize(
+    "url, expected",
+    [
+        ("https://user:secret@host.example/sse", "https://host.example/..."),
+        ("https://host.example:8443/mcp?token=abc#frag", "https://host.example:8443/..."),
+        ("https://user:secret@[::1]:8443/sse?token=abc", "https://[::1]:8443/..."),
+        ("https://host.example/sse", "https://host.example/..."),
+        ("https://host.example", "https://host.example"),
+        ("https://host.example/", "https://host.example/"),
+    ],
+)
+def test_redact_url_strips_credentials_and_query(url: str, expected: str) -> None:
+    assert mcp_mod._redact_url(url) == expected

tests/tools/test_sandbox.py

@@ -57,6 +57,37 @@ class TestBwrapBackend:
         tmpfs_targets = {tokens[i + 1] for i in tmpfs_indices}
         assert str(ws.parent) in tmpfs_targets
 
+    def test_tmp_dir_mounted_as_tmpfs(self, tmp_path):
+        """Regression coverage for #1948: commands need writable scratch space."""
+        ws = tmp_path / "project"
+        result = wrap_command("bwrap", "touch /tmp/probe", str(ws), str(ws))
+        tokens = _parse(result)
+
+        tmpfs_indices = [i for i, t in enumerate(tokens) if t == "--tmpfs"]
+        tmpfs_targets = {tokens[i + 1] for i in tmpfs_indices}
+        assert "/tmp" in tmpfs_targets
+
+    def test_parent_mask_precedes_workspace_recreation(self, tmp_path):
+        ws = tmp_path / "project"
+        result = wrap_command("bwrap", "ls", str(ws), str(ws))
+        tokens = _parse(result)
+
+        parent_mask = next(
+            i for i, t in enumerate(tokens)
+            if t == "--tmpfs" and tokens[i + 1] == str(ws.parent)
+        )
+        workspace_dir = next(
+            i for i, t in enumerate(tokens)
+            if t == "--dir" and tokens[i + 1] == str(ws)
+        )
+        workspace_bind = next(
+            i for i, t in enumerate(tokens)
+            if t == "--bind" and tokens[i + 1] == str(ws) and tokens[i + 2] == str(ws)
+        )
+        chdir = tokens.index("--chdir")
+
+        assert parent_mask < workspace_dir < workspace_bind < chdir
+
     def test_cwd_inside_workspace(self, tmp_path):
         ws = tmp_path / "project"
         sub = ws / "src" / "lib"

tests/utils/test_file_edit_events.py

@@ -412,6 +412,41 @@ def test_streaming_tracker_applies_canonical_call_id_to_final_tool(tmp_path: Pat
     asyncio.run(run())
 
 
+def test_streaming_tracker_does_not_remap_non_file_edit_final_tool(tmp_path: Path) -> None:
+    events: list[dict] = []
+
+    async def emit(batch: list[dict]) -> None:
+        events.extend(batch)
+
+    async def run() -> None:
+        tracker = StreamingFileEditTracker(workspace=tmp_path, tools={}, emit=emit)
+        await tracker.update({
+            "index": 0,
+            "name": "read_file",
+            "arguments_delta": '{"path":"matched.md"}',
+        })
+        await tracker.update({
+            "index": 1,
+            "name": "write_file",
+            "arguments_delta": '{"path":"matched.md","content":"one\\n',
+        })
+        read_final = SimpleNamespace(
+            id="read-unique",
+            name="read_file",
+            arguments={"path": "matched.md"},
+        )
+        write_final = SimpleNamespace(
+            id="write-final",
+            name="write_file",
+            arguments={"path": "matched.md", "content": "one\n"},
+        )
+        tracker.apply_final_call_ids([read_final, write_final])
+        assert read_final.id == "read-unique"
+        assert write_final.id == "idx:1"
+
+    asyncio.run(run())
+
+
 def test_streaming_tracker_does_not_restore_duplicate_canonical_ids(tmp_path: Path) -> None:
     events: list[dict] = []
 

tests/webui/test_session_list_index.py

@@ -101,6 +101,7 @@ def test_webui_session_list_uses_webui_transcript_activity_for_sort(
     old_session.created_at = datetime(2026, 6, 15, 10, 0, 0)
     old_session.updated_at = datetime(2026, 6, 15, 10, 0, 0)
     old_session.add_message("user", "old metadata")
+    old_session.messages[-1]["timestamp"] = "2026-06-15T10:00:00"
     old_session.updated_at = datetime(2026, 6, 15, 10, 0, 0)
     manager.save(old_session)
 
@@ -108,6 +109,7 @@ def test_webui_session_list_uses_webui_transcript_activity_for_sort(
     newer_metadata.created_at = datetime(2026, 6, 15, 11, 0, 0)
     newer_metadata.updated_at = datetime(2026, 6, 15, 11, 0, 0)
     newer_metadata.add_message("user", "newer metadata")
+    newer_metadata.messages[-1]["timestamp"] = "2026-06-15T11:00:00"
     newer_metadata.updated_at = datetime(2026, 6, 15, 11, 0, 0)
     manager.save(newer_metadata)
 
@@ -141,6 +143,7 @@ def test_webui_session_list_rescans_when_transcript_changes(
     session.created_at = datetime(2026, 6, 15, 10, 0, 0)
     session.updated_at = datetime(2026, 6, 15, 10, 0, 0)
     session.add_message("user", "preview")
+    session.messages[-1]["timestamp"] = "2026-06-15T10:00:00"
     session.updated_at = datetime(2026, 6, 15, 10, 0, 0)
     manager.save(session)
 
@@ -169,6 +172,32 @@ def test_webui_session_list_rescans_when_transcript_changes(
     assert rows[0]["updated_at"].startswith("2026-06-15T12:30:00")
 
 
+def test_webui_session_list_sorts_by_message_activity_not_maintenance_timestamp(
+    tmp_path: Path,
+) -> None:
+    manager = SessionManager(tmp_path)
+    old = manager.get_or_create("websocket:old")
+    old.created_at = datetime(2026, 6, 1, 10, 0, 0)
+    old.add_message("user", "old first visible activity")
+    old.messages[-1]["timestamp"] = "2026-06-01T10:00:00"
+    old.add_message("assistant", "automation result")
+    old.messages[-1]["timestamp"] = "2026-06-05T10:00:00"
+    old.updated_at = datetime(2026, 6, 30, 17, 40, 0)
+    manager.save(old)
+
+    newer = manager.get_or_create("websocket:newer")
+    newer.created_at = datetime(2026, 6, 4, 10, 0, 0)
+    newer.add_message("user", "newer real activity")
+    newer.messages[-1]["timestamp"] = "2026-06-04T10:00:00"
+    newer.updated_at = datetime(2026, 6, 4, 10, 0, 0)
+    manager.save(newer)
+
+    rows = list_webui_sessions(manager)
+
+    assert [row["key"] for row in rows] == ["websocket:old", "websocket:newer"]
+    assert rows[0]["updated_at"] == "2026-06-05T10:00:00"
+
+
 def list_webui_sessions(manager: SessionManager) -> list[dict]:
     return session_list_index.list_webui_sessions(manager)
 

tests/webui/test_settings_api.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import json
+from types import SimpleNamespace
 
 import httpx
 import pytest
@@ -12,6 +13,7 @@ from nanobot.webui.settings_api import (
     WebUISettingsError,
     _oauth_provider_status,
     create_model_configuration,
+    login_oauth_provider,
     provider_models_payload,
     settings_payload,
     settings_usage_payload,
@@ -844,6 +846,39 @@ def test_openai_codex_oauth_status_rejects_unavailable_token(
     assert status["account"] is None
 
 
+def test_openai_codex_oauth_login_passes_configured_proxy(
+    tmp_path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    proxy = "http://127.0.0.1:23458"
+    config_path = tmp_path / "config.json"
+    save_config(
+        Config.model_validate({"providers": {"openaiCodex": {"proxy": "${CODEX_PROXY_TEST}"}}}),
+        config_path,
+    )
+    monkeypatch.setenv("CODEX_PROXY_TEST", proxy)
+    monkeypatch.setattr("nanobot.config.loader._current_config_path", config_path)
+
+    import oauth_cli_kit
+
+    captured: dict[str, str | None] = {}
+
+    def fake_get_token(*, proxy=None):
+        captured["get_proxy"] = proxy
+        raise RuntimeError("no-token")
+
+    def fake_login(*, print_fn, prompt_fn, proxy=None):
+        captured["login_proxy"] = proxy
+        return SimpleNamespace(access="access-token", account_id="acct-test")
+
+    monkeypatch.setattr(oauth_cli_kit, "get_token", fake_get_token)
+    monkeypatch.setattr(oauth_cli_kit, "login_oauth_interactive", fake_login)
+
+    login_oauth_provider({"provider": ["openai-codex"]})
+
+    assert captured == {"get_proxy": proxy, "login_proxy": proxy}
+
+
 def test_provider_models_payload_fetches_openai_compatible_models(
     tmp_path,
     monkeypatch: pytest.MonkeyPatch,

webui/src/components/thread/PromptRail.tsx

@@ -1,11 +1,4 @@
-import {
-  type RefObject,
-  useCallback,
-  useEffect,
-  useMemo,
-  useRef,
-  useState,
-} from "react";
+import { type RefObject, useCallback, useEffect, useMemo, useRef, useState } from "react";
 
 import { cn } from "@/lib/utils";
 import type { UIMessage } from "@/lib/types";
@@ -29,6 +22,7 @@ interface MeasuredPrompt extends PromptAnchor {
 }
 
 interface PromptMarker {
+  answerPreview: string;
   count: number;
   ids: string[];
   label: string;
@@ -43,10 +37,11 @@ const DENSE_BUCKET_HEIGHT_PX = 12;
 const DENSE_BUCKET_FALLBACK_COUNT = 32;
 const DENSE_BUCKET_MAX_COUNT = 42;
 const MARKER_MIN_GAP_PX = 9;
-const MARKER_BASE_WIDTH_PX = 16;
-const MARKER_MAX_WIDTH_PX = 28;
+const MARKER_BASE_WIDTH_PX = 9;
+const MARKER_STACK_GAP_PX = 16;
+const RAIL_FALLBACK_HEIGHT_PX = 300;
 const MEASURE_RETRY_FRAMES = 4;
-const RAIL_REVEAL_MS = 1400;
+const HOVER_MARKER_WIDTHS_PX = [28, 22, 16, 11];
 
 export function PromptRail({
   bottomOffset,
@@ -57,22 +52,12 @@ export function PromptRail({
   const promptAnchors = useMemo(() => userPromptAnchors(messages), [messages]);
   const [markers, setMarkers] = useState<PromptMarker[]>([]);
   const [activePromptId, setActivePromptId] = useState<string | null>(null);
-  const [revealed, setRevealed] = useState(false);
-  const revealTimeoutRef = useRef<number | null>(null);
-
-  const revealTemporarily = useCallback(() => {
-    setRevealed(true);
-    if (revealTimeoutRef.current !== null) {
-      window.clearTimeout(revealTimeoutRef.current);
-    }
-    revealTimeoutRef.current = window.setTimeout(() => {
-      setRevealed(false);
-      revealTimeoutRef.current = null;
-    }, RAIL_REVEAL_MS);
-  }, []);
+  const [focusedMarkerIndex, setFocusedMarkerIndex] = useState<number | null>(null);
 
   const updateMarkers = useCallback(() => {
     const scrollEl = scrollRef.current;
+    const nextRailHeight = railRef.current?.clientHeight ?? 0;
+
     if (!scrollEl || promptAnchors.length < MIN_PROMPTS_FOR_RAIL) {
       setMarkers([]);
       setActivePromptId(null);
@@ -87,7 +72,8 @@ export function PromptRail({
     }
 
     const measured = measurePrompts(scrollEl, promptAnchors, scrollRange);
-    setMarkers(groupPromptMarkers(measured, railRef.current?.clientHeight ?? 0));
+    const grouped = groupPromptMarkers(measured, nextRailHeight);
+    setMarkers(distributeMarkerPositions(grouped, nextRailHeight));
     setActivePromptId(activePromptForScroll(measured, scrollEl.scrollTop));
   }, [promptAnchors, scrollRef]);
 
@@ -112,7 +98,6 @@ export function PromptRail({
     let frame = 0;
     const schedule = () => {
       window.cancelAnimationFrame(frame);
-      revealTemporarily();
       frame = window.requestAnimationFrame(updateMarkers);
     };
 
@@ -123,7 +108,7 @@ export function PromptRail({
       scrollEl.removeEventListener("scroll", schedule);
       window.removeEventListener("resize", schedule);
     };
-  }, [revealTemporarily, scrollRef, updateMarkers]);
+  }, [scrollRef, updateMarkers]);
 
   useEffect(() => {
     const scrollEl = scrollRef.current;
@@ -134,77 +119,72 @@ export function PromptRail({
     return () => observer.disconnect();
   }, [scrollRef, updateMarkers]);
 
-  useEffect(() => {
-    return () => {
-      if (revealTimeoutRef.current !== null) {
-        window.clearTimeout(revealTimeoutRef.current);
-      }
-    };
-  }, []);
-
   if (markers.length === 0) return null;
 
-  const maxMarkerCount = Math.max(...markers.map((marker) => marker.count));
-  const activeMarkerIndex = markers.findIndex((marker) =>
-    marker.ids.includes(activePromptId ?? ""),
-  );
-
   return (
     <div
       ref={railRef}
       aria-label="User prompt navigation"
       className={cn(
-        "group pointer-events-auto absolute right-4 top-14 z-20 hidden w-8 opacity-70 md:block",
-        "transition-opacity duration-200 hover:opacity-100",
+        "group pointer-events-auto absolute left-7 top-3 z-20 hidden w-9 opacity-100 md:block",
+        "transition-opacity duration-200",
         "motion-safe:animate-in motion-safe:fade-in-0 motion-safe:duration-200",
       )}
+      onPointerLeave={() => setFocusedMarkerIndex(null)}
       style={{ bottom: Math.max(80, bottomOffset) }}
     >
       {markers.map((marker, index) => {
         const active = marker.ids.includes(activePromptId ?? "");
-        const nearActive = activeMarkerIndex < 0 || Math.abs(index - activeMarkerIndex) <= 1;
+        const hoverDistance =
+          focusedMarkerIndex === null ? null : Math.abs(index - focusedMarkerIndex);
         return (
           <button
             key={marker.ids.join("|")}
             type="button"
             aria-label={`Jump to prompt: ${marker.label}`}
             onClick={() => jumpToPrompt(scrollRef.current, marker.ids[marker.ids.length - 1])}
+            onBlur={() => setFocusedMarkerIndex(null)}
+            onFocus={() => setFocusedMarkerIndex(index)}
+            onPointerEnter={() => setFocusedMarkerIndex(index)}
+            onPointerLeave={() => setFocusedMarkerIndex(null)}
             className={cn(
-              "group/marker absolute right-0 h-5 -translate-y-1/2 overflow-visible rounded-full",
+              "group/marker absolute left-0 h-4 w-9 -translate-y-1/2 overflow-visible rounded-sm",
               "focus-visible:outline-none focus-visible:ring-2 focus-visible:ring-blue-400/60",
             )}
-            style={{
-              top: `${marker.topPercent}%`,
-              width: markerWidth(marker.count, maxMarkerCount, active),
-            }}
+            style={{ top: `${marker.topPercent}%` }}
           >
             <span
               aria-hidden
+              data-testid="prompt-rail-marker"
               className={cn(
-                "absolute right-0 top-1/2 h-[3px] w-full -translate-y-1/2 rounded-full",
-                "bg-foreground/20 transition-[background-color,opacity,transform,height] duration-200",
-                "group-hover/marker:bg-blue-500/70 group-hover/marker:opacity-100 group-hover/marker:scale-x-110",
-                "group-focus-visible/marker:bg-blue-500 group-focus-visible/marker:opacity-100 group-focus-visible/marker:scale-x-110",
-                marker.count > 1 && "bg-foreground/30",
-                active && "h-1 bg-foreground/65 opacity-80 shadow-sm",
-                !active && nearActive && "opacity-25 group-hover:opacity-55",
-                !active && !nearActive && !revealed && "opacity-0 group-hover:opacity-40",
-                !active && !nearActive && revealed && "opacity-35",
+                "absolute left-0 top-1/2 h-0.5 -translate-y-1/2 rounded-full",
+                "transition-[width,background-color,opacity,height] duration-150",
+                railMarkerTone(hoverDistance, active),
               )}
+              style={{
+                height: markerHeight(hoverDistance),
+                width: markerWidth(hoverDistance),
+              }}
             />
             <span
               aria-hidden
               className={cn(
-                "pointer-events-none absolute right-9 top-1/2 z-30 w-64 -translate-y-1/2 rounded-lg px-3 py-2 text-left",
-                "bg-background/95 text-xs leading-5 text-foreground shadow-lg ring-1 ring-border/80 backdrop-blur",
-                "opacity-0 translate-x-1 transition-[opacity,transform] duration-150",
-                "group-hover/marker:opacity-100 group-hover/marker:translate-x-0",
-                "group-focus-visible/marker:opacity-100 group-focus-visible/marker:translate-x-0",
+                "pointer-events-none absolute left-10 top-1/2 z-30 w-[34rem] max-w-[calc(100vw-4rem)] -translate-y-1/2 rounded-[20px] px-4 py-3 text-left",
+                "border border-border/70 bg-popover/95 text-popover-foreground shadow-[0_18px_45px_rgba(0,0,0,0.12)] backdrop-blur-xl",
+                "dark:border-white/10 dark:bg-[#2f2f2f]/95 dark:text-white dark:shadow-[0_18px_45px_rgba(0,0,0,0.45)]",
+                "-translate-x-2 scale-[0.98] opacity-0 transition-[opacity,transform] duration-150",
+                "group-hover/marker:translate-x-0 group-hover/marker:scale-100 group-hover/marker:opacity-100",
+                "group-focus-visible/marker:translate-x-0 group-focus-visible/marker:scale-100 group-focus-visible/marker:opacity-100",
               )}
             >
-              <span className="block max-h-24 overflow-hidden whitespace-pre-wrap break-words">
+              <span className="line-clamp-2 whitespace-pre-wrap break-words text-[15px] font-semibold leading-6">
                 {marker.preview}
               </span>
+              {marker.answerPreview ? (
+                <span className="mt-1.5 line-clamp-3 whitespace-pre-wrap break-words text-[14px] leading-6 text-muted-foreground dark:text-white/55">
+                  {marker.answerPreview}
+                </span>
+              ) : null}
             </span>
           </button>
         );
@@ -250,10 +230,12 @@ function groupPromptMarkers(
       last.count += 1;
       last.ids.push(prompt.id);
       last.label = groupedPromptLabel(last.count, prompt.label);
-      last.preview = groupedPromptPreview(last.count, prompt.preview);
+      last.answerPreview = prompt.answerPreview;
+      last.preview = prompt.preview;
       continue;
     }
     groups.push({
+      answerPreview: prompt.answerPreview,
       count: 1,
       ids: [prompt.id],
       label: prompt.label,
@@ -298,14 +280,30 @@ function bucketPromptMarkers(
       label: bucket.length === 1
         ? latest.label
         : groupedPromptLabel(bucket.length, latest.label),
-      preview: bucket.length === 1
-        ? latest.preview
-        : groupedPromptPreview(bucket.length, latest.preview),
+      answerPreview: latest.answerPreview,
+      preview: latest.preview,
       topPercent,
     }];
   });
 }
 
+function distributeMarkerPositions(markers: PromptMarker[], railHeight: number): PromptMarker[] {
+  const height = railHeight > 0 ? railHeight : RAIL_FALLBACK_HEIGHT_PX;
+  if (markers.length <= 1) {
+    return markers.map((marker) => ({ ...marker, topPercent: 50 }));
+  }
+
+  const availableHeight = Math.max(0, height - MARKER_STACK_GAP_PX);
+  const stepPx = Math.min(MARKER_STACK_GAP_PX, availableHeight / (markers.length - 1));
+  const stackHeight = stepPx * (markers.length - 1);
+  const firstMarkerPx = (height - stackHeight) / 2;
+
+  return markers.map((marker, index) => ({
+    ...marker,
+    topPercent: ((firstMarkerPx + stepPx * index) / height) * 100,
+  }));
+}
+
 function activePromptForScroll(
   measured: MeasuredPrompt[],
   scrollTop: number,
@@ -327,16 +325,26 @@ function groupedPromptLabel(count: number, latestLabel: string): string {
   return `${count} prompts, latest: ${latestLabel}`;
 }
 
-function groupedPromptPreview(count: number, latestPreview: string): string {
-  return `${count} prompts\n\n${latestPreview}`;
+function markerWidth(hoverDistance: number | null): number {
+  if (hoverDistance === null) return MARKER_BASE_WIDTH_PX;
+  return HOVER_MARKER_WIDTHS_PX[hoverDistance] ?? MARKER_BASE_WIDTH_PX;
 }
 
-function markerWidth(count: number, maxCount: number, active: boolean): number {
-  if (maxCount <= 1) return active ? 34 : MARKER_BASE_WIDTH_PX;
-  const density = Math.log2(count + 1) / Math.log2(maxCount + 1);
-  const width = MARKER_BASE_WIDTH_PX
-    + (MARKER_MAX_WIDTH_PX - MARKER_BASE_WIDTH_PX) * density;
-  return Math.round(active ? width + 4 : width);
+function markerHeight(hoverDistance: number | null): number {
+  return hoverDistance === 0 ? 3 : 2;
+}
+
+function railMarkerTone(hoverDistance: number | null, active: boolean): string {
+  if (hoverDistance === 0) {
+    return "bg-[#222222] opacity-100 dark:bg-white";
+  }
+  if (hoverDistance !== null && hoverDistance < HOVER_MARKER_WIDTHS_PX.length) {
+    return "bg-[#d0d0d0] opacity-100 dark:bg-white/35";
+  }
+  if (active) {
+    return "bg-[#6f6f6f] opacity-100 dark:bg-white/55";
+  }
+  return "bg-[#d8d8d8] opacity-100 dark:bg-white/25";
 }
 
 function clamp(value: number, min: number, max: number): number {

webui/src/components/thread/promptNavigation.ts

@@ -1,6 +1,7 @@
 import type { UIMessage } from "@/lib/types";
 
 export interface PromptAnchor {
+  answerPreview: string;
   id: string;
   label: string;
   preview: string;
@@ -10,9 +11,10 @@ export interface PromptAnchor {
 
 export function userPromptAnchors(messages: UIMessage[]): PromptAnchor[] {
   let index = 0;
-  return messages.flatMap((message) => {
+  return messages.flatMap((message, messageIndex) => {
     if (message.role !== "user") return [];
     const anchor: PromptAnchor = {
+      answerPreview: nextAssistantPreview(messages, messageIndex),
       id: message.id,
       label: promptLabel(message.content, index),
       preview: promptPreview(message.content, index),
@@ -27,13 +29,34 @@ export function userPromptAnchors(messages: UIMessage[]): PromptAnchor[] {
 export function promptLabel(content: string, index: number): string {
   const text = content.replace(/\s+/g, " ").trim();
   if (!text) return `Prompt ${index + 1}`;
-  return text.length > 80 ? `${text.slice(0, 77)}...` : text;
+  return truncatePreview(text, 80);
 }
 
 export function promptPreview(content: string, index: number): string {
-  const text = content.replace(/\n{3,}/g, "\n\n").trim();
+  const text = compactPreview(content);
   if (!text) return `Prompt ${index + 1}`;
-  return text.length > 320 ? `${text.slice(0, 317)}...` : text;
+  return truncatePreview(text, 320);
+}
+
+function nextAssistantPreview(messages: UIMessage[], promptIndex: number): string {
+  for (let index = promptIndex + 1; index < messages.length; index += 1) {
+    const message = messages[index];
+    if (message.role === "user") return "";
+    if (message.role !== "assistant") continue;
+
+    const preview = truncatePreview(compactPreview(message.content), 240);
+    if (preview) return preview;
+  }
+
+  return "";
+}
+
+function compactPreview(content: string): string {
+  return content.replace(/\n{3,}/g, "\n\n").trim();
+}
+
+function truncatePreview(text: string, maxLength: number): string {
+  return text.length > maxLength ? `${text.slice(0, maxLength - 3)}...` : text;
 }
 
 export function jumpToPrompt(scrollEl: HTMLElement | null, promptId: string | undefined): void {

webui/src/lib/nanobot-client.ts

@@ -425,6 +425,13 @@ export class NanobotClient {
     for (const handler of this.statusHandlers) handler(status);
   }
 
+  private clearRunStatusesForReconnect(): void {
+    if (this.runStartedAtByChatId.size === 0) return;
+    const chatIds = [...this.runStartedAtByChatId.keys()];
+    this.runStartedAtByChatId.clear();
+    for (const chatId of chatIds) this.emitRunStatus(chatId, null);
+  }
+
   private handleOpen(): void {
     this.setStatus("open");
     this.reconnectAttempts = 0;
@@ -629,6 +636,7 @@ export class NanobotClient {
   }
 
   private scheduleReconnect(): void {
+    this.clearRunStatusesForReconnect();
     this.setStatus("reconnecting");
     const attempt = this.reconnectAttempts++;
     // Exponential backoff: 0.5s, 1s, 2s, 4s, capped.

webui/src/tests/nanobot-client.test.ts

@@ -188,6 +188,32 @@ describe("NanobotClient", () => {
     expect(client.getRunStartedAt("chat-strip")).toBeNull();
   });
 
+  it("clears stale run strip when reconnecting after a dropped socket", async () => {
+    const client = new NanobotClient({
+      url: "ws://test",
+      reconnect: true,
+      maxBackoffMs: 10,
+      socketFactory: (url) => new FakeSocket(url) as unknown as WebSocket,
+    });
+    const handler = vi.fn();
+    client.onRunStatus(handler);
+    client.connect();
+    lastSocket().fakeOpen();
+    lastSocket().fakeMessage({
+      event: "goal_status",
+      chat_id: "chat-strip",
+      status: "running",
+      started_at: 12_345,
+    });
+
+    lastSocket().close();
+
+    expect(client.getRunStartedAt("chat-strip")).toBeNull();
+    expect(handler).toHaveBeenLastCalledWith("chat-strip", null);
+    await vi.advanceTimersByTimeAsync(20);
+    expect(FakeSocket.instances.length).toBeGreaterThan(1);
+  });
+
   it("clears run strip when a turn_end arrives without idle", () => {
     const client = new NanobotClient({
       url: "ws://test",

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

@@ -91,6 +91,23 @@ function makeLongMessages(count: number): UIMessage[] {
   }));
 }
 
+function makePromptExchangeMessages(count: number): UIMessage[] {
+  return Array.from({ length: count }, (_, index) => ([
+    {
+      id: `m${index}`,
+      role: "user" as const,
+      content: `message ${index}`,
+      createdAt: index * 2,
+    },
+    {
+      id: `a${index}`,
+      role: "assistant" as const,
+      content: `answer ${index}`,
+      createdAt: index * 2 + 1,
+    },
+  ])).flat();
+}
+
 function ViewportWithPromptNavigator({ messages }: { messages: UIMessage[] }) {
   const viewportRef = useRef<ThreadViewportHandle | null>(null);
   return (
@@ -604,7 +621,7 @@ describe("ThreadViewport", () => {
       screen.queryByText(`message ${firstVisible - 1}`),
     ).not.toBeInTheDocument();
     expect(screen.getByText(`message ${firstVisible}`)).toBeInTheDocument();
-    expect(screen.getByText("message 299")).toBeInTheDocument();
+    expect(screen.getAllByText("message 299").length).toBeGreaterThan(0);
   });
 
   it("automatically requests older transcript pages near the top", () => {
@@ -635,7 +652,7 @@ describe("ThreadViewport", () => {
   });
 
   it("renders a prompt rail that jumps to user messages", async () => {
-    const promptMessages = makeLongMessages(5);
+    const promptMessages = makePromptExchangeMessages(5);
     const { container } = render(
       <ThreadViewport
         messages={promptMessages}
@@ -670,9 +687,31 @@ describe("ThreadViewport", () => {
     });
 
     expect(screen.getByLabelText("User prompt navigation")).toBeInTheDocument();
+    const promptMarkers = screen.getAllByRole("button", { name: /Jump to prompt:/ });
+    const markerTops = promptMarkers.map((marker) => Number.parseFloat(marker.style.top));
+    expect(markerTops[2]).toBeCloseTo(50);
+    expect(markerTops[1] - markerTops[0]).toBeCloseTo(16 / 3);
+    expect(markerTops[4] - markerTops[0]).toBeCloseTo(64 / 3);
+
+    const railMarkers = screen.getAllByTestId("prompt-rail-marker");
+    expect(railMarkers).toHaveLength(promptMarkers.length);
+    expect(railMarkers.every((marker) => marker.style.width === "9px")).toBe(true);
+
+    fireEvent.pointerEnter(promptMarkers[2]);
+    expect(railMarkers.map((marker) => marker.style.width)).toEqual([
+      "16px",
+      "22px",
+      "28px",
+      "22px",
+      "16px",
+    ]);
+
+    fireEvent.pointerLeave(promptMarkers[2]);
+    expect(railMarkers.every((marker) => marker.style.width === "9px")).toBe(true);
 
     const targetPrompt = screen.getByRole("button", { name: "Jump to prompt: message 3" });
     expect(within(targetPrompt).getByText("message 3")).toBeInTheDocument();
+    expect(within(targetPrompt).getByText("answer 3")).toBeInTheDocument();
 
     fireEvent.click(targetPrompt);