fix(agent): make detached exec portable on Windows

feat(agent): add runtime budget convergence notices
feat(agent): support detached exec services
2026-08-08 21:38:40 +03:00 · 2026-06-28 11:05:10 +00:00 · 2026-06-28 05:11:04 +00:00 · 2026-06-28 05:09:39 +00:00 · 2026-06-28 05:09:39 +00:00 · 2026-06-28 05:09:39 +00:00
81 changed files with 1665 additions and 2748 deletions
@@ -240,7 +240,6 @@ 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 |
 |----------|---------|-------------|
@@ -633,37 +632,20 @@ 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. `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.
+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.

 **1. Login:**
 ```bash
 nanobot provider login openai-codex
 ```

-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`):
+**2. Set model** (merge into `~/.nanobot/config.json`):
 ```json
 {
  "modelPresets": {
    "codex": {
      "provider": "openai_codex",
-      "model": "gpt-5.1-codex",
-      "reasoningEffort": "high"
+      "model": "openai-codex/gpt-5.1-codex"
    }
  },
  "agents": {
@@ -674,9 +656,7 @@ The proxy applies to Codex OAuth token refresh, interactive token exchange, and
 }
 ```

-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:**
+**3. Chat:**
 ```bash
 nanobot agent -m "Hello!"

@@ -695,17 +675,7 @@ 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.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"
-```
+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.

 **1. Login:**
 ```bash
@@ -2004,7 +1974,6 @@ 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
@@ -61,12 +61,9 @@ 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
@@ -425,32 +422,6 @@ 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:
@@ -34,26 +34,6 @@ 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}"
@@ -72,8 +52,7 @@ class AutoCompact:
                continue
            if key in active_session_keys:
                continue
-            updated_at = info.get("updated_at")
-            if self._is_expired(updated_at, now) and self._has_compactable_idle_tail(key):
+            if self._is_expired(info.get("updated_at"), now):
                self._archiving.add(key)
                schedule_background(self._archive(key))

@@ -36,23 +36,6 @@ 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)
@@ -78,9 +61,7 @@ class ContextGovernor:
        messages: list[dict[str, Any]],
        compacted_tool_call_ids: set[str],
    ) -> list[dict[str, Any]]:
-        updated = self.strip_placeholder_assistant_messages(messages)
-        updated = self.strip_malformed_tool_calls(updated)
-        updated = self.drop_orphan_tool_results(updated)
+        updated = self.drop_orphan_tool_results(messages)
        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)
@@ -135,99 +116,6 @@ 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]],
@@ -57,11 +57,7 @@ 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,
-    replay_max_messages_for_context,
-)
+from nanobot.session.manager import Session, SessionManager
 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
@@ -205,6 +201,7 @@ 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,
@@ -218,7 +215,6 @@ 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

@@ -228,7 +224,6 @@ 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
@@ -297,7 +292,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 = replay_max_messages_for_context(self.context_window_tokens)
+        self._max_messages = max_messages if max_messages > 0 else 120
        self._running = False
        self._mcp_servers = mcp_servers or {}
        self._mcp_stacks: dict[str, AsyncExitStack] = {}
@@ -395,10 +390,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,
@@ -426,7 +421,6 @@ 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(
@@ -440,9 +434,6 @@ 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
@@ -1852,17 +1843,13 @@ 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:
@@ -1876,13 +1863,5 @@ 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)
@@ -33,6 +33,7 @@ if TYPE_CHECKING:
    from nanobot.providers.base import LLMProvider
    from nanobot.session.manager import SessionManager

+
 # ---------------------------------------------------------------------------
 # MemoryStore — pure file I/O layer
 # ---------------------------------------------------------------------------
@@ -1005,6 +1006,7 @@ 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 ""

@@ -1016,11 +1018,12 @@ class Consolidator:
                metadata={},
                last_consolidated=0,
            )
-            result = probe.retain_recent_legal_suffix(max_suffix, extend_to_user=True)
+            dropped, already_consolidated = probe.retain_recent_legal_suffix(max_suffix, extend_to_user=True)
            messages_to_keep = probe.messages
-            messages_to_remove = result.dropped[result.already_consolidated_count:]
+            messages_to_remove = dropped[already_consolidated:]

            if not messages_to_remove and not messages_to_keep:
+                session.updated_at = datetime.now()
                self.sessions.save(session)
                return ""

@@ -1043,6 +1046,7 @@ class Consolidator:

            session.messages = messages_to_keep
            session.last_consolidated = 0
+            session.updated_at = datetime.now()
            self.sessions.save(session)

            if messages_to_remove:
@@ -50,6 +50,7 @@ from nanobot.utils.runtime import (
    build_finalization_retry_message,
    build_goal_continue_message,
    build_length_recovery_message,
+    build_runtime_budget_notice_message,
    is_blank_text,
    repeated_external_lookup_error,
    repeated_workspace_violation_error,
@@ -67,6 +68,7 @@ _MAX_EMPTY_RETRIES = 2
 _MAX_LENGTH_RECOVERIES = 3
 _MAX_INJECTIONS_PER_TURN = 3
 _MAX_INJECTION_CYCLES = 5
+_BUDGET_NOTICE_MIN_ITERATIONS = 20
 # Backward-compatible module attribute for tests/extensions that monkeypatch
 # the former single-file tracker hook. Runtime uses prepare_file_edit_trackers.
 prepare_file_edit_tracker = _prepare_file_edit_tracker
@@ -357,6 +359,7 @@ class AgentRunner:
        length_recovery_count = 0
        had_injections = False
        injection_cycles = 0
+        budget_notice_level_sent = 0
        compacted_tool_call_ids: set[str] = set()
        governance_config = ContextGovernanceConfig(
            provider=self.provider,
@@ -389,15 +392,7 @@ class AgentRunner:
                    spec.session_key or "default",
                )
                try:
-                    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.drop_orphan_tool_results(messages)
                    messages_for_model = ContextGovernor.backfill_missing_tool_results(
                        messages_for_model
                    )
@@ -519,6 +514,12 @@ class AgentRunner:
                )
                if _drained:
                    had_injections = True
+                budget_notice_level_sent = self._append_runtime_budget_notice_if_needed(
+                    spec,
+                    messages,
+                    completed_iterations=iteration + 1,
+                    sent_level=budget_notice_level_sent,
+                )
                await hook.after_iteration(context)
                continue

@@ -733,8 +734,6 @@ 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:
@@ -877,94 +876,8 @@ 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,
@@ -1036,6 +949,53 @@ class AgentRunner:
        retry_messages.append(build_budget_exhausted_finalization_message())
        return retry_messages

+    @classmethod
+    def _append_runtime_budget_notice_if_needed(
+        cls,
+        spec: AgentRunSpec,
+        messages: list[dict[str, Any]],
+        *,
+        completed_iterations: int,
+        sent_level: int,
+    ) -> int:
+        level = cls._runtime_budget_notice_level(
+            max_iterations=spec.max_iterations,
+            completed_iterations=completed_iterations,
+        )
+        if level <= sent_level:
+            return sent_level
+
+        remaining_iterations = max(0, spec.max_iterations - completed_iterations)
+        messages.append(build_runtime_budget_notice_message(
+            level=level,
+            max_iterations=spec.max_iterations,
+            used_iterations=completed_iterations,
+            remaining_iterations=remaining_iterations,
+        ))
+        return level
+
+    @staticmethod
+    def _runtime_budget_notice_level(
+        *,
+        max_iterations: int,
+        completed_iterations: int,
+    ) -> int:
+        """Return the convergence-warning level for a long tool loop."""
+        if max_iterations < _BUDGET_NOTICE_MIN_ITERATIONS:
+            return 0
+
+        remaining_iterations = max_iterations - completed_iterations
+        if remaining_iterations <= 0:
+            return 0
+
+        convergence_threshold = max(5, (max_iterations + 9) // 10)
+        final_threshold = max(3, (max_iterations + 32) // 33)
+        if remaining_iterations <= final_threshold:
+            return 2
+        if remaining_iterations <= convergence_threshold:
+            return 1
+        return 0
+
    @staticmethod
    def _max_iterations_fallback(spec: AgentRunSpec) -> str:
        if spec.max_iterations_message:
@@ -118,22 +118,6 @@ 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."""
@@ -351,10 +335,6 @@ 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'])

@@ -17,6 +17,13 @@ from nanobot.agent.tools.schema import (
    StringSchema,
    tool_parameters_schema,
 )
+from nanobot.agent.verification_state import (
+    VerificationAnalysis,
+    analyze_verification_result,
+    append_verification_feedback,
+    record_verification_observation,
+)
+from nanobot.utils.helpers import build_structured_output_summary

 DEFAULT_YIELD_MS = 1000
 MAX_YIELD_MS = 30_000
@@ -37,6 +44,7 @@ class _SessionPoll:
    terminated: bool = False
    stdin_closed: bool = False
    truncated_chars: int = 0
+    analysis: VerificationAnalysis | None = None


@dataclass(slots=True)
@@ -147,7 +155,19 @@ class _ExecSession:
            output = "".join(self._chunks)
            self._chunks.clear()

-        output, truncated = _truncate_output(output, max_output_chars)
+        analysis = analyze_verification_result(
+            command=self.command,
+            output=output,
+            exit_code=self.process.returncode,
+            timed_out=self._timed_out,
+        )
+        output, truncated = _truncate_output(
+            output,
+            max_output_chars,
+            analysis=analysis,
+            exit_code=self.process.returncode,
+            elapsed_s=max(0.0, time.monotonic() - self.started_at),
+        )
        return _SessionPoll(
            output=output,
            done=self.process.returncode is not None,
@@ -157,6 +177,7 @@ class _ExecSession:
            terminated=terminated,
            stdin_closed=stdin_closed,
            truncated_chars=truncated,
+            analysis=analysis,
        )

    async def kill(self) -> None:
@@ -320,15 +341,33 @@ def clamp_session_int(value: int | None, default: int, minimum: int, maximum: in
    return min(max(value, minimum), maximum)


-def _truncate_output(output: str, max_output_chars: int) -> tuple[str, int]:
+def _truncate_output(
+    output: str,
+    max_output_chars: int,
+    *,
+    analysis: VerificationAnalysis | None = None,
+    exit_code: int | None = None,
+    elapsed_s: float | None = None,
+) -> tuple[str, int]:
    if len(output) <= max_output_chars:
        return output, 0
-    half = max_output_chars // 2
    omitted = len(output) - max_output_chars
    return (
-        output[:half]
-        + f"\n\n... ({omitted:,} chars truncated) ...\n\n"
-        + output[-half:],
+        build_structured_output_summary(
+            "[tool output truncated]",
+            output,
+            max_chars=max_output_chars,
+            metadata=[
+                ("original_size_chars", len(output)),
+                ("exit_code", exit_code if exit_code is not None else "running"),
+                ("elapsed_s", f"{elapsed_s:.1f}" if elapsed_s is not None else "unknown"),
+            ],
+            analysis=analysis,
+            guidance=(
+                "Use the structured summary first. Poll again for new output "
+                "or rerun a narrower command instead of reading broad logs."
+            ),
+        ),
        omitted,
    )

@@ -351,6 +390,20 @@ def format_session_poll(session_id: str, poll: _SessionPoll) -> str:
    return "\n".join(parts) if parts else "(no output yet)"


+def _format_poll_with_verification(session_id: str, poll: _SessionPoll) -> str:
+    result = format_session_poll(session_id, poll)
+    if not poll.done:
+        return result
+    analysis = poll.analysis or analyze_verification_result(
+        command="",
+        output=result,
+        exit_code=poll.exit_code,
+        timed_out=poll.timed_out,
+    )
+    record_verification_observation(current_request_session_key(), analysis)
+    return append_verification_feedback(result, analysis)
+
+
@tool_parameters(
    tool_parameters_schema(
        session_id=StringSchema("Session id returned by exec when yield_time_ms is used."),
@@ -492,7 +545,7 @@ class WriteStdinTool(Tool):
                max_output_chars=output_limit,
                owner_session_key=current_request_session_key(),
            )
-            return format_session_poll(session_id, poll)
+            return _format_poll_with_verification(session_id, poll)
        except KeyError:
            return f"Error: exec session not found: {session_id}"
        except Exception as exc:
@@ -532,10 +585,10 @@ class WriteStdinTool(Tool):
                joined = "".join(aggregate)
                if wait_for in joined:
                    poll.output = joined
-                    return format_session_poll(session_id, poll)
+                    return _format_poll_with_verification(session_id, poll)
            if poll.done or remaining_ms <= 0:
                poll.output = "".join(aggregate)
-                result = format_session_poll(session_id, poll)
+                result = _format_poll_with_verification(session_id, poll)
                if wait_for not in poll.output:
                    result += f"\nWait target not observed: {wait_for!r}"
                return result
@@ -23,6 +23,11 @@ from typing import TYPE_CHECKING, Any
 from nanobot.agent.tools.base import Tool, tool_parameters
 from nanobot.agent.tools.context import ContextAware, RequestContext
 from nanobot.agent.tools.schema import StringSchema, tool_parameters_schema
+from nanobot.agent.verification_state import (
+    clear_verification_observation,
+    format_completion_gate_message,
+    latest_verification_observation,
+)
 from nanobot.bus.runtime_events import GoalStateChanged, RuntimeEventBus, RuntimeEventContext
 from nanobot.session.goal_state import (
    GOAL_STATE_KEY,
@@ -97,8 +102,7 @@ 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 use this for a single current-turn answer, including one that uses spawn subagents.",
+            "do not delay this tool call to over-plan, research, or decide execution details.",
            max_length=12_000,
        ),
        ui_summary=StringSchema(
@@ -140,8 +144,6 @@ 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. "
@@ -190,6 +192,29 @@ class LongTaskTool(Tool, _GoalToolsMixin):
            max_length=8000,
            nullable=True,
        ),
+        verification_summary=StringSchema(
+            "For coding or file-producing tasks, summarize how the work was verified. "
+            "Mention the most relevant test/check command and whether it passed. "
+            "If no verification was possible, say why.",
+            max_length=4000,
+            nullable=True,
+        ),
+        commands_run=StringSchema(
+            "Optional concise list of verification/build commands run before completion.",
+            max_length=4000,
+            nullable=True,
+        ),
+        artifacts_created=StringSchema(
+            "Optional concise list of files, outputs, or artifacts created.",
+            max_length=4000,
+            nullable=True,
+        ),
+        remaining_failures=StringSchema(
+            "Known unresolved failures, if intentionally stopping before success. "
+            "Leave empty when verification passes.",
+            max_length=4000,
+            nullable=True,
+        ),
        required=[],
    )
 )
@@ -225,30 +250,67 @@ class CompleteGoalTool(Tool, _GoalToolsMixin):
        return (
            "End bookkeeping for the active sustained goal. "
            "Use when the objective is fully achieved and verified—recap what was delivered. "
+            "For coding/file-producing tasks, run the smallest reliable verification first and include "
+            "verification_summary / commands_run / artifacts_created. "
            "Also call when the user cancels, redirects, or replaces the goal: recap must reflect "
            "what actually happened (not necessarily success). "
+            "If recent verification failed and no later verification passed, this tool will ask you to "
+            "continue fixing unless remaining_failures describes an intentional incomplete stop. "
            "If no goal is active, the tool reports that and leaves metadata unchanged."
        )

-    async def execute(self, recap: str | None = None, **kwargs: Any) -> str:
+    async def execute(
+        self,
+        recap: str | None = None,
+        verification_summary: str | None = None,
+        commands_run: str | None = None,
+        artifacts_created: str | None = None,
+        remaining_failures: str | None = None,
+        **kwargs: Any,
+    ) -> str:
        sess = self._session()
        if sess is None:
            return "Error: complete_goal requires an active chat session."
+
+        session_key = self._request_ctx.get().session_key if self._request_ctx.get() else None
+        observation = latest_verification_observation(session_key)
+        if (
+            observation is not None
+            and observation.analysis.status == "failed"
+            and not _has_meaningful_remaining_failures(remaining_failures)
+        ):
+            return format_completion_gate_message(observation)
+
        prior = parse_goal_state(goal_state_raw(sess.metadata))
        if not isinstance(prior, dict) or prior.get("status") != "active":
            return "No active goal to complete."

        ended = _iso_now()
-        sess.metadata[GOAL_STATE_KEY] = {
+        completed = {
            **prior,
            "status": "completed",
            "completed_at": ended,
            "recap": (recap or "").strip(),
        }
+        if verification_summary:
+            completed["verification_summary"] = verification_summary.strip()
+        if commands_run:
+            completed["commands_run"] = commands_run.strip()
+        if artifacts_created:
+            completed["artifacts_created"] = artifacts_created.strip()
+        if remaining_failures:
+            completed["remaining_failures"] = remaining_failures.strip()
+        sess.metadata[GOAL_STATE_KEY] = completed
        discard_legacy_goal_state_key(sess.metadata)
        self._sessions.save(sess)
+        clear_verification_observation(session_key)
        await self._publish_goal_state_changed(sess.metadata)
        tail = (recap or "").strip()
        if tail:
            return f"Goal marked complete ({ended}). Recap:\n{tail}"
        return f"Goal marked complete ({ended})."
+
+
+def _has_meaningful_remaining_failures(value: str | None) -> bool:
+    text = (value or "").strip().lower()
+    return bool(text and text not in {"none", "no", "n/a", "na", "no remaining failures"})
@@ -1,7 +1,6 @@
 """MCP client: connects to MCP servers and wraps their tools as native nanobot tools."""

 import asyncio
-import json
 import os
 import re
 import shutil
@@ -166,31 +165,12 @@ 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 {_redact_url(str(request.url))} ({error})",
+            f"Blocked unsafe MCP URL {request.url} ({error})",
            request=request,
        )

@@ -333,52 +313,6 @@ 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."""

@@ -406,6 +340,8 @@ 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:
@@ -460,63 +396,17 @@ class MCPToolWrapper(_MCPWrapperBase):
                )
                return f"(MCP tool call failed: {type(exc).__name__})"
            else:
-                # Success — extract text and persist any image content as artifacts.
-                return self._render_call_result(result.content, kwargs)
+                # 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)"

        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."""
@@ -793,7 +683,7 @@ async def connect_mcp_servers(
                    logger.warning(
                        "MCP server '{}': blocked unsafe URL {} ({})",
                        name,
-                        _redact_url(cfg.url),
+                        cfg.url,
                        error,
                    )
                    await server_stack.aclose()
@@ -814,7 +704,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, _redact_url(cfg.url))
+                    logger.warning("MCP server '{}': {} unreachable, skipping", name, cfg.url)
                    await server_stack.aclose()
                    return name, None

@@ -841,7 +731,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, _redact_url(cfg.url))
+                    logger.warning("MCP server '{}': {} unreachable, skipping", name, cfg.url)
                    await server_stack.aclose()
                    return name, None

@@ -438,9 +438,6 @@ 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}")
@@ -6,14 +6,17 @@ import asyncio
 import os
 import re
 import shutil
+import subprocess
 import sys
+import time
+import uuid
 from contextlib import suppress
 from dataclasses import dataclass
 from pathlib import Path
 from typing import Any

 from loguru import logger
-from pydantic import Field
+from pydantic import AliasChoices, Field

 from nanobot.agent.tools.base import Tool, tool_parameters
 from nanobot.agent.tools.context import current_request_session_key
@@ -33,12 +36,19 @@ from nanobot.agent.tools.schema import (
    StringSchema,
    tool_parameters_schema,
 )
+from nanobot.agent.verification_state import (
+    analyze_verification_result,
+    append_verification_feedback,
+    record_verification_observation,
+)
 from nanobot.config.paths import get_media_dir
 from nanobot.config_base import Base
 from nanobot.security.workspace_access import current_scope_allows_loopback, current_tool_workspace
 from nanobot.security.workspace_policy import is_path_within
+from nanobot.utils.helpers import build_structured_output_summary

 _IS_WINDOWS = sys.platform == "win32"
+_DETACHED_EXIT_GRACE_S = 1.0 if _IS_WINDOWS else 0.2


 # Policy note appended to recoverable workspace-boundary guard errors.
@@ -55,6 +65,13 @@ class ExecToolConfig(Base):
    """Shell exec tool configuration."""
    enable: bool = True
    timeout: int = Field(default=60, ge=0)  # Hard timeout (s); 0 = no limit. Not capped by the per-call max.
+    allow_local_service_access: bool = Field(
+        default=False,
+        validation_alias=AliasChoices(
+            "allowLocalServiceAccess",
+            "allow_local_service_access",
+        ),
+    )  # allow shell commands to reach literal localhost/loopback services
    path_prepend: str = ""
    path_append: str = ""
    sandbox: str = ""
@@ -126,6 +143,16 @@ class _PreparedCommand:
            maximum=MAX_OUTPUT_CHARS,
            nullable=True,
        ),
+        detach=BooleanSchema(
+            description=(
+                "Run the command as a detached background process that can "
+                "survive after the agent finishes. Use for local servers, "
+                "dev servers, mock APIs, or other services that must remain "
+                "available for later commands or external verification."
+            ),
+            default=False,
+            nullable=True,
+        ),
    )
 )
 class ExecTool(Tool):
@@ -149,6 +176,7 @@ class ExecTool(Tool):
            working_dir=ctx.workspace,
            timeout=cfg.timeout,
            restrict_to_workspace=ctx.config.restrict_to_workspace,
+            allow_local_service_access=cfg.allow_local_service_access,
            webui_allow_local_service_access=ctx.config.webui_allow_local_service_access,
            sandbox=cfg.sandbox,
            path_prepend=cfg.path_prepend,
@@ -165,6 +193,7 @@ class ExecTool(Tool):
        deny_patterns: list[str] | None = None,
        allow_patterns: list[str] | None = None,
        restrict_to_workspace: bool = False,
+        allow_local_service_access: bool = False,
        webui_allow_local_service_access: bool = True,
        allow_local_preview_access: bool | None = None,
        sandbox: str = "",
@@ -197,6 +226,7 @@ class ExecTool(Tool):
        ]
        self.allow_patterns = allow_patterns or []
        self.restrict_to_workspace = restrict_to_workspace
+        self.allow_local_service_access = allow_local_service_access
        if allow_local_preview_access is not None:
            webui_allow_local_service_access = allow_local_preview_access
        self.webui_allow_local_service_access = webui_allow_local_service_access
@@ -236,8 +266,11 @@ class ExecTool(Tool):
            "Use -y or --yes flags to avoid interactive prompts. "
            "For long-running or interactive commands, pass yield_time_ms; "
            "if the command keeps running, exec returns a session_id that can "
-            "be polled or written to with write_stdin. Output is truncated at "
-            "10 000 chars; timeout defaults to 60s."
+            "be polled or written to with write_stdin. For services that "
+            "must remain available after you finish, pass detach=true instead "
+            "of yield_time_ms; detached output is written to a log file and "
+            "the tool returns a pid. Output is truncated at 10 000 chars; "
+            "timeout defaults to 60s."
        )

    @property
@@ -251,6 +284,7 @@ class ExecTool(Tool):
        login: bool | None = None, yield_time_ms: int | None = None,
        max_output_chars: int | None = None,
        max_output_tokens: int | None = None,
+        detach: bool | None = False,
        **kwargs: Any,
    ) -> str:
        command = command or cmd
@@ -264,10 +298,14 @@ class ExecTool(Tool):
        if isinstance(prepared, str):
            return prepared

+        if detach:
+            return await self._execute_detached(prepared)
+
        if yield_time_ms is not None:
            return await self._execute_session(prepared, yield_time_ms, max_output_chars)

        try:
+            started_at = time.monotonic()
            process = await self._spawn(
                prepared.command,
                prepared.cwd,
@@ -283,7 +321,15 @@ class ExecTool(Tool):
                )
            except asyncio.TimeoutError:
                await self._kill_process(process)
-                return f"Error: Command timed out after {prepared.timeout} seconds"
+                result = f"Error: Command timed out after {prepared.timeout} seconds"
+                analysis = analyze_verification_result(
+                    command=prepared.command,
+                    output=result,
+                    exit_code=None,
+                    timed_out=True,
+                )
+                record_verification_observation(current_request_session_key(), analysis)
+                return append_verification_feedback(result, analysis)
            except asyncio.CancelledError:
                await self._kill_process(process)
                raise
@@ -301,17 +347,35 @@ class ExecTool(Tool):
            output_parts.append(f"\nExit code: {process.returncode}")

            result = "\n".join(output_parts) if output_parts else "(no output)"
+            elapsed_s = max(0.0, time.monotonic() - started_at)
+
+            analysis = analyze_verification_result(
+                command=prepared.command,
+                output=result,
+                exit_code=process.returncode,
+            )

            max_len = clamp_session_int(max_output_chars, self._MAX_OUTPUT, 1000, MAX_OUTPUT_CHARS)
            if len(result) > max_len:
-                half = max_len // 2
-                result = (
-                    result[:half]
-                    + f"\n\n... ({len(result) - max_len:,} chars truncated) ...\n\n"
-                    + result[-half:]
+                result = build_structured_output_summary(
+                    "[tool output truncated]",
+                    result,
+                    max_chars=max_len,
+                    metadata=[
+                        ("original_size_chars", len(result)),
+                        ("exit_code", process.returncode),
+                        ("duration_s", f"{elapsed_s:.1f}"),
+                    ],
+                    analysis=analysis,
+                    guidance=(
+                        "Use the structured summary first. Rerun a narrower "
+                        "command, grep a specific failure, or inspect the "
+                        "named artifact instead of rerunning broad noisy logs."
+                    ),
                )

-            return result
+            record_verification_observation(current_request_session_key(), analysis)
+            return append_verification_feedback(result, analysis)

        except Exception as e:
            return f"Error executing command: {str(e)}"
@@ -339,10 +403,71 @@ class ExecTool(Tool):
                    MAX_OUTPUT_CHARS,
                ),
            )
-            return format_session_poll(session_id, poll)
+            result = format_session_poll(session_id, poll)
+            if poll.done:
+                analysis = analyze_verification_result(
+                    command=prepared.command,
+                    output=result,
+                    exit_code=poll.exit_code,
+                    timed_out=poll.timed_out,
+                )
+                record_verification_observation(current_request_session_key(), analysis)
+                return append_verification_feedback(result, analysis)
+            return result
        except Exception as exc:
            return f"Error executing command: {exc}"

+    async def _execute_detached(self, prepared: _PreparedCommand) -> str:
+        log_dir = Path(prepared.cwd) / ".nanobot" / "exec-logs"
+        try:
+            log_dir.mkdir(parents=True, exist_ok=True)
+            log_path = log_dir / f"detached-{uuid.uuid4().hex[:12]}.log"
+        except Exception as exc:
+            return f"Error preparing detached command log directory: {exc}"
+
+        log_handle = None
+        try:
+            log_handle = open(log_path, "ab", buffering=0)
+            process = await self._spawn(
+                prepared.command,
+                prepared.cwd,
+                prepared.env,
+                prepared.shell_program,
+                prepared.login,
+                stdout=log_handle,
+                stderr=log_handle,
+                start_new_session=not _IS_WINDOWS,
+                creationflags=subprocess.CREATE_NEW_PROCESS_GROUP if _IS_WINDOWS else 0,
+            )
+        except Exception as exc:
+            return f"Error starting detached command: {exc}"
+        finally:
+            if log_handle is not None:
+                with suppress(Exception):
+                    log_handle.close()
+
+        try:
+            exit_code = await asyncio.wait_for(process.wait(), timeout=_DETACHED_EXIT_GRACE_S)
+        except asyncio.TimeoutError:
+            return (
+                "Detached process started.\n"
+                f"pid: {process.pid}\n"
+                f"cwd: {prepared.cwd}\n"
+                f"log: {log_path}\n"
+                "Poll the log or run a health check to verify the service is ready."
+            )
+
+        log_text = ""
+        with suppress(Exception):
+            log_text = log_path.read_text(encoding="utf-8", errors="replace")
+        if len(log_text) > 4000:
+            log_text = log_text[-4000:]
+        return (
+            f"Detached process exited immediately with code {exit_code}.\n"
+            f"log: {log_path}\n"
+            f"{log_text}"
+        )
+
    def _resolve_timeout(self, timeout: int | None) -> int | None:
        """Resolve the effective hard timeout in seconds (None = no limit).

@@ -464,6 +589,10 @@ class ExecTool(Tool):
        login: bool = False,
        *,
        stdin: int = asyncio.subprocess.DEVNULL,
+        stdout: Any = asyncio.subprocess.PIPE,
+        stderr: Any = asyncio.subprocess.PIPE,
+        start_new_session: bool = False,
+        creationflags: int = 0,
    ) -> asyncio.subprocess.Process:
        """Launch *command* in a platform-appropriate shell."""
        if _IS_WINDOWS:
@@ -471,18 +600,20 @@ class ExecTool(Tool):
                return await asyncio.create_subprocess_exec(
                    "powershell", "-NoProfile", "-Command", command,
                    stdin=stdin,
-                    stdout=asyncio.subprocess.PIPE,
-                    stderr=asyncio.subprocess.PIPE,
+                    stdout=stdout,
+                    stderr=stderr,
                    cwd=cwd,
                    env=env,
+                    creationflags=creationflags,
                )
            return await asyncio.create_subprocess_shell(
                command,
                stdin=stdin,
-                stdout=asyncio.subprocess.PIPE,
-                stderr=asyncio.subprocess.PIPE,
+                stdout=stdout,
+                stderr=stderr,
                cwd=cwd,
                env=env,
+                creationflags=creationflags,
            )
        shell_program = shell_program or shutil.which("bash") or "/bin/bash"
        args = [shell_program]
@@ -493,10 +624,11 @@ class ExecTool(Tool):
        return await asyncio.create_subprocess_exec(
            *args,
            stdin=stdin,
-            stdout=asyncio.subprocess.PIPE,
-            stderr=asyncio.subprocess.PIPE,
+            stdout=stdout,
+            stderr=stderr,
            cwd=cwd,
            env=env,
+            start_new_session=start_new_session,
        )

    @staticmethod
@@ -614,11 +746,12 @@ class ExecTool(Tool):
                return "Error: Command blocked by allowlist filter (not in allowlist)"

        from nanobot.security.network import contains_internal_url
+        allow_loopback = self.allow_local_service_access or current_scope_allows_loopback(
+            enabled=self.webui_allow_local_service_access,
+        )
        if contains_internal_url(
            cmd,
-            allow_loopback=current_scope_allows_loopback(
-                enabled=self.webui_allow_local_service_access,
-            ),
+            allow_loopback=allow_loopback,
        ):
            # The runner turns this marker into a non-retryable security hint.
            return "Error: Command blocked by safety guard (internal/private URL detected)"
@@ -63,9 +63,6 @@ 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."
@@ -0,0 +1,292 @@
+"""Lightweight verification-result detection for coding workflows."""
+
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass
+from typing import Literal
+
+VerificationStatus = Literal["passed", "failed"]
+
+
+@dataclass(frozen=True, slots=True)
+class VerificationAnalysis:
+    """Structured summary of a command that appears to be verification."""
+
+    status: VerificationStatus
+    command: str
+    exit_code: int | None
+    failed_tests: tuple[str, ...] = ()
+    primary_errors: tuple[str, ...] = ()
+    missing_artifacts: tuple[str, ...] = ()
+    timed_out: bool = False
+
+
+@dataclass(frozen=True, slots=True)
+class VerificationObservation:
+    """Latest verification signal observed for a session."""
+
+    analysis: VerificationAnalysis
+    sequence: int
+
+
+_OBSERVATIONS: dict[str, VerificationObservation] = {}
+_SEQUENCE = 0
+
+_TEST_COMMAND_RE = re.compile(
+    r"(?ix)"
+    r"("
+    r"\bpytest\b|\bpy\.test\b|\bunittest\b|\bnosetests\b|"
+    r"\btest_outputs\.py\b|\brun_tests?(?:\.sh|\.py)?\b|"
+    r"\bnpm\s+(?:run\s+)?test\b|\byarn\s+test\b|\bpnpm\s+test\b|"
+    r"\bcargo\s+test\b|\bgo\s+test\b|\bctest\b|"
+    r"\bmake\s+(?:[^;&|]*\s+)?test\b"
+    r")"
+)
+_ARTIFACT_CHECK_COMMAND_RE = re.compile(
+    r"(?ix)"
+    r"("
+    r"\bcmp\b|"
+    r"\bdiff\b|"
+    r"\bsha(?:1|224|256|384|512)?sum\b|"
+    r"\bmd5sum\b|"
+    r"\bgcc\b.*(?:&&|;).*\./|"
+    r"\bclang\b.*(?:&&|;).*\./|"
+    r"\bpython3?\b.*<<['\"]?PY\b.*\bassert\b"
+    r")"
+)
+_COMPARISON_COMMAND_RE = re.compile(r"(?i)\b(?:cmp|diff)\b")
+_FAILURE_RE = re.compile(
+    r"(?im)"
+    r"("
+    r"^FAILED\s+|"
+    r"\b\d+\s+failed\b|"
+    r"\bAssertionError\b|"
+    r"\bFileNotFoundError\b|"
+    r"\bTimeoutError\b|"
+    r"\bcommand not found\b|"
+    r"\bError:\s+Command timed out\b|"
+    r"\bFAILURES?\b|"
+    r"\bTEST FAILED\b"
+    r")"
+)
+_SUCCESS_RE = re.compile(
+    r"(?im)"
+    r"("
+    r"\b\d+\s+passed\b|"
+    r"\bOK\b|"
+    r"\bTEST PASSED\b|"
+    r"\bExit code:\s*0\b"
+    r")"
+)
+_ARTIFACT_SUCCESS_RE = re.compile(
+    r"(?im)"
+    r"("
+    r"\b(?:cmp|diff|test|verify)_exit:\s*0\b|"
+    r"^\s*(?:cmp|diff|match|same|image|ppm|stdout|stderr|out|err)[\w.-]*:\s*0\s*$"
+    r")"
+)
+_ARTIFACT_FAILURE_RE = re.compile(
+    r"(?im)"
+    r"("
+    r"\b(?:cmp|diff|test|verify)_exit:\s*[1-9]\d*\b|"
+    r"^\s*(?:cmp|diff|match|same|image|ppm|stdout|stderr|out|err)[\w.-]*:\s*[1-9]\d*\s*$"
+    r")"
+)
+_FAILED_TEST_RE = re.compile(r"(?m)^FAILED\s+([^\s]+)")
+_PYTEST_SHORT_RE = re.compile(r"(?m)^_{3,}\s+([A-Za-z0-9_./:-]+)\s+_{3,}$")
+_ERROR_LINE_RE = re.compile(
+    r"(?m)"
+    r"^\s*(?:E\s+)?("
+    r"(?:AssertionError|FileNotFoundError|TimeoutError|ValueError|TypeError|RuntimeError)"
+    r"(?::[^\n]*)?|"
+    r"assert\s+[^\n]+|"
+    r"[^:\n]+:\s+line\s+\d+:\s+[^:\n]+:\s+command not found|"
+    r"Error:\s+[^\n]+|"
+    r"TEST FAILED[^\n]*"
+    r")"
+)
+_MISSING_PATH_RE = re.compile(
+    r"(?i)"
+    r"(?:No such file or directory:\s*['\"]([^'\"]+)['\"]|"
+    r"(?:file|path)\s+([^\s'\"]+)\s+does not exist|"
+    r"cannot open file\s+['\"]([^'\"]+)['\"])"
+)
+
+
+def analyze_verification_result(
+    *,
+    command: str,
+    output: str,
+    exit_code: int | None,
+    timed_out: bool = False,
+) -> VerificationAnalysis | None:
+    """Return a verification summary when a command/output looks like a test."""
+
+    command = " ".join((command or "").split())
+    looks_like_test_command = bool(_TEST_COMMAND_RE.search(command))
+    looks_like_artifact_check = bool(_ARTIFACT_CHECK_COMMAND_RE.search(command))
+    looks_like_comparison_command = bool(_COMPARISON_COMMAND_RE.search(command))
+    looks_like_verification = looks_like_test_command or looks_like_artifact_check
+    failure_seen = bool(_FAILURE_RE.search(output))
+    success_seen = bool(_SUCCESS_RE.search(output))
+    artifact_success_seen = bool(_ARTIFACT_SUCCESS_RE.search(output)) and (
+        looks_like_comparison_command or bool(re.search(r"\b(?:test|verify)_exit:\s*0\b", output, flags=re.I))
+    )
+    artifact_failure_seen = bool(_ARTIFACT_FAILURE_RE.search(output)) and (
+        looks_like_comparison_command or bool(re.search(r"\b(?:test|verify)_exit:\s*[1-9]\d*\b", output, flags=re.I))
+    )
+
+    if not looks_like_test_command and not failure_seen:
+        if not (looks_like_artifact_check and artifact_success_seen and exit_code == 0):
+            return None
+
+    if (
+        (timed_out and looks_like_verification)
+        or (exit_code not in (None, 0) and (looks_like_verification or failure_seen))
+        or failure_seen
+        or artifact_failure_seen
+    ):
+        return VerificationAnalysis(
+            status="failed",
+            command=command,
+            exit_code=exit_code,
+            failed_tests=_unique(_FAILED_TEST_RE.findall(output), limit=8),
+            primary_errors=_extract_primary_errors(output),
+            missing_artifacts=_extract_missing_artifacts(output),
+            timed_out=timed_out,
+        )
+
+    if looks_like_test_command and exit_code == 0 and success_seen:
+        return VerificationAnalysis(
+            status="passed",
+            command=command,
+            exit_code=exit_code,
+        )
+
+    if looks_like_artifact_check and exit_code == 0 and artifact_success_seen:
+        return VerificationAnalysis(
+            status="passed",
+            command=command,
+            exit_code=exit_code,
+        )
+
+    return None
+
+
+def append_verification_feedback(output: str, analysis: VerificationAnalysis | None) -> str:
+    """Append model-facing feedback for failed verification results."""
+
+    if analysis is None or analysis.status != "failed":
+        return output
+
+    lines = [
+        "",
+        "[Verification Feedback]",
+        "Verification status: failed.",
+        "Do not call complete_goal or present the task as finished until this is fixed and a verification passes.",
+    ]
+    if analysis.command:
+        lines.append(f"Command: {analysis.command[:240]}")
+    if analysis.exit_code is not None:
+        lines.append(f"Exit code: {analysis.exit_code}")
+    if analysis.timed_out:
+        lines.append("Failure type: command timeout")
+    if analysis.failed_tests:
+        lines.append("Failed tests:")
+        lines.extend(f"- {item}" for item in analysis.failed_tests)
+    if analysis.primary_errors:
+        lines.append("Primary errors:")
+        lines.extend(f"- {item}" for item in analysis.primary_errors)
+    if analysis.missing_artifacts:
+        lines.append("Missing artifacts:")
+        lines.extend(f"- {item}" for item in analysis.missing_artifacts)
+    lines.append("Next action: inspect the failing assertion, fix the implementation or artifact, then rerun the most specific verification command.")
+    lines.append("[/Verification Feedback]")
+    return output.rstrip() + "\n" + "\n".join(lines)
+
+
+def record_verification_observation(session_key: str | None, analysis: VerificationAnalysis | None) -> None:
+    """Remember the latest verification signal for a session."""
+
+    if not session_key or analysis is None:
+        return
+    global _SEQUENCE
+    _SEQUENCE += 1
+    _OBSERVATIONS[session_key] = VerificationObservation(
+        analysis=analysis,
+        sequence=_SEQUENCE,
+    )
+
+
+def latest_verification_observation(session_key: str | None) -> VerificationObservation | None:
+    if not session_key:
+        return None
+    return _OBSERVATIONS.get(session_key)
+
+
+def clear_verification_observation(session_key: str | None) -> None:
+    if session_key:
+        _OBSERVATIONS.pop(session_key, None)
+
+
+def format_completion_gate_message(observation: VerificationObservation) -> str:
+    """Build the complete_goal soft-gate message for unresolved failures."""
+
+    analysis = observation.analysis
+    lines = [
+        "Recent verification appears to have failed, so the goal is not marked complete yet.",
+        "Continue fixing the task and rerun verification before completing.",
+    ]
+    if analysis.command:
+        lines.append(f"Last failed verification command: {analysis.command[:240]}")
+    if analysis.failed_tests:
+        lines.append("Failed tests: " + ", ".join(analysis.failed_tests[:5]))
+    if analysis.primary_errors:
+        lines.append("Primary error: " + analysis.primary_errors[0])
+    if analysis.missing_artifacts:
+        lines.append("Missing artifact: " + analysis.missing_artifacts[0])
+    lines.append(
+        "If you are intentionally stopping with known failures, call complete_goal again with remaining_failures describing them honestly."
+    )
+    return "\n".join(lines)
+
+
+def _extract_primary_errors(output: str) -> tuple[str, ...]:
+    candidates: list[str] = []
+    for match in _ERROR_LINE_RE.findall(output):
+        text = " ".join(match.split())
+        if text and text not in candidates:
+            candidates.append(text[:240])
+        if len(candidates) >= 8:
+            break
+    if not candidates:
+        for match in _PYTEST_SHORT_RE.findall(output):
+            text = " ".join(match.split())
+            if text and text not in candidates:
+                candidates.append(text[:240])
+            if len(candidates) >= 4:
+                break
+    return tuple(candidates)
+
+
+def _extract_missing_artifacts(output: str) -> tuple[str, ...]:
+    paths: list[str] = []
+    for groups in _MISSING_PATH_RE.findall(output):
+        path = next((item for item in groups if item), "")
+        if path and path not in paths:
+            paths.append(path[:240])
+        if len(paths) >= 8:
+            break
+    return tuple(paths)
+
+
+def _unique(items: list[str], *, limit: int) -> tuple[str, ...]:
+    out: list[str] = []
+    for item in items:
+        text = " ".join(item.split())
+        if text and text not in out:
+            out.append(text[:240])
+        if len(out) >= limit:
+            break
+    return tuple(out)
@@ -129,13 +129,6 @@ 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):
@@ -174,10 +167,6 @@ 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
@@ -1234,38 +1223,14 @@ class WeixinChannel(BaseChannel):
    async def send_delta(
        self, chat_id: str, delta: str, metadata: dict[str, Any] | None = None
    ) -> None:
-        """Deliver a streamed reply to WeChat.
+        """Weixin iLink does not support native streaming deltas.

-        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.
+        We only hook ``_stream_end`` so buffered tool hints are flushed even
+        when the final answer carries the ``_streamed`` flag and bypasses
+        :meth:`send`.
        """
-        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)
+        if metadata and metadata.get("_stream_end"):
+            await self._flush_tool_hints(chat_id)

    async def _start_typing(self, chat_id: str, context_token: str = "") -> None:
        """Start typing indicator immediately when a message is received."""
@@ -499,30 +499,6 @@ 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")
@@ -556,9 +532,6 @@ 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]
@@ -1744,11 +1744,6 @@ _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."""
@@ -1780,51 +1775,9 @@ 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)
@@ -1836,8 +1789,6 @@ 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")
@@ -1861,23 +1812,14 @@ 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(proxy=proxy)
+            token = get_token()
        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]")
@@ -4,7 +4,6 @@ from __future__ import annotations

 import asyncio
 import os
-import subprocess
 import sys
 import time
 from contextlib import suppress
@@ -51,7 +50,7 @@ BUILTIN_COMMAND_SPECS: tuple[BuiltinCommandSpec, ...] = (
    BuiltinCommandSpec(
        "/restart",
        "Restart nanobot",
-        "Restart the bot process.",
+        "Restart the bot process in place.",
        "rotate-cw",
    ),
    BuiltinCommandSpec(
@@ -131,15 +130,6 @@ 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,
@@ -148,7 +138,7 @@ async def cmd_stop(ctx: CommandContext) -> OutboundMessage:


 async def cmd_restart(ctx: CommandContext) -> OutboundMessage:
-    """Restart the process."""
+    """Restart the process in-place via os.execv."""
    msg = ctx.msg
    set_restart_notice_to_env(
        channel=msg.channel,
@@ -158,19 +148,7 @@ async def cmd_restart(ctx: CommandContext) -> OutboundMessage:

    async def _do_restart():
        await asyncio.sleep(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)
+        os.execv(sys.executable, [sys.executable, "-m", "nanobot"] + sys.argv[1:])

    asyncio.create_task(_do_restart())
    return OutboundMessage(
@@ -648,7 +626,7 @@ async def cmd_history(ctx: CommandContext) -> OutboundMessage:

 _GOAL_PROMPT_TEMPLATE = """The user declared a sustained objective for this thread.

-Inspect or clarify if needed, then call `long_task` with the refined objective (and optional short ui_summary). Work proceeds as normal assistant turns using your usual tools. When the objective is fully done and verified, call `complete_goal` with a brief recap. If the user later cancels or changes direction, still call `complete_goal` with an honest recap (then `long_task` again only after there is no active goal). Do not use `long_task` / `complete_goal` for trivial one-shot answers.
+Inspect or clarify if needed, then call `long_task` with the refined objective (and optional short ui_summary). Work proceeds as normal assistant turns using your usual tools. When the objective is fully done and verified, call `complete_goal` with a brief recap plus verification_summary / commands_run / artifacts_created when applicable. If the user later cancels or changes direction, still call `complete_goal` with an honest recap (then `long_task` again only after there is no active goal). Do not use `long_task` / `complete_goal` for trivial one-shot answers.

 Goal:
 {goal}
@@ -7,7 +7,6 @@ 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
@@ -80,10 +79,6 @@ 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)
@@ -157,23 +152,6 @@ 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", {})
@@ -154,6 +154,10 @@ 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,
@@ -179,7 +183,6 @@ 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
@@ -314,7 +317,6 @@ 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)


@@ -357,25 +357,6 @@ 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:
@@ -641,7 +622,7 @@ class CronService:

    def list_jobs(self, include_disabled: bool = False) -> list[CronJob]:
        """List all jobs."""
-        store = self._require_store()
+        store = self._load_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'))

@@ -703,7 +684,7 @@ class CronService:
        _normalize_agent_turn_job(job)
        self._enforce_agent_binding(job)
        if self._running:
-            store = self._require_store()
+            store = self._load_store()
            store.jobs.append(job)
            self._save_store()
            self._arm_timer()
@@ -715,7 +696,7 @@ class CronService:

    def register_system_job(self, job: CronJob) -> CronJob:
        """Register an internal system job (idempotent on restart)."""
-        store = self._require_store()
+        store = self._load_store()
        now = _now_ms()
        job.state = CronJobState(next_run_at_ms=_compute_next_run(job.schedule, now))
        job.created_at_ms = now
@@ -729,7 +710,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._require_store()
+        store = self._load_store()
        job = next((j for j in store.jobs if j.id == job_id), None)
        if job is None:
            return "not_found"
@@ -754,7 +735,7 @@ class CronService:

    def enable_job(self, job_id: str, enabled: bool = True) -> CronJob | None:
        """Enable or disable a job."""
-        store = self._require_store()
+        store = self._load_store()
        for job in store.jobs:
            if job.id == job_id:
                job.enabled = enabled
@@ -789,7 +770,7 @@ class CronService:
        For ``channel`` and ``to``, pass an explicit value (including ``None``)
        to update; omit (sentinel ``...``) to leave unchanged.
        """
-        store = self._require_store()
+        store = self._load_store()
        job = next((j for j in store.jobs if j.id == job_id), None)
        if job is None:
            return "not_found"
@@ -834,7 +815,7 @@ class CronService:
        was_running = self._running
        self._running = True
        try:
-            store = self._require_store()
+            store = self._load_store()
            for job in store.jobs:
                if job.id == job_id:
                    if self._is_unbound_agent_job(job):
@@ -854,12 +835,12 @@ class CronService:

    def get_job(self, job_id: str) -> CronJob | None:
        """Get a job by ID."""
-        store = self._require_store()
+        store = self._load_store()
        return next((j for j in store.jobs if j.id == job_id), None)

    def status(self) -> dict:
        """Get service status."""
-        store = self._require_store()
+        store = self._load_store()
        return {
            "enabled": self._running,
            "jobs": len(store.jobs),
@@ -54,18 +54,6 @@ 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 = (
@@ -58,11 +58,6 @@ 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:
@@ -84,10 +79,7 @@ def _make_provider_core(
    if backend == "openai_codex":
        from nanobot.providers.openai_codex_provider import OpenAICodexProvider

-        provider = OpenAICodexProvider(
-            default_model=model,
-            proxy=getattr(p, "proxy", None) if p else None,
-        )
+        provider = OpenAICodexProvider(default_model=model)
    elif backend == "azure_openai":
        from nanobot.providers.azure_openai_provider import AzureOpenAIProvider

@@ -132,7 +124,6 @@ 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()
@@ -227,7 +218,6 @@ 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)
@@ -247,7 +237,6 @@ 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),
    )

@@ -2,7 +2,6 @@

 from __future__ import annotations

-import os
 import time
 import webbrowser
 from collections.abc import Awaitable, Callable
@@ -30,12 +29,6 @@ _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,
@@ -75,16 +68,11 @@ 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(
-            device_code_url,
+            DEFAULT_GITHUB_DEVICE_CODE_URL,
            headers={"Accept": "application/json", "User-Agent": USER_AGENT},
-            data={"client_id": client_id, "scope": GITHUB_COPILOT_SCOPE},
+            data={"client_id": GITHUB_COPILOT_CLIENT_ID, "scope": GITHUB_COPILOT_SCOPE},
        )
        response.raise_for_status()
        payload = response.json()
@@ -108,10 +96,10 @@ def login_github_copilot(
        token_expires_in = _LONG_LIVED_TOKEN_SECONDS
        while time.time() < deadline:
            poll = client.post(
-                access_token_url,
+                DEFAULT_GITHUB_ACCESS_TOKEN_URL,
                headers={"Accept": "application/json", "User-Agent": USER_AGENT},
                data={
-                    "client_id": client_id,
+                    "client_id": GITHUB_COPILOT_CLIENT_ID,
                    "device_code": device_code,
                    "grant_type": "urn:ietf:params:oauth:grant-type:device_code",
                },
@@ -144,7 +132,7 @@ def login_github_copilot(
            raise RuntimeError("GitHub device flow timed out.")

        user = client.get(
-            user_url,
+            DEFAULT_GITHUB_USER_URL,
            headers={
                "Authorization": f"Bearer {access_token}",
                "Accept": "application/vnd.github+json",
@@ -176,7 +164,7 @@ class GitHubCopilotProvider(OpenAICompatProvider):
        self._copilot_expires_at: float = 0.0
        super().__init__(
            api_key="no-key",
-            api_base=_resolve("NANOBOT_COPILOT_BASE_URL", DEFAULT_COPILOT_BASE_URL),
+            api_base=DEFAULT_COPILOT_BASE_URL,
            default_model=default_model,
            extra_headers={
                "Editor-Version": EDITOR_VERSION,
@@ -198,7 +186,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(
-                _resolve("NANOBOT_COPILOT_TOKEN_URL", DEFAULT_COPILOT_TOKEN_URL),
+                DEFAULT_COPILOT_TOKEN_URL,
                headers=_copilot_headers(github_token.access),
            )
            response.raise_for_status()
@@ -2,6 +2,7 @@

 from __future__ import annotations

+import ast
 import asyncio
 import hashlib
 import json
@@ -26,6 +27,25 @@ from nanobot.providers.openai_responses import (

 DEFAULT_CODEX_URL = "https://chatgpt.com/backend-api/codex/responses"
 DEFAULT_ORIGINATOR = "nanobot"
+_RESPONSE_FAILED_PREFIX = "Response failed:"
+_RETRYABLE_RESPONSE_FAILED_TOKENS = frozenset({
+    "overloaded",
+    "overloaded_error",
+    "rate_limit_exceeded",
+    "request_limit_exceeded",
+    "requests_limit_exceeded",
+    "server_error",
+    "server_is_overloaded",
+    "service_unavailable",
+    "temporarily_unavailable",
+    "too_many_requests",
+})
+_NON_RETRYABLE_RESPONSE_FAILED_TOKENS = frozenset({
+    "content_filter",
+    "content_policy_violation",
+    "cyber_policy",
+    "safety_violation",
+})


 class OpenAICodexProvider(LLMProvider):
@@ -33,14 +53,9 @@ class OpenAICodexProvider(LLMProvider):

    supports_progress_deltas = True

-    def __init__(
-        self,
-        default_model: str = "openai-codex/gpt-5.1-codex",
-        proxy: str | None = None,
-    ):
+    def __init__(self, default_model: str = "openai-codex/gpt-5.1-codex"):
        super().__init__(api_key=None, api_base=None)
        self.default_model = default_model
-        self.proxy = proxy or None

    async def _call_codex(
        self,
@@ -57,6 +72,9 @@ 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,
@@ -76,13 +94,9 @@ 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,
@@ -93,7 +107,6 @@ 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,
@@ -206,17 +219,12 @@ 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()
-    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 httpx.AsyncClient(timeout=idle_timeout_s, verify=verify) as client:
        async with client.stream("POST", url, headers=headers, json=body) as response:
            if response.status_code != 200:
                text = await response.aread()
@@ -258,6 +266,8 @@ def _codex_error_response(exc: Exception) -> LLMResponse:

    status_code = getattr(exc, "status_code", None)
    error_kind: str | None = None
+    error_type = getattr(exc, "error_type", None)
+    error_code = getattr(exc, "error_code", None)
    default_detail: str | None = None
    should_retry: bool | None = getattr(exc, "should_retry", None)

@@ -277,12 +287,20 @@ def _codex_error_response(exc: Exception) -> LLMResponse:
        error_kind = "http"
        default_detail = "HTTP request failed"

+    failed_type, failed_code = _extract_response_failed_error(detail)
+    if failed_type or failed_code:
+        error_kind = error_kind or "provider"
+        error_type = failed_type or error_type
+        error_code = failed_code or error_code
+        if should_retry is None:
+            should_retry = _should_retry_response_failed(error_type, error_code, detail)
+
    if status_code is not None and should_retry is None:
        retry_content = None if int(status_code) == 429 and isinstance(exc, _CodexHTTPError) else detail
        should_retry = _should_retry_status(
            int(status_code),
-            getattr(exc, "error_type", None),
-            getattr(exc, "error_code", None),
+            error_type,
+            error_code,
            retry_content,
        )

@@ -295,13 +313,56 @@ def _codex_error_response(exc: Exception) -> LLMResponse:
        retry_after=retry_after,
        error_status_code=int(status_code) if status_code is not None else None,
        error_kind=error_kind,
-        error_type=getattr(exc, "error_type", None),
-        error_code=getattr(exc, "error_code", None),
+        error_type=error_type,
+        error_code=error_code,
        error_retry_after_s=retry_after,
        error_should_retry=should_retry,
    )


+def _extract_response_failed_error(detail: str) -> tuple[str | None, str | None]:
+    """Extract provider semantic error fields from Responses SSE failures."""
+    if _RESPONSE_FAILED_PREFIX not in detail:
+        return None, None
+
+    payload = detail.split(_RESPONSE_FAILED_PREFIX, 1)[1].strip()
+    if not payload:
+        return None, None
+
+    parsed: Any = None
+    try:
+        parsed = json.loads(payload)
+    except Exception:
+        try:
+            parsed = ast.literal_eval(payload)
+        except Exception:
+            parsed = None
+
+    error_type, error_code = LLMProvider._extract_error_type_code(parsed or payload)
+    return error_type, error_code
+
+
+def _should_retry_response_failed(
+    error_type: str | None,
+    error_code: str | None,
+    detail: str,
+) -> bool | None:
+    semantic_tokens = {
+        token for token in (
+            LLMProvider._normalize_error_token(error_type),
+            LLMProvider._normalize_error_token(error_code),
+        )
+        if token is not None
+    }
+    if any(token in _NON_RETRYABLE_RESPONSE_FAILED_TOKENS for token in semantic_tokens):
+        return False
+    if any(token in _RETRYABLE_RESPONSE_FAILED_TOKENS for token in semantic_tokens):
+        return True
+    if LLMProvider._is_transient_error(detail):
+        return True
+    return None
+
+
 def _codex_log_summary(exc_type: str, response: LLMResponse) -> str:
    """Return a bounded diagnostic summary without request body or raw upstream payload."""
    if response.error_status_code is not None:
@@ -358,7 +358,6 @@ 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
@@ -367,7 +366,6 @@ 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)
@@ -398,14 +396,7 @@ class OpenAICompatProvider(LLMProvider):

        timeout_s = _openai_compat_timeout_s()
        http_client: httpx.AsyncClient | None = None
-        if self._proxy:
-            http_client = httpx.AsyncClient(
-                timeout=timeout_s,
-                proxy=self._proxy,
-                trust_env=False,
-                follow_redirects=True,
-            )
-        elif self._is_local:
+        if 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
@@ -27,8 +27,6 @@ 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*$')
@@ -45,15 +43,6 @@ _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.

@@ -110,12 +99,6 @@ 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."""
@@ -149,7 +132,7 @@ class Session:

    def get_history(
        self,
-        max_messages: int = FILE_MAX_MESSAGES,
+        max_messages: int = 120,
        *,
        max_tokens: int = 0,
        extend_to_user: bool = False,
@@ -160,7 +143,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 FILE_MAX_MESSAGES
+        max_messages = max_messages if max_messages > 0 else 120
        start_idx = recent_message_start_index(
            unconsolidated,
            max_messages,
@@ -295,26 +278,22 @@ class Session:
        max_messages: int,
        *,
        extend_to_user: bool = False,
-    ) -> RetentionResult:
+    ) -> tuple[list[dict], int]:
        """Keep a legal recent suffix, optionally extending it back to a user turn.

-        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.
+        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.
        """
        if max_messages <= 0:
            dropped = list(self.messages)
            lc = self.last_consolidated
            self.clear()
-            return RetentionResult(
-                dropped=dropped,
-                already_consolidated_count=min(lc, len(dropped)),
-            )
+            return dropped, min(lc, len(dropped))
        if len(self.messages) <= max_messages:
-            return RetentionResult(
-                dropped=[],
-                already_consolidated_count=0,
-            )
+            return [], 0

        original = list(self.messages)
        before_lc = self.last_consolidated
@@ -380,10 +359,7 @@ class Session:
        self.messages = retained
        self.last_consolidated = new_lc
        self.updated_at = datetime.now()
-        return RetentionResult(
-            dropped=dropped,
-            already_consolidated_count=already_consolidated,
-        )
+        return dropped, already_consolidated

    def enforce_file_cap(
        self,
@@ -394,17 +370,17 @@ class Session:
        if limit <= 0 or len(self.messages) <= limit:
            return

-        result = self.retain_recent_legal_suffix(limit)
-        if not result.dropped:
+        dropped, already_consolidated = self.retain_recent_legal_suffix(limit)
+        if not dropped:
            return

-        archive_chunk = result.dropped[result.already_consolidated_count:]
+        archive_chunk = dropped[already_consolidated:]
        if archive_chunk and on_archive:
            on_archive(archive_chunk)
        logger.info(
            "Session file cap hit for {}: dropped {}, raw-archived {}, kept {}",
            self.key,
-            len(result.dropped),
+            len(dropped),
            len(archive_chunk),
            len(self.messages),
        )
@@ -563,10 +539,9 @@ class SessionManager:
                logger.info("Recovered session {} from corrupt file ({} messages)", key, len(repaired.messages))
            return repaired

-    def _repair(self, key: str, *, path: Path | None = None) -> Session | None:
+    def _repair(self, key: str) -> Session | None:
        """Attempt to recover a session from a corrupt JSONL file."""
-        if path is None:
-            path = self._get_session_path(key)
+        path = self._get_session_path(key)
        if not path.exists():
            return None

@@ -640,6 +615,7 @@ class SessionManager:
        the most recent writes.
        """
        path = self._get_session_path(session.key)
+        path.parent.mkdir(parents=True, exist_ok=True)
        tmp_path = path.with_suffix(".jsonl.tmp")

        try:
@@ -918,7 +894,7 @@ class SessionManager:
                                }
                            )
            except Exception:
-                repaired = self._repair(fallback_key, path=path)
+                repaired = self._repair(fallback_key)
                if repaired is not None:
                    sessions.append(
                        {
@@ -26,7 +26,7 @@ Those belong to the execution phase after the marker is set.

 - **`long_task`** — Register **one** sustained objective per thread. Call it promptly once the user has asked for a sustained task. The `goal` should follow the idempotent-goal rules below, but it should be produced quickly from the user's request—not after a long hidden planning pass.

- **`complete_goal`** — Close bookkeeping for the **current** active goal. Call when work is **done**, **and also** when the user **cancels**, **changes direction**, or **replaces** the objective: use **`recap`** to state honestly what happened (e.g. cancelled, partially done, superseded). Then you may call **`long_task`** again for a **new** objective after the session shows no active goal (or after the user agrees to replace).
+- **`complete_goal`** — Close bookkeeping for the **current** active goal. Call when work is **done**, **and also** when the user **cancels**, **changes direction**, or **replaces** the objective: use **`recap`** to state honestly what happened (e.g. cancelled, partially done, superseded). For coding or file-producing tasks, include **`verification_summary`**, **`commands_run`**, and **`artifacts_created`** when possible; if stopping with known unresolved issues, fill **`remaining_failures`** honestly. Then you may call **`long_task`** again for a **new** objective after the session shows no active goal (or after the user agrees to replace).

 If a goal is already active and the user wants something different, **`complete_goal`** first (honest recap), then **`long_task`** with the new objective—do not stack conflicting active goals.

@@ -68,7 +68,7 @@ Use this when the goal is to **build or reshape a codebase** (app, service, tool

 1. **Modular layout** — Split into **meaningful modules** (directories + files with clear responsibilities: entrypoints, domain logic, config, infra, CLI/UI routes, etc.). **Do not** default to dumping an entire project into one giant source file unless the user explicitly wants a minimal single-file artifact.
 2. **Conventional structure** — Follow normal practice for that stack (separation of concerns, sensible naming, config vs code, reusable helpers). Aim for reviewable increments, not unreadable blobs.
-3. **Verify as you go** — Run/format/lint/tests the project affords after meaningful chunks so the tree stays truthful; bake **checks or manual steps into the goal** when they matter.
+3. **Verify as you go** — Run/format/lint/tests the project affords after meaningful chunks so the tree stays truthful; bake **checks or manual steps into the goal** when they matter. Before `complete_goal`, run the smallest reliable verification you can and summarize it in `verification_summary`.

 ## Look things up instead of guessing

@@ -5,7 +5,4 @@ Task: {{ task }}
 Result:
 {{ result }}

-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.
+Summarize this naturally for the user. Keep it brief (1-2 sentences). Do not mention technical details like "subagent" or task IDs.
@@ -4,15 +4,6 @@

 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' %}

@@ -529,9 +529,6 @@ 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:
@@ -290,7 +290,8 @@ def current_time_str(timezone: str | None = None) -> str:


 _UNSAFE_CHARS = re.compile(r'[<>:"/\\|?*]')
-_TOOL_RESULT_PREVIEW_CHARS = 1200
+_TOOL_RESULT_SUMMARY_MAX_EDGE_CHARS = 800
+_TOOL_RESULT_SUMMARY_MIN_EDGE_CHARS = 80
 _TOOL_RESULTS_DIR = ".nanobot/tool-results"
 _TOOL_RESULT_RETENTION_SECS = 7 * 24 * 60 * 60
 _TOOL_RESULT_MAX_BUCKETS = 32
@@ -404,22 +405,106 @@ def stringify_text_blocks(content: list[dict[str, Any]]) -> str | None:
    return "\n".join(parts)


-def _render_tool_result_reference(
-    filepath: Path,
+def build_structured_output_summary(
+    title: str,
+    text: str,
    *,
-    original_size: int,
-    preview: str,
-    truncated_preview: bool,
+    max_chars: int,
+    metadata: list[tuple[str, Any]] | None = None,
+    analysis: Any | None = None,
+    guidance: str | None = None,
 ) -> str:
-    result = (
-        f"[tool output persisted]\n"
-        f"Full output saved to: {filepath}\n"
-        f"Original size: {original_size} chars\n"
-        f"Preview:\n{preview}"
+    """Return a compact, structured head/tail summary for oversized tool output."""
+
+    if max_chars <= 0:
+        return text
+    edge_chars = min(
+        _TOOL_RESULT_SUMMARY_MAX_EDGE_CHARS,
+        max(_TOOL_RESULT_SUMMARY_MIN_EDGE_CHARS, max_chars // 3),
+    )
+    while True:
+        head = text[:edge_chars]
+        if len(text) > edge_chars * 2:
+            tail: str | None = text[-edge_chars:]
+            omitted_middle_chars = len(text) - len(head) - len(tail)
+        else:
+            tail = None
+            omitted_middle_chars = 0
+        result = _render_structured_output_summary(
+            title,
+            metadata=metadata or [],
+            guidance=guidance,
+            analysis=analysis,
+            head=head,
+            tail=tail,
+            omitted_middle_chars=omitted_middle_chars,
+        )
+        if len(result) <= max_chars or edge_chars <= _TOOL_RESULT_SUMMARY_MIN_EDGE_CHARS:
+            return truncate_text(result, max_chars)
+        overflow = len(result) - max_chars
+        edge_chars = max(
+            _TOOL_RESULT_SUMMARY_MIN_EDGE_CHARS,
+            edge_chars - max(overflow // 2 + 1, 16),
+        )
+
+
+def _render_structured_output_summary(
+    title: str,
+    *,
+    metadata: list[tuple[str, Any]],
+    guidance: str | None,
+    analysis: Any | None,
+    head: str,
+    tail: str | None,
+    omitted_middle_chars: int,
+) -> str:
+    lines = [title]
+    lines.extend(f"{key}: {value}" for key, value in metadata)
+    if omitted_middle_chars:
+        lines.append(f"truncation: {omitted_middle_chars:,} chars truncated from the middle")
+    if guidance:
+        lines.append(f"guidance: {guidance}")
+    lines.extend(_verification_summary_lines(analysis))
+    lines.extend(["head:", head])
+    if tail is not None:
+        lines.extend(["tail:", tail])
+    return "\n".join(lines)
+
+
+def _verification_summary_lines(analysis: Any | None) -> list[str]:
+    if analysis is None or getattr(analysis, "status", None) != "failed":
+        return []
+    lines = ["verification_status: failed"]
+    if getattr(analysis, "timed_out", False):
+        lines.append("failure_type: command timeout")
+    if getattr(analysis, "failed_tests", ()):
+        lines.append("failed_tests:")
+        lines.extend(f"- {item}" for item in analysis.failed_tests)
+    if getattr(analysis, "primary_errors", ()):
+        lines.append("primary_errors:")
+        lines.extend(f"- {item}" for item in analysis.primary_errors)
+    if getattr(analysis, "missing_artifacts", ()):
+        lines.append("missing_artifacts:")
+        lines.extend(f"- {item}" for item in analysis.missing_artifacts)
+    return lines
+
+
+def _build_tool_result_reference(filepath: Path, text: str, *, max_chars: int) -> str:
+    return build_structured_output_summary(
+        "[tool output persisted]",
+        text,
+        max_chars=max_chars,
+        metadata=[
+            ("tool_output_id", filepath.stem),
+            ("original_size_chars", len(text)),
+            ("storage", "internal audit artifact"),
+        ],
+        guidance=(
+            "Use this head/tail summary first. Avoid reading persisted "
+            "tool-output files wholesale; rerun a narrower command when "
+            "more detail is needed."
+        ),
    )
-    if truncated_preview:
-        result += "\n...\n(Read the saved file if you need the full output.)"
-    return result


 def _bucket_mtime(path: Path) -> float:
@@ -494,13 +579,7 @@ def maybe_persist_tool_result(
        else:
            _write_text_atomic(path, text_payload)

-    preview = text_payload[:_TOOL_RESULT_PREVIEW_CHARS]
-    return _render_tool_result_reference(
-        path,
-        original_size=len(text_payload),
-        preview=preview,
-        truncated_preview=len(text_payload) > _TOOL_RESULT_PREVIEW_CHARS,
-    )
+    return _build_tool_result_reference(path, text_payload, max_chars=max_chars)


 def split_message(content: str, max_len: int = 2000) -> list[str]:
@@ -42,6 +42,27 @@ SUSTAINED_GOAL_CONTINUE_PROMPT = (
    "objective using your tools, or call complete_goal if the work is truly finished."
 )

+RUNTIME_BUDGET_CONVERGENCE_PROMPT = """\
+[Runtime Budget Notice]
+You have used {used_iterations} of {max_iterations} model/tool iterations for this turn. \
+{remaining_iterations} iteration(s) remain before NanoBot must finalize without more tools.
+
+Switch to convergence mode: stop broad exploration, choose the smallest high-signal command or edit, \
+verify the likely solution, and preserve enough budget for a final answer. For coding or \
+file-producing tasks, do not mark the work complete until the smallest reliable verification passes, \
+or clearly state remaining failures.
+[/Runtime Budget Notice]"""
+
+RUNTIME_BUDGET_FINAL_PROMPT = """\
+[Runtime Budget Notice]
+Only {remaining_iterations} of {max_iterations} model/tool iteration(s) remain before NanoBot must \
+finalize without more tools.
+
+Finalize the solution path now: avoid new broad searches or builds unless essential, make the \
+smallest final fix or artifact, run one targeted verification if possible, then answer honestly with \
+the evidence or remaining failures.
+[/Runtime Budget Notice]"""
+

 def empty_tool_result_message(tool_name: str) -> str:
    """Short prompt-safe marker for tools that completed without visible output."""
@@ -88,6 +109,25 @@ def build_goal_continue_message(custom: str | None = None) -> dict[str, str]:
    return {"role": "user", "content": custom or SUSTAINED_GOAL_CONTINUE_PROMPT}


+def build_runtime_budget_notice_message(
+    *,
+    level: int,
+    max_iterations: int,
+    used_iterations: int,
+    remaining_iterations: int,
+) -> dict[str, str]:
+    """Prompt the model to converge as the generic tool-iteration budget runs low."""
+    template = RUNTIME_BUDGET_FINAL_PROMPT if level >= 2 else RUNTIME_BUDGET_CONVERGENCE_PROMPT
+    return {
+        "role": "user",
+        "content": template.format(
+            max_iterations=max_iterations,
+            used_iterations=used_iterations,
+            remaining_iterations=remaining_iterations,
+        ),
+    }
+
+
 def external_lookup_signature(tool_name: str, arguments: Any) -> str | None:
    """Stable signature for repeated external lookups we want to throttle."""
    if not isinstance(arguments, dict):
@@ -35,15 +35,10 @@ def format_tool_hints(tool_calls: list, max_length: int = 40) -> str:

    formatted = []
    for tc in tool_calls:
-        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)
+        fmt = _TOOL_FORMATS.get(tc.name)
        if fmt:
            formatted.append(_fmt_known(tc, fmt, max_length))
-        elif name.startswith("mcp_"):
+        elif tc.name.startswith("mcp_"):
            formatted.append(_fmt_mcp(tc, max_length))
        else:
            formatted.append(_fmt_fallback(tc, max_length))
@@ -26,11 +26,10 @@ from nanobot.session.manager import (
    _metadata_title,
 )

-_INDEX_VERSION = 2
+_INDEX_VERSION = 1
 _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]]:
@@ -215,45 +214,14 @@ 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": _visible_activity_updated_at(
-            session.updated_at.isoformat(),
-            visible_message_at,
-            activity_updated_at,
-        ),
+        "updated_at": _latest_updated_at(session.updated_at.isoformat(), activity_updated_at),
        "title": _metadata_title(session.metadata),
        "preview": _preview_from_messages(session.messages),
        "file": path.name,
@@ -276,37 +244,31 @@ 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)
-                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
+                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
            signature = _file_signature(path)
            created_at_s = data.get("created_at")
            updated_at_s = data.get("updated_at")
@@ -320,11 +282,7 @@ def _scan_session_row(session_manager: SessionManager, path: Path) -> dict[str,
            return {
                "key": key,
                "created_at": created_at_s,
-                "updated_at": _visible_activity_updated_at(
-                    updated_at_s,
-                    visible_message_at,
-                    activity_updated_at,
-                ),
+                "updated_at": _latest_updated_at(updated_at_s, activity_updated_at),
                "title": _metadata_title(data.get("metadata", {})),
                "preview": preview or fallback_preview,
                "file": path.name,
@@ -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, resolve_config_env_vars, save_config
+from nanobot.config.loader import get_config_path, load_config, save_config
 from nanobot.config.schema import ModelPresetConfig, ProviderConfig
 from nanobot.providers.image_generation import (
    get_image_gen_provider,
@@ -1166,19 +1166,14 @@ 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(proxy=proxy)
+            token = get_token()
        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)
@@ -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.6,<1.0.0",
+    "oauth-cli-kit>=0.1.3,<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",
@@ -269,15 +269,7 @@ if [ "${NANOBOT_SKIP_WIZARD:-}" = "1" ]; then
  exit 0
 fi

-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 "Starting setup wizard..."
+run_nanobot onboard --wizard

 info "Done. Try: $(nanobot_try_command) agent -m \"Hello!\""
@@ -38,6 +38,7 @@ 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,
@@ -63,6 +64,7 @@ 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:
@@ -77,8 +79,8 @@ def make_loop(
    if patch_deps:
        with patch("nanobot.agent.loop.ContextBuilder"), \
             patch("nanobot.agent.loop.SessionManager"), \
-             patch("nanobot.agent.loop.SubagentManager") as mock_sub_mgr:
-            mock_sub_mgr.return_value.cancel_by_session = AsyncMock(return_value=0)
+             patch("nanobot.agent.loop.SubagentManager") as MockSubMgr:
+            MockSubMgr.return_value.cancel_by_session = AsyncMock(return_value=0)
            return AgentLoop(**kwargs)
    return AgentLoop(**kwargs)

@@ -91,6 +91,7 @@ def _make_fake_compact(

        tail = list(session.messages[session.last_consolidated:])
        if not tail:
+            session.updated_at = datetime.now()
            loop.sessions.save(session)
            return ""

@@ -102,14 +103,15 @@ def _make_fake_compact(
            metadata={},
            last_consolidated=0,
        )
-        result = probe.retain_recent_legal_suffix(
+        dropped, already_consolidated = probe.retain_recent_legal_suffix(
            max_suffix,
            extend_to_user=True,
        )
        kept = probe.messages
-        archive_msgs = result.dropped[result.already_consolidated_count:]
+        archive_msgs = dropped[already_consolidated:]

        if not archive_msgs and not kept:
+            session.updated_at = datetime.now()
            loop.sessions.save(session)
            return ""

@@ -130,6 +132,7 @@ def _make_fake_compact(

        session.messages = kept
        session.last_consolidated = 0
+        session.updated_at = datetime.now()
        loop.sessions.save(session)
        return s

@@ -1018,28 +1021,27 @@ class TestProactiveAutoCompact:
        await self._run_check_expired(loop)
        assert _fake_compact.state["count"] == 1

-        # Second tick: should NOT re-schedule because the session has no removable tail.
+        # Second tick: should NOT re-schedule (updated_at is fresh after clear)
        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_session_does_not_schedule_idle_compact(self, tmp_path):
-        """Empty expired sessions have no removable tail and should not schedule."""
+    async def test_empty_skip_refreshes_updated_at_prevents_reschedule(self, tmp_path):
+        """Empty session skip refreshes updated_at, preventing immediate re-scheduling."""
        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)

-        _fake_compact = _make_fake_compact(loop)
-        loop.consolidator.compact_idle_session = _fake_compact
+        loop.consolidator.compact_idle_session = _make_fake_compact(loop)

+        # 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()

@@ -200,11 +200,8 @@ class TestCheckExpired:
        """Expired session should trigger schedule_background."""
        ac = _make_autocompact(ttl=15)
        mock_sm = MagicMock(spec=SessionManager)
-        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
+        old_ts = (datetime.now() - timedelta(minutes=20)).isoformat()
+        mock_sm.list_sessions.return_value = [{"key": "cli:old", "updated_at": old_ts}]
        ac.sessions = mock_sm

        scheduled = []
@@ -276,24 +273,6 @@ 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
@@ -430,11 +430,9 @@ 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)
@@ -447,7 +445,6 @@ 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(
@@ -521,10 +518,8 @@ class TestCompactIdleSession:
        assert entries[0]["session_key"] == "cli:test"

    @pytest.mark.asyncio
-    async def test_empty_session_does_not_refresh_timestamp(
-        self, real_consolidator
-    ):
-        """Empty session with old updated_at does not look active after compaction."""
+    async def test_empty_session_refreshes_timestamp(self, real_consolidator):
+        """Empty session with old updated_at → refreshed after call, returns ''."""
        from datetime import datetime, timedelta

        sessions = real_consolidator.sessions
@@ -537,8 +532,7 @@ class TestCompactIdleSession:
        assert result == ""

        reloaded = sessions.get_or_create("cli:empty")
-        assert reloaded.updated_at == old_ts
-        assert reloaded.metadata == {}
+        assert reloaded.updated_at > old_ts

    @pytest.mark.asyncio
    async def test_nothing_summary_not_stored(self, real_consolidator, mock_provider):
@@ -1,4 +1,4 @@
-"""Tests for the internal max_messages replay cap."""
+"""Tests for max_messages config wiring into session history replay."""

 from __future__ import annotations

@@ -11,27 +11,20 @@ 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.providers.factory import ProviderSnapshot
-from nanobot.session.manager import (
-    FILE_MAX_MESSAGES,
-    Session,
-    replay_max_messages_for_context,
-)
+from nanobot.session.manager import Session
+
+DEFAULT_MAX_MESSAGES = 120


-def _make_loop(
-    tmp_path: Path,
-    context_window_tokens: int = 200_000,
-) -> AgentLoop:
+def _make_loop(tmp_path: Path, max_messages: int = DEFAULT_MAX_MESSAGES) -> 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",
-        context_window_tokens=context_window_tokens,
+        max_messages=max_messages,
    )


@@ -58,44 +51,24 @@ def _tool_round(call_id: str) -> list[dict]:


 class TestMaxMessagesInit:
-    """Verify AgentLoop derives the internal replay cap correctly."""
+    """Verify AgentLoop stores the config value correctly."""

-    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:
+    def test_default_is_builtin_limit(self, tmp_path: Path) -> None:
        loop = _make_loop(tmp_path)
-        assert loop._max_messages == FILE_MAX_MESSAGES
+        assert loop._max_messages == DEFAULT_MAX_MESSAGES

-    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_positive_value_stored(self, tmp_path: Path) -> None:
+        loop = _make_loop(tmp_path, max_messages=25)
+        assert loop._max_messages == 25

-    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_zero_uses_builtin_limit(self, tmp_path: Path) -> None:
+        loop = _make_loop(tmp_path, max_messages=0)
+        assert loop._max_messages == DEFAULT_MAX_MESSAGES

-        assert loop._max_messages == 327
-        loop._refresh_provider_snapshot()
-        assert loop._max_messages == FILE_MAX_MESSAGES
+    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


 class TestGetHistoryWithMaxMessages:
@@ -104,7 +77,7 @@ class TestGetHistoryWithMaxMessages:
    def test_default_uses_builtin_limit(self) -> None:
        session = _populated_session(80)
        history = session.get_history()
-        assert len(history) <= FILE_MAX_MESSAGES
+        assert len(history) <= DEFAULT_MAX_MESSAGES

    def test_explicit_max_messages_limits_output(self) -> None:
        session = _populated_session(40)  # 80 messages total
@@ -120,7 +93,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) <= FILE_MAX_MESSAGES
+        assert len(history) <= DEFAULT_MAX_MESSAGES

    def test_small_session_unaffected(self) -> None:
        """When session has fewer messages than max_messages, all are returned."""
@@ -130,13 +103,12 @@ class TestGetHistoryWithMaxMessages:


 class TestMaxMessagesIntegration:
-    """Verify AgentLoop passes the replay cap into get_history calls."""
+    """Verify the config flows from AgentLoop into get_history calls."""

    @pytest.mark.asyncio
-    async def test_process_message_passes_limit_to_history_call(self, tmp_path: Path) -> None:
+    async def test_process_message_passes_config_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)
-        loop._max_messages = 25
+        loop = _make_loop(tmp_path, max_messages=25)
        loop.provider.chat_with_retry = AsyncMock(
            return_value=LLMResponse(content="ok", tool_calls=[], usage={})
        )
@@ -155,11 +127,8 @@ class TestMaxMessagesIntegration:
        assert mock_hist.call_args.kwargs["extend_to_user"] is False

    @pytest.mark.asyncio
-    async def test_default_limit_passes_context_derived_limit_to_history_call(
-        self,
-        tmp_path: Path,
-    ) -> None:
-        loop = _make_loop(tmp_path)
+    async def test_zero_config_passes_builtin_limit_to_history_call(self, tmp_path: Path) -> None:
+        loop = _make_loop(tmp_path, max_messages=0)
        loop.provider.chat_with_retry = AsyncMock(
            return_value=LLMResponse(content="ok", tool_calls=[], usage={})
        )
@@ -173,7 +142,7 @@ class TestMaxMessagesIntegration:
            )

        assert result is not None
-        assert mock_hist.call_args.kwargs["max_messages"] == FILE_MAX_MESSAGES
+        assert mock_hist.call_args.kwargs["max_messages"] == DEFAULT_MAX_MESSAGES
        assert mock_hist.call_args.kwargs["extend_to_user"] is False

    @pytest.mark.asyncio
@@ -182,8 +151,7 @@ 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)
-        loop._max_messages = 6
+        loop = _make_loop(tmp_path, max_messages=6)
        loop.provider.chat_with_retry = AsyncMock(
            return_value=LLMResponse(content="ok", tool_calls=[], usage={})
        )
@@ -214,3 +182,31 @@ 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)
@@ -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, ToolCallRequest
+from nanobot.providers.base import LLMResponse

 _MAX_TOOL_RESULT_CHARS = AgentDefaults().max_tool_result_chars

@@ -877,162 +877,3 @@ 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
@@ -48,7 +48,13 @@ async def test_runner_persists_large_tool_results_for_follow_up_calls(tmp_path):
    assert result.final_content == "done"
    tool_message = next(msg for msg in captured_second_call if msg.get("role") == "tool")
    assert "[tool output persisted]" in tool_message["content"]
-    assert "tool-results" in tool_message["content"]
+    assert "tool_output_id: call_big" in tool_message["content"]
+    assert "original_size_chars: 20000" in tool_message["content"]
+    assert "head:" in tool_message["content"]
+    assert "tail:" in tool_message["content"]
+    assert "Read the saved file" not in tool_message["content"]
+    assert str(tmp_path) not in tool_message["content"]
+    assert len(tool_message["content"]) <= 2048
    assert (tmp_path / ".nanobot" / "tool-results" / "test_runner" / "call_big.txt").exists()


@@ -76,6 +82,8 @@ def test_persist_tool_result_prunes_old_session_buckets(tmp_path):
    )

    assert "[tool output persisted]" in persisted
+    assert "tool_output_id: call_big" in persisted
+    assert "tool-results" not in persisted
    assert not old_bucket.exists()
    assert recent_bucket.exists()
    assert (root / "current_session" / "call_big.txt").exists()
@@ -358,3 +358,79 @@ async def test_runner_blocks_repeated_external_fetches():
        if msg.get("role") == "tool" and msg.get("tool_call_id") == "call_3"
    ][0]
    assert "repeated external lookup blocked" in blocked_tool_message["content"]
+
+
+@pytest.mark.asyncio
+async def test_runner_adds_budget_notice_near_long_tool_budget():
+    provider = MagicMock()
+    captured_final_call: list[dict] = []
+    call_count = {"n": 0}
+
+    async def chat_with_retry(*, messages, **kwargs):
+        call_count["n"] += 1
+        if call_count["n"] <= 16:
+            return LLMResponse(
+                content="working",
+                tool_calls=[ToolCallRequest(id=f"call_{call_count['n']}", name="work", arguments={})],
+                usage={},
+            )
+        captured_final_call[:] = messages
+        return LLMResponse(content="done", tool_calls=[], usage={})
+
+    provider.chat_with_retry = chat_with_retry
+    tools = MagicMock()
+    tools.get_definitions.return_value = []
+    tools.execute = AsyncMock(return_value="tool result")
+
+    result = await AgentRunner(provider).run(AgentRunSpec(
+        initial_messages=[{"role": "user", "content": "finish a large task"}],
+        tools=tools,
+        model="test-model",
+        max_iterations=20,
+        max_tool_result_chars=_MAX_TOOL_RESULT_CHARS,
+    ))
+
+    assert result.final_content == "done"
+    notices = [
+        msg["content"]
+        for msg in captured_final_call
+        if msg.get("role") == "user" and "[Runtime Budget Notice]" in str(msg.get("content"))
+    ]
+    assert len(notices) == 1
+    assert "15 of 20 model/tool iterations" in notices[0]
+    assert "Switch to convergence mode" in notices[0]
+    assert tools.execute.await_count == 16
+
+
+@pytest.mark.asyncio
+async def test_runner_budget_notice_does_not_affect_short_runs():
+    provider = MagicMock()
+    captured_final_call: list[dict] = []
+    call_count = {"n": 0}
+
+    async def chat_with_retry(*, messages, **kwargs):
+        call_count["n"] += 1
+        if call_count["n"] <= 2:
+            return LLMResponse(
+                content="working",
+                tool_calls=[ToolCallRequest(id=f"call_{call_count['n']}", name="work", arguments={})],
+                usage={},
+            )
+        captured_final_call[:] = messages
+        return LLMResponse(content="done", tool_calls=[], usage={})
+
+    provider.chat_with_retry = chat_with_retry
+    tools = MagicMock()
+    tools.get_definitions.return_value = []
+    tools.execute = AsyncMock(return_value="tool result")
+
+    result = await AgentRunner(provider).run(AgentRunSpec(
+        initial_messages=[{"role": "user", "content": "small task"}],
+        tools=tools,
+        model="test-model",
+        max_iterations=4,
+        max_tool_result_chars=_MAX_TOOL_RESULT_CHARS,
+    ))
+
+    assert result.final_content == "done"
+    assert all("[Runtime Budget Notice]" not in str(msg.get("content")) for msg in captured_final_call)
@@ -1,6 +1,7 @@
 """Tests for atomic session save and corrupt-file repair."""

 import json
+import shutil
 from datetime import datetime
 from pathlib import Path

@@ -36,6 +37,17 @@ class TestAtomicSave:
        tmp_files = list(mgr.sessions_dir.glob("*.tmp"))
        assert tmp_files == []

+    def test_save_recreates_deleted_sessions_dir(self, tmp_path: Path):
+        mgr = SessionManager(tmp_path)
+        shutil.rmtree(mgr.sessions_dir)
+
+        session = Session(key="test:recreate")
+        session.add_message("user", "hello")
+        mgr.save(session)
+
+        path = mgr._get_session_path("test:recreate")
+        assert path.exists()
+
    def test_tmp_file_cleaned_up_on_write_failure(self, tmp_path: Path):
        mgr = SessionManager(tmp_path)
        session = Session(key="test:fail")
@@ -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}"})

-    result = session.retain_recent_legal_suffix(4)
+    dropped, already_cons = session.retain_recent_legal_suffix(4)

-    assert len(result.dropped) == 6
-    assert [m["content"] for m in result.dropped] == [f"msg{i}" for i in range(6)]
+    assert len(dropped) == 6
+    assert [m["content"] for m in dropped] == [f"msg{i}" for i in range(6)]
    assert len(session.messages) == 4
-    assert result.already_consolidated_count == 0
+    assert already_cons == 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}"})

-    result = session.retain_recent_legal_suffix(4)
+    dropped, already_cons = session.retain_recent_legal_suffix(4)

-    assert result.dropped == []
-    assert result.already_consolidated_count == 0
+    assert dropped == []
+    assert already_cons == 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

-    result = session.retain_recent_legal_suffix(0)
+    dropped, already_cons = session.retain_recent_legal_suffix(0)

-    assert len(result.dropped) == 5
-    assert result.already_consolidated_count == 3
+    assert len(dropped) == 5
+    assert already_cons == 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

-    result = session.retain_recent_legal_suffix(4)
+    dropped, already_cons = 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 result.already_consolidated_count == 9
+    assert already_cons == 9
@@ -1,7 +1,7 @@
 """Tests for tool hint formatting (nanobot.utils.tool_hints)."""

-from nanobot.providers.base import ToolCallRequest
 from nanobot.utils.tool_hints import format_tool_hints
+from nanobot.providers.base import ToolCallRequest


 def _tc(name: str, args) -> ToolCallRequest:
@@ -306,22 +306,3 @@ 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"
@@ -0,0 +1,175 @@
+from __future__ import annotations
+
+from nanobot.agent.verification_state import (
+    analyze_verification_result,
+    append_verification_feedback,
+)
+
+
+def test_analyze_pytest_failure_extracts_actionable_summary():
+    output = """\
+FAILED ../tests/test_outputs.py::test_regex_matches_dates - AssertionError: Expected dates
+E       AssertionError: Expected ['2025-01-09'], but got ['bad']
+E       FileNotFoundError: [Errno 2] No such file or directory: '/app/out.txt'
+============================== 1 failed in 0.05s ===============================
+Exit code: 1
+"""
+
+    analysis = analyze_verification_result(
+        command="pytest /tests/test_outputs.py",
+        output=output,
+        exit_code=1,
+    )
+
+    assert analysis is not None
+    assert analysis.status == "failed"
+    assert analysis.failed_tests == ("../tests/test_outputs.py::test_regex_matches_dates",)
+    assert any("AssertionError" in item for item in analysis.primary_errors)
+    assert "/app/out.txt" in analysis.missing_artifacts
+
+
+def test_append_verification_feedback_tells_agent_not_to_finish():
+    analysis = analyze_verification_result(
+        command="python /app/test_outputs.py",
+        output="FAILED test_outputs.py::test_file\nAssertionError: missing\nExit code: 1",
+        exit_code=1,
+    )
+
+    feedback = append_verification_feedback("raw output\nExit code: 1", analysis)
+
+    assert "[Verification Feedback]" in feedback
+    assert "Do not call complete_goal" in feedback
+    assert "Next action" in feedback
+
+
+def test_analyze_passing_test_records_success_without_feedback():
+    analysis = analyze_verification_result(
+        command="pytest",
+        output="============================== 3 passed in 0.10s ==============================\nExit code: 0",
+        exit_code=0,
+    )
+
+    assert analysis is not None
+    assert analysis.status == "passed"
+    assert append_verification_feedback("ok", analysis) == "ok"
+
+
+def test_analyze_command_not_found_as_failed_check():
+    output = """\
+STDERR:
+/usr/bin/bash: line 1: python3: command not found
+
+Exit code: 127
+"""
+
+    analysis = analyze_verification_result(
+        command="python3 - <<'PY'\nprint('quick verification')\nPY",
+        output=output,
+        exit_code=127,
+    )
+
+    assert analysis is not None
+    assert analysis.status == "failed"
+    assert any("command not found" in item for item in analysis.primary_errors)
+
+
+def test_analyze_artifact_comparison_success_records_pass():
+    output = """\
+run_exit:0
+0d115b98  /app/image.ppm
+0d115b98  /tmp/orig.ppm
+cmp_exit:0
+      7      21    1024
+
+Exit code: 0
+"""
+
+    analysis = analyze_verification_result(
+        command=(
+            "cd /usr/bin && gcc -static -o /app/reversed_final /app/mystery.c -lm "
+            "&& (cd /app && ./reversed_final >/tmp/final_out 2>/tmp/final_err); "
+            "sha256sum /app/image.ppm /tmp/orig.ppm; "
+            "cmp -s /app/image.ppm /tmp/orig.ppm; echo cmp_exit:$?"
+        ),
+        output=output,
+        exit_code=0,
+    )
+
+    assert analysis is not None
+    assert analysis.status == "passed"
+    assert append_verification_feedback("ok", analysis) == "ok"
+
+
+def test_analyze_plain_checksum_without_success_marker_is_ignored():
+    analysis = analyze_verification_result(
+        command="sha256sum /app/image.ppm /tmp/orig.ppm",
+        output="0d115b98  /app/image.ppm\n0d115b98  /tmp/orig.ppm\nExit code: 0",
+        exit_code=0,
+    )
+
+    assert analysis is None
+
+
+def test_analyze_named_comparison_markers_record_pass():
+    output = """\
+ppm:0
+stderr:0
+stdout:0
+      4      26    1011
+1821 mystery.c
+
+Exit code: 0
+"""
+
+    analysis = analyze_verification_result(
+        command=(
+            "gcc -static -O2 -o reversed mystery.c -lm\n"
+            "./reversed > vrout.txt 2> vrerr.txt\n"
+            "cp image.ppm rev.ppm\n"
+            "./mystery > voout.txt 2> voerr.txt\n"
+            "cmp image.ppm rev.ppm\n"
+            "printf 'ppm:%s\\n' $?\n"
+            "cmp voerr.txt vrerr.txt\n"
+            "printf 'stderr:%s\\n' $?\n"
+            "cmp voout.txt vrout.txt\n"
+            "printf 'stdout:%s\\n' $?"
+        ),
+        output=output,
+        exit_code=0,
+    )
+
+    assert analysis is not None
+    assert analysis.status == "passed"
+
+
+def test_analyze_named_comparison_marker_failure_records_failed():
+    output = """\
+ppm:0
+stderr:1
+stdout:0
+
+Exit code: 0
+"""
+
+    analysis = analyze_verification_result(
+        command=(
+            "cmp image.ppm rev.ppm; printf 'ppm:%s\\n' $?; "
+            "cmp voerr.txt vrerr.txt; printf 'stderr:%s\\n' $?; "
+            "cmp voout.txt vrout.txt; printf 'stdout:%s\\n' $?"
+        ),
+        output=output,
+        exit_code=0,
+    )
+
+    assert analysis is not None
+    assert analysis.status == "failed"
+
+
+def test_analyze_plain_run_status_marker_without_comparison_is_ignored():
+    analysis = analyze_verification_result(
+        command="gcc -static -O2 -o reversed mystery.c -lm && ./reversed",
+        output="rc:0\nExit code: 0",
+        exit_code=0,
+    )
+
+    assert analysis is None
@@ -13,6 +13,11 @@ from nanobot.agent.tools.long_task import (
    CompleteGoalTool,
    LongTaskTool,
 )
+from nanobot.agent.verification_state import (
+    VerificationAnalysis,
+    clear_verification_observation,
+    record_verification_observation,
+)
 from nanobot.bus.queue import MessageBus
 from nanobot.bus.runtime_events import RuntimeEventBus
 from nanobot.session.goal_state import GOAL_STATE_KEY
@@ -192,6 +197,66 @@ async def test_complete_goal_without_active_is_noop_message(tmp_path):
    assert "No active" in out


+@pytest.mark.asyncio
+async def test_complete_goal_blocks_unresolved_verification_failure(tmp_path):
+    sm = SessionManager(tmp_path)
+    lt, cg = _tools(sm)
+    await lt.execute(goal="Fix the tests")
+    record_verification_observation(
+        "websocket:c1",
+        VerificationAnalysis(
+            status="failed",
+            command="pytest /tests/test_outputs.py",
+            exit_code=1,
+            failed_tests=("test_outputs.py::test_output",),
+            primary_errors=("AssertionError: wrong output",),
+        ),
+    )
+
+    out = await cg.execute(recap="Done.")
+
+    assert "not marked complete" in out
+    assert "test_outputs.py::test_output" in out
+    assert sm.get_or_create("websocket:c1").metadata[GOAL_STATE_KEY]["status"] == "active"
+    clear_verification_observation("websocket:c1")
+
+
+@pytest.mark.asyncio
+async def test_complete_goal_allows_after_later_successful_verification(tmp_path):
+    sm = SessionManager(tmp_path)
+    lt, cg = _tools(sm)
+    await lt.execute(goal="Fix the tests")
+    record_verification_observation(
+        "websocket:c1",
+        VerificationAnalysis(
+            status="failed",
+            command="pytest /tests/test_outputs.py",
+            exit_code=1,
+            failed_tests=("test_outputs.py::test_output",),
+        ),
+    )
+    record_verification_observation(
+        "websocket:c1",
+        VerificationAnalysis(
+            status="passed",
+            command="pytest /tests/test_outputs.py",
+            exit_code=0,
+        ),
+    )
+
+    out = await cg.execute(
+        recap="Done.",
+        verification_summary="pytest /tests/test_outputs.py passed",
+        commands_run="pytest /tests/test_outputs.py",
+        artifacts_created="/app/out.txt",
+    )
+
+    assert "marked complete" in out
+    blob = sm.get_or_create("websocket:c1").metadata[GOAL_STATE_KEY]
+    assert blob["status"] == "completed"
+    assert blob["verification_summary"] == "pytest /tests/test_outputs.py passed"
+
+
@pytest.mark.asyncio
 async def test_long_task_skips_ws_publish_without_bus(tmp_path):
    sm = SessionManager(tmp_path)
@@ -236,12 +236,10 @@ class TestModifyRestricted:

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

    @pytest.mark.asyncio
    async def test_modify_none_value_for_restricted_int(self):
@@ -7,7 +7,6 @@ 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
@@ -483,49 +482,3 @@ 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"
@@ -1,59 +0,0 @@
-"""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
@@ -1764,44 +1764,6 @@ 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()
@@ -1,8 +1,6 @@
 from __future__ import annotations

 import asyncio
-import sys
-import types
 from types import SimpleNamespace
 from unittest.mock import AsyncMock, MagicMock

@@ -88,23 +86,6 @@ 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 = {}
@@ -320,60 +301,6 @@ 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()
@@ -5,7 +5,6 @@ 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
@@ -20,7 +19,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, provider_signature
+from nanobot.providers.factory import ProviderSnapshot, make_provider
 from nanobot.providers.openai_codex_provider import _strip_model_prefix
 from nanobot.providers.registry import find_by_name
 from nanobot.webui.metadata import (
@@ -435,154 +434,6 @@ 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"
@@ -834,54 +685,6 @@ 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

@@ -949,7 +752,7 @@ def test_make_provider_passes_extra_headers_to_custom_provider():
                        "x-session-affinity": "sticky-session",
                    },
                }
-            }
+            },
        }
    )

@@ -4,7 +4,6 @@ from __future__ import annotations

 import asyncio
 import os
-import sys
 import time
 from types import SimpleNamespace
 from unittest.mock import AsyncMock, MagicMock, patch
@@ -45,8 +44,7 @@ class TestRestartCommand:
            RESTART_STARTED_AT_ENV,
        )

-        loop, _bus = _make_loop()
-        loop.restart_mode = "exec"
+        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)

@@ -78,75 +76,10 @@ 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:
@@ -1,81 +0,0 @@
-"""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
@@ -2,8 +2,6 @@ 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

@@ -95,41 +93,6 @@ 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

@@ -283,3 +246,16 @@ def test_load_config_accepts_legacy_local_preview_access(tmp_path) -> None:
    config = load_config(config_path)

    assert config.tools.webui_allow_local_service_access is False
+
+
+def test_load_config_accepts_exec_local_service_access(tmp_path) -> None:
+    config_path = tmp_path / "config.json"
+    config_path.write_text(
+        json.dumps({"tools": {"exec": {"allowLocalServiceAccess": True}}}),
+        encoding="utf-8",
+    )
+
+    config = load_config(config_path)
+
+    assert config.tools.exec.allow_local_service_access is True
+    assert not hasattr(config.tools, "allow_local_service_access")
@@ -8,7 +8,6 @@ from nanobot.config.loader import (
    resolve_config_env_vars,
    save_config,
 )
-from nanobot.config.schema import Config


 class TestResolveEnvVars:
@@ -128,31 +127,6 @@ 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
@@ -1,15 +0,0 @@
-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")
@@ -10,12 +10,11 @@ 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 CronJob, CronPayload, CronSchedule
+from nanobot.cron.types import CronSchedule


 def _seeded_store(tmp_path: Path) -> tuple[CronService, Path]:
@@ -42,29 +41,6 @@ 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."""
@@ -172,126 +148,6 @@ 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."""
@@ -1,143 +0,0 @@
-"""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"
@@ -21,12 +21,9 @@ 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",
-        fake_token,
+        lambda: SimpleNamespace(account_id="acct", access="token"),
    )


@@ -80,12 +77,7 @@ async def test_codex_request_non_200_populates_http_metadata(monkeypatch) -> Non
            request=request,
        )

-    def fake_client(
-        *,
-        timeout: int,
-        verify: bool,
-        **_kwargs: object,
-    ) -> httpx.AsyncClient:
+    def fake_client(*, timeout: int, verify: bool) -> httpx.AsyncClient:
        assert timeout == 90
        assert verify is True
        return original_client(transport=httpx.MockTransport(handler), timeout=timeout)
@@ -114,12 +106,7 @@ 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,
-        **_kwargs: object,
-    ) -> httpx.AsyncClient:
+    def fake_client(*, timeout: int, verify: bool) -> httpx.AsyncClient:
        seen["timeout"] = timeout
        return original_client(transport=httpx.MockTransport(handler), timeout=timeout)

@@ -130,39 +117,6 @@ 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] = []
@@ -174,12 +128,11 @@ 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,
    ):
-        _ = proxy, on_thinking_delta, on_tool_call_delta
+        _ = on_thinking_delta, on_tool_call_delta
        bodies.append(body)
        return "ok", [], "stop", {}, None

@@ -233,40 +186,6 @@ 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)
@@ -384,6 +303,37 @@ async def test_codex_http_error_preserves_status_and_retry_after(monkeypatch) ->
    assert response.error_should_retry is True


+def test_codex_response_failed_server_error_is_retryable() -> None:
+    response = _codex_error_response(
+        RuntimeError(
+            "Response failed: {'type': 'server_error', 'code': 'server_error', "
+            "'message': 'The server had an error while processing your request.'}"
+        )
+    )
+
+    assert response.finish_reason == "error"
+    assert response.error_kind == "provider"
+    assert response.error_type == "server_error"
+    assert response.error_code == "server_error"
+    assert response.error_should_retry is True
+    assert provider_base.LLMProvider._is_transient_response(response) is True
+
+
+def test_codex_response_failed_cyber_policy_is_not_retryable() -> None:
+    response = _codex_error_response(
+        RuntimeError(
+            "Response failed: {'type': 'invalid_request_error', 'code': 'cyber_policy', "
+            "'message': 'Request denied.'}"
+        )
+    )
+
+    assert response.error_kind == "provider"
+    assert response.error_type == "invalid_request_error"
+    assert response.error_code == "cyber_policy"
+    assert response.error_should_retry is False
+    assert provider_base.LLMProvider._is_transient_response(response) is False
+
+
@pytest.mark.asyncio
 async def test_codex_http_diagnostic_log_omits_raw_body(monkeypatch) -> None:
    log_capture = _capture_codex_warnings(monkeypatch)
@@ -490,12 +440,9 @@ 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",
-        fake_token,
+        lambda: SimpleNamespace(account_id="acct", access="token"),
    )

    async def fake_request(
@@ -503,12 +450,11 @@ 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, proxy, on_tool_call_delta
+        _ = url, headers, verify, on_tool_call_delta
        assert body["reasoning"] == {"summary": "auto", "effort": "medium"}
        if on_content_delta:
            await on_content_delta("answer")
@@ -4,7 +4,6 @@ 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


@@ -55,32 +54,3 @@ 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
@@ -1,40 +0,0 @@
-"""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"
@@ -9,7 +9,11 @@ from unittest.mock import patch
 import pytest

 from nanobot.agent.tools.shell import ExecTool
-from nanobot.security.workspace_access import bind_workspace_scope, build_workspace_scope, reset_workspace_scope
+from nanobot.security.workspace_access import (
+    bind_workspace_scope,
+    build_workspace_scope,
+    reset_workspace_scope,
+)


 def _fake_resolve_private(hostname, port, family=0, type_=0):
@@ -68,6 +72,21 @@ def test_exec_core_full_workspace_scope_blocks_loopback(tmp_path):
    assert "internal/private" in error


+def test_exec_explicit_local_service_access_allows_loopback(tmp_path):
+    tool = ExecTool(working_dir=str(tmp_path), allow_local_service_access=True)
+    with patch("nanobot.security.network.socket.getaddrinfo", _fake_resolve_localhost):
+        error = tool._guard_command("curl http://localhost:8765/", str(tmp_path))
+    assert error is None
+
+
+def test_exec_explicit_local_service_access_still_blocks_metadata(tmp_path):
+    tool = ExecTool(working_dir=str(tmp_path), allow_local_service_access=True)
+    with patch("nanobot.security.network.socket.getaddrinfo", _fake_resolve_private):
+        error = tool._guard_command("curl http://169.254.169.254/latest/meta-data/", str(tmp_path))
+    assert error is not None
+    assert "internal/private" in error
+
+
 def test_exec_full_workspace_scope_blocks_loopback_when_local_service_disabled(tmp_path):
    tool = ExecTool(working_dir=str(tmp_path), webui_allow_local_service_access=False)
    scope = build_workspace_scope(tmp_path, "full", source_channel="websocket")
@@ -104,6 +104,84 @@ def test_exec_one_shot_accepts_max_output_tokens_alias(tmp_path):
    assert "Exit code: 0" in result


+def test_exec_detach_starts_background_process(tmp_path):
+    async def run() -> str:
+        tool = ExecTool(working_dir=str(tmp_path), timeout=5)
+        ready_path = tmp_path / "ready.txt"
+        command = _python_command(
+            "import pathlib, time; "
+            "pathlib.Path('ready.txt').write_text('ok'); "
+            "time.sleep(0.6)"
+        )
+        result = await tool.execute(command=command, detach=True)
+        for _ in range(20):
+            if ready_path.exists():
+                break
+            await asyncio.sleep(0.05)
+        return result
+
+    result = asyncio.run(run())
+
+    assert "Detached process started." in result
+    assert "pid:" in result
+    assert "log:" in result
+    assert (tmp_path / "ready.txt").read_text() == "ok"
+
+
+def test_exec_detach_reports_immediate_exit(tmp_path):
+    async def run() -> str:
+        tool = ExecTool(working_dir=str(tmp_path), timeout=5)
+        command = _python_command("print('boom'); raise SystemExit(7)")
+        return await tool.execute(command=command, detach=True)
+
+    result = asyncio.run(run())
+
+    assert "Detached process exited immediately with code 7" in result
+    assert "boom" in result
+
+
+def test_exec_long_output_summary_includes_failure_signals(tmp_path):
+    async def run() -> str:
+        tool = ExecTool(working_dir=str(tmp_path), timeout=5)
+        command = _python_command(
+            "print('A' * 3000); "
+            "print('FAILED ../tests/test_outputs.py::test_artifact - AssertionError: missing output'); "
+            "print(\"FileNotFoundError: [Errno 2] No such file or directory: '/app/out.txt'\"); "
+            "print('B' * 3000); "
+            "raise SystemExit(1)"
+        )
+        return await tool.execute(command=command, max_output_tokens=2500)
+
+    result = asyncio.run(run())
+
+    assert "[tool output truncated]" in result
+    assert "chars truncated" in result
+    assert "failed_tests:" in result
+    assert "../tests/test_outputs.py::test_artifact" in result
+    assert "missing_artifacts:" in result
+    assert "/app/out.txt" in result
+    assert "head:" in result
+    assert "tail:" in result
+    assert "[Verification Feedback]" in result
+
+
+def test_exec_adds_verification_feedback_for_test_failures(tmp_path):
+    async def run() -> str:
+        tool = ExecTool(working_dir=str(tmp_path), timeout=5)
+        command = _python_command(
+            "print('FAILED test_outputs.py::test_answer - AssertionError: wrong'); "
+            "print('AssertionError: wrong'); raise SystemExit(1)"
+        )
+        return await tool.execute(command=command)
+
+    result = asyncio.run(run())
+
+    assert "Exit code: 1" in result
+    assert "[Verification Feedback]" in result
+    assert "Do not call complete_goal" in result
+    assert "test_outputs.py::test_answer" in result
+
+
 def test_exec_accepts_supported_shell_parameter(tmp_path):
    async def run() -> str:
        tool = ExecTool(working_dir=str(tmp_path), timeout=5)
@@ -235,6 +313,35 @@ def test_write_stdin_accepts_max_output_tokens_alias(tmp_path):
    assert "Session terminated." in cleanup


+def test_write_stdin_long_output_summary_includes_failure_signals(tmp_path):
+    async def run() -> str:
+        manager = ExecSessionManager()
+        exec_tool = ExecTool(working_dir=str(tmp_path), timeout=5, session_manager=manager)
+        command = _python_command(
+            "print('A' * 3000); "
+            "print('FAILED test_outputs.py::test_file - AssertionError: bad'); "
+            "print(\"FileNotFoundError: [Errno 2] No such file or directory: '/app/missing.txt'\"); "
+            "print('B' * 3000); "
+            "raise SystemExit(1)"
+        )
+        return await exec_tool.execute(
+            command=command,
+            yield_time_ms=1000,
+            max_output_tokens=2500,
+        )
+
+    result = asyncio.run(run())
+
+    assert "[tool output truncated]" in result
+    assert "chars truncated" in result
+    assert "failed_tests:" in result
+    assert "test_outputs.py::test_file" in result
+    assert "missing_artifacts:" in result
+    assert "/app/missing.txt" in result
+    assert "Exit code: 1" in result
+    assert "[Verification Feedback]" in result
+
+
 def test_write_stdin_preserves_completed_session_output_until_polled(tmp_path):
    async def run() -> tuple[str, str]:
        manager = ExecSessionManager()
@@ -1,10 +1,8 @@
 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
@@ -38,12 +36,6 @@ 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}
@@ -58,7 +50,6 @@ def _fake_mcp_module(
        TextContent=_FakeTextContent,
        TextResourceContents=_FakeTextResourceContents,
        BlobResourceContents=_FakeBlobResourceContents,
-        ImageContent=_FakeImageContent,
    )

    class _FakeStdioServerParameters:
@@ -304,60 +295,6 @@ 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:
@@ -1240,18 +1177,3 @@ 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
@@ -57,37 +57,6 @@ 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"
@@ -660,10 +660,12 @@ async def test_exec_head_tail_truncation(tmp_path) -> None:
    else:
        command = f"{shlex.quote(sys.executable)} {shlex.quote(str(script_file))}"
    result = await tool.execute(command=command)
+    assert "[tool output truncated]" in result
    assert "chars truncated" in result
-    # Head portion should start with As
-    assert result.startswith("A")
-    # Tail portion should end with the exit code which comes after Bs
+    assert "head:" in result
+    assert "tail:" in result
+    assert "A" * 80 in result
+    assert "B" * 80 in result
    assert "Exit code:" in result


@@ -412,41 +412,6 @@ 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] = []

@@ -101,7 +101,6 @@ 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)

@@ -109,7 +108,6 @@ 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)

@@ -143,7 +141,6 @@ 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)

@@ -172,32 +169,6 @@ 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)

@@ -1,7 +1,6 @@
 from __future__ import annotations

 import json
-from types import SimpleNamespace

 import httpx
 import pytest
@@ -13,7 +12,6 @@ from nanobot.webui.settings_api import (
    WebUISettingsError,
    _oauth_provider_status,
    create_model_configuration,
-    login_oauth_provider,
    provider_models_payload,
    settings_payload,
    settings_usage_payload,
@@ -846,39 +844,6 @@ 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,
@@ -1,4 +1,11 @@
-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";
@@ -22,7 +29,6 @@ interface MeasuredPrompt extends PromptAnchor {
 }

 interface PromptMarker {
-  answerPreview: string;
  count: number;
  ids: string[];
  label: string;
@@ -37,11 +43,10 @@ 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 = 9;
-const MARKER_STACK_GAP_PX = 16;
-const RAIL_FALLBACK_HEIGHT_PX = 300;
+const MARKER_BASE_WIDTH_PX = 16;
+const MARKER_MAX_WIDTH_PX = 28;
 const MEASURE_RETRY_FRAMES = 4;
-const HOVER_MARKER_WIDTHS_PX = [28, 22, 16, 11];
+const RAIL_REVEAL_MS = 1400;

 export function PromptRail({
  bottomOffset,
@@ -52,12 +57,22 @@ export function PromptRail({
  const promptAnchors = useMemo(() => userPromptAnchors(messages), [messages]);
  const [markers, setMarkers] = useState<PromptMarker[]>([]);
  const [activePromptId, setActivePromptId] = useState<string | null>(null);
-  const [focusedMarkerIndex, setFocusedMarkerIndex] = useState<number | 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 updateMarkers = useCallback(() => {
    const scrollEl = scrollRef.current;
-    const nextRailHeight = railRef.current?.clientHeight ?? 0;
-
    if (!scrollEl || promptAnchors.length < MIN_PROMPTS_FOR_RAIL) {
      setMarkers([]);
      setActivePromptId(null);
@@ -72,8 +87,7 @@ export function PromptRail({
    }

    const measured = measurePrompts(scrollEl, promptAnchors, scrollRange);
-    const grouped = groupPromptMarkers(measured, nextRailHeight);
-    setMarkers(distributeMarkerPositions(grouped, nextRailHeight));
+    setMarkers(groupPromptMarkers(measured, railRef.current?.clientHeight ?? 0));
    setActivePromptId(activePromptForScroll(measured, scrollEl.scrollTop));
  }, [promptAnchors, scrollRef]);

@@ -98,6 +112,7 @@ export function PromptRail({
    let frame = 0;
    const schedule = () => {
      window.cancelAnimationFrame(frame);
+      revealTemporarily();
      frame = window.requestAnimationFrame(updateMarkers);
    };

@@ -108,7 +123,7 @@ export function PromptRail({
      scrollEl.removeEventListener("scroll", schedule);
      window.removeEventListener("resize", schedule);
    };
-  }, [scrollRef, updateMarkers]);
+  }, [revealTemporarily, scrollRef, updateMarkers]);

  useEffect(() => {
    const scrollEl = scrollRef.current;
@@ -119,72 +134,77 @@ 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 left-7 top-3 z-20 hidden w-9 opacity-100 md:block",
-        "transition-opacity duration-200",
+        "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",
        "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 hoverDistance =
-          focusedMarkerIndex === null ? null : Math.abs(index - focusedMarkerIndex);
+        const nearActive = activeMarkerIndex < 0 || Math.abs(index - activeMarkerIndex) <= 1;
        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 left-0 h-4 w-9 -translate-y-1/2 overflow-visible rounded-sm",
+              "group/marker absolute right-0 h-5 -translate-y-1/2 overflow-visible rounded-full",
              "focus-visible:outline-none focus-visible:ring-2 focus-visible:ring-blue-400/60",
            )}
-            style={{ top: `${marker.topPercent}%` }}
+            style={{
+              top: `${marker.topPercent}%`,
+              width: markerWidth(marker.count, maxMarkerCount, active),
+            }}
          >
            <span
              aria-hidden
-              data-testid="prompt-rail-marker"
              className={cn(
-                "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),
+                "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",
              )}
-              style={{
-                height: markerHeight(hoverDistance),
-                width: markerWidth(hoverDistance),
-              }}
            />
            <span
              aria-hidden
              className={cn(
-                "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",
+                "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",
              )}
            >
-              <span className="line-clamp-2 whitespace-pre-wrap break-words text-[15px] font-semibold leading-6">
+              <span className="block max-h-24 overflow-hidden whitespace-pre-wrap break-words">
                {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>
        );
@@ -230,12 +250,10 @@ function groupPromptMarkers(
      last.count += 1;
      last.ids.push(prompt.id);
      last.label = groupedPromptLabel(last.count, prompt.label);
-      last.answerPreview = prompt.answerPreview;
-      last.preview = prompt.preview;
+      last.preview = groupedPromptPreview(last.count, prompt.preview);
      continue;
    }
    groups.push({
-      answerPreview: prompt.answerPreview,
      count: 1,
      ids: [prompt.id],
      label: prompt.label,
@@ -280,30 +298,14 @@ function bucketPromptMarkers(
      label: bucket.length === 1
        ? latest.label
        : groupedPromptLabel(bucket.length, latest.label),
-      answerPreview: latest.answerPreview,
-      preview: latest.preview,
+      preview: bucket.length === 1
+        ? latest.preview
+        : groupedPromptPreview(bucket.length, 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,
@@ -325,26 +327,16 @@ function groupedPromptLabel(count: number, latestLabel: string): string {
  return `${count} prompts, latest: ${latestLabel}`;
 }

-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 groupedPromptPreview(count: number, latestPreview: string): string {
+  return `${count} prompts\n\n${latestPreview}`;
 }

-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 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 clamp(value: number, min: number, max: number): number {
@@ -1,7 +1,6 @@
 import type { UIMessage } from "@/lib/types";

 export interface PromptAnchor {
-  answerPreview: string;
  id: string;
  label: string;
  preview: string;
@@ -11,10 +10,9 @@ export interface PromptAnchor {

 export function userPromptAnchors(messages: UIMessage[]): PromptAnchor[] {
  let index = 0;
-  return messages.flatMap((message, messageIndex) => {
+  return messages.flatMap((message) => {
    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),
@@ -29,34 +27,13 @@ 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 truncatePreview(text, 80);
+  return text.length > 80 ? `${text.slice(0, 77)}...` : text;
 }

 export function promptPreview(content: string, index: number): string {
-  const text = compactPreview(content);
+  const text = content.replace(/\n{3,}/g, "\n\n").trim();
  if (!text) return `Prompt ${index + 1}`;
-  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;
+  return text.length > 320 ? `${text.slice(0, 317)}...` : text;
 }

 export function jumpToPrompt(scrollEl: HTMLElement | null, promptId: string | undefined): void {
@@ -425,13 +425,6 @@ 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;
@@ -636,7 +629,6 @@ export class NanobotClient {
  }

  private scheduleReconnect(): void {
-    this.clearRunStatusesForReconnect();
    this.setStatus("reconnecting");
    const attempt = this.reconnectAttempts++;
    // Exponential backoff: 0.5s, 1s, 2s, 4s, capped.
@@ -188,32 +188,6 @@ 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",
@@ -91,23 +91,6 @@ 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 (
@@ -621,7 +604,7 @@ describe("ThreadViewport", () => {
      screen.queryByText(`message ${firstVisible - 1}`),
    ).not.toBeInTheDocument();
    expect(screen.getByText(`message ${firstVisible}`)).toBeInTheDocument();
-    expect(screen.getAllByText("message 299").length).toBeGreaterThan(0);
+    expect(screen.getByText("message 299")).toBeInTheDocument();
  });

  it("automatically requests older transcript pages near the top", () => {
@@ -652,7 +635,7 @@ describe("ThreadViewport", () => {
  });

  it("renders a prompt rail that jumps to user messages", async () => {
-    const promptMessages = makePromptExchangeMessages(5);
+    const promptMessages = makeLongMessages(5);
    const { container } = render(
      <ThreadViewport
        messages={promptMessages}
@@ -687,31 +670,9 @@ 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);
Author	SHA1	Message	Date
Ubuntu	de9c5f03ae	fix(agent): make detached exec portable on Windows	2026-06-28 11:05:10 +00:00
Ubuntu	1cd3431639	feat(agent): add runtime budget convergence notices	2026-06-28 05:11:04 +00:00
Ubuntu	c62d0d5fa7	feat(agent): support detached exec services	2026-06-28 05:09:39 +00:00
Ubuntu	9474498e3e	fix(agent): flag missing verification commands	2026-06-28 05:09:39 +00:00
Ubuntu	abf930a381	feat(agent): structure long tool output summaries	2026-06-28 05:09:39 +00:00
Ubuntu	c2b1453b2e	fix(agent): harden persisted output handling	2026-06-28 05:09:39 +00:00
Ubuntu	3fccd9ab9a	Revert "docs(agent): capture eval reliability lessons" This reverts commit f56e0bea2ee0fa3dc533be1b4d4b33a5f302de10.	2026-06-28 05:09:39 +00:00
Ubuntu	12610138af	docs(agent): capture eval reliability lessons	2026-06-28 05:09:39 +00:00
Ubuntu	5b9eba4318	feat(agent): add verification gates and provider recovery	2026-06-28 05:09:39 +00:00