refactor(providers): declare Responses capabilities

.github/workflows/ci.yml

@@ -173,7 +173,7 @@ jobs:
 
       - name: Test WebUI
         working-directory: webui
-        run: bun run test
+        run: bun run test:coverage
 
       - name: Build WebUI
         working-directory: webui

docs/guides/configure-langfuse-observability.md

@@ -27,7 +27,7 @@ nanobot agent -m "Hello!"
 Install Langfuse:
 
 ```bash
-python -m pip install langfuse
+nanobot plugins enable langfuse
 ```
 
 ## Minimal working example

docs/provider-cookbook.md

@@ -549,7 +549,7 @@ This recipe applies after the agent works and you want observability for OpenAI-
 Install the optional package in the same Python environment that runs nanobot:
 
 ```bash
-python -m pip install langfuse
+nanobot plugins enable langfuse
 ```
 
 Set the environment variables before starting nanobot:

docs/webui.md

@@ -288,6 +288,13 @@ The gateway refuses to start with `host` set to `"0.0.0.0"` unless `token` or
 `http://<your-ip>:8765` from the other device and enter the secret in the login
 form.
 
+Plain HTTP is enough for basic WebUI access, but browsers expose microphone
+capture only in secure contexts. Voice input works on same-machine localhost;
+from another device, serve the WebUI over HTTPS with a certificate that device
+trusts. Configure [`sslCertfile` and `sslKeyfile`](./websocket.md#tlsssl) on the
+WebSocket channel and open `https://<your-host>:8765`, or terminate HTTPS at a
+reverse proxy and use that proxy's HTTPS URL.
+
 Remote WebUI clients with a valid token can view and use Apps. Actions that
 install missing nanobot support packages, such as adding a channel dependency,
 are blocked by default. To let trusted remote administrators change the Python
@@ -322,6 +329,10 @@ If the page does not open, check these in order:
 4. You are opening port `8765`, not the gateway health port.
 5. LAN access uses `host: "0.0.0.0"` and a token or token issue secret.
 
+If voice input asks for a secure connection, use HTTPS with a certificate the
+device trusts. Browsers do not expose microphone capture to
+`http://<your-ip>` origins.
+
 For detailed diagnostics, see
 [`troubleshooting.md#webui-problems`](./troubleshooting.md#webui-problems).
 For frontend development, see [`../webui/README.md`](../webui/README.md).

nanobot/agent/autocompact.py

@@ -8,7 +8,7 @@ from typing import TYPE_CHECKING, Any, Callable, Coroutine, cast
 
 from loguru import logger
 
-from nanobot.session.manager import Session, SessionManager
+from nanobot.session.manager import MIN_COMPACTED_REPLAY_MESSAGES, Session, SessionManager
 
 if TYPE_CHECKING:
     from nanobot.agent.memory import Consolidator
@@ -16,7 +16,7 @@ if TYPE_CHECKING:
 
 
 class AutoCompact:
-    _RECENT_SUFFIX_MESSAGES = 8
+    _RECENT_SUFFIX_MESSAGES = MIN_COMPACTED_REPLAY_MESSAGES
     _INTERNAL_SESSION_PREFIXES = ("dream:",)
 
     def __init__(self, sessions: SessionManager, consolidator: Consolidator,
@@ -45,25 +45,9 @@ class AutoCompact:
             return False
         return idle_seconds >= self._ttl * 60
 
-    def _has_compactable_idle_tail(self, key: str) -> bool:
+    def _has_unarchived_messages(self, key: str) -> bool:
         session = self.sessions.get_or_create(key)
-        tail = list(session.messages[session.last_consolidated:])
+        return session.last_consolidated < len(session.messages)
-        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:
@@ -88,7 +72,7 @@ class AutoCompact:
             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(updated_at, now) and self._has_unarchived_messages(key):
                 session = self.sessions.get_or_create(key)
                 try:
                     runtime = resolve_runtime(session)

nanobot/agent/automation_turns.py

@@ -140,10 +140,3 @@ class AutomationTurnCoordinator:
             if pending_id:
                 pending_ids.add(pending_id)
         return pending_ids
-
-    async def publish_next_deferred(self, session_key: str) -> bool:
-        return await publish_next_deferred_turn(
-            deferred_queues=self.deferred_queues,
-            publish_inbound=self._publish_inbound,
-            session_key=session_key,
-        )

nanobot/agent/context.py

@@ -13,7 +13,11 @@ from nanobot.agent.tools import mcp as mcp_tools
 from nanobot.agent.tools import sessions as session_tools
 from nanobot.agent.tools.registry import ToolRegistry
 from nanobot.apps.cli import utils as cli_app_utils
-from nanobot.bus.events import InboundMessage
+from nanobot.bus.events import (
+    INBOUND_META_RUNTIME_CONTROL,
+    RUNTIME_CONTROL_SESSION_DISCARD,
+    InboundMessage,
+)
 from nanobot.runtime_context import (
     RUNTIME_CONTEXT_END,
     RUNTIME_CONTEXT_MESSAGE_META,
@@ -47,6 +51,9 @@ async def close_mcp(state: Any) -> None:
 
 
 async def handle_runtime_control(state: Any, msg: InboundMessage, tools: ToolRegistry) -> bool:
+    if msg.metadata.get(INBOUND_META_RUNTIME_CONTROL) == RUNTIME_CONTROL_SESSION_DISCARD:
+        await state.discard_session(msg.session_key)
+        return True
     for handler in (
         image_generation_tools.handle_runtime_control,
         mcp_tools.handle_runtime_control,
@@ -79,6 +86,7 @@ class ContextBuilder:
         channel: str | None = None,
         session_summary: str | None = None,
         workspace: Path | None = None,
+        include_memory: bool = True,
         include_memory_recent_history: bool = True,
         session_key: str | None = None,
         unified_session: bool = False,
@@ -93,6 +101,7 @@ class ContextBuilder:
 
         parts.append(render_template("agent/tool_contract.md"))
 
+        if include_memory:
             memory = self.memory.read_memory()
             if memory and not self._is_template_content(memory, "memory/MEMORY.md"):
                 parts.append(f"# Memory\n\n## Long-term Memory\n{memory}")
@@ -219,6 +228,7 @@ class ContextBuilder:
         session_summary: str | None = None,
         runtime_context_blocks: Sequence[RuntimeContextBlock] | None = None,
         workspace: Path | None = None,
+        include_memory: bool = True,
         include_memory_recent_history: bool = True,
         session_key: str | None = None,
         unified_session: bool = False,
@@ -238,6 +248,7 @@ class ContextBuilder:
                     channel=channel,
                     session_summary=session_summary,
                     workspace=root,
+                    include_memory=include_memory,
                     include_memory_recent_history=include_memory_recent_history,
                     session_key=session_key,
                     unified_session=unified_session,

nanobot/agent/loop.py

@@ -398,6 +398,7 @@ class AgentLoop:
         self._mcp_connecting = False
         self._runtime_context_providers: list[RuntimeContextProvider] = []
         self._active_tasks: dict[str, set[asyncio.Task[Any]]] = {}
+        self._discarding_sessions: set[str] = set()
         self._background_tasks: set[asyncio.Task[Any]] = set()
         self._close_mcp_lock = asyncio.Lock()
         self._session_locks: weakref.WeakValueDictionary[str, asyncio.Lock] = (
@@ -721,6 +722,7 @@ class AgentLoop:
             session_summary=ctx.pending_summary,
             workspace=scope.project_path,
             runtime_context_blocks=ctx.runtime_context_blocks,
+            include_memory=ctx.session.policy.persist,
             include_memory_recent_history=not ctx.ephemeral,
             session_key=ctx.session.key,
             unified_session=self._unified_session,
@@ -786,9 +788,9 @@ class AgentLoop:
             logger.warning("Command '{}' matched but dispatch returned None", raw)
 
     async def _cancel_active_tasks(self, key: str) -> int:
-        """Cancel and await all active tasks and subagents for *key*.
+        """Cancel and await all active work for *key*.
 
-        Returns the total number of cancelled tasks + subagents.
+        Returns the total number of cancelled tasks, subagents, and exec sessions.
         """
         tasks = tuple(self._active_tasks.pop(key, set()))
         cancelled = sum(1 for t in tasks if not t.done() and t.cancel())
@@ -796,7 +798,17 @@ class AgentLoop:
             with suppress(asyncio.CancelledError, Exception):
                 await t
         sub_cancelled = await self.subagents.cancel_by_session(key)
-        return cancelled + sub_cancelled
+        exec_cancelled = await self._exec_session_manager.terminate_by_owner(key)
+        return cancelled + sub_cancelled + exec_cancelled
+
+    async def discard_session(self, key: str) -> None:
+        """Stop active work for *key* and forget its cached session."""
+        self._discarding_sessions.add(key)
+        try:
+            self.sessions.invalidate(key)
+            await self._cancel_active_tasks(key)
+        finally:
+            self._discarding_sessions.discard(key)
 
     def _effective_session_key(self, msg: InboundMessage) -> str:
         """Return the session key used for task routing and mid-turn injections."""
@@ -1161,6 +1173,11 @@ class AgentLoop:
                 effective_key = self._effective_session_key(msg)
                 if await agent_context.handle_runtime_control(self, msg, self.tools):
                     continue
+                if (
+                    msg.require_existing_session
+                    and self.sessions.get_cached(effective_key) is None
+                ):
+                    continue
                 if self.commands.is_priority(raw):
                     await self._dispatch_command_inline(
                         msg, effective_key, raw,
@@ -1279,6 +1296,8 @@ class AgentLoop:
                     # _emit_checkpoint during tool execution; materializing
                     # it into session history now makes it visible in the
                     # next conversation turn.
+                    if session_key in self._discarding_sessions:
+                        raise
                     try:
                         key = self._effective_session_key(msg)
                         session = self.sessions.get_or_create(key)
@@ -1556,6 +1575,7 @@ class AgentLoop:
         had_injections: bool,
         streamed_content: bool,
         *,
+        log_content: bool = True,
         turn_latency_ms: int | None = None,
     ) -> OutboundMessage | None:
         """Assemble the final outbound message from turn results."""
@@ -1564,8 +1584,11 @@ class AgentLoop:
             if not had_injections or stop_reason == "empty_final_response":
                 return None
 
+        if log_content:
             preview = final_content[:120] + "..." if len(final_content) > 120 else final_content
             logger.info("Response to {}:{}: {}", msg.channel, msg.sender_id, preview)
+        else:
+            logger.info("Response to {}:{}: [content hidden]", msg.channel, msg.sender_id)
 
         event = None
         meta = dict(msg.metadata or {})
@@ -1594,17 +1617,33 @@ class AgentLoop:
             ctx.msg = dataclasses.replace(msg, content=new_content, media=image_paths)
             msg = ctx.msg
 
-        preview = msg.content[:80] + "..." if len(msg.content) > 80 else msg.content
-        if ctx.kind is TurnKind.SYSTEM:
-            logger.info("Processing system message from {}", msg.sender_id)
-        else:
-            logger.info("Processing message from {}:{}: {}", msg.channel, msg.sender_id, preview)
-
-        # Session is already fetched by the caller (_process_message) but
-        # ensure it exists in case this handler is invoked independently.
         if ctx.session is None:
+            if msg.require_existing_session:
+                ctx.session = self.sessions.get_cached(ctx.session_key)
+                if ctx.session is None:
+                    raise RuntimeError("required session is not active")
+            else:
                 ctx.session = self.sessions.get_or_create(ctx.session_key)
         session = ctx.session
+        ctx.ephemeral = ctx.ephemeral or not session.policy.persist
+        tools = ctx.tools or self.tools
+        if session.policy.disabled_tools:
+            restricted = ToolRegistry()
+            for name in tools.tool_names:
+                tool = tools.get(name)
+                if name not in session.policy.disabled_tools and tool:
+                    restricted.register(tool)
+            tools = restricted
+        ctx.tools = tools
+
+        if ctx.kind is TurnKind.SYSTEM:
+            logger.info("Processing system message from {}", msg.sender_id)
+        elif session.policy.log_content:
+            preview = msg.content[:80] + "..." if len(msg.content) > 80 else msg.content
+            logger.info("Processing message from {}:{}: {}", msg.channel, msg.sender_id, preview)
+        else:
+            logger.info("Processing message from {}:{}: [content hidden]", msg.channel, msg.sender_id)
+
         self._remember_unified_session_route(
             session,
             msg,
@@ -1907,6 +1946,7 @@ class AgentLoop:
             ctx.stop_reason,
             ctx.had_injections,
             ctx.streamed_content,
+            log_content=ctx.require_session().policy.log_content,
             turn_latency_ms=ctx.turn_latency_ms,
         )
         if ctx.ephemeral and ctx.outbound is not None:

nanobot/agent/memory.py

@@ -21,7 +21,7 @@ from typing import TYPE_CHECKING, Any, Callable, Iterator, cast
 from loguru import logger
 
 from nanobot.runtime_context import public_history_messages
-from nanobot.session.manager import Session, SessionManager
+from nanobot.session.manager import MIN_COMPACTED_REPLAY_MESSAGES, Session, SessionManager
 from nanobot.utils.gitstore import GitStore
 from nanobot.utils.helpers import (
     content_with_media_breadcrumbs,
@@ -858,14 +858,13 @@ class Consolidator:
         return last_boundary
 
     @staticmethod
-    def _full_unconsolidated_history(
+    def _full_replay_history(
         session: Session,
     ) -> list[dict[str, Any]]:
-        """Return the whole unconsolidated tail for consolidation decisions."""
+        """Return all messages that can reach the next model prompt."""
-        unconsolidated_count = len(session.messages) - session.last_consolidated
+        if not session.messages:
-        if unconsolidated_count <= 0:
             return []
-        return session.get_history(max_messages=unconsolidated_count)
+        return session.get_history(max_messages=len(session.messages))
 
     @staticmethod
     def _replay_overflow_boundary(
@@ -948,8 +947,8 @@ class Consolidator:
         *,
         runtime: LLMRuntime,
     ) -> tuple[int, str]:
-        """Estimate prompt size from the full unconsolidated session tail."""
+        """Estimate prompt size from the full replayable session history."""
-        history = self._full_unconsolidated_history(session)
+        history = self._full_replay_history(session)
         channel = session.key.split(":", 1)[0] if ":" in session.key else None
         # Include archived summary in estimation so the budget accounts for it.
         meta = session.metadata.get("_last_summary")
@@ -1160,42 +1159,37 @@ class Consolidator:
         session_key: str,
         *,
         runtime: LLMRuntime,
-        max_suffix: int = 8,
+        max_suffix: int = MIN_COMPACTED_REPLAY_MESSAGES,
     ) -> str | None:
-        """Archive an idle prefix and hide it from replay without deleting it."""
+        """Archive the full idle tail while keeping recent messages replayable.
+
+        ``max_suffix`` remains accepted for SDK compatibility. Replay retention
+        is now derived independently from archive progress using the project-wide
+        compacted-session window.
+        """
+        if max_suffix != MIN_COMPACTED_REPLAY_MESSAGES:
+            logger.debug(
+                "Idle-session compact for {} uses the fixed replay window ({}, requested {})",
+                session_key,
+                MIN_COMPACTED_REPLAY_MESSAGES,
+                max_suffix,
+            )
         lock = self.get_lock(session_key)
         async with lock:
             self.sessions.invalidate(session_key)
             session = self.sessions.get_or_create(session_key)
 
-            messages_to_summarize = list(session.messages[session.last_consolidated:])
+            archive_start = session.last_consolidated
-            if not messages_to_summarize:
+            messages_to_archive = list(session.messages[archive_start:])
-                self.sessions.save(session)
+            if not messages_to_archive:
-                return ""
-
-            probe = Session(
-                key=session.key,
-                messages=messages_to_summarize.copy(),
-                created_at=session.created_at,
-                updated_at=session.updated_at,
-                metadata={},
-                last_consolidated=0,
-            )
-            result = probe.retain_recent_legal_suffix(max_suffix, extend_to_user=True)
-            visible_suffix = probe.messages
-            messages_to_remove = result.dropped
-
-            if not messages_to_remove:
-                self.sessions.save(session)
                 return ""
 
             last_active = session.updated_at
-            # The visible suffix informs the summary but stays out of raw fallback.
+            archive_end = archive_start + len(messages_to_archive)
             summary = await self.archive(
-                messages_to_remove,
+                messages_to_archive,
                 runtime=runtime,
                 session_key=session_key,
-                summary_messages=messages_to_summarize,
             )
 
             if summary and summary != "(nothing)":
@@ -1204,16 +1198,22 @@ class Consolidator:
                     "last_active": last_active.isoformat(),
                 }
 
-            # Preserve history and advance only the replay boundary.
+            # A turn can append while the provider call is in flight. Advance only
-            session.last_consolidated = len(session.messages) - len(visible_suffix)
+            # through the captured batch so new messages remain eligible next time.
+            session.last_consolidated = archive_end
             session.provider_state = None
             self.sessions.save(session)
 
+            visible = session.get_history(
+                max_messages=MIN_COMPACTED_REPLAY_MESSAGES,
+                extend_to_user=True,
+            )
+
             logger.info(
                 "Idle-session compact for {}: archived={}, visible={}, retained={}, summary={}",
                 session_key,
-                len(messages_to_remove),
+                len(messages_to_archive),
-                len(visible_suffix),
+                len(visible),
                 len(session.messages),
                 bool(summary),
             )

nanobot/agent/tools/filesystem.py

@@ -785,22 +785,6 @@ def _best_window(old_text: str, content: str) -> tuple[float, int, list[str], li
     return best_ratio, best_start, best_window_lines, hints
 
 
-def _find_match(content: str, old_text: str) -> tuple[str | None, int]:
-    """Locate old_text in content with a multi-level fallback chain:
-
-    1. Exact substring match
-    2. Line-trimmed sliding window (handles indentation differences)
-    3. Smart quote normalization (curly ↔ straight quotes)
-
-    Both inputs should use LF line endings (caller normalises CRLF).
-    Returns (matched_fragment, count) or (None, 0).
-    """
-    matches = _find_matches(content, old_text)
-    if not matches:
-        return None, 0
-    return matches[0].text, len(matches)
-
-
 @tool_parameters(
     tool_parameters_schema(
         path=StringSchema("The file path to edit"),

nanobot/agent/tools/path_utils.py

@@ -3,15 +3,7 @@
 from pathlib import Path
 
 from nanobot.config.paths import get_media_dir
-from nanobot.security.workspace_policy import (
+from nanobot.security.workspace_policy import resolve_allowed_path
-    is_path_within,
-    resolve_allowed_path,
-)
-
-
-def is_under(path: Path, directory: Path) -> bool:
-    """Return True when path resolves under directory."""
-    return is_path_within(path, directory)
 
 
 def resolve_workspace_path(

nanobot/agent/tools/web.py

@@ -453,12 +453,15 @@ class WebSearchTool(Tool):
 
     async def _search_olostep(self, query: str, n: int) -> str:
         try:
-            from olostep import (  # pyright: ignore[reportMissingImports]
+            from olostep import (  # pyright: ignore[reportMissingImports, reportMissingTypeStubs]
                 AsyncOlostep,  # pyright: ignore[reportUnknownVariableType]
-                Olostep_BaseError,  # pyright: ignore[reportUnknownVariableType]
+                Olostep_BaseError,  # pyright: ignore[reportAttributeAccessIssue, reportUnknownVariableType]
             )
         except ImportError:
-            return ToolResult.error("Error: olostep package not installed. Run: pip install olostep")
+            return ToolResult.error(
+                "Error: Olostep support is not installed. "
+                "Run `nanobot plugins enable olostep`."
+            )
         async_olostep = cast(Any, AsyncOlostep)
         olostep_base_error = cast(type[Exception], Olostep_BaseError)
         api_key = self.config.api_key or os.environ.get("OLOSTEP_API_KEY", "")

nanobot/apps/cli/utils.py

@@ -12,15 +12,6 @@ def session_extra(metadata: Mapping[str, Any] | None) -> dict[str, Any]:
     return {"cli_apps": cli_apps} if isinstance(cli_apps, list) and cli_apps else {}
 
 
-def runtime_lines(message: Any, workspace: Path, *, skip: bool = False) -> list[str]:
-    """Return model-visible CLI app annotations for the current turn."""
-    if skip:
-        return []
-    text = message.content if isinstance(getattr(message, "content", None), str) else ""
-    metadata = message.metadata if isinstance(getattr(message, "metadata", None), Mapping) else None
-    return runtime_lines_for_request(text, metadata, workspace)
-
-
 def runtime_lines_for_request(
     text: str,
     metadata: Mapping[str, Any] | None,

nanobot/bus/events.py

@@ -18,6 +18,7 @@ INBOUND_META_RUNTIME_CONTROL = "_runtime_control"
 RUNTIME_CONTROL_ACK = "_ack"
 RUNTIME_CONTROL_MCP_RELOAD = "mcp_reload"
 RUNTIME_CONTROL_IMAGE_GENERATION_RELOAD = "image_generation_reload"
+RUNTIME_CONTROL_SESSION_DISCARD = "session_discard"
 
 
 @dataclass
@@ -32,6 +33,7 @@ class InboundMessage:
     media: list[str] = field(default_factory=list)  # Media URLs
     metadata: dict[str, Any] = field(default_factory=dict)  # Channel-specific data
     session_key_override: str | None = None  # Optional override for thread-scoped sessions
+    require_existing_session: bool = False
 
     @property
     def session_key(self) -> str:

nanobot/channels/base.py

@@ -101,6 +101,31 @@ class BaseChannel(ABC):
         """
         pass
 
+    def progress_transport_defaults(self) -> tuple[bool, bool] | None:
+        """Return channel-owned defaults for progress and tool-hint messages.
+
+        ``None`` keeps the global channel policy. Channels should override this
+        only when their transport requires different defaults.
+        """
+        return None
+
+    def should_retry_send_error(self, error: Exception) -> bool:
+        """Return whether the channel manager may retry a failed delivery.
+
+        Channels with protocol-level business errors can override this hook to
+        prevent retries that cannot succeed until external state changes.
+        Transport and unexpected errors remain retryable by default.
+        """
+        return True
+
+    def start_error_message(self, error: Exception) -> str | None:
+        """Return an actionable public message for a channel startup failure.
+
+        Channel-specific exception handling stays in the owning channel. Returning
+        ``None`` keeps the manager's generic fallback.
+        """
+        return None
+
     async def send_delta(
         self,
         chat_id: str,
@@ -237,6 +262,7 @@ class BaseChannel(ABC):
         session_key: str | None = None,
         is_dm: bool = False,
         authorization_id: str | None = None,
+        require_existing_session: bool = False,
     ) -> None:
         """Handle a message after checking its authorization subject.
 
@@ -289,6 +315,7 @@ class BaseChannel(ABC):
             media=media or [],
             metadata=meta,
             session_key_override=session_key,
+            require_existing_session=require_existing_session,
         )
 
         await self.bus.publish_inbound(msg)

nanobot/channels/feishu/runtime.py

@@ -470,15 +470,6 @@ def _extract_post_content(content_json: dict[str, Any]) -> tuple[str, list[str]]
     return "", []
 
 
-def _extract_post_text(content_json: dict[str, Any]) -> str:  # pyright: ignore[reportUnusedFunction]
-    """Extract plain text from Feishu post (rich text) message content.
-
-    Legacy wrapper for _extract_post_content, returns only text.
-    """
-    text, _ = _extract_post_content(content_json)
-    return text
-
-
 # =============================================================================
 # QR scan-to-create onboarding
 #

nanobot/channels/feishu/tests/test_feishu_reaction.py

@@ -238,20 +238,6 @@ class TestStreamEndReactionCleanup:
 
         ch._remove_reaction.assert_not_called()
 
-    @pytest.mark.asyncio
-    async def test_no_removal_when_both_ids_missing(self):
-        ch = _make_channel()
-        ch._stream_bufs["oc_chat1"] = _FeishuStreamBuf(
-            text="Done", card_id="card_1", sequence=3, last_edit=0.0,
-        )
-        ch._client.cardkit.v1.card_element.content.return_value = MagicMock(success=MagicMock(return_value=True))
-        ch._client.cardkit.v1.card.settings.return_value = MagicMock(success=MagicMock(return_value=True))
-        ch._remove_reaction = AsyncMock()
-
-        await ch.send_delta("oc_chat1", "", stream_end=True)
-
-        ch._remove_reaction.assert_not_called()
-
     @pytest.mark.asyncio
     async def test_no_removal_when_not_stream_end(self):
         ch = _make_channel()

nanobot/channels/manager.py

@@ -187,11 +187,15 @@ class ChannelManager:
         channel = cls(section, self.bus, **kwargs)
         if runtime_name and runtime_name != channel.name:
             channel.name = runtime_name
+        progress_default, tool_hints_default = channel.progress_transport_defaults() or (
+            self.config.channels.send_progress,
+            self.config.channels.send_tool_hints,
+        )
         channel.send_progress = self._resolve_bool_override(
-            section, "send_progress", self.config.channels.send_progress,
+            section, "send_progress", progress_default,
         )
         channel.send_tool_hints = self._resolve_bool_override(
-            section, "send_tool_hints", self.config.channels.send_tool_hints,
+            section, "send_tool_hints", tool_hints_default,
         )
         channel.show_reasoning = self._resolve_bool_override(
             section, "show_reasoning", self.config.channels.show_reasoning,
@@ -347,8 +351,12 @@ class ChannelManager:
             await channel.start()
         except asyncio.CancelledError:
             raise
-        except Exception:
+        except Exception as exc:
-            errors[name] = "Channel failed to start. Check gateway logs."
+            public_error = channel.start_error_message(exc)
+            errors[name] = public_error or "Channel failed to start. Check gateway logs."
+            if public_error:
+                logger.error("Failed to start channel {}: {}", name, public_error)
+            else:
                 logger.exception("Failed to start channel {}", name)
 
     def _start_channel_task(self, name: str, channel: BaseChannel) -> asyncio.Task[None]:
@@ -912,6 +920,14 @@ class ChannelManager:
             except asyncio.CancelledError:
                 raise  # Propagate cancellation for graceful shutdown
             except Exception as e:
+                if not channel.should_retry_send_error(e):
+                    logger.error(
+                        "Send to {} failed with a non-retryable {}: {}",
+                        msg.channel,
+                        type(e).__name__,
+                        e,
+                    )
+                    return
                 loop = asyncio.get_running_loop()
                 exhausted = (
                     attempt >= max_attempts

nanobot/channels/matrix/runtime.py

@@ -24,10 +24,12 @@ try:
     import nh3
     from mistune import HTMLRenderer, create_markdown
     from nio import (
+        Api,
         AsyncClient,
         AsyncClientConfig,
         InviteEvent,
         JoinError,
+        JoinResponse,
         KeyVerificationCancel,
         KeyVerificationEvent,
         KeyVerificationKey,
@@ -43,6 +45,7 @@ try:
         RoomSendResponse,
         RoomTypingError,
         SyncError,
+        SyncResponse,
         ToDeviceError,
         UploadError,
     )
@@ -701,6 +704,7 @@ class MatrixChannel(BaseChannel):
         client.add_response_callback(self._on_sync_error, SyncError)
         client.add_response_callback(self._on_join_error, JoinError)
         client.add_response_callback(self._on_send_error, RoomSendError)
+        client.add_response_callback(self._on_sync_invite_fallback, SyncResponse)
 
     def _is_sas_sender_allowed(self, sender: str) -> bool:
         return bool(sender and self.is_allowed(sender))
@@ -782,6 +786,49 @@ class MatrixChannel(BaseChannel):
                 with suppress(Exception):
                     self.client.stop_sync_forever()
 
+    async def _join_room_safe(self, room_id: str) -> bool:
+        """Join a room, sending a non-empty POST body.
+
+        nio's ``Api.join()`` produces a POST with no body. Some homeservers
+        (notably Continuwuity) reject empty bodies with ``M_BAD_JSON``.
+        Sending ``"{}"`` satisfies both strict and lenient servers.
+        """
+        client = self._require_client()
+        method, path = Api.join(client.access_token, room_id)
+        try:
+            resp = cast(
+                JoinResponse | JoinError,
+                await client._send(  # type: ignore[reportPrivateUsage, reportUnknownMemberType]
+                    JoinResponse, method, path, data="{}"
+                ),
+            )
+        except Exception:
+            self.logger.error("Matrix join request exception for room={}", room_id, exc_info=True)
+            return False
+        if isinstance(resp, JoinError):
+            self.logger.error("Matrix auto-join failed for room={}: {}", room_id, resp)
+            return False
+        self.logger.info("Matrix auto-join succeeded: {}", room_id)
+        return True
+
+    async def _on_sync_invite_fallback(self, response: SyncResponse) -> None:
+        """Safety net: join pending invites that the event callback may have missed.
+
+        Some homeservers (e.g. Continuwuity) deliver each invite only once.
+        If ``_on_room_invite`` fires but the join fails, the sync token
+        advances and the invite is never re-delivered. This callback inspects
+        the same ``SyncResponse`` for pending invites and joins them, acting
+        as a fallback alongside the event-based callback.
+        """
+        if not response.rooms or not response.rooms.invite:
+            return
+        for room_id, invite_info in response.rooms.invite.items():
+            for event in cast(list[Any], invite_info.invite_state):
+                sender = getattr(event, "sender", None)
+                if sender and self.is_allowed(cast(str, sender)):
+                    await self._join_room_safe(room_id)
+                    break
+
     async def _on_join_error(self, response: JoinError) -> None:
         self._log_response_error("join", response)
 
@@ -838,8 +885,7 @@ class MatrixChannel(BaseChannel):
 
     async def _on_room_invite(self, room: MatrixRoom, event: InviteEvent) -> None:
         if self.is_allowed(event.sender):
-            client = self._require_client()
+            await self._join_room_safe(room.room_id)
-            await client.join(room.room_id)
 
     def _is_direct_room(self, room: MatrixRoom) -> bool:
         count = getattr(room, "member_count", None)

nanobot/channels/matrix/tests/test_matrix_channel.py

@@ -4,13 +4,14 @@ import asyncio
 import sys
 from pathlib import Path
 from types import SimpleNamespace
+from urllib.parse import unquote
 
 import pytest
 
 pytest.importorskip("nio")
 pytest.importorskip("nh3")
 pytest.importorskip("mistune")
-from nio import RoomSendResponse, SyncError
+from nio import JoinResponse, RoomSendResponse, SyncError
 
 import nanobot.channels.matrix.runtime as matrix_module
 from nanobot.bus.events import OutboundMessage
@@ -104,6 +105,15 @@ class _FakeAsyncClient:
     async def join(self, room_id: str) -> None:
         self.join_calls.append(room_id)
 
+    async def _send(self, response_class, method, path, data=None, **kwargs):
+        """Minimal mock for nio's ``_send`` used by ``_join_room_safe``."""
+        if response_class is JoinResponse and method == "POST" and "/join/" in path:
+            encoded = path.split("/join/")[1].split("?")[0]
+            room_id = unquote(encoded)
+            self.join_calls.append(room_id)
+            return JoinResponse(room_id=room_id)
+        return response_class()
+
     async def accept_key_verification(self, transaction_id: str):
         self.operation_calls.append(f"accept:{transaction_id}")
         self.accept_key_verification_calls.append(transaction_id)
@@ -308,7 +318,7 @@ async def test_start_skips_load_store_when_device_id_missing(
     assert clients[0].load_store_called is False
     assert len(clients[0].callbacks) == 3
     assert clients[0].to_device_callbacks == []
-    assert len(clients[0].response_callbacks) == 3
+    assert len(clients[0].response_callbacks) == 4
 
     await channel.stop()
 
@@ -590,6 +600,7 @@ async def test_room_invite_joins_when_sender_allowed() -> None:
 
     assert client.join_calls == ["!room:matrix.org"]
 
+
 @pytest.mark.asyncio
 async def test_room_invite_respects_allow_list_when_configured() -> None:
     channel = MatrixChannel(_make_config(allow_from=["@bob:matrix.org"]), MessageBus())
@@ -604,6 +615,61 @@ async def test_room_invite_respects_allow_list_when_configured() -> None:
     assert client.join_calls == []
 
 
+@pytest.mark.asyncio
+async def test_on_sync_invite_fallback_joins_pending_invites() -> None:
+    """_on_sync_invite_fallback joins rooms from sync invite_state for allowed senders."""
+    channel = MatrixChannel(
+        _make_config(allow_from=["@alice:matrix.org"]), MessageBus()
+    )
+    client = _FakeAsyncClient("", "", "", None)
+    channel.client = client
+
+    invite_event = SimpleNamespace(sender="@alice:matrix.org")
+    invite_info = SimpleNamespace(invite_state=[invite_event])
+    rooms = SimpleNamespace(invite={"!room:matrix.org": invite_info})
+    response = SimpleNamespace(rooms=rooms)
+
+    await channel._on_sync_invite_fallback(response)
+
+    assert client.join_calls == ["!room:matrix.org"]
+
+
+@pytest.mark.asyncio
+async def test_on_sync_invite_fallback_skips_when_no_invites() -> None:
+    """_on_sync_invite_fallback is a no-op when sync has no invites."""
+    channel = MatrixChannel(
+        _make_config(allow_from=["@alice:matrix.org"]), MessageBus()
+    )
+    client = _FakeAsyncClient("", "", "", None)
+    channel.client = client
+
+    rooms = SimpleNamespace(invite={})
+    response = SimpleNamespace(rooms=rooms)
+
+    await channel._on_sync_invite_fallback(response)
+
+    assert client.join_calls == []
+
+
+@pytest.mark.asyncio
+async def test_on_sync_invite_fallback_skips_denied_sender() -> None:
+    """_on_sync_invite_fallback respects the allow list."""
+    channel = MatrixChannel(
+        _make_config(allow_from=["@bob:matrix.org"]), MessageBus()
+    )
+    client = _FakeAsyncClient("", "", "", None)
+    channel.client = client
+
+    invite_event = SimpleNamespace(sender="@alice:matrix.org")
+    invite_info = SimpleNamespace(invite_state=[invite_event])
+    rooms = SimpleNamespace(invite={"!room:matrix.org": invite_info})
+    response = SimpleNamespace(rooms=rooms)
+
+    await channel._on_sync_invite_fallback(response)
+
+    assert client.join_calls == []
+
+
 @pytest.mark.asyncio
 async def test_on_message_sets_typing_for_allowed_sender() -> None:
     channel = MatrixChannel(_make_config(), MessageBus())

nanobot/channels/mattermost/runtime.py

@@ -658,11 +658,6 @@ class MattermostChannel(BaseChannel):
         resp.raise_for_status()
         return cast(dict[str, Any], resp.json())
 
-    async def _api_put(self, path: str, json_data: dict[str, Any]) -> dict[str, Any]:
-        resp = await self._require_http_client().put(path, json=json_data)
-        resp.raise_for_status()
-        return cast(dict[str, Any], resp.json())
-
     async def _create_post(
         self,
         channel_id: str,
@@ -681,9 +676,6 @@ class MattermostChannel(BaseChannel):
             body["file_ids"] = file_ids
         return await self._api_post("/api/v4/posts", body)
 
-    async def _edit_post(self, post_id: str, message: str) -> dict[str, Any]:
-        return await self._api_put(f"/api/v4/posts/{post_id}", {"id": post_id, "message": message})
-
     async def _upload_file(self, channel_id: str, file_path: str) -> str | None:
         path = Path(file_path)
         if not path.exists():

nanobot/channels/msteams/runtime.py

@@ -811,11 +811,6 @@ class MSTeamsChannel(BaseChannel):
         except Exception as e:
             self.logger.warning("Failed to save conversation refs: {}", e)
 
-    def _save_refs(self, *, prune: bool = True) -> None:
-        """Persist conversation references."""
-        with self._refs_guard:
-            self._save_refs_locked(prune=prune)
-
     async def _get_access_token(self) -> str:
         """Fetch an access token for Bot Framework / Azure Bot auth."""
 

nanobot/channels/msteams/tests/test_msteams.py

@@ -228,7 +228,8 @@ def test_save_prunes_unsupported_conversation_refs(make_channel, tmp_path, monke
         ),
     }
 
-    ch._save_refs()
+    with ch._refs_guard:
+        ch._save_refs_locked()
 
     assert set(ch._conversation_refs.keys()) == {"conv-valid"}
 
@@ -378,7 +379,8 @@ def test_save_uses_atomic_replace_and_keeps_existing_file_on_replace_error(make_
         raise OSError("replace failed")
 
     monkeypatch.setattr(msteams_module.os, "replace", _raise_replace)
-    ch._save_refs()
+    with ch._refs_guard:
+        ch._save_refs_locked()
 
     persisted = json.loads(refs_path.read_text(encoding="utf-8"))
     assert set(persisted.keys()) == {"conv-old"}
@@ -934,7 +936,8 @@ def test_save_refs_prunes_webchat_and_stale_refs(make_channel):
         ),
     }
 
-    ch._save_refs()
+    with ch._refs_guard:
+        ch._save_refs_locked()
 
     assert set(ch._conversation_refs) == {"teams-good"}
     saved = json.loads(ch._refs_path.read_text(encoding="utf-8"))

nanobot/channels/signal/runtime.py

@@ -431,6 +431,7 @@ class SignalChannel(BaseChannel):
         session_key: str | None = None,
         is_dm: bool = False,
         authorization_id: str | None = None,
+        require_existing_session: bool = False,
     ) -> None:
         """Handle an inbound message whose policy has already been checked.
 
@@ -453,6 +454,7 @@ class SignalChannel(BaseChannel):
                 media=media or [],
                 metadata=meta,
                 session_key_override=session_key,
+                require_existing_session=require_existing_session,
             )
         )
 

nanobot/channels/websocket/runtime.py

@@ -20,7 +20,10 @@ from websockets.asyncio.server import ServerConnection, serve, unix_serve
 from websockets.exceptions import ConnectionClosed
 from websockets.http11 import Request as WsRequest
 
-from nanobot.bus.events import OUTBOUND_META_AGENT_UI, OutboundMessage
+from nanobot.bus.events import (
+    OUTBOUND_META_AGENT_UI,
+    OutboundMessage,
+)
 from nanobot.bus.outbound_events import (
     GoalStateSyncEvent,
     GoalStatusEvent,
@@ -30,7 +33,6 @@ from nanobot.bus.outbound_events import (
     TurnEndEvent,
     TurnModelUpdatedEvent,
     outbound_event_from_message,
-    outbound_message_for_event,
 )
 from nanobot.bus.queue import MessageBus
 from nanobot.channels.base import BaseChannel
@@ -49,6 +51,7 @@ from nanobot.security.workspace_access import (
 from nanobot.session.goal_state import goal_state_ws_blob
 from nanobot.session.webui_turns import (
     clear_websocket_turn_if_current,
+    clear_websocket_turns,
     mark_websocket_turn_transcript_persistence_failed,
     register_queued_websocket_turn_if_idle,
     websocket_turn_id,
@@ -81,6 +84,8 @@ from nanobot.webui.session_access import (
     WebuiSessionAccess,
     session_mentions_runtime_context,
 )
+from nanobot.webui.sidebar_state import write_webui_sidebar_state
+from nanobot.webui.temporary_chats import TemporaryChatError
 from nanobot.webui.transcript import WEBUI_TRANSCRIPT_INCOMPLETE_KEY
 from nanobot.webui.transcription_ws import webui_transcription_event
 from nanobot.webui.websocket_logging import websockets_server_logger
@@ -279,21 +284,6 @@ class WebSocketConfig(Base):
         )
 
 
-def publish_runtime_model_update(
-    bus: MessageBus,
-    model: str,
-    model_preset: str | None,
-) -> None:
-    """Enqueue a runtime model snapshot for websocket subscribers (fan-out in-channel)."""
-    bus.outbound.put_nowait(
-        outbound_message_for_event(
-            channel="websocket",
-            chat_id="*",
-            event=RuntimeModelUpdatedEvent(model=model, model_preset=model_preset),
-        )
-    )
-
-
 def _parse_inbound_payload(raw: str) -> str | None:
     """Parse a client frame into text; return None for empty or unrecognized content."""
     text = raw.strip()
@@ -393,6 +383,7 @@ class WebSocketChannel(BaseChannel):
         self._ingress = gateway.ingress
         self._transcripts = gateway.transcripts
         self._workspaces = gateway.workspaces
+        self._temporary_chats = gateway.temporary_chats
         self._session_access = (
             WebuiSessionAccess(gateway.session_manager)
             if gateway.session_manager is not None
@@ -411,6 +402,33 @@ class WebSocketChannel(BaseChannel):
         self._subs.setdefault(chat_id, set()).add(connection)
         self._conn_chats.setdefault(connection, set()).add(chat_id)
 
+    def _detach(self, connection: ServerConnection, chat_id: str) -> None:
+        chats = self._conn_chats.get(connection)
+        if chats is not None:
+            chats.discard(chat_id)
+            if not chats:
+                self._conn_chats.pop(connection, None)
+        subscribers = self._subs.get(chat_id)
+        if subscribers is not None:
+            subscribers.discard(connection)
+            if not subscribers:
+                self._subs.pop(chat_id, None)
+
+    def _clear_stream_buffers(self, chat_id: str) -> None:
+        for key in tuple(self._stream_text_buffers):
+            if key[0] == chat_id:
+                self._stream_text_buffers.pop(key, None)
+
+    async def _discard_connection_owned_chat(
+        self,
+        connection: ServerConnection,
+        chat_id: str,
+    ) -> None:
+        await self._temporary_chats.discard(connection, chat_id)
+        self._detach(connection, chat_id)
+        clear_websocket_turns(chat_id)
+        self._clear_stream_buffers(chat_id)
+
     async def send_webui_protocol_error(
         self,
         connection: ServerConnection,
@@ -439,16 +457,16 @@ class WebSocketChannel(BaseChannel):
         )
         await self._hydrate_after_subscribe(fork_id)
 
-    def _cleanup_connection(self, connection: ServerConnection) -> None:
+    async def _cleanup_connection(self, connection: ServerConnection) -> None:
         """Remove *connection* from every subscription set; safe to call multiple times."""
-        chat_ids = self._conn_chats.pop(connection, set())
+        chat_ids = tuple(self._conn_chats.get(connection, ()))
         for cid in chat_ids:
-            subs = self._subs.get(cid)
+            if self._temporary_chats.owns(connection, cid):
-            if subs is None:
+                await self._discard_connection_owned_chat(connection, cid)
-                continue
+            else:
-            subs.discard(connection)
+                self._detach(connection, cid)
-            if not subs:
+        for cid in self._temporary_chats.chat_ids_for_owner(connection):
-                self._subs.pop(cid, None)
+            await self._discard_connection_owned_chat(connection, cid)
         self._conn_default.pop(connection, None)
         self._webui_connections.discard(connection)
 
@@ -501,7 +519,7 @@ class WebSocketChannel(BaseChannel):
         try:
             await connection.send(raw)
         except ConnectionClosed:
-            self._cleanup_connection(connection)
+            await self._cleanup_connection(connection)
         except Exception as e:
             self.logger.warning("failed to send {} event: {}", event, e)
 
@@ -728,7 +746,7 @@ class WebSocketChannel(BaseChannel):
         except Exception as e:
             self.logger.debug("connection ended: {}", e)
         finally:
-            self._cleanup_connection(connection)
+            await self._cleanup_connection(connection)
 
     # -- Inbound WebSocket envelopes ---------------------------------------
 
@@ -763,23 +781,84 @@ class WebSocketChannel(BaseChannel):
             )
             await self._hydrate_after_subscribe(new_id)
             return
+        if t == "new_temporary_chat":
+            try:
+                new_id = self._temporary_chats.create(
+                    connection,
+                    trusted_webui=connection in self._webui_connections,
+                )
+            except TemporaryChatError as exc:
+                await self._send_event(connection, "error", detail=exc.detail)
+                return
+            self._attach(connection, new_id)
+            await self._send_event(
+                connection,
+                "attached",
+                chat_id=new_id,
+                temporary=True,
+            )
+            return
         if t == "fork_chat":
             await handle_webui_fork_chat(self, connection, envelope)
             return
+        if t == "discard_temporary_chat":
+            cid = envelope.get("chat_id")
+            if not _is_valid_chat_id(cid):
+                await self._send_event(connection, "error", detail="invalid temporary chat_id")
+                return
+            try:
+                await self._discard_connection_owned_chat(connection, cid)
+            except TemporaryChatError as exc:
+                await self._send_event(connection, "error", detail=exc.detail, chat_id=cid)
+            return
         if t == "attach":
             cid = envelope.get("chat_id")
             if not _is_valid_chat_id(cid):
                 await self._send_event(connection, "error", detail="invalid chat_id")
                 return
+            try:
+                self._temporary_chats.validate_attach(cid)
+            except TemporaryChatError as exc:
+                await self._send_event(connection, "error", detail=exc.detail, chat_id=cid)
+                return
             self._attach(connection, cid)
             await self._send_event(connection, "attached", chat_id=cid)
             await self._hydrate_after_subscribe(cid)
             return
+        if t == "set_sidebar_state":
+            if connection not in self._webui_connections:
+                await self._send_event(connection, "error", detail="access_denied")
+                return
+            state = envelope.get("state")
+            if not isinstance(state, dict):
+                await self._send_event(
+                    connection,
+                    "error",
+                    detail="invalid_sidebar_state",
+                )
+                return
+            try:
+                await asyncio.to_thread(
+                    write_webui_sidebar_state,
+                    cast(dict[str, Any], state),
+                )
+            except (OSError, ValueError):
+                await self._send_event(
+                    connection,
+                    "error",
+                    detail="invalid_sidebar_state",
+                )
+            return
         if t == "set_workspace_scope":
             cid = envelope.get("chat_id")
             if not _is_valid_chat_id(cid):
                 await self._send_event(connection, "error", detail="invalid chat_id")
                 return
+            try:
+                self._temporary_chats.validate_workspace_update(cid)
+            except TemporaryChatError as exc:
+                await self._send_event(connection, "error", detail=exc.detail, chat_id=cid)
+                return
             scope = await self._workspace_scope_or_error(
                 connection,
                 lambda: self._workspaces.scope_for_set_request(
@@ -848,6 +927,21 @@ class WebSocketChannel(BaseChannel):
                 )
                 return
 
+            try:
+                temporary_policy = self._temporary_chats.message_policy(
+                    connection,
+                    cid,
+                    content,
+                )
+            except TemporaryChatError as exc:
+                await self._send_event(
+                    connection,
+                    "error",
+                    detail=exc.detail,
+                    **rejection_fields,
+                )
+                return
+
             raw_media = envelope.get("media")
             media_paths: list[str] = []
             if raw_media is not None:
@@ -870,6 +964,8 @@ class WebSocketChannel(BaseChannel):
                         **rejection_fields,
                     )
                     return
+                if temporary_policy is not None:
+                    self._temporary_chats.register_media(connection, cid, media_paths)
 
             # Allow media-only turns (content may be empty when attachments are present).
             if not content.strip() and not media_paths:
@@ -882,16 +978,21 @@ class WebSocketChannel(BaseChannel):
                 return
             # Auto-attach on first use so clients can one-shot without a separate attach.
             self._attach(connection, cid)
+            if temporary_policy is None or temporary_policy.hydrate_transcript:
                 await self._hydrate_after_subscribe(cid)
 
             # Resolve after hydration so a concurrent downgrade cannot be overwritten.
             scope = await self._workspace_scope_or_error(
                 connection,
-                lambda: self._workspaces.scope_for_message(
+                lambda: (
+                    temporary_policy.workspace_scope
+                    if temporary_policy is not None
+                    else self._workspaces.scope_for_message(
                         envelope,
                         chat_id=cid,
                         chat_running=websocket_turn_wall_started_at(cid) is not None,
                         controls_available=self._workspace_controls_available(connection),
+                    )
                 ),
                 chat_id=cid,
                 turn_id=turn_id,
@@ -944,7 +1045,13 @@ class WebSocketChannel(BaseChannel):
                     metadata[WEBSOCKET_TURN_OWNER_METADATA_KEY] = queued_owner
             accepted = False
             try:
-                if is_webui:
+                if (
+                    is_webui
+                    and (
+                        temporary_policy is None
+                        or temporary_policy.persist_transcript
+                    )
+                ):
                     self._transcripts.append_user_message(
                         cid,
                         content,
@@ -973,6 +1080,16 @@ class WebSocketChannel(BaseChannel):
                     media=media_paths or None,
                     metadata=metadata,
                     is_dm=False,
+                    session_key=(
+                        temporary_policy.session_key
+                        if temporary_policy is not None
+                        else None
+                    ),
+                    require_existing_session=(
+                        temporary_policy.require_existing_session
+                        if temporary_policy is not None
+                        else False
+                    ),
                 )
                 accepted = True
             finally:
@@ -1033,6 +1150,7 @@ class WebSocketChannel(BaseChannel):
         self._conn_default.clear()
         self._webui_connections.clear()
         self._tokens.clear()
+        self._temporary_chats.close()
 
     async def _safe_send_to(
         self,
@@ -1045,7 +1163,7 @@ class WebSocketChannel(BaseChannel):
         try:
             await connection.send(raw)
         except ConnectionClosed:
-            self._cleanup_connection(connection)
+            await self._cleanup_connection(connection)
             self.logger.warning("connection gone{}", label)
         except Exception:
             self.logger.exception("send failed{}", label)
@@ -1062,6 +1180,8 @@ class WebSocketChannel(BaseChannel):
         transcript_overrides: dict[str, Any] | None = None,
     ) -> bool:
         """Persist one canonical turn event and retain unsafe owners on failure."""
+        if not self._temporary_chats.should_persist_transcript(chat_id):
+            return True
         persisted = self._transcripts.prepare_and_append(
             chat_id,
             event,

nanobot/channels/websocket/tests/test_websocket_channel.py

@@ -13,7 +13,9 @@ from websockets.exceptions import ConnectionClosed
 from websockets.frames import Close
 
 from nanobot.bus.events import (
+    INBOUND_META_RUNTIME_CONTROL,
     OUTBOUND_META_AGENT_UI,
+    RUNTIME_CONTROL_SESSION_DISCARD,
     OutboundMessage,
 )
 from nanobot.bus.outbound_events import (
@@ -32,11 +34,11 @@ from nanobot.channels.websocket.runtime import (
     _is_valid_chat_id,
     _parse_envelope,
     _parse_inbound_payload,
-    publish_runtime_model_update,
 )
 from nanobot.config.loader import load_config, save_config
 from nanobot.config.schema import Config, ModelPresetConfig
 from nanobot.runtime_context import RUNTIME_CONTEXT_INPUT_META, WEBUI_QUOTE_SOURCE
+from nanobot.security.workspace_access import WORKSPACE_SCOPE_METADATA_KEY
 from nanobot.session import webui_turns as wth
 from nanobot.session.manager import SessionManager
 from nanobot.webui.gateway_services import GatewayServices, build_gateway_services
@@ -193,6 +195,302 @@ def isolate_webui_workspace_state(tmp_path, monkeypatch) -> None:
     wth._WEBSOCKET_TURN_OWNERS.clear()
 
 
+async def _new_temporary_chat(
+    channel: WebSocketChannel,
+    connection: AsyncMock,
+) -> str:
+    channel._webui_connections.add(connection)
+    await channel._dispatch_envelope(
+        connection,
+        "webui-client",
+        {"type": "new_temporary_chat"},
+    )
+    payload = json.loads(connection.send.await_args.args[0])
+    assert payload["event"] == "attached"
+    assert payload["temporary"] is True
+    connection.send.reset_mock()
+    return payload["chat_id"]
+
+
+@pytest.mark.asyncio
+async def test_temporary_chat_is_transient_and_discarded(bus, tmp_path) -> None:
+    sessions = SessionManager(tmp_path)
+    selected_project = tmp_path / "selected-project"
+    selected_project.mkdir()
+    channel = WebSocketChannel(
+        {"enabled": True, "allowFrom": ["*"]},
+        bus,
+        gateway=_basic_handler(
+            bus,
+            session_manager=sessions,
+            workspace_path=tmp_path,
+        ),
+    )
+    connection = AsyncMock()
+    connection.remote_address = ("127.0.0.1", 5000)
+    chat_id = await _new_temporary_chat(channel, connection)
+    upload = tmp_path / "temporary-upload.txt"
+    upload.write_text("private attachment", encoding="utf-8")
+    channel.gateway.media.store_inbound_attachments = MagicMock(
+        return_value=([str(upload)], None),
+    )
+
+    await channel._dispatch_envelope(
+        connection,
+        "webui-client",
+        {
+            "type": "message",
+            "chat_id": chat_id,
+            "content": "read this",
+            "media": [{"data_url": "data:text/plain;base64,cHJpdmF0ZQ=="}],
+            "cli_apps": [{"name": "drawio"}],
+            "workspace_scope": {
+                "project_path": str(selected_project),
+                "access_mode": "full",
+            },
+            "turn_id": "turn-1",
+            "webui": True,
+        },
+    )
+
+    inbound = bus.publish_inbound.await_args_list[0].args[0]
+    assert inbound.session_key == f"websocket:{chat_id}"
+    assert inbound.session_key_override == f"websocket:{chat_id}"
+    assert inbound.require_existing_session is True
+    assert inbound.metadata["cli_apps"] == [{"name": "drawio"}]
+    assert inbound.metadata[WORKSPACE_SCOPE_METADATA_KEY] == {
+        "project_path": str(tmp_path.resolve()),
+        "access_mode": "restricted",
+    }
+    session = sessions.get_cached(inbound.session_key)
+    assert session is not None
+    assert session.policy.persist is False
+    assert upload.exists()
+    assert read_transcript_lines(inbound.session_key) == []
+    assert [payload["event"] for payload in _sent_ws_payloads(connection)] == [
+        "message_accepted",
+    ]
+
+    await channel._dispatch_envelope(
+        connection,
+        "webui-client",
+        {"type": "discard_temporary_chat", "chat_id": chat_id},
+    )
+
+    control = bus.publish_inbound.await_args_list[1].args[0]
+    assert bus.publish_inbound.await_count == 2
+    assert control.session_key == inbound.session_key
+    assert control.metadata[INBOUND_META_RUNTIME_CONTROL] == (
+        RUNTIME_CONTROL_SESSION_DISCARD
+    )
+    assert sessions.get_cached(inbound.session_key) is None
+    assert chat_id not in channel._subs
+    assert chat_id not in channel._conn_chats.get(connection, set())
+    assert not upload.exists()
+    assert read_transcript_lines(inbound.session_key) == []
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("content", ["/goal private", "/trigger later", "/dream"])
+async def test_temporary_chat_rejects_persistent_commands(bus, tmp_path, content) -> None:
+    sessions = SessionManager(tmp_path)
+    channel = WebSocketChannel(
+        {"enabled": True, "allowFrom": ["*"]},
+        bus,
+        gateway=_basic_handler(bus, session_manager=sessions, workspace_path=tmp_path),
+    )
+    connection = AsyncMock()
+    connection.remote_address = ("127.0.0.1", 5000)
+    chat_id = await _new_temporary_chat(channel, connection)
+
+    await channel._dispatch_envelope(connection, "webui-client", {
+        "type": "message",
+        "chat_id": chat_id,
+        "content": content,
+        "webui": True,
+    })
+
+    assert bus.publish_inbound.await_count == 0
+    assert sessions.get_cached(f"websocket:{chat_id}") is not None
+    assert json.loads(connection.send.await_args.args[0])["detail"] == (
+        "temporary_chat_command_rejected"
+    )
+
+
+@pytest.mark.asyncio
+async def test_disconnect_discards_temporary_chat(bus, tmp_path) -> None:
+    sessions = SessionManager(tmp_path)
+    channel = WebSocketChannel(
+        {"enabled": True, "allowFrom": ["*"]},
+        bus,
+        gateway=_basic_handler(
+            bus,
+            session_manager=sessions,
+            workspace_path=tmp_path,
+        ),
+    )
+    connection = AsyncMock()
+    chat_id = await _new_temporary_chat(channel, connection)
+
+    await channel._dispatch_envelope(
+        connection,
+        "webui-client",
+        {
+            "type": "message",
+            "chat_id": chat_id,
+            "content": "hello",
+            "webui": True,
+        },
+    )
+    await channel._cleanup_connection(connection)
+
+    session_key = f"websocket:{chat_id}"
+    control = bus.publish_inbound.await_args_list[-1].args[0]
+    assert control.session_key == session_key
+    assert control.metadata[INBOUND_META_RUNTIME_CONTROL] == (
+        RUNTIME_CONTROL_SESSION_DISCARD
+    )
+    assert sessions.get_cached(session_key) is None
+    assert chat_id not in channel._subs
+
+
+@pytest.mark.asyncio
+async def test_temporary_chat_creation_requires_authenticated_webui_connection(bus, tmp_path) -> None:
+    sessions = SessionManager(tmp_path)
+    channel = WebSocketChannel(
+        {"enabled": True, "allowFrom": ["*"]},
+        bus,
+        gateway=_basic_handler(bus, session_manager=sessions, workspace_path=tmp_path),
+    )
+    connection = AsyncMock()
+
+    await channel._dispatch_envelope(
+        connection,
+        "generic-websocket-client",
+        {"type": "new_temporary_chat"},
+    )
+
+    assert json.loads(connection.send.await_args.args[0])["detail"] == "access_denied"
+    assert sessions.list_sessions() == []
+
+
+@pytest.mark.asyncio
+async def test_temporary_chat_cannot_be_claimed_by_another_connection(bus, tmp_path) -> None:
+    sessions = SessionManager(tmp_path)
+    channel = WebSocketChannel(
+        {"enabled": True, "allowFrom": ["*"]},
+        bus,
+        gateway=_basic_handler(bus, session_manager=sessions, workspace_path=tmp_path),
+    )
+    owner = AsyncMock()
+    other = AsyncMock()
+    channel._webui_connections.add(other)
+    chat_id = await _new_temporary_chat(channel, owner)
+
+    await channel._dispatch_envelope(
+        other,
+        "other-webui-client",
+        {
+            "type": "message",
+            "chat_id": chat_id,
+            "content": "claim it",
+            "webui": True,
+        },
+    )
+
+    assert json.loads(other.send.await_args.args[0])["detail"] == (
+        "temporary_chat_unavailable"
+    )
+    assert bus.publish_inbound.await_count == 0
+    assert sessions.get_cached(f"websocket:{chat_id}") is not None
+
+
+@pytest.mark.asyncio
+async def test_temporary_chat_cannot_persist_workspace_scope(bus, tmp_path) -> None:
+    sessions = SessionManager(tmp_path)
+    channel = WebSocketChannel(
+        {"enabled": True, "allowFrom": ["*"]},
+        bus,
+        gateway=_basic_handler(bus, session_manager=sessions, workspace_path=tmp_path),
+    )
+    connection = AsyncMock()
+    chat_id = await _new_temporary_chat(channel, connection)
+
+    await channel._dispatch_envelope(
+        connection,
+        "webui-client",
+        {
+            "type": "set_workspace_scope",
+            "chat_id": chat_id,
+            "workspace_scope": {
+                "project_path": str(tmp_path),
+                "access_mode": "full",
+            },
+        },
+    )
+
+    payload = json.loads(connection.send.await_args.args[0])
+    assert payload["detail"] == "temporary_chat_workspace_rejected"
+    session = sessions.get_cached(f"websocket:{chat_id}")
+    assert session is not None
+    assert WORKSPACE_SCOPE_METADATA_KEY not in session.metadata
+    assert sessions.list_sessions() == []
+
+
+@pytest.mark.asyncio
+async def test_temporary_looking_id_does_not_define_session_policy(bus, tmp_path) -> None:
+    sessions = SessionManager(tmp_path)
+    channel = WebSocketChannel(
+        {"enabled": True, "allowFrom": ["*"]},
+        bus,
+        gateway=_basic_handler(bus, session_manager=sessions, workspace_path=tmp_path),
+    )
+    connection = AsyncMock()
+    channel._webui_connections.add(connection)
+
+    await channel._dispatch_envelope(
+        connection,
+        "webui-client",
+        {
+            "type": "message",
+            "chat_id": "temporary-looking-but-persistent",
+            "content": "/goal ordinary chat",
+            "webui": True,
+        },
+    )
+
+    inbound = bus.publish_inbound.await_args.args[0]
+    assert inbound.require_existing_session is False
+    assert inbound.session_key_override is None
+    session = sessions.get_cached("websocket:temporary-looking-but-persistent")
+    assert session is not None
+    assert session.policy.persist is True
+
+
+@pytest.mark.asyncio
+async def test_discard_temporary_chat_does_not_detach_persistent_chat(bus, tmp_path) -> None:
+    sessions = SessionManager(tmp_path)
+    channel = WebSocketChannel(
+        {"enabled": True, "allowFrom": ["*"]},
+        bus,
+        gateway=_basic_handler(bus, session_manager=sessions, workspace_path=tmp_path),
+    )
+    connection = AsyncMock()
+    channel._attach(connection, "ordinary-chat")
+
+    await channel._dispatch_envelope(
+        connection,
+        "webui-client",
+        {"type": "discard_temporary_chat", "chat_id": "ordinary-chat"},
+    )
+
+    assert json.loads(connection.send.await_args.args[0])["detail"] == (
+        "temporary_chat_unavailable"
+    )
+    assert connection in channel._subs["ordinary-chat"]
+    assert "ordinary-chat" in channel._conn_chats[connection]
+
+
 @pytest.mark.asyncio
 async def test_send_session_updated_broadcasts_to_other_webui_connections(bus) -> None:
     class Conn:
@@ -559,6 +857,34 @@ def test_only_bootstrap_tokens_mark_webui_connections(bus: MagicMock) -> None:
     assert client_connection not in channel._webui_connections
 
 
+@pytest.mark.asyncio
+async def test_webui_persists_sidebar_state_larger_than_http_request_line(
+    bus: MagicMock,
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr("nanobot.config.paths.get_data_dir", lambda: tmp_path)
+    channel = _ch(bus)
+    conn = AsyncMock()
+    channel._webui_connections.add(conn)
+    session_order = [f"websocket:{index:04d}-{'x' * 48}" for index in range(160)]
+    envelope = {
+        "type": "set_sidebar_state",
+        "state": {
+            "session_order": session_order,
+            "view": {"sort": "manual"},
+        },
+    }
+    assert len(json.dumps(envelope).encode()) > 8_192
+
+    await channel._dispatch_envelope(conn, "webui-client", envelope)
+
+    saved = json.loads((tmp_path / "webui" / "sidebar-state.json").read_text(encoding="utf-8"))
+    assert saved["session_order"] == session_order
+    assert saved["view"]["sort"] == "manual"
+    conn.send.assert_not_awaited()
+
+
 @pytest.mark.asyncio
 async def test_client_cannot_self_assert_webui_quote_context(bus: MagicMock) -> None:
     channel = _ch(bus)
@@ -1077,8 +1403,14 @@ async def test_send_broadcasts_runtime_model_updates() -> None:
     mock_ws = AsyncMock()
     channel._attach(mock_ws, "chat-1")
 
-    publish_runtime_model_update(bus, "openai/gpt-4.1", "fast")
+    await channel.send(
-    await channel.send(bus.outbound.get_nowait())
+        OutboundMessage(
+            channel="websocket",
+            chat_id="*",
+            content="",
+            event=RuntimeModelUpdatedEvent(model="openai/gpt-4.1", model_preset="fast"),
+        )
+    )
 
     payload = json.loads(mock_ws.send.call_args[0][0])
     assert payload["event"] == "runtime_model_updated"
@@ -1113,26 +1445,6 @@ async def test_send_scopes_turn_model_updates_to_the_subscribed_chat() -> None:
     chat_two.send.assert_not_awaited()
 
 
-@pytest.mark.asyncio
-async def test_runtime_model_update_publisher_uses_websocket_outbound_event() -> None:
-    bus = MessageBus()
-
-    publish_runtime_model_update(
-        bus,
-        "openai/gpt-4.1",
-        "fast",
-    )
-
-    event = bus.outbound.get_nowait()
-    assert event.channel == "websocket"
-    assert event.chat_id == "*"
-    assert event.content == ""
-    assert event.metadata == {}
-    assert isinstance(event.event, RuntimeModelUpdatedEvent)
-    assert event.event.model == "openai/gpt-4.1"
-    assert event.event.model_preset == "fast"
-
-
 @pytest.mark.asyncio
 async def test_send_stages_external_media_as_signed_url(monkeypatch, tmp_path) -> None:
     bus = MagicMock()

nanobot/channels/websocket/tests/test_websocket_http_routes.py

@@ -19,11 +19,6 @@ from nanobot.channels.websocket.runtime import WebSocketChannel, WebSocketConfig
 from nanobot.cron.service import CronService
 from nanobot.cron.types import CronJob, CronPayload, CronSchedule
 from nanobot.optional_features import InstallResult
-from nanobot.runtime_context import (
-    RUNTIME_CONTEXT_HISTORY_META,
-    RuntimeContextBlock,
-    append_runtime_context,
-)
 from nanobot.security.workspace_access import WORKSPACE_SCOPE_METADATA_KEY
 from nanobot.session.keys import UNIFIED_SESSION_KEY
 from nanobot.session.manager import Session, SessionManager
@@ -256,7 +251,7 @@ async def test_bootstrap_returns_token_for_localhost(
 
 
 @pytest.mark.asyncio
-async def test_sessions_routes_require_bearer_token(
+async def test_sessions_list_requires_bearer_token(
     bus: MagicMock, tmp_path: Path
 ) -> None:
     sm = _seed_session(tmp_path, key="websocket:abc")
@@ -278,14 +273,26 @@ async def test_sessions_routes_require_bearer_token(
         # Server stays an opaque source: filesystem paths must not leak to the wire.
         assert all("path" not in s for s in listing.json()["sessions"])
 
-        msgs = await _http_get(
+    finally:
-            "http://127.0.0.1:29902/api/sessions/websocket:abc/messages",
+        await channel.stop()
-            headers=auth,
+        await server_task
+
+
+@pytest.mark.asyncio
+async def test_legacy_session_messages_route_is_not_exposed(
+    bus: MagicMock, tmp_path: Path
+) -> None:
+    sm = _seed_session(tmp_path, key="websocket:legacy")
+    channel = _ch(bus, session_manager=sm, port=29919)
+    server_task = asyncio.create_task(channel.start())
+    try:
+        token = channel.gateway.tokens.issue_api_token(300)
+        response = await _http_get(
+            "http://127.0.0.1:29919/api/sessions/websocket:legacy/messages",
+            headers={"Authorization": f"Bearer {token}"},
         )
-        assert msgs.status_code == 200
+
-        body = msgs.json()
+        assert response.status_code == 404
-        assert body["key"] == "websocket:abc"
-        assert [m["role"] for m in body["messages"]] == ["user", "assistant"]
     finally:
         await channel.stop()
         await server_task
@@ -2280,6 +2287,7 @@ async def test_webui_sidebar_state_routes_are_config_dir_scoped(
         payload = {
             "pinned_keys": ["websocket:sidebar"],
             "archived_keys": ["websocket:old"],
+            "session_order": ["websocket:old", "websocket:sidebar"],
             "title_overrides": {"websocket:sidebar": "Pinned work"},
             "view": {"density": "compact", "show_archived": True},
         }
@@ -2291,6 +2299,7 @@ async def test_webui_sidebar_state_routes_are_config_dir_scoped(
         assert updated.status_code == 200
         body = updated.json()
         assert body["pinned_keys"] == ["websocket:sidebar"]
+        assert body["session_order"] == ["websocket:old", "websocket:sidebar"]
         assert body["title_overrides"] == {"websocket:sidebar": "Pinned work"}
         assert body["view"]["density"] == "compact"
 
@@ -2845,7 +2854,7 @@ async def test_session_delete_blocks_origin_automation_when_unified_enabled(
 
 
 @pytest.mark.asyncio
-async def test_session_routes_accept_percent_encoded_websocket_keys(
+async def test_session_delete_accepts_percent_encoded_websocket_keys(
     bus: MagicMock, tmp_path: Path
 ) -> None:
     sm = _seed_session(tmp_path, key="websocket:encoded-key")
@@ -2855,13 +2864,6 @@ async def test_session_routes_accept_percent_encoded_websocket_keys(
         token = channel.gateway.tokens.issue_api_token(300)
         auth = {"Authorization": f"Bearer {token}"}
 
-        msgs = await _http_get(
-            "http://127.0.0.1:29910/api/sessions/websocket%3Aencoded-key/messages",
-            headers=auth,
-        )
-        assert msgs.status_code == 200
-        assert msgs.json()["key"] == "websocket:encoded-key"
-
         path = sm._get_session_path("websocket:encoded-key")
         assert path.exists()
         deleted = await _http_get(
@@ -2876,41 +2878,6 @@ async def test_session_routes_accept_percent_encoded_websocket_keys(
         await server_task
 
 
-@pytest.mark.asyncio
-async def test_session_messages_hide_persisted_runtime_context(
-    bus: MagicMock, tmp_path: Path
-) -> None:
-    sm = SessionManager(tmp_path)
-    session = sm.get_or_create("websocket:runtime-context")
-    content, marker = append_runtime_context(
-        "visible user text",
-        [RuntimeContextBlock(source="goal", content="private goal context")],
-    )
-    session.add_message(
-        "user",
-        content,
-        **{RUNTIME_CONTEXT_HISTORY_META: marker},
-    )
-    sm.save(session)
-    channel = _ch(bus, session_manager=sm, port=29919)
-    server_task = asyncio.create_task(channel.start())
-    try:
-        token = channel.gateway.tokens.issue_api_token(300)
-        response = await _http_get(
-            "http://127.0.0.1:29919/api/sessions/websocket:runtime-context/messages",
-            headers={"Authorization": f"Bearer {token}"},
-        )
-
-        assert response.status_code == 200
-        message = response.json()["messages"][0]
-        assert message["content"] == "visible user text"
-        assert RUNTIME_CONTEXT_HISTORY_META not in message
-        assert "private goal context" not in response.text
-    finally:
-        await channel.stop()
-        await server_task
-
-
 @pytest.mark.asyncio
 async def test_webui_thread_resigns_assistant_media_urls(
     bus: MagicMock, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
@@ -3114,7 +3081,7 @@ async def test_webui_thread_negotiates_gzip_for_large_payloads(
 
 
 @pytest.mark.asyncio
-async def test_session_routes_reject_non_websocket_keys(
+async def test_session_delete_rejects_non_websocket_keys(
     bus: MagicMock, tmp_path: Path
 ) -> None:
     sm = _seed_many(
@@ -3131,14 +3098,6 @@ async def test_session_routes_reject_non_websocket_keys(
         token = channel.gateway.tokens.issue_api_token(300)
         auth = {"Authorization": f"Bearer {token}"}
 
-        # The webui list already hides non-websocket sessions; handcrafted URLs
-        # should hit the same boundary rather than exposing or deleting them.
-        msgs = await _http_get(
-            "http://127.0.0.1:29909/api/sessions/cli:direct/messages",
-            headers=auth,
-        )
-        assert msgs.status_code == 404
-
         doomed = sm._get_session_path("slack:C123")
         assert doomed.exists()
         deny_delete = await _http_get(
@@ -3153,7 +3112,7 @@ async def test_session_routes_reject_non_websocket_keys(
 
 
 @pytest.mark.asyncio
-async def test_session_routes_reject_invalid_key(
+async def test_session_delete_rejects_invalid_key(
     bus: MagicMock, tmp_path: Path
 ) -> None:
     sm = _seed_session(tmp_path)
@@ -3166,7 +3125,7 @@ async def test_session_routes_reject_invalid_key(
         # Invalid characters in the key -> regex match fails -> 404
         # (route doesn't match, falls through to channel 404).
         resp = await _http_get(
-            "http://127.0.0.1:29904/api/sessions/bad%20key/messages",
+            "http://127.0.0.1:29904/api/sessions/bad%20key/delete",
             headers=auth,
         )
         assert resp.status_code in {400, 404}

nanobot/channels/websocket/tests/test_websocket_media_route.py

@@ -1,11 +1,8 @@
-"""Tests for the signed ``/api/media/<sig>/<payload>`` route and its replay
+"""Tests for the signed ``/api/media/<sig>/<payload>`` route and WebUI replay.
-integration on ``/api/sessions/<key>/messages``.
 
-The route is the return path for images attached to persisted user turns:
+The route is the return path for local media rendered by the WebUI. These tests
-:meth:`WebSocketChannel.gateway.media.sign_media_path` mints URLs during session reads,
+cover URL signing and serving end-to-end plus the adversarial edges (bad
-and :meth:`GatewayHTTPHandler._handle_media_fetch` serves the bytes back.
+signatures, ``..`` traversal, non-existent files, non-image types).
-These tests cover the two halves end-to-end plus the adversarial edges
-(bad signatures, ``..`` traversal, non-existent files, non-image types).
 """
 
 from __future__ import annotations
@@ -20,11 +17,12 @@ from unittest.mock import AsyncMock, MagicMock, patch
 import pytest
 
 from nanobot.channels.websocket.runtime import WebSocketChannel, WebSocketConfig
-from nanobot.session.manager import Session, SessionManager
+from nanobot.session.manager import SessionManager
 from nanobot.webui.gateway_services import build_gateway_services
 from nanobot.webui.media_api import (
     b64url_decode,
     b64url_encode,
+    sign_media_path,
 )
 
 from .ws_test_client import InProcessHttpChannel
@@ -87,8 +85,16 @@ def _fake_media_dir(root: Path):
     return inner
 
 
+def _sign_media_path(channel: WebSocketChannel, path: Path) -> str | None:
+    return sign_media_path(
+        path,
+        secret=channel.gateway.media.secret,
+        media_dir=channel.gateway.media._media_dir,
+    )
+
+
 # ---------------------------------------------------------------------------
-# gateway.media.sign_media_path: the URL minter
+# media_api.sign_media_path: the URL minter
 # ---------------------------------------------------------------------------
 
 
@@ -108,10 +114,10 @@ def test_sign_media_path_rejects_paths_outside_media_root(
     media.mkdir()
     channel = _ch(bus, port=0)
     with patch("nanobot.webui.media_gateway.get_media_dir", return_value=media):
-        assert channel.gateway.media.sign_media_path(outside) is None
+        assert _sign_media_path(channel, outside) is None
         # Traversal via the media root is also rejected — the resolve() step
         # normalises ``..`` out before the relative_to check.
-        assert channel.gateway.media.sign_media_path(media / ".." / "secrets" / "cred.txt") is None
+        assert _sign_media_path(channel, media / ".." / "secrets" / "cred.txt") is None
 
 
 def test_sign_media_path_round_trips_via_hmac(
@@ -123,7 +129,7 @@ def test_sign_media_path_round_trips_via_hmac(
     (media / "a.png").write_bytes(_PNG_BYTES)
     channel = _ch(bus, port=0)
     with patch("nanobot.webui.media_gateway.get_media_dir", return_value=media):
-        url = channel.gateway.media.sign_media_path(media / "a.png")
+        url = _sign_media_path(channel, media / "a.png")
     assert url is not None
     assert url.startswith("/api/media/")
     sig, payload = url[len("/api/media/"):].split("/", 1)
@@ -238,7 +244,7 @@ async def test_media_route_serves_signed_file(
 
     channel = _ch(bus, port=29920)
     with patch("nanobot.webui.media_gateway.get_media_dir", return_value=media):
-        url_path = channel.gateway.media.sign_media_path(target)
+        url_path = _sign_media_path(channel, target)
         assert url_path is not None
         server_task = asyncio.create_task(channel.start())
         try:
@@ -270,7 +276,7 @@ async def test_media_route_serves_video_byte_ranges(
 
     channel = _ch(bus, port=29927)
     with patch("nanobot.webui.media_gateway.get_media_dir", return_value=media):
-        url_path = channel.gateway.media.sign_media_path(target)
+        url_path = _sign_media_path(channel, target)
         assert url_path is not None
         server_task = asyncio.create_task(channel.start())
         try:
@@ -301,7 +307,7 @@ async def test_media_route_serves_suffix_video_byte_ranges(
 
     channel = _ch(bus, port=29928)
     with patch("nanobot.webui.media_gateway.get_media_dir", return_value=media):
-        url_path = channel.gateway.media.sign_media_path(target)
+        url_path = _sign_media_path(channel, target)
         assert url_path is not None
         server_task = asyncio.create_task(channel.start())
         try:
@@ -329,7 +335,7 @@ async def test_media_route_rejects_unsatisfiable_byte_range(
 
     channel = _ch(bus, port=29929)
     with patch("nanobot.webui.media_gateway.get_media_dir", return_value=media):
-        url_path = channel.gateway.media.sign_media_path(target)
+        url_path = _sign_media_path(channel, target)
         assert url_path is not None
         server_task = asyncio.create_task(channel.start())
         try:
@@ -361,7 +367,7 @@ async def test_media_route_rejects_bad_signature(
 
     channel = _ch(bus, port=29921)
     with patch("nanobot.webui.media_gateway.get_media_dir", return_value=media):
-        good = channel.gateway.media.sign_media_path(media / "f.png")
+        good = _sign_media_path(channel, media / "f.png")
         assert good is not None
         _, payload = good[len("/api/media/"):].split("/", 1)
         # Forge a sig with a *different* secret.
@@ -426,7 +432,7 @@ async def test_media_route_404s_missing_file(
 
     channel = _ch(bus, port=29923)
     with patch("nanobot.webui.media_gateway.get_media_dir", return_value=media):
-        url_path = channel.gateway.media.sign_media_path(target)
+        url_path = _sign_media_path(channel, target)
         assert url_path is not None
         target.unlink()  # the file vanishes between signing and fetching
         server_task = asyncio.create_task(channel.start())
@@ -483,7 +489,7 @@ async def test_media_route_serves_svg_with_strict_csp(
 
     channel = _ch(bus, port=29928)
     with patch("nanobot.webui.media_gateway.get_media_dir", return_value=media):
-        url_path = channel.gateway.media.sign_media_path(target)
+        url_path = _sign_media_path(channel, target)
         assert url_path is not None
         server_task = asyncio.create_task(channel.start())
         try:
@@ -497,91 +503,3 @@ async def test_media_route_serves_svg_with_strict_csp(
     assert resp.headers.get("x-content-type-options") == "nosniff"
     assert "default-src 'none'" in resp.headers.get("content-security-policy", "")
     assert "sandbox" in resp.headers.get("content-security-policy", "")
-
-
-# ---------------------------------------------------------------------------
-# /api/sessions/<key>/messages: media_urls hydration on session read
-# ---------------------------------------------------------------------------
-
-
-@pytest.mark.asyncio
-async def test_session_messages_exposes_signed_media_urls(
-    bus: MagicMock, tmp_path: Path
-) -> None:
-    """The read path must map persisted ``media`` paths onto signed URLs
-    and strip the raw path — the client never learns the server's layout."""
-    media = tmp_path / "media"
-    media.mkdir()
-    img = media / "u.png"
-    img.write_bytes(_PNG_BYTES)
-
-    sm = SessionManager(tmp_path / "ws_state")
-    sess = Session(key="websocket:media-hydrate")
-    sess.add_message("user", "look at this", media=[str(img)])
-    sess.add_message("assistant", "nice")
-    sm.save(sess)
-
-    channel = _ch(bus, session_manager=sm, port=29925)
-    with patch("nanobot.webui.media_gateway.get_media_dir", return_value=media):
-        server_task = asyncio.create_task(channel.start())
-        try:
-            token = channel.gateway.tokens.issue_api_token(300)
-            auth = {"Authorization": f"Bearer {token}"}
-            resp = await _http_get(
-                "http://127.0.0.1:29925/api/sessions/websocket:media-hydrate/messages",
-                headers=auth,
-            )
-            body = resp.json()
-            # The signed URL round-trips end-to-end: fetching it yields the same bytes.
-            user_msg = next(m for m in body["messages"] if m["role"] == "user")
-            urls = user_msg["media_urls"]
-            assert isinstance(urls, list) and len(urls) == 1
-            assert urls[0]["name"] == "u.png"
-            assert urls[0]["url"].startswith("/api/media/")
-            # Raw paths must not leak to the wire.
-            assert "media" not in user_msg
-
-            # And the URL actually works.
-            fetched = await _http_get(f"http://127.0.0.1:29925{urls[0]['url']}")
-            assert fetched.status_code == 200
-            assert fetched.content == _PNG_BYTES
-        finally:
-            await channel.stop()
-            await server_task
-
-
-@pytest.mark.asyncio
-async def test_session_messages_skips_vanished_media(
-    bus: MagicMock, tmp_path: Path
-) -> None:
-    """Paths that no longer resolve inside the media root produce no URL —
-    the message is still delivered, just without the preview."""
-    media = tmp_path / "media"
-    media.mkdir()
-
-    sm = SessionManager(tmp_path / "ws_state")
-    sess = Session(key="websocket:vanished")
-    sess.add_message("user", "missing pic", media=[str(media / "absent.png")])
-    sm.save(sess)
-
-    channel = _ch(bus, session_manager=sm, port=29926)
-    with patch("nanobot.webui.media_gateway.get_media_dir", return_value=media):
-        server_task = asyncio.create_task(channel.start())
-        try:
-            token = channel.gateway.tokens.issue_api_token(300)
-            resp = await _http_get(
-                "http://127.0.0.1:29926/api/sessions/websocket:vanished/messages",
-                headers={"Authorization": f"Bearer {token}"},
-            )
-            user_msg = next(m for m in resp.json()["messages"] if m["role"] == "user")
-            # absent.png lives inside the media root so it *does* get a signed
-            # URL (we don't stat the file at signing time — that would slow
-            # the listing). Fetching the URL is where the 404 surfaces.
-            urls = user_msg.get("media_urls") or []
-            assert len(urls) == 1
-            fetched = await _http_get(f"http://127.0.0.1:29926{urls[0]['url']}")
-            assert fetched.status_code == 404
-            assert "media" not in user_msg
-        finally:
-            await channel.stop()
-            await server_task

nanobot/channels/weixin/connect.py

@@ -47,7 +47,10 @@ class WeixinConnectStore:
         if not session_id:
             raise ChannelConnectError("missing WeChat connect session")
         if action == "poll":
-            return await self.poll(session_id)
+            return await self.poll(
+                session_id,
+                verify_code=(query_first(query, "verify_code") or "").strip(),
+            )
         if action == "cancel":
             return await self.cancel(session_id)
         raise ChannelConnectError(f"unsupported WeChat connect action: {action}", status=404)
@@ -91,7 +94,7 @@ class WeixinConnectStore:
         )
         return self._start_payload(self._sessions[session_id])
 
-    async def poll(self, session_id: str) -> dict[str, Any]:
+    async def poll(self, session_id: str, *, verify_code: str = "") -> dict[str, Any]:
         await self._cleanup()
         session = self._sessions.get(session_id)
         if session is None:
@@ -105,6 +108,7 @@ class WeixinConnectStore:
             status_data = await session.channel.connect_poll_qr_code(
                 base_url=session.current_poll_base_url,
                 qrcode_id=session.qrcode_id,
+                verify_code=verify_code,
             )
         except Exception as exc:
             if session.channel.connect_poll_error_is_retryable(exc):
@@ -120,6 +124,8 @@ class WeixinConnectStore:
 
         status_payload = status_data
         status = status_payload.get("status", "")
+        from nanobot.channels.weixin.runtime import MAX_QR_REFRESH_COUNT
+
         if status == "confirmed":
             if self._sessions.get(session_id) is not session:
                 return {
@@ -157,9 +163,66 @@ class WeixinConnectStore:
                 )
             return self._pending_payload(session)
 
-        if status == "expired":
+        if status == "need_verifycode":
-            from nanobot.channels.weixin.runtime import MAX_QR_REFRESH_COUNT
+            return self._pending_payload(
+                session,
+                challenge="verify_code",
+                message=(
+                    "That verification code did not match. Enter the new number shown in WeChat."
+                    if verify_code
+                    else "Enter the number shown in WeChat to continue."
+                ),
+                verification_failed=bool(verify_code),
+            )
 
+        if status == "verify_code_blocked":
+            session.refresh_count += 1
+            if session.refresh_count > MAX_QR_REFRESH_COUNT:
+                self._sessions.pop(session_id, None)
+                await self._close_channel(session.channel)
+                return {
+                    "session_id": session_id,
+                    "status": "failed",
+                    "message": "Too many incorrect verification attempts. Try again later.",
+                }
+            try:
+                session.qrcode_id, session.qr_url = (
+                    await session.channel.connect_fetch_qr_code()
+                )
+            except Exception as exc:
+                self._sessions.pop(session_id, None)
+                await self._close_channel(session.channel)
+                return {
+                    "session_id": session_id,
+                    "status": "failed",
+                    "message": f"Could not refresh WeChat QR code: {exc}",
+                }
+            session.current_poll_base_url = session.channel.connect_base_url
+            return self._pending_payload(
+                session,
+                message="Verification was blocked. Scan the refreshed QR code to try again.",
+            )
+
+        if status == "binded_redirect":
+            if not session.channel.connect_load_state():
+                self._sessions.pop(session_id, None)
+                await self._close_channel(session.channel)
+                return {
+                    "session_id": session_id,
+                    "status": "failed",
+                    "message": (
+                        "WeChat reports an existing binding, but no local credentials were found."
+                    ),
+                }
+            self._sessions.pop(session_id, None)
+            await self._close_channel(session.channel)
+            return {
+                "session_id": session_id,
+                "status": "succeeded",
+                "message": "WeChat is already connected to this nanobot instance.",
+            }
+
+        if status == "expired":
             session.refresh_count += 1
             if session.refresh_count > MAX_QR_REFRESH_COUNT:
                 self._sessions.pop(session_id, None)
@@ -238,15 +301,25 @@ class WeixinConnectStore:
         }
 
     @staticmethod
-    def _pending_payload(session: WeixinConnectSession) -> dict[str, Any]:
+    def _pending_payload(
-        return {
+        session: WeixinConnectSession,
+        *,
+        challenge: str = "",
+        message: str = "Waiting for WeChat scan.",
+        verification_failed: bool = False,
+    ) -> dict[str, Any]:
+        payload: dict[str, Any] = {
             "session_id": session.id,
             "status": "pending",
             "qr_url": session.qr_url,
             "interval_ms": 2000,
             "expires_at_ms": int((session.created_wall + 600) * 1000),
-            "message": "Waiting for WeChat scan.",
+            "message": message,
         }
+        if challenge:
+            payload["challenge"] = challenge
+            payload["verification_failed"] = verification_failed
+        return payload
 
 
 __all__ = ["WeixinConnectStore"]

nanobot/channels/weixin/manifest.py

@@ -10,6 +10,20 @@ SETUP_SPEC = ChannelSetupSpec(
     fields={
         "token": field("secret"),
         "allowFrom": field("list"),
+        "baseUrl": field(default="https://ilinkai.weixin.qq.com"),
+        "cdnBaseUrl": field(default="https://novac2c.cdn.weixin.qq.com/c2c"),
+        "routeTag": field(),
+        "stateDir": field(),
+        "pollTimeout": field("int", default=35),
+        "sendProgress": field("bool", default=False),
+        "sendToolHints": field("bool", default=False),
+        "replyProgressMessages": field("bool", default=False),
+        "replyProgressMaxMessages": field("int", default=2),
+        "contextMessageBudget": field("int", default=8),
+        "streaming": field("bool", default=True),
+        "blockStreaming": field("bool", default=False),
+        "blockStreamingMinChars": field("int", default=1200),
+        "blockStreamingMaxMessages": field("int", default=3),
     },
     required=(required("token"),),
     official_url="https://weixin.qq.com/",

nanobot/channels/weixin/tests/test_connect.py

@@ -147,3 +147,129 @@ async def test_weixin_cancel_wins_over_inflight_confirmation(
     assert cancelled["status"] == "cancelled"
     assert completed["status"] == "cancelled"
     assert not (state_dir / "account.json").exists()
+
+
+@pytest.mark.asyncio
+async def test_weixin_connect_store_handles_verification_code(
+    tmp_path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    state_dir = tmp_path / "weixin-state"
+    config_path = tmp_path / "config.json"
+    save_config(
+        Config.model_validate({"channels": {"weixin": {"stateDir": str(state_dir)}}}),
+        config_path,
+    )
+    monkeypatch.setattr("nanobot.config.loader._current_config_path", config_path)
+
+    async def fake_fetch_qr_code(self: WeixinChannel) -> tuple[str, str]:
+        return "qr-verify", "https://qr.example/verify"
+
+    responses = [
+        {"status": "need_verifycode"},
+        {
+            "status": "confirmed",
+            "bot_token": "verified-token",
+            "ilink_user_id": "wx-user",
+        },
+    ]
+
+    async def fake_api_get_with_base(
+        self: WeixinChannel,
+        *,
+        params: dict[str, Any],
+        **_kwargs: Any,
+    ) -> dict[str, str]:
+        if len(responses) == 1:
+            assert params == {"qrcode": "qr-verify", "verify_code": "1234"}
+        return responses.pop(0)
+
+    monkeypatch.setattr(WeixinChannel, "_fetch_qr_code", fake_fetch_qr_code)
+    monkeypatch.setattr(WeixinChannel, "_api_get_with_base", fake_api_get_with_base)
+
+    store = WeixinConnectStore()
+    started = await store.start()
+    challenged = await store.poll(started["session_id"])
+    completed = await store.handle(
+        "poll",
+        {
+            "session_id": [started["session_id"]],
+            "verify_code": ["1234"],
+        },
+    )
+
+    assert challenged["status"] == "pending"
+    assert challenged["challenge"] == "verify_code"
+    assert completed["status"] == "succeeded"
+
+
+@pytest.mark.asyncio
+async def test_weixin_connect_store_treats_existing_binding_as_success(
+    tmp_path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    state_dir = tmp_path / "weixin-state"
+    state_dir.mkdir()
+    (state_dir / "account.json").write_text(
+        json.dumps({"token": "working-token"}),
+        encoding="utf-8",
+    )
+    config_path = tmp_path / "config.json"
+    save_config(
+        Config.model_validate({"channels": {"weixin": {"stateDir": str(state_dir)}}}),
+        config_path,
+    )
+    monkeypatch.setattr("nanobot.config.loader._current_config_path", config_path)
+
+    async def fake_fetch_qr_code(self: WeixinChannel) -> tuple[str, str]:
+        return "qr-existing", "https://qr.example/existing"
+
+    async def fake_api_get_with_base(
+        self: WeixinChannel,
+        **_kwargs: Any,
+    ) -> dict[str, str]:
+        return {"status": "binded_redirect"}
+
+    monkeypatch.setattr(WeixinChannel, "_fetch_qr_code", fake_fetch_qr_code)
+    monkeypatch.setattr(WeixinChannel, "_api_get_with_base", fake_api_get_with_base)
+
+    store = WeixinConnectStore()
+    started = await store.start(force=True)
+    completed = await store.poll(started["session_id"])
+
+    assert completed["status"] == "succeeded"
+    assert "already connected" in completed["message"]
+    assert json.loads((state_dir / "account.json").read_text())["token"] == "working-token"
+
+
+@pytest.mark.asyncio
+async def test_weixin_connect_store_rejects_existing_binding_without_local_credentials(
+    tmp_path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    state_dir = tmp_path / "weixin-state"
+    config_path = tmp_path / "config.json"
+    save_config(
+        Config.model_validate({"channels": {"weixin": {"stateDir": str(state_dir)}}}),
+        config_path,
+    )
+    monkeypatch.setattr("nanobot.config.loader._current_config_path", config_path)
+
+    async def fake_fetch_qr_code(self: WeixinChannel) -> tuple[str, str]:
+        return "qr-missing", "https://qr.example/missing"
+
+    async def fake_api_get_with_base(
+        self: WeixinChannel,
+        **_kwargs: Any,
+    ) -> dict[str, str]:
+        return {"status": "binded_redirect"}
+
+    monkeypatch.setattr(WeixinChannel, "_fetch_qr_code", fake_fetch_qr_code)
+    monkeypatch.setattr(WeixinChannel, "_api_get_with_base", fake_api_get_with_base)
+
+    store = WeixinConnectStore()
+    started = await store.start(force=True)
+    completed = await store.poll(started["session_id"])
+
+    assert completed["status"] == "failed"
+    assert "no local credentials" in completed["message"]

nanobot/channels/weixin/tests/test_weixin_channel.py

@@ -17,6 +17,7 @@ from nanobot.channels.weixin.runtime import (
     ITEM_TEXT,
     MESSAGE_TYPE_BOT,
     WEIXIN_CHANNEL_VERSION,
+    WeixinAuthError,
     WeixinChannel,
     WeixinConfig,
     _decrypt_aes_ecb,
@@ -67,11 +68,11 @@ def test_make_headers_includes_route_tag_when_configured() -> None:
     assert headers["Authorization"] == "Bearer token"
     assert headers["SKRouteTag"] == "123"
     assert headers["iLink-App-Id"] == "bot"
-    assert headers["iLink-App-ClientVersion"] == str((2 << 16) | (1 << 8) | 1)
+    assert headers["iLink-App-ClientVersion"] == str((2 << 16) | (4 << 8) | 6)
 
 
 def test_channel_version_matches_reference_plugin_version() -> None:
-    assert WEIXIN_CHANNEL_VERSION == "2.1.1"
+    assert WEIXIN_CHANNEL_VERSION == "2.4.6"
 
 
 def test_save_and_load_state_persists_context_tokens(tmp_path) -> None:
@@ -159,6 +160,29 @@ def test_save_state_persists_explicit_config_token_over_stale_state(tmp_path) ->
     assert saved["get_updates_buf"] == "current-cursor"
 
 
+def test_save_state_preserves_qr_replacement_of_configured_token(tmp_path) -> None:
+    config = WeixinConfig(
+        enabled=True,
+        allow_from=["*"],
+        token="configured-token",
+        state_dir=str(tmp_path),
+    )
+    old_runtime = WeixinChannel(config, MessageBus())
+    old_runtime._token = "configured-token"
+
+    replacement = WeixinChannel(config, MessageBus())
+    replacement.connect_commit_account(
+        token="replacement-token",
+        base_url="https://new.example",
+    )
+
+    old_runtime._save_state()
+
+    saved = json.loads((tmp_path / "account.json").read_text())
+    assert saved["token"] == "replacement-token"
+    assert saved["base_url"] == "https://new.example"
+
+
 def test_save_state_with_empty_runtime_token_preserves_persisted_account(tmp_path) -> None:
     channel = WeixinChannel(
         WeixinConfig(enabled=True, allow_from=["*"], state_dir=str(tmp_path)),
@@ -442,15 +466,15 @@ async def test_send_without_context_token_raises() -> None:
 
 
 @pytest.mark.asyncio
-async def test_send_raises_when_session_is_paused() -> None:
+async def test_send_raises_when_authentication_is_required() -> None:
     channel, _bus = _make_channel()
     channel._client = object()
     channel._token = "token"
     channel._context_tokens["wx-user"] = "ctx-2"
-    channel._pause_session(60)
+    channel._auth_required = True
     channel._send_text = AsyncMock()
 
-    with pytest.raises(RuntimeError, match="session paused"):
+    with pytest.raises(WeixinAuthError, match="bot token is stale"):
         await channel.send(
             type("Msg", (), {"chat_id": "wx-user", "content": "pong", "media": [], "metadata": {}})()
         )
@@ -525,20 +549,21 @@ async def test_send_still_sends_text_when_typing_ticket_missing() -> None:
 
 
 @pytest.mark.asyncio
-async def test_poll_once_pauses_session_on_expired_errcode() -> None:
+async def test_poll_once_requires_login_on_stale_token() -> None:
     channel, _bus = _make_channel()
     channel._client = SimpleNamespace(timeout=None)
     channel._token = "token"
     channel._api_post = AsyncMock(return_value={"ret": 0, "errcode": -14, "errmsg": "expired"})
 
+    with pytest.raises(WeixinAuthError, match="no replacement credentials"):
         await channel._poll_once()
 
-    assert channel._session_pause_remaining_s() > 0
+    assert channel._auth_required is True
 
 
 @pytest.mark.asyncio
-async def test_poll_once_reloads_refreshed_state_after_session_pause(
+async def test_poll_once_reloads_refreshed_state_after_stale_token(
-    tmp_path, monkeypatch: pytest.MonkeyPatch
+    tmp_path,
 ) -> None:
     channel = WeixinChannel(
         WeixinConfig(enabled=True, allow_from=["*"], state_dir=str(tmp_path)),
@@ -550,8 +575,13 @@ async def test_poll_once_reloads_refreshed_state_after_session_pause(
         json.dumps({"token": "new-token", "base_url": "https://new.example"}),
         encoding="utf-8",
     )
-    channel._session_pause_until = time.time() + 10
+    channel._client = object()
-    monkeypatch.setattr(weixin_mod.asyncio, "sleep", AsyncMock())
+    channel._api_post = AsyncMock(
+        side_effect=[
+            {"ret": 0, "errcode": -14, "errmsg": "stale"},
+            {"ret": 0},
+        ]
+    )
 
     await channel._poll_once()
 
@@ -560,8 +590,8 @@ async def test_poll_once_reloads_refreshed_state_after_session_pause(
 
 
 @pytest.mark.asyncio
-async def test_poll_once_keeps_explicit_token_after_session_pause(
+async def test_poll_once_keeps_explicit_token_and_requires_login(
-    tmp_path, monkeypatch: pytest.MonkeyPatch
+    tmp_path,
 ) -> None:
     channel = WeixinChannel(
         WeixinConfig(
@@ -577,24 +607,132 @@ async def test_poll_once_keeps_explicit_token_after_session_pause(
         json.dumps({"token": "stale-token", "base_url": "https://stale.example"}),
         encoding="utf-8",
     )
-    channel._session_pause_until = time.time() + 10
+    channel._client = object()
-    monkeypatch.setattr(weixin_mod.asyncio, "sleep", AsyncMock())
+    channel._api_post = AsyncMock(
+        return_value={"ret": 0, "errcode": -14, "errmsg": "stale"}
+    )
 
+    with pytest.raises(WeixinAuthError, match="no replacement credentials"):
         await channel._poll_once()
 
     assert channel._token == "configured-token"
     assert channel.config.base_url == "https://ilinkai.weixin.qq.com"
 
 
+@pytest.mark.asyncio
+async def test_poll_once_loads_qr_replacement_for_configured_token(tmp_path) -> None:
+    config = WeixinConfig(
+        enabled=True,
+        allow_from=["*"],
+        token="configured-token",
+        state_dir=str(tmp_path),
+    )
+    replacement = WeixinChannel(config, MessageBus())
+    replacement.connect_commit_account(
+        token="replacement-token",
+        base_url="https://new.example",
+    )
+
+    channel = WeixinChannel(config, MessageBus())
+    channel._token = "configured-token"
+    channel._client = object()
+    channel._api_post = AsyncMock(
+        side_effect=[
+            {"ret": 0, "errcode": -14, "errmsg": "stale"},
+            {"ret": 0},
+        ]
+    )
+
+    await channel._poll_once()
+
+    assert channel._token == "replacement-token"
+    assert channel.config.base_url == "https://new.example"
+
+
+@pytest.mark.asyncio
+async def test_start_uses_qr_replacement_for_configured_token(tmp_path) -> None:
+    config = WeixinConfig(
+        enabled=True,
+        allow_from=["*"],
+        token="configured-token",
+        state_dir=str(tmp_path),
+    )
+    connector = WeixinChannel(config, MessageBus())
+    connector.connect_commit_account(
+        token="replacement-token",
+        base_url="https://new.example",
+    )
+
+    channel = WeixinChannel(config, MessageBus())
+    observed_tokens: list[str] = []
+
+    async def stop_after_first_poll() -> None:
+        observed_tokens.append(channel._token)
+        channel._running = False
+
+    channel._notify_lifecycle = AsyncMock()  # type: ignore[method-assign]
+    channel._poll_once = stop_after_first_poll  # type: ignore[method-assign]
+
+    await channel.start()
+    await channel.stop()
+
+    assert observed_tokens == ["replacement-token"]
+    assert channel.config.base_url == "https://new.example"
+
+
+@pytest.mark.asyncio
+async def test_manager_surfaces_actionable_weixin_auth_error_without_traceback(
+    tmp_path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    from nanobot.channels import manager as manager_mod
+
+    channel = WeixinChannel(
+        WeixinConfig(enabled=True, allow_from=["*"], state_dir=str(tmp_path)),
+        MessageBus(),
+    )
+    channel.start = AsyncMock(  # type: ignore[method-assign]
+        side_effect=WeixinAuthError(
+            "getupdates",
+            errcode=-14,
+            errmsg="stale",
+        )
+    )
+    errors: list[str] = []
+    tracebacks: list[str] = []
+    monkeypatch.setattr(
+        manager_mod.logger,
+        "error",
+        lambda message, *args: errors.append(message.format(*args)),
+    )
+    monkeypatch.setattr(
+        manager_mod.logger,
+        "exception",
+        lambda message, *args: tracebacks.append(message.format(*args)),
+    )
+    manager = manager_mod.ChannelManager.__new__(manager_mod.ChannelManager)
+    manager._channel_errors = {}
+
+    await manager._start_channel("weixin", channel)
+
+    assert manager._channel_errors["weixin"] == (
+        "WeChat login expired. Scan again to reconnect."
+    )
+    assert errors == [
+        "Failed to start channel weixin: WeChat login expired. Scan again to reconnect."
+    ]
+    assert tracebacks == []
+
+
 @pytest.mark.asyncio
 async def test_qr_login_refreshes_expired_qr_and_then_succeeds(
     no_qr_poll_delay,
 ) -> None:
     channel, _bus = _make_channel()
     channel._running = True
-    channel._save_state = lambda: None
+    channel._save_state = lambda **_kwargs: None
     channel._print_qr_code = lambda url: None
-    channel._api_get = AsyncMock(
+    channel._api_post = AsyncMock(
         side_effect=[
             {"qrcode": "qr-1", "qrcode_img_content": "url-1"},
             {"qrcode": "qr-2", "qrcode_img_content": "url-2"},
@@ -627,7 +765,7 @@ async def test_qr_login_returns_false_after_too_many_expired_qr_codes(
     channel, _bus = _make_channel()
     channel._running = True
     channel._print_qr_code = lambda url: None
-    channel._api_get = AsyncMock(
+    channel._api_post = AsyncMock(
         side_effect=[
             {"qrcode": "qr-1", "qrcode_img_content": "url-1"},
             {"qrcode": "qr-2", "qrcode_img_content": "url-2"},
@@ -655,7 +793,7 @@ async def test_qr_login_switches_polling_base_url_on_redirect_status(
 ) -> None:
     channel, _bus = _make_channel()
     channel._running = True
-    channel._save_state = lambda: None
+    channel._save_state = lambda **_kwargs: None
     channel._print_qr_code = lambda url: None
     channel._fetch_qr_code = AsyncMock(return_value=("qr-1", "url-1"))
 
@@ -689,7 +827,7 @@ async def test_qr_login_redirect_without_host_keeps_current_polling_base_url(
 ) -> None:
     channel, _bus = _make_channel()
     channel._running = True
-    channel._save_state = lambda: None
+    channel._save_state = lambda **_kwargs: None
     channel._print_qr_code = lambda url: None
     channel._fetch_qr_code = AsyncMock(return_value=("qr-1", "url-1"))
 
@@ -723,7 +861,7 @@ async def test_qr_login_resets_redirect_base_url_after_qr_refresh(
 ) -> None:
     channel, _bus = _make_channel()
     channel._running = True
-    channel._save_state = lambda: None
+    channel._save_state = lambda **_kwargs: None
     channel._print_qr_code = lambda url: None
     channel._fetch_qr_code = AsyncMock(side_effect=[("qr-1", "url-1"), ("qr-2", "url-2")])
 
@@ -1015,7 +1153,7 @@ async def test_qr_login_treats_temporary_connect_error_as_wait_and_recovers(
 ) -> None:
     channel, _bus = _make_channel()
     channel._running = True
-    channel._save_state = lambda: None
+    channel._save_state = lambda **_kwargs: None
     channel._print_qr_code = lambda url: None
     channel._fetch_qr_code = AsyncMock(return_value=("qr-1", "url-1"))
 
@@ -1045,7 +1183,7 @@ async def test_qr_login_treats_5xx_gateway_response_error_as_wait_and_recovers(
 ) -> None:
     channel, _bus = _make_channel()
     channel._running = True
-    channel._save_state = lambda: None
+    channel._save_state = lambda **_kwargs: None
     channel._print_qr_code = lambda url: None
     channel._fetch_qr_code = AsyncMock(return_value=("qr-1", "url-1"))
 
@@ -1080,6 +1218,32 @@ def test_decrypt_aes_ecb_strips_valid_pkcs7_padding() -> None:
     assert decrypted == plaintext
 
 
+def test_missing_aes_dependency_recommends_weixin_plugin(monkeypatch) -> None:
+    real_import = __import__
+
+    def fake_import(name, *args, **kwargs):
+        if name.startswith(("Crypto", "cryptography")):
+            raise ImportError("missing AES dependency")
+        return real_import(name, *args, **kwargs)
+
+    warnings: list[str] = []
+    monkeypatch.setattr("builtins.__import__", fake_import)
+    monkeypatch.setattr(
+        weixin_mod.logger,
+        "warning",
+        lambda message, *args: warnings.append(message.format(*args)),
+    )
+    key_b64 = "MDEyMzQ1Njc4OWFiY2RlZg=="
+    data = b"unencrypted media"
+
+    assert _encrypt_aes_ecb(data, key_b64) == data
+    assert _decrypt_aes_ecb(data, key_b64) == data
+    assert warnings == [
+        "Cannot encrypt media. Run `nanobot plugins enable weixin` to install WeChat support.",
+        "Cannot decrypt media. Run `nanobot plugins enable weixin` to install WeChat support.",
+    ]
+
+
 class _DummyDownloadResponse:
     def __init__(self, content: bytes, status_code: int = 200) -> None:
         self.content = content
@@ -1412,7 +1576,7 @@ async def test_send_text_raises_on_api_error() -> None:
         return_value={"errcode": -14, "errmsg": "session expired"}
     )
 
-    with pytest.raises(RuntimeError, match="WeChat send text error.*-14"):
+    with pytest.raises(WeixinAuthError, match="WeChat sendmessage failed.*errcode=-14"):
         await channel._send_text("wx-user", "hello", "ctx-expired")
 
     channel._api_post.assert_awaited_once()
@@ -1445,7 +1609,7 @@ async def test_send_text_raises_on_nonzero_ret_even_when_errcode_zero() -> None:
         return_value={"ret": -100, "errcode": 0, "errmsg": "internal error"}
     )
 
-    with pytest.raises(RuntimeError, match="WeChat send text error.*ret=-100.*errcode=0"):
+    with pytest.raises(RuntimeError, match="WeChat sendmessage failed.*ret=-100.*errcode=0"):
         await channel._send_text("wx-user", "hello", "ctx-ok")
 
     channel._api_post.assert_awaited_once()

nanobot/channels/weixin/tests/test_weixin_hardening.py

@@ -0,0 +1,441 @@
+from __future__ import annotations
+
+import asyncio
+import json
+import time
+from unittest.mock import AsyncMock
+
+import httpx
+import pytest
+
+from nanobot.bus.events import OutboundMessage
+from nanobot.bus.outbound_events import ProgressEvent
+from nanobot.bus.queue import MessageBus
+from nanobot.channels.manager import ChannelManager
+from nanobot.channels.weixin.manifest import SETUP_SPEC
+from nanobot.channels.weixin.runtime import (
+    ITEM_TOOL_CALL_RESULT,
+    ITEM_TOOL_CALL_START,
+    WEIXIN_MAX_MESSAGE_LEN,
+    WeixinAPIError,
+    WeixinAuthError,
+    WeixinChannel,
+    WeixinConfig,
+    WeixinQuotaError,
+    sanitize_weixin_markdown,
+    split_weixin_message,
+)
+from nanobot.config.schema import Config
+
+
+def _channel(**config: object) -> WeixinChannel:
+    return WeixinChannel(
+        WeixinConfig.model_validate(
+            {"enabled": True, "allowFrom": ["*"], **config}
+        ),
+        MessageBus(),
+    )
+
+
+def _ready_channel(**config: object) -> WeixinChannel:
+    channel = _channel(**config)
+    channel._client = object()
+    channel._token = "bot-token"
+    channel._context_tokens["wx-user"] = "ctx-1"
+    channel._context_token_at["wx-user"] = time.time()
+    channel._typing_tickets["wx-user"] = {
+        "ticket": "",
+        "next_fetch_at": time.time() + 3600,
+    }
+    return channel
+
+
+def test_weixin_defaults_protect_context_quota() -> None:
+    config = WeixinConfig()
+
+    assert WEIXIN_MAX_MESSAGE_LEN == 1800
+    assert config.send_progress is False
+    assert config.send_tool_hints is False
+    assert config.reply_progress_messages is False
+    assert config.context_message_budget == 8
+    assert config.block_streaming is False
+
+
+def test_weixin_webui_manifest_covers_runtime_configuration() -> None:
+    runtime_fields = set(WeixinConfig().model_dump(mode="json", by_alias=True))
+
+    assert set(SETUP_SPEC.fields) == runtime_fields - {"enabled"}
+
+
+def test_reply_progress_opt_in_enables_progress_transport() -> None:
+    config = WeixinConfig(reply_progress_messages=True)
+
+    assert config.send_progress is True
+    assert config.send_tool_hints is True
+
+
+@pytest.mark.parametrize(
+    ("section", "send_progress", "send_tool_hints"),
+    [
+        ({"enabled": True}, False, False),
+        ({"enabled": True, "replyProgressMessages": True}, True, True),
+        ({"enabled": True, "sendProgress": True, "sendToolHints": False}, True, False),
+    ],
+)
+def test_channel_manager_preserves_weixin_quota_defaults(
+    section: dict[str, object],
+    send_progress: bool,
+    send_tool_hints: bool,
+) -> None:
+    manager = ChannelManager.__new__(ChannelManager)
+    manager.config = Config.model_validate({"channels": {"weixin": section}})
+    manager.bus = MessageBus()
+
+    channel = manager._build_channel("weixin", WeixinChannel, section)
+
+    assert channel.send_progress is send_progress
+    assert channel.send_tool_hints is send_tool_hints
+
+
+@pytest.mark.asyncio
+async def test_channel_manager_does_not_retry_permanent_weixin_error(monkeypatch) -> None:
+    manager = ChannelManager.__new__(ChannelManager)
+    manager.config = Config.model_validate({"channels": {"sendMaxRetries": 3}})
+    manager.bus = MessageBus()
+    channel = _channel()
+    channel.send = AsyncMock(
+        side_effect=WeixinAPIError(
+            "sendmessage",
+            errcode=-1,
+            errmsg="business rejection",
+            retryable=False,
+        )
+    )
+    sleep = AsyncMock()
+    monkeypatch.setattr("nanobot.channels.manager.asyncio.sleep", sleep)
+
+    await manager._send_with_retry(
+        channel,
+        OutboundMessage(channel="weixin", chat_id="wx-user", content="test"),
+    )
+
+    channel.send.assert_awaited_once()
+    sleep.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_weixin_http_clients_ignore_system_proxy(tmp_path, monkeypatch) -> None:
+    captured: list[dict[str, object]] = []
+
+    class FakeClient:
+        async def aclose(self) -> None:
+            return None
+
+    def make_client(**kwargs: object) -> FakeClient:
+        captured.append(kwargs)
+        return FakeClient()
+
+    monkeypatch.setattr("nanobot.channels.weixin.runtime.httpx.AsyncClient", make_client)
+
+    connect_channel = _channel(stateDir=str(tmp_path / "connect"))
+    connect_channel.connect_open_client()
+    await connect_channel.connect_close_client()
+
+    login_channel = _channel(stateDir=str(tmp_path / "login"))
+    login_channel._qr_login = AsyncMock(return_value=True)
+    assert await login_channel.login() is True
+
+    start_channel = _channel(token="configured-token", stateDir=str(tmp_path / "start"))
+
+    async def stop_after_poll() -> None:
+        start_channel._running = False
+
+    start_channel._notify_lifecycle = AsyncMock()
+    start_channel._poll_once = AsyncMock(side_effect=stop_after_poll)
+    await start_channel.start()
+    await start_channel.stop()
+
+    assert len(captured) == 3
+    assert all(kwargs["trust_env"] is False for kwargs in captured)
+
+
+def test_markdown_sanitizer_preserves_code_and_escapes_bare_angles() -> None:
+    content = "before <tag> `x<y>`\n```python\na<b\n```\n![drop](https://x.test/a.png)"
+
+    sanitized = sanitize_weixin_markdown(content)
+
+    assert "before ＜tag＞" in sanitized
+    assert "`x<y>`" in sanitized
+    assert "a<b" in sanitized
+    assert "![drop]" not in sanitized
+
+
+def test_markdown_split_balances_fences_and_stays_within_limit() -> None:
+    chunks = split_weixin_message("```python\n" + ("x" * 4000) + "\n```")
+
+    assert len(chunks) >= 3
+    assert all(len(chunk) <= WEIXIN_MAX_MESSAGE_LEN for chunk in chunks)
+    assert all(chunk.count("```") % 2 == 0 for chunk in chunks)
+
+
+@pytest.mark.asyncio
+async def test_qr_fetch_posts_known_local_tokens(tmp_path) -> None:
+    state_dir = tmp_path / "weixin"
+    state_dir.mkdir()
+    (state_dir / "account.json").write_text(
+        json.dumps({"token": "persisted-token"}),
+        encoding="utf-8",
+    )
+    channel = _channel(stateDir=str(state_dir))
+    channel._api_post = AsyncMock(
+        return_value={"qrcode": "qr-1", "qrcode_img_content": "https://qr.test/1"}
+    )
+
+    assert await channel._fetch_qr_code() == ("qr-1", "https://qr.test/1")
+    channel._api_post.assert_awaited_once_with(
+        "ilink/bot/get_bot_qrcode?bot_type=3",
+        {"local_token_list": ["persisted-token"]},
+        auth=False,
+        include_base_info=False,
+    )
+
+
+@pytest.mark.asyncio
+async def test_qr_fetch_retries_without_rejected_local_tokens(tmp_path) -> None:
+    state_dir = tmp_path / "weixin"
+    state_dir.mkdir()
+    (state_dir / "account.json").write_text(
+        json.dumps({"token": "invalid-token"}),
+        encoding="utf-8",
+    )
+    channel = _channel(stateDir=str(state_dir))
+    channel._api_post = AsyncMock(
+        side_effect=[
+            {"ret": -3},
+            {"ret": 0, "qrcode": "qr-1", "qrcode_img_content": "https://qr.test/1"},
+        ]
+    )
+
+    assert await channel._fetch_qr_code() == ("qr-1", "https://qr.test/1")
+    assert [call.args[1] for call in channel._api_post.await_args_list] == [
+        {"local_token_list": ["invalid-token"]},
+        {"local_token_list": []},
+    ]
+
+
+@pytest.mark.asyncio
+async def test_qr_fetch_does_not_retry_invalid_request_without_local_tokens(tmp_path) -> None:
+    channel = _channel(stateDir=str(tmp_path / "weixin"))
+    channel._api_post = AsyncMock(return_value={"ret": -3})
+
+    with pytest.raises(WeixinAPIError, match="get_bot_qrcode failed.*ret=-3"):
+        await channel._fetch_qr_code()
+
+    channel._api_post.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_lifecycle_notifications_are_best_effort() -> None:
+    channel = _ready_channel()
+    channel._api_post = AsyncMock(return_value={"ret": 0})
+
+    await channel._notify_lifecycle("start")
+    await channel._notify_lifecycle("stop")
+
+    assert [call.args[0] for call in channel._api_post.await_args_list] == [
+        "ilink/bot/msg/notifystart",
+        "ilink/bot/msg/notifystop",
+    ]
+
+
+def test_business_errors_have_explicit_retry_contracts() -> None:
+    channel = _channel()
+
+    with pytest.raises(WeixinQuotaError) as quota:
+        channel._raise_for_api_error("sendmessage", {"ret": -2})
+    with pytest.raises(WeixinAuthError) as auth:
+        channel._raise_for_api_error("getupdates", {"errcode": -14})
+    with pytest.raises(WeixinAPIError) as rejected:
+        channel._raise_for_api_error("sendmessage", {"ret": -100})
+
+    assert channel.should_retry_send_error(quota.value) is False
+    assert channel.should_retry_send_error(auth.value) is False
+    assert channel.should_retry_send_error(rejected.value) is False
+    assert channel.should_retry_send_error(httpx.ReadTimeout("slow")) is True
+
+    request = httpx.Request("POST", "https://ilinkai.weixin.qq.com/send")
+    for status_code in (408, 425, 429, 503):
+        response = httpx.Response(status_code, request=request)
+        error = httpx.HTTPStatusError(
+            "retryable response",
+            request=request,
+            response=response,
+        )
+        assert channel.should_retry_send_error(error) is True
+
+    rejected_response = httpx.Response(400, request=request)
+    rejected_http = httpx.HTTPStatusError(
+        "bad request",
+        request=request,
+        response=rejected_response,
+    )
+    assert channel.should_retry_send_error(rejected_http) is False
+
+
+def test_error_classification_checks_ret_and_errcode_independently() -> None:
+    channel = _channel()
+
+    with pytest.raises(WeixinQuotaError):
+        channel._raise_for_api_error(
+            "sendmessage",
+            {"ret": -2, "errcode": -100},
+        )
+    with pytest.raises(WeixinAuthError):
+        channel._raise_for_api_error(
+            "getupdates",
+            {"ret": -14, "errcode": -100},
+        )
+
+
+@pytest.mark.asyncio
+async def test_stop_cancels_inflight_long_poll() -> None:
+    channel = _channel(token="configured-token")
+    poll_started = asyncio.Event()
+    poll_cancelled = asyncio.Event()
+
+    class FakeClient:
+        async def aclose(self) -> None:
+            return None
+
+    async def blocking_poll() -> None:
+        poll_started.set()
+        try:
+            await asyncio.Event().wait()
+        except asyncio.CancelledError:
+            poll_cancelled.set()
+            raise
+
+    channel._new_http_client = lambda _timeout: FakeClient()  # type: ignore[method-assign]
+    channel._notify_lifecycle = AsyncMock()
+    channel._poll_once = blocking_poll  # type: ignore[method-assign]
+
+    start_task = asyncio.create_task(channel.start())
+    await asyncio.wait_for(poll_started.wait(), timeout=1)
+    await asyncio.wait_for(channel.stop(), timeout=1)
+    await asyncio.wait_for(start_task, timeout=1)
+
+    assert poll_cancelled.is_set()
+    assert channel._poll_task is None
+
+
+@pytest.mark.asyncio
+async def test_retry_reuses_client_id_and_skips_completed_chunks() -> None:
+    channel = _ready_channel()
+    request = httpx.Request("POST", "https://ilinkai.weixin.qq.com/ilink/bot/sendmessage")
+    channel._api_post = AsyncMock(
+        side_effect=[
+            {"ret": 0},
+            httpx.ReadTimeout("ambiguous timeout", request=request),
+            {"ret": 0},
+        ]
+    )
+    msg = OutboundMessage(
+        channel="weixin",
+        chat_id="wx-user",
+        content="x" * (WEIXIN_MAX_MESSAGE_LEN + 200),
+    )
+
+    with pytest.raises(httpx.ReadTimeout):
+        await channel.send(msg)
+    await channel.send(msg)
+
+    bodies = [call.args[1] for call in channel._api_post.await_args_list]
+    client_ids = [body["msg"]["client_id"] for body in bodies]
+    assert client_ids[0] != client_ids[1]
+    assert client_ids[1] == client_ids[2]
+    assert channel._context_send_counts["ctx-1"] == 2
+
+
+@pytest.mark.asyncio
+async def test_quota_rejection_defers_final_until_fresh_context() -> None:
+    channel = _ready_channel()
+    channel._api_post = AsyncMock(side_effect=[{"ret": -2}, {"ret": 0}])
+    msg = OutboundMessage(
+        channel="weixin",
+        chat_id="wx-user",
+        content="deferred answer",
+    )
+
+    with pytest.raises(WeixinQuotaError):
+        await channel.send(msg)
+    first_client_id = channel._api_post.await_args_list[0].args[1]["msg"]["client_id"]
+    assert "wx-user" in channel._deferred_outbound
+
+    channel._context_tokens["wx-user"] = "ctx-2"
+    channel._context_token_at["wx-user"] = time.time()
+    await channel._retry_deferred_messages("wx-user")
+
+    second_client_id = channel._api_post.await_args_list[1].args[1]["msg"]["client_id"]
+    assert second_client_id == first_client_id
+    assert "wx-user" not in channel._deferred_outbound
+
+
+@pytest.mark.asyncio
+async def test_local_context_budget_stops_before_extra_api_call() -> None:
+    channel = _ready_channel(contextMessageBudget=1)
+    channel._api_post = AsyncMock(return_value={"ret": 0})
+
+    await channel._send_text("wx-user", "one", "ctx-1")
+    with pytest.raises(WeixinQuotaError, match="local safety budget"):
+        await channel._send_text("wx-user", "two", "ctx-1")
+
+    channel._api_post.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_bounded_block_streaming_reserves_one_final_message() -> None:
+    channel = _ready_channel(
+        blockStreaming=True,
+        blockStreamingMinChars=200,
+        blockStreamingMaxMessages=3,
+    )
+    channel._send_text = AsyncMock()
+
+    await channel.send_delta("wx-user", "a" * 250, stream_id="stream-1")
+    await channel.send_delta("wx-user", "b" * 250, stream_id="stream-1")
+    await channel.send_delta("wx-user", "c" * 250, stream_id="stream-1")
+    await channel.send_delta("wx-user", "done", stream_id="stream-1", stream_end=True)
+
+    assert channel._send_text.await_count == 3
+    assert "stream-1" not in channel._stream_buffers
+    assert "stream-1" not in channel._stream_sent_counts
+
+
+@pytest.mark.asyncio
+async def test_structured_progress_is_capped_and_uses_one_run_id() -> None:
+    channel = _ready_channel(
+        replyProgressMessages=True,
+        replyProgressMaxMessages=2,
+    )
+    channel._send_message_item = AsyncMock()
+    events = [
+        {"phase": "start", "call_id": "call-1", "name": "read_file"},
+        {"phase": "end", "call_id": "call-1", "name": "read_file"},
+        {"phase": "start", "call_id": "call-2", "name": "exec"},
+    ]
+
+    await channel.send(
+        OutboundMessage(
+            channel="weixin",
+            chat_id="wx-user",
+            content="read_file",
+            event=ProgressEvent(content="read_file", tool_hint=True, tool_events=events),
+        )
+    )
+
+    assert channel._send_message_item.await_count == 2
+    first = channel._send_message_item.await_args_list[0]
+    second = channel._send_message_item.await_args_list[1]
+    assert first.args[1]["type"] == ITEM_TOOL_CALL_START
+    assert second.args[1]["type"] == ITEM_TOOL_CALL_RESULT
+    assert first.kwargs["run_id"] == second.kwargs["run_id"]

nanobot/channels/weixin/webui/WeixinConnectFlow.tsx

@@ -1,25 +1,148 @@
+import { useState } from "react";
 import { useTranslation } from "react-i18next";
 
-import { channelTranslator } from "@/channel-plugins/i18n";
+import {
+  channelTranslator,
+  type ChannelTranslator,
+} from "@/channel-plugins/i18n";
 import type { ChannelPluginConnectFlowProps } from "@/channel-plugins/types";
-import { ChannelQrConnectFlow } from "@/components/settings/channels/ChannelQrConnectFlow";
+import {
+  ChannelQrConnectFlow,
+  type ChannelQrConnectPendingContext,
+} from "@/components/settings/channels/ChannelQrConnectFlow";
+import { Button } from "@/components/ui/button";
+import { Input } from "@/components/ui/input";
+import type { ChannelConnectPayload } from "@/lib/types";
+
+type WeixinVerificationPayload = ChannelConnectPayload & {
+  challenge: "verify_code";
+  verification_failed?: boolean;
+};
+
+export const WEIXIN_AUTH_EXPIRED_MESSAGE =
+  "WeChat login expired. Scan again to reconnect.";
+
+function isVerificationChallenge(
+  payload: ChannelConnectPayload,
+): payload is WeixinVerificationPayload {
+  return (
+    "challenge" in payload
+    && payload.challenge === "verify_code"
+    && (
+      !("verification_failed" in payload)
+      || typeof payload.verification_failed === "boolean"
+    )
+  );
+}
+
+function weixinConnectMessage(
+  payload: ChannelConnectPayload,
+  tx: ChannelTranslator,
+): string {
+  if (payload.status === "succeeded") {
+    return tx("custom.connected", "WeChat is connected.");
+  }
+  if (payload.status === "expired") {
+    return tx("custom.expired", WEIXIN_AUTH_EXPIRED_MESSAGE);
+  }
+  if (payload.status === "failed") {
+    return payload.message
+      ?? tx("custom.failed", "Unable to connect WeChat. Try again.");
+  }
+  if (payload.status === "cancelled") {
+    return tx("custom.stopped", "WeChat login stopped.");
+  }
+  if (isVerificationChallenge(payload)) {
+    return payload.verification_failed
+      ? tx(
+        "custom.verifyMismatch",
+        "That code did not match. Enter the new number shown in WeChat.",
+      )
+      : tx(
+        "custom.verifyDescription",
+        "Enter the number shown in WeChat to continue.",
+      );
+  }
+  return tx("custom.waiting", "Waiting for WeChat scan...");
+}
 
 export function WeixinConnectFlow({
   token,
+  feature,
   idleLabel,
   connectRequestId,
   onFeaturesUpdate,
 }: ChannelPluginConnectFlowProps) {
   const { t } = useTranslation();
   const tx = channelTranslator(t, "weixin");
+  const [verificationCode, setVerificationCode] = useState("");
+  const authExpired = feature.runtime_error === WEIXIN_AUTH_EXPIRED_MESSAGE;
+  const scanAgainLabel = t("settings.channels.scanAgain", {
+    defaultValue: "Scan again",
+  });
+
+  const renderVerification = ({
+    connect,
+    busy,
+    poll,
+  }: ChannelQrConnectPendingContext) => {
+    if (!isVerificationChallenge(connect)) return null;
+    return (
+      <form
+        className="mt-3 space-y-2"
+        onSubmit={(event) => {
+          event.preventDefault();
+          const code = verificationCode.trim();
+          if (!code) return;
+          void poll({ verify_code: code }).then((payload) => {
+            if (payload && !isVerificationChallenge(payload)) {
+              setVerificationCode("");
+            }
+          });
+        }}
+      >
+        <div className="text-[12px] font-semibold text-foreground">
+          {tx("custom.verifyTitle", "Verification required")}
+        </div>
+        <p className="text-[12px] leading-5 text-muted-foreground">
+          {weixinConnectMessage(connect, tx)}
+        </p>
+        <div className="flex gap-2">
+          <Input
+            value={verificationCode}
+            onChange={(event) => setVerificationCode(event.target.value)}
+            inputMode="numeric"
+            autoComplete="one-time-code"
+            placeholder={tx("custom.verifyPlaceholder", "Code")}
+            className="h-8 max-w-40"
+            aria-invalid={connect.verification_failed || undefined}
+          />
+          <Button
+            type="submit"
+            size="sm"
+            className="h-8 rounded-full px-3 text-[12px] font-semibold"
+            disabled={busy || !verificationCode.trim()}
+          >
+            {tx("custom.verifySubmit", "Verify")}
+          </Button>
+        </div>
+      </form>
+    );
+  };
+
   return (
     <ChannelQrConnectFlow
       token={token}
       channelName="weixin"
-      idleLabel={idleLabel}
+      startOptions={{ force: authExpired }}
+      idleLabel={authExpired ? scanAgainLabel : idleLabel}
       connectRequestId={connectRequestId}
       forceOnRepeat
       onFeaturesUpdate={onFeaturesUpdate}
+      pausePolling={isVerificationChallenge}
+      suppressSucceeded={feature.runtime_status === "failed"}
+      renderPending={renderVerification}
+      resolveMessage={(payload) => weixinConnectMessage(payload, tx)}
       labels={{
         qrAlt: tx("custom.qrAlt", "WeChat login QR code"),
         scanTitle: tx("custom.scanTitle", "Scan with WeChat"),
@@ -31,7 +154,7 @@ export function WeixinConnectFlow({
         connected: tx("custom.connected", "WeChat is connected."),
         stopped: tx("custom.stopped", "WeChat login stopped."),
         connecting: tx("custom.connecting", "Connecting..."),
-        scanAgain: t("settings.channels.scanAgain", { defaultValue: "Scan again" }),
+        scanAgain: scanAgainLabel,
         connect: t("settings.channels.connect", { defaultValue: "Connect" }),
       }}
     />

nanobot/channels/weixin/webui/WeixinPanel.tsx

@@ -0,0 +1,553 @@
+import { useCallback, useEffect, useMemo, useRef, useState, type ReactNode } from "react";
+import { Check, ChevronDown, ExternalLink, Loader2, Plus } from "lucide-react";
+import { useTranslation } from "react-i18next";
+
+import { channelFieldMessageKey, channelTranslator } from "@/channel-plugins/i18n";
+import { channelLocaleMessages } from "@/channel-plugins/locale-registry";
+import type { ChannelPluginPanelProps } from "@/channel-plugins/types";
+import { ToggleButton } from "@/components/settings/ToggleButton";
+import {
+  chatAppGuideUrl,
+  docsUrlWithBase,
+  type ChannelConfigField,
+} from "@/components/settings/channels/catalog";
+import {
+  CredentialForm,
+  channelValuesForSave,
+  defaultChannelFieldValues,
+} from "@/components/settings/channels/CredentialForm";
+import { Button } from "@/components/ui/button";
+import { useLogoFallback } from "@/hooks/useLogoFallback";
+import { normalizeLocale } from "@/i18n/config";
+import { configureChannel } from "@/lib/api";
+import { logoFallbackUrls } from "@/lib/provider-brand";
+import type {
+  ChannelRuntimeStatus,
+  ChannelSetupContractField,
+  NanobotFeatureInfo,
+} from "@/lib/types";
+import { cn } from "@/lib/utils";
+
+import {
+  WEIXIN_AUTH_EXPIRED_MESSAGE,
+  WeixinConnectFlow,
+} from "./WeixinConnectFlow";
+
+export const WEIXIN_PRIMARY_FIELD_KEYS = [
+  "channels.weixin.sendProgress",
+  "channels.weixin.sendToolHints",
+  "channels.weixin.streaming",
+] as const;
+
+export const WEIXIN_ADVANCED_FIELD_KEYS = [
+  "channels.weixin.allowFrom",
+  "channels.weixin.token",
+  "channels.weixin.replyProgressMessages",
+  "channels.weixin.replyProgressMaxMessages",
+  "channels.weixin.contextMessageBudget",
+  "channels.weixin.blockStreaming",
+  "channels.weixin.blockStreamingMinChars",
+  "channels.weixin.blockStreamingMaxMessages",
+  "channels.weixin.baseUrl",
+  "channels.weixin.cdnBaseUrl",
+  "channels.weixin.routeTag",
+  "channels.weixin.stateDir",
+  "channels.weixin.pollTimeout",
+] as const;
+
+export function WeixinPanel({
+  token,
+  feature,
+  actionKey,
+  chatAppsDocsUrl,
+  showBrandLogos,
+  onAction,
+  onFeaturesUpdate,
+}: ChannelPluginPanelProps) {
+  const { t, i18n } = useTranslation();
+  const tx = (key: string, fallback: string) => t(key, { defaultValue: fallback });
+  const channelTx = channelTranslator(t, "weixin");
+  const runtimeError = weixinRuntimeError(feature.runtime_error, channelTx);
+  const displayName = channelTx("displayName", "WeChat");
+  const enabledBusy = actionKey === `enable:${feature.name}`;
+  const disabledBusy = actionKey === `disable:${feature.name}`;
+  const channelBusy = enabledBusy || disabledBusy;
+  const channelChecked =
+    feature.runtime_status === "running" || feature.runtime_status === "starting";
+  const missingSupport = feature.enabled && !feature.installed;
+  const alwaysEnabled = feature.capabilities?.includes("always_enabled") ?? false;
+  const toggleChecked = alwaysEnabled || channelChecked;
+  const channelToggleDisabled =
+    alwaysEnabled
+    || channelBusy
+    || (!feature.install_supported && !feature.installed && !feature.enabled);
+  const [connectRequestId, setConnectRequestId] = useState(0);
+  const [visibleSecrets, setVisibleSecrets] = useState<Record<string, boolean>>({});
+  const [touchedFields, setTouchedFields] = useState<Set<string>>(() => new Set());
+  const [saving, setSaving] = useState(false);
+  const [saveRevision, setSaveRevision] = useState(0);
+  const [attemptedRevision, setAttemptedRevision] = useState(0);
+  const [saveState, setSaveState] = useState<"idle" | "saved">("idle");
+  const [saveError, setSaveError] = useState<string | null>(null);
+  const configValuesKey = JSON.stringify(feature.config_values ?? {});
+  const setupFieldsKey = JSON.stringify(feature.setup?.fields ?? []);
+  const configuredFields = useMemo(
+    () => new Set(feature.configured_fields ?? []),
+    [feature.configured_fields],
+  );
+  const onLabel = tx("settings.values.on", "On");
+  const offLabel = tx("settings.values.off", "Off");
+  const setupFields = weixinSetupFields(
+    feature,
+    i18n.resolvedLanguage ?? i18n.language,
+  );
+  const primaryFields = localizeBooleanFields(setupFields.primary, onLabel, offLabel);
+  const advancedFields = localizeBooleanFields(setupFields.advanced, onLabel, offLabel);
+  const editableFields = [...primaryFields, ...advancedFields];
+  const docsUrl = docsUrlWithBase(chatAppGuideUrl("wechat"), chatAppsDocsUrl)
+    ?? chatAppGuideUrl("wechat");
+  const [fieldValues, setFieldValues] = useState<Record<string, string>>(() =>
+    defaultChannelFieldValues(editableFields, feature.config_values),
+  );
+  const fieldValuesRef = useRef(fieldValues);
+  const touchedFieldsRef = useRef(touchedFields);
+  const editableFieldsRef = useRef(editableFields);
+  const saveContextRef = useRef({
+    token,
+    enabled: feature.enabled,
+    onFeaturesUpdate,
+  });
+  editableFieldsRef.current = editableFields;
+  saveContextRef.current = {
+    token,
+    enabled: feature.enabled,
+    onFeaturesUpdate,
+  };
+
+  useEffect(() => {
+    const nextValues = defaultChannelFieldValues(editableFields, feature.config_values);
+    for (const key of touchedFieldsRef.current) {
+      nextValues[key] = fieldValuesRef.current[key] ?? "";
+    }
+    fieldValuesRef.current = nextValues;
+    setFieldValues(nextValues);
+    setVisibleSecrets({});
+  }, [configValuesKey, setupFieldsKey]);
+
+  useEffect(() => {
+    if (saveState !== "saved") return;
+    const timeout = window.setTimeout(() => setSaveState("idle"), 1500);
+    return () => window.clearTimeout(timeout);
+  }, [saveState]);
+
+  const saveSettings = useCallback(async (
+    values: Record<string, string>,
+    savedFields: Set<string>,
+  ) => {
+    const context = saveContextRef.current;
+    setSaving(true);
+    setSaveError(null);
+    setSaveState("idle");
+    try {
+      const payload = await configureChannel(
+        context.token,
+        "weixin",
+        channelValuesForSave(editableFieldsRef.current, values),
+        { enable: context.enabled },
+      );
+      const remainingFields = new Set(touchedFieldsRef.current);
+      for (const key of savedFields) {
+        if (fieldValuesRef.current[key] === values[key]) remainingFields.delete(key);
+      }
+      touchedFieldsRef.current = remainingFields;
+      setTouchedFields(remainingFields);
+      setSaveState(remainingFields.size ? "idle" : "saved");
+      if (payload.nanobot_features) context.onFeaturesUpdate(payload.nanobot_features);
+    } catch (err) {
+      setSaveError((err as Error).message);
+    } finally {
+      setSaving(false);
+    }
+  }, []);
+
+  useEffect(() => {
+    if (
+      !editableFields.length
+      || !touchedFields.size
+      || saving
+      || saveRevision <= attemptedRevision
+    ) return;
+    const timeout = window.setTimeout(() => {
+      setAttemptedRevision(saveRevision);
+      void saveSettings(
+        { ...fieldValuesRef.current },
+        new Set(touchedFieldsRef.current),
+      );
+    }, 500);
+    return () => window.clearTimeout(timeout);
+  }, [
+    attemptedRevision,
+    editableFields.length,
+    saveRevision,
+    saveSettings,
+    saving,
+    touchedFields.size,
+  ]);
+
+  const setFieldValue = (key: string, value: string) => {
+    if (fieldValuesRef.current[key] === value) return;
+    const nextValues = { ...fieldValuesRef.current, [key]: value };
+    const nextTouchedFields = new Set(touchedFieldsRef.current).add(key);
+    fieldValuesRef.current = nextValues;
+    touchedFieldsRef.current = nextTouchedFields;
+    setFieldValues(nextValues);
+    setTouchedFields(nextTouchedFields);
+    setSaveError(null);
+    setSaveState("idle");
+    setSaveRevision((current) => current + 1);
+  };
+
+  const toggleAriaLabel = t("settings.channels.toggleChannel", {
+    name: displayName,
+    defaultValue: "{{name}} channel",
+  });
+
+  return (
+    <aside className="min-h-full rounded-[20px] bg-settings-surface p-5">
+      <div className="flex items-start justify-between gap-4">
+        <div className="flex min-w-0 items-start gap-3">
+          <WeixinLogo showBrandLogos={showBrandLogos} />
+          <div className="min-w-0 flex-1">
+            <h3 className="truncate text-[18px] font-semibold leading-6 text-foreground">
+              {displayName}
+            </h3>
+            <p className="mt-1 text-[13px] leading-5 text-muted-foreground">
+              {channelTx("description", "Use nanobot from WeChat conversations.")}
+            </p>
+            {missingSupport && feature.install_supported ? (
+              <Button
+                type="button"
+                size="sm"
+                variant="secondary"
+                disabled={enabledBusy}
+                onClick={() => onAction("enable", feature.name)}
+                className="mt-2 h-8 rounded-full px-3 text-[12px] font-semibold"
+              >
+                {enabledBusy ? (
+                  <Loader2 className="mr-1.5 h-3.5 w-3.5 animate-spin" aria-hidden />
+                ) : (
+                  <Plus className="mr-1.5 h-3.5 w-3.5" aria-hidden />
+                )}
+                {tx("settings.nanobotFeatures.installSupport", "Install support")}
+              </Button>
+            ) : null}
+          </div>
+        </div>
+        <div className="flex shrink-0 items-center gap-2 pt-1">
+          <WeixinStatusBadge status={feature.runtime_status}>
+            {weixinStatusLabel(feature, tx)}
+          </WeixinStatusBadge>
+          {channelBusy ? (
+            <Loader2 className="h-3.5 w-3.5 animate-spin text-muted-foreground" aria-hidden />
+          ) : null}
+          <ToggleButton
+            checked={toggleChecked}
+            disabled={channelToggleDisabled}
+            ariaLabel={toggleAriaLabel}
+            label={toggleChecked ? onLabel : offLabel}
+            onChange={(checked) => {
+              if (checked && !channelChecked && feature.configured === false) {
+                setConnectRequestId((current) => current + 1);
+                return;
+              }
+              onAction(checked ? "enable" : "disable", feature.name);
+            }}
+          />
+        </div>
+      </div>
+
+      {runtimeError ? (
+        <div className="mt-4 rounded-[12px] border border-destructive/20 bg-destructive/5 px-3 py-2 text-[12px] leading-5 text-destructive">
+          {runtimeError}
+        </div>
+      ) : null}
+
+      <div className="mt-4 space-y-4">
+        <WeixinConnectFlow
+          token={token}
+          feature={feature}
+          idleLabel={channelTx("setup.primaryAction", "Connect WeChat")}
+          connectRequestId={connectRequestId}
+          onFeaturesUpdate={onFeaturesUpdate}
+        />
+
+        {primaryFields.length ? (
+          <CredentialForm
+            fields={primaryFields}
+            values={fieldValues}
+            configuredFields={configuredFields}
+            visibleSecrets={visibleSecrets}
+            onChange={setFieldValue}
+            onToggleSecret={(key) => {
+              setVisibleSecrets((current) => ({ ...current, [key]: !current[key] }));
+            }}
+            compact
+          />
+        ) : null}
+
+        <div
+          role="status"
+          aria-live="polite"
+          aria-atomic="true"
+          className={cn(
+            "flex items-center justify-end gap-1.5 text-[11px] leading-4 text-muted-foreground",
+            !saving && saveState !== "saved" && "sr-only",
+          )}
+        >
+          {saving ? (
+            <>
+              <Loader2 className="h-3 w-3 animate-spin" aria-hidden />
+              {tx("settings.actions.saving", "Saving")}
+            </>
+          ) : saveState === "saved" ? (
+            <>
+              <Check className="h-3 w-3" aria-hidden />
+              {tx("settings.channels.savedSettings", "Saved settings.")}
+            </>
+          ) : null}
+        </div>
+
+        {saveError ? (
+          <div
+            role="alert"
+            className="rounded-[12px] border border-destructive/20 bg-destructive/5 px-3 py-2 text-[12px] leading-5 text-destructive"
+          >
+            {saveError}
+          </div>
+        ) : null}
+
+        {advancedFields.length ? (
+          <details className="group text-[12px] leading-5 text-muted-foreground">
+            <summary className="cursor-pointer list-none text-[12px] font-semibold text-foreground">
+              <span className="inline-flex items-center gap-1.5">
+                {tx("settings.channels.advanced", "Advanced")}
+                <ChevronDown
+                  className="h-3.5 w-3.5 transition-transform group-open:rotate-180"
+                  aria-hidden
+                />
+              </span>
+            </summary>
+            <div className="mt-3">
+              <CredentialForm
+                fields={advancedFields}
+                values={fieldValues}
+                configuredFields={configuredFields}
+                visibleSecrets={visibleSecrets}
+                onChange={setFieldValue}
+                onToggleSecret={(key) => {
+                  setVisibleSecrets((current) => ({ ...current, [key]: !current[key] }));
+                }}
+                compact
+              />
+            </div>
+          </details>
+        ) : null}
+
+        <div className="flex justify-end">
+          <WeixinGuideLink
+            url={docsUrl}
+            label={channelTx("setup.docsLabel", "Open WeChat setup")}
+          />
+        </div>
+      </div>
+    </aside>
+  );
+}
+
+function weixinSetupFields(
+  feature: NanobotFeatureInfo,
+  locale: string,
+): { primary: ChannelConfigField[]; advanced: ChannelConfigField[] } {
+  const fields = feature.setup?.fields ?? [];
+  const fieldsByKey = new Map(fields.map((field) => [field.key, field]));
+  const messages = channelLocaleMessages("weixin", normalizeLocale(locale))?.setup;
+  const knownKeys = new Set<string>([
+    ...WEIXIN_PRIMARY_FIELD_KEYS,
+    ...WEIXIN_ADVANCED_FIELD_KEYS,
+  ]);
+  const extraKeys = fields
+    .map((field) => field.key)
+    .filter((key) => !knownKeys.has(key));
+  const hydrate = (keys: readonly string[]) => keys.flatMap((key) => {
+    const field = fieldsByKey.get(key);
+    if (!field) return [];
+    const copy = messages?.fields?.[channelFieldMessageKey("weixin", key)];
+    return [weixinConfigField(field, copy)];
+  });
+
+  return {
+    primary: hydrate(WEIXIN_PRIMARY_FIELD_KEYS),
+    advanced: hydrate([...WEIXIN_ADVANCED_FIELD_KEYS, ...extraKeys]),
+  };
+}
+
+function weixinConfigField(
+  field: ChannelSetupContractField,
+  copy: { label: string; placeholder?: string; help?: string; choices?: Record<string, string> }
+    | undefined,
+): ChannelConfigField {
+  const choices = field.kind === "bool" ? ["true", "false"] : field.choices;
+  return {
+    key: field.key,
+    label: copy?.label ?? fieldLabel(field.field),
+    placeholder: copy?.placeholder,
+    help: copy?.help,
+    secret: field.kind === "secret",
+    optional: !field.required,
+    inputType: field.kind === "int" ? "number" : undefined,
+    defaultValue: field.default_value,
+    options:
+      field.kind === "enum" || field.kind === "bool"
+        ? choices.map((choice) => ({
+            value: choice,
+            label: copy?.choices?.[choice] ?? fieldLabel(choice),
+          }))
+        : undefined,
+  };
+}
+
+function fieldLabel(value: string): string {
+  const spaced = value
+    .replace(/([a-z0-9])([A-Z])/g, "$1 $2")
+    .replace(/[_-]+/g, " ")
+    .trim();
+  return spaced ? spaced[0].toUpperCase() + spaced.slice(1) : value;
+}
+
+function WeixinLogo({ showBrandLogos }: { showBrandLogos: boolean }) {
+  const logoUrls = useMemo(() => logoFallbackUrls("https://weixin.qq.com/favicon.ico"), []);
+  const { logoUrl, onLogoError, onLogoLoad } = useLogoFallback(logoUrls);
+  if (showBrandLogos && logoUrl) {
+    return (
+      <span className="grid h-10 w-10 shrink-0 place-items-center rounded-[12px] bg-background">
+        <img
+          src={logoUrl}
+          alt=""
+          decoding="async"
+          loading="lazy"
+          className="h-5.5 w-5.5 max-h-6 max-w-6 object-contain"
+          onLoad={onLogoLoad}
+          onError={onLogoError}
+        />
+      </span>
+    );
+  }
+  return (
+    <span
+      className="flex h-10 w-10 shrink-0 items-center justify-center rounded-[12px] bg-background text-[11px] font-bold"
+      style={{ color: "#07C160" }}
+      aria-hidden
+    >
+      WX
+    </span>
+  );
+}
+
+function WeixinGuideLink({ url, label }: { url: string; label: string }) {
+  const logoUrls = useMemo(() => logoFallbackUrls("https://weixin.qq.com/favicon.ico"), []);
+  const { logoUrl, onLogoError, onLogoLoad } = useLogoFallback(logoUrls);
+  return (
+    <a
+      href={url}
+      target="_blank"
+      rel="noreferrer"
+      className="inline-flex max-w-full items-center gap-2 rounded-full bg-background/80 py-1 pl-1 pr-2.5 text-[11.5px] font-semibold text-foreground transition-colors hover:bg-background"
+    >
+      <span
+        className="grid h-5 w-5 shrink-0 place-items-center overflow-hidden rounded-full bg-muted/70 text-[9px] font-bold"
+        style={{ color: "#07C160" }}
+        aria-hidden
+      >
+        {logoUrl ? (
+          <img
+            src={logoUrl}
+            alt=""
+            decoding="async"
+            loading="lazy"
+            className="h-3.5 w-3.5 object-contain"
+            onLoad={onLogoLoad}
+            onError={onLogoError}
+          />
+        ) : (
+          "WX"
+        )}
+      </span>
+      <span className="truncate">{label}</span>
+      <ExternalLink className="h-3.5 w-3.5 shrink-0 text-muted-foreground" aria-hidden />
+    </a>
+  );
+}
+
+function WeixinStatusBadge({
+  children,
+  status,
+}: {
+  children: ReactNode;
+  status?: ChannelRuntimeStatus;
+}) {
+  return (
+    <span className={cn(
+      "shrink-0 rounded-full px-2 py-0.5 text-[11px] font-medium leading-4",
+      status === "failed"
+        ? "bg-destructive/10 text-destructive"
+        : status === "running"
+          ? "bg-emerald-500/10 text-emerald-700 dark:text-emerald-200"
+          : "bg-muted/75 text-muted-foreground",
+    )}>
+      {children}
+    </span>
+  );
+}
+
+function weixinStatusLabel(
+  feature: NanobotFeatureInfo,
+  tx: (key: string, fallback: string) => string,
+): string {
+  if (feature.runtime_status === "failed") {
+    return tx("settings.channels.runtimeFailed", "Failed");
+  }
+  if (feature.runtime_status === "starting") {
+    return tx("settings.channels.runtimeStarting", "Starting");
+  }
+  if (feature.runtime_status === "running") return tx("settings.values.on", "On");
+  if (feature.enabled) return tx("settings.channels.runtimeStopped", "Not running");
+  return tx("settings.values.off", "Off");
+}
+
+function weixinRuntimeError(
+  error: string | undefined,
+  tx: (key: string, fallback: string) => string,
+): string | undefined {
+  if (error === WEIXIN_AUTH_EXPIRED_MESSAGE) {
+    return tx("custom.expired", error);
+  }
+  return error;
+}
+
+function localizeBooleanFields(
+  fields: ChannelConfigField[],
+  onLabel: string,
+  offLabel: string,
+): ChannelConfigField[] {
+  return fields.map((field) => {
+    const values = new Set(field.options?.map((option) => option.value));
+    if (values.size !== 2 || !values.has("true") || !values.has("false")) return field;
+    return {
+      ...field,
+      options: field.options?.map((option) => ({
+        ...option,
+        label: option.value === "true" ? onLabel : offLabel,
+      })),
+    };
+  });
+}

nanobot/channels/weixin/webui/index.tsx

@@ -2,8 +2,14 @@ import type { ChannelUiContribution } from "@/channel-plugins/types";
 import { chatAppGuideUrl } from "@/components/settings/channels/catalog";
 
 import { WeixinConnectFlow } from "./WeixinConnectFlow";
+import {
+  WEIXIN_ADVANCED_FIELD_KEYS,
+  WEIXIN_PRIMARY_FIELD_KEYS,
+  WeixinPanel,
+} from "./WeixinPanel";
 
 export default {
+  Panel: WeixinPanel,
   ConnectFlow: WeixinConnectFlow,
   canConnectBeforeConfigured: true,
   aliases: {
@@ -18,10 +24,8 @@ export default {
       mode: "connect",
       command: "nanobot channels login weixin",
       docsUrl: chatAppGuideUrl("wechat"),
-      manualFields: [
+      fields: WEIXIN_PRIMARY_FIELD_KEYS.map((key) => ({ key })),
-        { key: "channels.weixin.allowFrom" },
+      manualFields: WEIXIN_ADVANCED_FIELD_KEYS.map((key) => ({ key })),
-        { key: "channels.weixin.token" },
-      ],
     },
   },
 } satisfies ChannelUiContribution;

nanobot/channels/weixin/webui/locales/en.json

@@ -20,7 +20,21 @@
       "token": {
         "label": "Token",
         "placeholder": "Saved by QR login"
-      }
+      },
+      "sendProgress": { "label": "Send progress" },
+      "sendToolHints": { "label": "Send tool hints" },
+      "streaming": { "label": "Use streaming API" },
+      "replyProgressMessages": { "label": "Send structured progress" },
+      "replyProgressMaxMessages": { "label": "Structured progress limit" },
+      "contextMessageBudget": { "label": "Context message budget" },
+      "blockStreaming": { "label": "Send response blocks" },
+      "blockStreamingMinChars": { "label": "Minimum block size" },
+      "blockStreamingMaxMessages": { "label": "Block message limit" },
+      "baseUrl": { "label": "API URL" },
+      "cdnBaseUrl": { "label": "CDN URL" },
+      "routeTag": { "label": "Route tag" },
+      "stateDir": { "label": "State directory" },
+      "pollTimeout": { "label": "Poll timeout" }
     }
   },
   "custom": {
@@ -30,6 +44,13 @@
     "waiting": "Waiting for WeChat scan...",
     "connected": "WeChat is connected.",
     "stopped": "WeChat login stopped.",
-    "connecting": "Connecting..."
+    "connecting": "Connecting...",
+    "verifyTitle": "Verification required",
+    "verifyDescription": "Enter the number shown in WeChat to continue.",
+    "verifyMismatch": "That code did not match. Enter the new number shown in WeChat.",
+    "expired": "WeChat login expired. Scan again to reconnect.",
+    "failed": "Unable to connect WeChat. Try again.",
+    "verifyPlaceholder": "Code",
+    "verifySubmit": "Verify"
   }
 }

nanobot/channels/weixin/webui/locales/es.json

@@ -20,7 +20,21 @@
       "token": {
         "label": "Token",
         "placeholder": "Guardado al iniciar sesión por QR"
-      }
+      },
+      "sendProgress": { "label": "Enviar progreso" },
+      "sendToolHints": { "label": "Enviar indicaciones de herramientas" },
+      "streaming": { "label": "Usar API de streaming" },
+      "replyProgressMessages": { "label": "Enviar progreso estructurado" },
+      "replyProgressMaxMessages": { "label": "Límite de progreso estructurado" },
+      "contextMessageBudget": { "label": "Presupuesto de mensajes por contexto" },
+      "blockStreaming": { "label": "Enviar respuestas por bloques" },
+      "blockStreamingMinChars": { "label": "Tamaño mínimo del bloque" },
+      "blockStreamingMaxMessages": { "label": "Límite de mensajes por bloques" },
+      "baseUrl": { "label": "URL de la API" },
+      "cdnBaseUrl": { "label": "URL de la CDN" },
+      "routeTag": { "label": "Etiqueta de ruta" },
+      "stateDir": { "label": "Directorio de estado" },
+      "pollTimeout": { "label": "Tiempo de espera de consulta" }
     }
   },
   "custom": {
@@ -30,6 +44,13 @@
     "waiting": "Esperando el escaneo de WeChat...",
     "connected": "WeChat está conectado.",
     "stopped": "Inicio de WeChat detenido.",
-    "connecting": "Conectando..."
+    "connecting": "Conectando...",
+    "verifyTitle": "Se requiere verificación",
+    "verifyDescription": "Introduce el número que aparece en WeChat para continuar.",
+    "verifyMismatch": "El código no coincide. Introduce el nuevo número que aparece en WeChat.",
+    "expired": "El inicio de sesión de WeChat caducó. Escanea de nuevo para volver a conectarte.",
+    "failed": "No se pudo conectar WeChat. Inténtalo de nuevo.",
+    "verifyPlaceholder": "Código",
+    "verifySubmit": "Verificar"
   }
 }

nanobot/channels/weixin/webui/locales/fr.json

@@ -20,7 +20,21 @@
       "token": {
         "label": "Jeton",
         "placeholder": "Enregistré après la connexion QR"
-      }
+      },
+      "sendProgress": { "label": "Envoyer la progression" },
+      "sendToolHints": { "label": "Envoyer les indications d’outils" },
+      "streaming": { "label": "Utiliser l’API de streaming" },
+      "replyProgressMessages": { "label": "Envoyer la progression structurée" },
+      "replyProgressMaxMessages": { "label": "Limite de progression structurée" },
+      "contextMessageBudget": { "label": "Budget de messages du contexte" },
+      "blockStreaming": { "label": "Envoyer la réponse par blocs" },
+      "blockStreamingMinChars": { "label": "Taille minimale d’un bloc" },
+      "blockStreamingMaxMessages": { "label": "Limite de messages par blocs" },
+      "baseUrl": { "label": "URL de l’API" },
+      "cdnBaseUrl": { "label": "URL du CDN" },
+      "routeTag": { "label": "Étiquette de routage" },
+      "stateDir": { "label": "Répertoire d’état" },
+      "pollTimeout": { "label": "Délai d’interrogation" }
     }
   },
   "custom": {
@@ -30,6 +44,13 @@
     "waiting": "En attente du scan WeChat...",
     "connected": "WeChat est connecté.",
     "stopped": "Connexion WeChat arrêtée.",
-    "connecting": "Connexion..."
+    "connecting": "Connexion...",
+    "verifyTitle": "Vérification requise",
+    "verifyDescription": "Saisissez le nombre affiché dans WeChat pour continuer.",
+    "verifyMismatch": "Le code ne correspond pas. Saisissez le nouveau nombre affiché dans WeChat.",
+    "expired": "La connexion WeChat a expiré. Scannez à nouveau pour vous reconnecter.",
+    "failed": "Impossible de connecter WeChat. Réessayez.",
+    "verifyPlaceholder": "Code",
+    "verifySubmit": "Vérifier"
   }
 }

nanobot/channels/weixin/webui/locales/id.json

@@ -20,7 +20,21 @@
       "token": {
         "label": "Token",
         "placeholder": "Disimpan saat login QR"
-      }
+      },
+      "sendProgress": { "label": "Kirim progres" },
+      "sendToolHints": { "label": "Kirim petunjuk alat" },
+      "streaming": { "label": "Gunakan API streaming" },
+      "replyProgressMessages": { "label": "Kirim progres terstruktur" },
+      "replyProgressMaxMessages": { "label": "Batas progres terstruktur" },
+      "contextMessageBudget": { "label": "Anggaran pesan konteks" },
+      "blockStreaming": { "label": "Kirim respons per blok" },
+      "blockStreamingMinChars": { "label": "Ukuran blok minimum" },
+      "blockStreamingMaxMessages": { "label": "Batas pesan blok" },
+      "baseUrl": { "label": "URL API" },
+      "cdnBaseUrl": { "label": "URL CDN" },
+      "routeTag": { "label": "Tag rute" },
+      "stateDir": { "label": "Direktori status" },
+      "pollTimeout": { "label": "Batas waktu polling" }
     }
   },
   "custom": {
@@ -30,6 +44,13 @@
     "waiting": "Menunggu pemindaian WeChat...",
     "connected": "WeChat sudah terhubung.",
     "stopped": "Login WeChat dihentikan.",
-    "connecting": "Menghubungkan..."
+    "connecting": "Menghubungkan...",
+    "verifyTitle": "Verifikasi diperlukan",
+    "verifyDescription": "Masukkan angka yang ditampilkan di WeChat untuk melanjutkan.",
+    "verifyMismatch": "Kode tidak cocok. Masukkan angka baru yang ditampilkan di WeChat.",
+    "expired": "Login WeChat telah kedaluwarsa. Pindai lagi untuk menghubungkan kembali.",
+    "failed": "Tidak dapat menghubungkan WeChat. Coba lagi.",
+    "verifyPlaceholder": "Kode",
+    "verifySubmit": "Verifikasi"
   }
 }

nanobot/channels/weixin/webui/locales/ja.json

@@ -20,7 +20,21 @@
       "token": {
         "label": "トークン",
         "placeholder": "QR ログインで保存"
-      }
+      },
+      "sendProgress": { "label": "進捗を送信" },
+      "sendToolHints": { "label": "ツールのヒントを送信" },
+      "streaming": { "label": "ストリーミング API を使用" },
+      "replyProgressMessages": { "label": "構造化された進捗を送信" },
+      "replyProgressMaxMessages": { "label": "構造化進捗の上限" },
+      "contextMessageBudget": { "label": "コンテキストのメッセージ予算" },
+      "blockStreaming": { "label": "応答をブロック単位で送信" },
+      "blockStreamingMinChars": { "label": "最小ブロックサイズ" },
+      "blockStreamingMaxMessages": { "label": "ブロックメッセージの上限" },
+      "baseUrl": { "label": "API URL" },
+      "cdnBaseUrl": { "label": "CDN URL" },
+      "routeTag": { "label": "ルートタグ" },
+      "stateDir": { "label": "状態ディレクトリ" },
+      "pollTimeout": { "label": "ポーリングタイムアウト" }
     }
   },
   "custom": {
@@ -30,6 +44,13 @@
     "waiting": "WeChat のスキャンを待っています...",
     "connected": "WeChat に接続しました。",
     "stopped": "WeChat ログインを停止しました。",
-    "connecting": "接続中..."
+    "connecting": "接続中...",
+    "verifyTitle": "確認が必要です",
+    "verifyDescription": "WeChat に表示された数字を入力してください。",
+    "verifyMismatch": "コードが一致しません。WeChat に表示された新しい数字を入力してください。",
+    "expired": "WeChat のログイン期限が切れました。再接続するにはもう一度スキャンしてください。",
+    "failed": "WeChat に接続できません。もう一度お試しください。",
+    "verifyPlaceholder": "コード",
+    "verifySubmit": "確認"
   }
 }

nanobot/channels/weixin/webui/locales/ko.json

@@ -20,7 +20,21 @@
       "token": {
         "label": "토큰",
         "placeholder": "QR 로그인으로 저장됨"
-      }
+      },
+      "sendProgress": { "label": "진행 상황 보내기" },
+      "sendToolHints": { "label": "도구 힌트 보내기" },
+      "streaming": { "label": "스트리밍 API 사용" },
+      "replyProgressMessages": { "label": "구조화된 진행 상황 보내기" },
+      "replyProgressMaxMessages": { "label": "구조화된 진행 메시지 한도" },
+      "contextMessageBudget": { "label": "컨텍스트 메시지 예산" },
+      "blockStreaming": { "label": "응답을 블록으로 보내기" },
+      "blockStreamingMinChars": { "label": "최소 블록 크기" },
+      "blockStreamingMaxMessages": { "label": "블록 메시지 한도" },
+      "baseUrl": { "label": "API URL" },
+      "cdnBaseUrl": { "label": "CDN URL" },
+      "routeTag": { "label": "경로 태그" },
+      "stateDir": { "label": "상태 디렉터리" },
+      "pollTimeout": { "label": "폴링 제한 시간" }
     }
   },
   "custom": {
@@ -30,6 +44,13 @@
     "waiting": "WeChat 스캔을 기다리는 중...",
     "connected": "WeChat이 연결되었습니다.",
     "stopped": "WeChat 로그인이 중지되었습니다.",
-    "connecting": "연결 중..."
+    "connecting": "연결 중...",
+    "verifyTitle": "인증 필요",
+    "verifyDescription": "계속하려면 WeChat에 표시된 숫자를 입력하세요.",
+    "verifyMismatch": "코드가 일치하지 않습니다. WeChat에 표시된 새 숫자를 입력하세요.",
+    "expired": "WeChat 로그인이 만료되었습니다. 다시 연결하려면 다시 스캔하세요.",
+    "failed": "WeChat에 연결할 수 없습니다. 다시 시도하세요.",
+    "verifyPlaceholder": "코드",
+    "verifySubmit": "인증"
   }
 }

nanobot/channels/weixin/webui/locales/pt-BR.json

@@ -20,7 +20,21 @@
       "token": {
         "label": "Token",
         "placeholder": "Salvo pelo login via QR"
-      }
+      },
+      "sendProgress": { "label": "Enviar progresso" },
+      "sendToolHints": { "label": "Enviar dicas de ferramentas" },
+      "streaming": { "label": "Usar API de streaming" },
+      "replyProgressMessages": { "label": "Enviar progresso estruturado" },
+      "replyProgressMaxMessages": { "label": "Limite de progresso estruturado" },
+      "contextMessageBudget": { "label": "Orçamento de mensagens do contexto" },
+      "blockStreaming": { "label": "Enviar resposta em blocos" },
+      "blockStreamingMinChars": { "label": "Tamanho mínimo do bloco" },
+      "blockStreamingMaxMessages": { "label": "Limite de mensagens em blocos" },
+      "baseUrl": { "label": "URL da API" },
+      "cdnBaseUrl": { "label": "URL da CDN" },
+      "routeTag": { "label": "Etiqueta de rota" },
+      "stateDir": { "label": "Diretório de estado" },
+      "pollTimeout": { "label": "Tempo limite da consulta" }
     }
   },
   "custom": {
@@ -30,6 +44,13 @@
     "waiting": "Aguardando leitura do WeChat...",
     "connected": "WeChat está conectado.",
     "stopped": "Login do WeChat interrompido.",
-    "connecting": "Conectando..."
+    "connecting": "Conectando...",
+    "verifyTitle": "Verificação necessária",
+    "verifyDescription": "Digite o número exibido no WeChat para continuar.",
+    "verifyMismatch": "O código não corresponde. Digite o novo número exibido no WeChat.",
+    "expired": "O login do WeChat expirou. Escaneie novamente para reconectar.",
+    "failed": "Não foi possível conectar o WeChat. Tente novamente.",
+    "verifyPlaceholder": "Código",
+    "verifySubmit": "Verificar"
   }
 }

nanobot/channels/weixin/webui/locales/vi.json

@@ -20,7 +20,21 @@
       "token": {
         "label": "Token",
         "placeholder": "Được lưu khi đăng nhập QR"
-      }
+      },
+      "sendProgress": { "label": "Gửi tiến trình" },
+      "sendToolHints": { "label": "Gửi gợi ý công cụ" },
+      "streaming": { "label": "Sử dụng API phát trực tiếp" },
+      "replyProgressMessages": { "label": "Gửi tiến trình có cấu trúc" },
+      "replyProgressMaxMessages": { "label": "Giới hạn tiến trình có cấu trúc" },
+      "contextMessageBudget": { "label": "Ngân sách tin nhắn ngữ cảnh" },
+      "blockStreaming": { "label": "Gửi phản hồi theo khối" },
+      "blockStreamingMinChars": { "label": "Kích thước khối tối thiểu" },
+      "blockStreamingMaxMessages": { "label": "Giới hạn tin nhắn theo khối" },
+      "baseUrl": { "label": "URL API" },
+      "cdnBaseUrl": { "label": "URL CDN" },
+      "routeTag": { "label": "Thẻ định tuyến" },
+      "stateDir": { "label": "Thư mục trạng thái" },
+      "pollTimeout": { "label": "Thời gian chờ thăm dò" }
     }
   },
   "custom": {
@@ -30,6 +44,13 @@
     "waiting": "Đang chờ quét WeChat...",
     "connected": "WeChat đã kết nối.",
     "stopped": "Đăng nhập WeChat đã dừng.",
-    "connecting": "Đang kết nối..."
+    "connecting": "Đang kết nối...",
+    "verifyTitle": "Cần xác minh",
+    "verifyDescription": "Nhập số hiển thị trong WeChat để tiếp tục.",
+    "verifyMismatch": "Mã không khớp. Nhập số mới hiển thị trong WeChat.",
+    "expired": "Đăng nhập WeChat đã hết hạn. Hãy quét lại để kết nối lại.",
+    "failed": "Không thể kết nối WeChat. Hãy thử lại.",
+    "verifyPlaceholder": "Mã",
+    "verifySubmit": "Xác minh"
   }
 }

nanobot/channels/weixin/webui/locales/zh-CN.json

@@ -21,7 +21,21 @@
       "token": {
         "label": "令牌",
         "placeholder": "二维码登录后自动保存"
-      }
+      },
+      "sendProgress": { "label": "发送进度消息" },
+      "sendToolHints": { "label": "发送工具提示" },
+      "streaming": { "label": "使用流式 API" },
+      "replyProgressMessages": { "label": "发送结构化进度" },
+      "replyProgressMaxMessages": { "label": "结构化进度消息上限" },
+      "contextMessageBudget": { "label": "上下文消息预算" },
+      "blockStreaming": { "label": "分块发送回复" },
+      "blockStreamingMinChars": { "label": "最小分块字符数" },
+      "blockStreamingMaxMessages": { "label": "分块消息上限" },
+      "baseUrl": { "label": "API 地址" },
+      "cdnBaseUrl": { "label": "CDN 地址" },
+      "routeTag": { "label": "路由标签" },
+      "stateDir": { "label": "状态目录" },
+      "pollTimeout": { "label": "轮询超时" }
     }
   },
   "custom": {
@@ -31,6 +45,13 @@
     "waiting": "正在等待微信扫码...",
     "connected": "微信已连接。",
     "stopped": "微信登录已停止。",
-    "connecting": "正在连接..."
+    "connecting": "正在连接...",
+    "verifyTitle": "需要验证",
+    "verifyDescription": "输入手机微信中显示的数字以继续。",
+    "verifyMismatch": "验证码不匹配，请输入微信中显示的新数字。",
+    "expired": "微信登录已过期，请重新扫码连接。",
+    "failed": "无法连接微信，请重试。",
+    "verifyPlaceholder": "验证码",
+    "verifySubmit": "验证"
   }
 }

nanobot/channels/weixin/webui/locales/zh-TW.json

@@ -21,7 +21,21 @@
       "token": {
         "label": "權杖",
         "placeholder": "二維碼登入後自動儲存"
-      }
+      },
+      "sendProgress": { "label": "傳送進度訊息" },
+      "sendToolHints": { "label": "傳送工具提示" },
+      "streaming": { "label": "使用串流 API" },
+      "replyProgressMessages": { "label": "傳送結構化進度" },
+      "replyProgressMaxMessages": { "label": "結構化進度訊息上限" },
+      "contextMessageBudget": { "label": "上下文訊息預算" },
+      "blockStreaming": { "label": "分塊傳送回覆" },
+      "blockStreamingMinChars": { "label": "最小分塊字元數" },
+      "blockStreamingMaxMessages": { "label": "分塊訊息上限" },
+      "baseUrl": { "label": "API 位址" },
+      "cdnBaseUrl": { "label": "CDN 位址" },
+      "routeTag": { "label": "路由標籤" },
+      "stateDir": { "label": "狀態目錄" },
+      "pollTimeout": { "label": "輪詢逾時" }
     }
   },
   "custom": {
@@ -31,6 +45,13 @@
     "waiting": "正在等待微信掃碼...",
     "connected": "微信已連接。",
     "stopped": "微信登入已停止。",
-    "connecting": "正在連接..."
+    "connecting": "正在連接...",
+    "verifyTitle": "需要驗證",
+    "verifyDescription": "輸入手機微信中顯示的數字以繼續。",
+    "verifyMismatch": "驗證碼不符，請輸入微信中顯示的新數字。",
+    "expired": "微信登入已過期，請重新掃碼連線。",
+    "failed": "無法連接微信，請重試。",
+    "verifyPlaceholder": "驗證碼",
+    "verifySubmit": "驗證"
   }
 }

nanobot/cli/onboard.py

@@ -32,6 +32,7 @@ from nanobot.cli.models import (
 )
 from nanobot.config.loader import get_config_path, load_config, resolve_config_env_vars
 from nanobot.config.schema import Config, ModelPresetConfig
+from nanobot.providers.oauth_guidance import OAUTH_CLI_KIT_MISSING_MESSAGE
 
 console = Console()
 
@@ -1674,7 +1675,7 @@ def _quick_start_oauth_login(config: Config, provider_name: str) -> bool:
             login_oauth_interactive,
         )
     except ImportError:
-        console.print("[red]oauth_cli_kit not installed. Run: pip install oauth-cli-kit[/red]")
+        console.print(f"[red]{OAUTH_CLI_KIT_MISSING_MESSAGE}[/red]")
         return False
 
     try:

nanobot/cli/provider.py

@@ -12,6 +12,7 @@ import typer
 from rich.console import Console
 
 from nanobot import __logo__
+from nanobot.providers.oauth_guidance import OAUTH_CLI_KIT_MISSING_MESSAGE
 
 if TYPE_CHECKING:
     from nanobot.providers.registry import ProviderSpec
@@ -74,7 +75,7 @@ def _required_module_attribute(module_name: str, attribute: str) -> object:
 
 
 def _load_openai_oauth_client() -> tuple[_GetOAuthToken, _LoginOAuthInteractive]:
-    """Load the optional untyped OAuth client behind a typed boundary."""
+    """Load the untyped OAuth client behind a typed boundary."""
     return (
         cast(_GetOAuthToken, _required_module_attribute("oauth_cli_kit", "get_token")),
         cast(
@@ -85,7 +86,7 @@ def _load_openai_oauth_client() -> tuple[_GetOAuthToken, _LoginOAuthInteractive]
 
 
 def _load_openai_oauth_storage() -> tuple[_OAuthProviderConfig, _FileTokenStorageFactory]:
-    """Load the optional untyped OAuth storage API behind a typed boundary."""
+    """Load the untyped OAuth storage API behind a typed boundary."""
     return (
         cast(
             _OAuthProviderConfig,
@@ -241,7 +242,7 @@ def _login_openai_codex() -> None:
             f"[green]✓ Authenticated with OpenAI Codex[/green]  [dim]{token.account_id}[/dim]"
         )
     except ImportError:
-        console.print("[red]oauth_cli_kit not installed. Run: pip install oauth-cli-kit[/red]")
+        console.print(f"[red]{OAUTH_CLI_KIT_MISSING_MESSAGE}[/red]")
         raise typer.Exit(1)
 
 
@@ -250,7 +251,7 @@ def _logout_openai_codex() -> None:
     try:
         provider_config, storage_factory = _load_openai_oauth_storage()
     except ImportError:
-        console.print("[red]oauth_cli_kit not installed. Run: pip install oauth-cli-kit[/red]")
+        console.print(f"[red]{OAUTH_CLI_KIT_MISSING_MESSAGE}[/red]")
         raise typer.Exit(1)
 
     storage = storage_factory(token_filename=provider_config.token_filename)
@@ -309,7 +310,7 @@ def _logout_github_copilot() -> None:
     try:
         from nanobot.providers.github_copilot_provider import get_storage
     except ImportError:
-        console.print("[red]oauth_cli_kit not installed. Run: pip install oauth-cli-kit[/red]")
+        console.print(f"[red]{OAUTH_CLI_KIT_MISSING_MESSAGE}[/red]")
         raise typer.Exit(1)
 
     storage = get_storage()

nanobot/cron/service.py

@@ -25,9 +25,6 @@ from nanobot.cron.types import (
     CronSchedule,
     CronStore,
 )
-from nanobot.utils.run_records import (
-    safe_run_record_name,
-)
 from nanobot.utils.run_records import (
     write_run_record as write_automation_run_record,
 )
@@ -440,10 +437,6 @@ class CronService:
             tmp_path.unlink(missing_ok=True)
             raise
 
-    @staticmethod
-    def _safe_run_record_name(run_id: str) -> str:
-        return safe_run_record_name(run_id)
-
     def write_run_record(self, run_id: str, record: dict[str, Any]) -> None:
         """Write an internal audit record for one cron execution."""
         write_automation_run_record(self._run_records_dir, run_id, record)

nanobot/cron/session_turns.py

@@ -7,7 +7,6 @@ from typing import Any, Mapping
 from nanobot.cron.types import CronJob
 from nanobot.session.automation_turns import (
     AutomationTurnSpec,
-    automation_history_overrides_for_spec,
     automation_trigger,
 )
 
@@ -63,11 +62,6 @@ def cron_run_id(metadata: Mapping[str, Any] | None) -> str | None:
     return value if isinstance(value, str) and value else None
 
 
-def cron_history_overrides(metadata: Mapping[str, Any] | None) -> tuple[str | None, dict[str, Any]]:
-    """Return session-history text/metadata overrides for a cron turn."""
-    return automation_history_overrides_for_spec(metadata, CRON_AUTOMATION_SPEC)
-
-
 def is_bound_cron_job(job: CronJob) -> bool:
     """True for session-bound cron jobs with complete delivery context."""
     payload = job.payload

nanobot/providers/factory.py

@@ -24,7 +24,6 @@ class ProviderSnapshot:
 @dataclass(frozen=True)
 class _ProviderSetup:
     model: str
-    provider_name: str
     provider_config: ProviderConfig | None
     spec: ProviderSpec | None
     backend: str
@@ -100,7 +99,6 @@ def _resolve_provider_setup(
 
     return _ProviderSetup(
         model=model,
-        provider_name=provider_name,
         provider_config=p,
         spec=spec,
         backend=backend,
@@ -136,7 +134,6 @@ def _make_provider_core(
         model=model,
     )
     model = setup.model
-    provider_name = setup.provider_name
     p = setup.provider_config
     spec = setup.spec
     backend = setup.backend
@@ -201,7 +198,7 @@ def _make_provider_core(
             extra_headers=_provider_extra_headers(spec, p),
             spec=spec,
             extra_body=p.extra_body if p else None,
-            api_type=p.api_type if p and provider_name == "openai" else "auto",
+            api_type=p.api_type if p else "auto",
             extra_query=p.extra_query if p else None,
             proxy=p.proxy if p else None,
         )

nanobot/providers/oauth_guidance.py

@@ -0,0 +1,6 @@
+"""Shared recovery guidance for OAuth dependency failures."""
+
+OAUTH_CLI_KIT_MISSING_MESSAGE = (
+    "This nanobot installation is missing the required oauth-cli-kit package. "
+    "Reinstall or upgrade nanobot-ai using the same installation method."
+)

nanobot/providers/openai_compat_provider.py

@@ -49,7 +49,7 @@ from nanobot.providers.openai_responses import (
 if TYPE_CHECKING:
     from openai import AsyncOpenAI as AsyncOpenAIType
 
-    from nanobot.providers.registry import ProviderSpec
+    from nanobot.providers.registry import ProviderSpec, ResponsesCapabilities
 
 # Module-level placeholder — set lazily by _ensure_client on first real
 # use, or replaced by tests via ``patch(...)``.  Kept as a plain name so
@@ -496,7 +496,12 @@ class OpenAICompatProvider(LLMProvider):
         self.extra_headers = extra_headers or {}
         self._spec = spec
         self._extra_body = dict(extra_body or {})
-        self._api_type = api_type if spec and spec.name == "openai" else "auto"
+        responses = spec.responses if spec is not None else None
+        self._api_type = (
+            api_type
+            if responses is not None and responses.allows_api_type_override
+            else "auto"
+        )
         self._extra_query = extra_query or {}
         self._proxy = proxy or None
         self._native_compaction_available = True
@@ -586,7 +591,7 @@ class OpenAICompatProvider(LLMProvider):
                     if os.environ.get("LANGFUSE_SECRET_KEY"):
                         logger.warning(
                             "LANGFUSE_SECRET_KEY is set but langfuse is not installed; "
-                            "install with `pip install langfuse` to enable tracing"
+                            "run `nanobot plugins enable langfuse` to enable tracing"
                         )
                     from openai import AsyncOpenAI as _AsyncOpenAI
                 AsyncOpenAI = _AsyncOpenAI
@@ -987,35 +992,33 @@ class OpenAICompatProvider(LLMProvider):
         """Choose Responses for providers/models that explicitly support it."""
         if self._api_type == "chat_completions":
             return False
-        spec_name = self._spec.name if self._spec is not None else None
+        capabilities = self._responses_capabilities()
-        model_name = self._request_model_name(model or self.default_model).lower()
+        if capabilities is None:
-        supported_models = {
-            supported.lower()
-            for supported in getattr(self._spec, "responses_models", ())
-        }
-        model_responses = any(
-            model_name == supported or model_name.endswith(f"/{supported}")
-            for supported in supported_models
-        )
-        provider_responses = spec_name in ("openai", "github_copilot")
-        if not provider_responses and not model_responses:
             return False
-        if self._responses_is_required():
+        model_name = self._request_model_name(model or self.default_model).lower()
-            # Explicit Responses-only request fields are mandatory; do not
+        if self._api_type == "responses":
+            # Explicit configuration means Responses is mandatory; do not
             # consult the circuit breaker or fall back to Chat Completions.
             return True
-        if provider_responses and (self._spec is None or self._spec.name != "github_copilot"):
+
-            if not _is_direct_openai_base(self._effective_base):
+        explicitly_supported = capabilities.matches_model(model_name)
+        if self._hosted_web_search_enabled() and (
+            capabilities.auto_route or explicitly_supported
+        ):
+            # Provider-hosted tools require Responses on models that the
+            # capability profile declares eligible for that transport.
+            return True
+        if (
+            capabilities.requires_direct_openai_base
+            and not _is_direct_openai_base(self._effective_base)
+        ):
             return False
 
-        wants = False
+        wants_auto_route = capabilities.auto_route and (
-        if model_responses:
+            (reasoning_effort is not None and reasoning_effort.lower() != "none")
-            wants = True
+            or any(token in model_name for token in ("gpt-5", "o1", "o3", "o4"))
-        elif reasoning_effort and reasoning_effort.lower() != "none":
+        )
-            wants = True
+        if not explicitly_supported and not wants_auto_route:
-        elif any(token in model_name for token in ("gpt-5", "o1", "o3", "o4")):
-            wants = True
-        if not wants:
             return False
 
         return self._responses_circuit_allows_probe(model, reasoning_effort)
@@ -1039,6 +1042,9 @@ class OpenAICompatProvider(LLMProvider):
             )
         )
 
+    def _responses_capabilities(self) -> ResponsesCapabilities | None:
+        return self._spec.responses if self._spec is not None else None
+
     def _responses_state_provider(self) -> str:
         spec_name = self._spec.name if self._spec is not None else "custom"
         effective_base = self._effective_base or "https://api.openai.com/v1"
@@ -1061,14 +1067,20 @@ class OpenAICompatProvider(LLMProvider):
     def supports_native_compaction(self, model: str | None = None) -> bool:
         """Enable server compaction only on direct OpenAI Responses endpoints."""
         _ = model
+        capabilities = self._responses_capabilities()
         if (
             not self._native_compaction_available
             or self._api_type == "chat_completions"
+            or capabilities is None
+            or not capabilities.supports_native_compaction
         ):
             return False
-        if self._spec is not None and self._spec.name != "openai":
+        if (
+            capabilities.requires_direct_openai_base
+            and not _is_direct_openai_base(self._effective_base)
+        ):
             return False
-        return _is_direct_openai_base(self._effective_base)
+        return True
 
     def _responses_circuit_allows_probe(
         self,
@@ -1156,7 +1168,10 @@ class OpenAICompatProvider(LLMProvider):
                     self._sanitize_empty_content(sanitized_state.pending_messages)
                 )
             )
-        preserve_reasoning = bool(self._spec and self._spec.name == "deepseek")
+        capabilities = self._responses_capabilities()
+        preserve_reasoning = (
+            capabilities is not None and capabilities.reasoning_replay == "plaintext"
+        )
         instructions, input_items, replayed = prepare_responses_input(
             sanitized_messages,
             state=sanitized_state,
@@ -1187,10 +1202,15 @@ class OpenAICompatProvider(LLMProvider):
                 "compact_threshold": compact_threshold,
             }]
 
-        if self._supports_temperature(model_name, reasoning_effort):
+        supports_temperature = self._supports_temperature(model_name, reasoning_effort)
+        if supports_temperature:
             body["temperature"] = temperature
 
-        if not self._supports_temperature(model_name, reasoning_effort) and not preserve_reasoning:
+        if (
+            not supports_temperature
+            and capabilities is not None
+            and capabilities.reasoning_replay == "encrypted"
+        ):
             body["include"] = ["reasoning.encrypted_content"]
         if reasoning_effort and reasoning_effort.lower() != "none":
             body["reasoning"] = {"effort": reasoning_effort}
@@ -1840,10 +1860,8 @@ class OpenAICompatProvider(LLMProvider):
                     self._record_responses_success(model, reasoning_effort)
                     return result
                 except Exception as responses_error:
-                    if self._spec and self._spec.name == "github_copilot":
+                    capabilities = self._responses_capabilities()
-                        # Copilot gateway exposes GPT-5/o-series only via /responses;
+                    if capabilities is not None and not capabilities.allows_chat_fallback:
-                        # falling back to /chat/completions cannot succeed and would
-                        # hide the real error.
                         raise
                     if self._responses_is_required():
                         raise
@@ -1936,10 +1954,8 @@ class OpenAICompatProvider(LLMProvider):
                         )
                     return result
                 except Exception as responses_error:
-                    if self._spec and self._spec.name == "github_copilot":
+                    capabilities = self._responses_capabilities()
-                        # Copilot gateway exposes GPT-5/o-series only via /responses;
+                    if capabilities is not None and not capabilities.allows_chat_fallback:
-                        # falling back to /chat/completions cannot succeed and would
-                        # hide the real error.
                         raise
                     if self._responses_is_required():
                         raise

nanobot/providers/registry.py

@@ -13,7 +13,7 @@ Every entry writes out all fields so you can copy-paste as a template.
 from __future__ import annotations
 
 from dataclasses import dataclass
-from typing import Any
+from typing import Any, Literal
 
 from pydantic.alias_generators import to_snake
 
@@ -28,6 +28,32 @@ class ProviderModelSpec:
     context_window: int | None = None
 
 
+@dataclass(frozen=True)
+class ResponsesCapabilities:
+    """Provider capabilities for the shared OpenAI Responses execution path.
+
+    ``reasoning_replay`` selects whether multi-turn reasoning is retained as
+    encrypted server content, plaintext local history, or not requested.
+    """
+
+    models: tuple[str, ...] = ()
+    auto_route: bool = False
+    requires_direct_openai_base: bool = False
+    allows_api_type_override: bool = False
+    reasoning_replay: Literal["none", "encrypted", "plaintext"] = "none"
+    supports_native_compaction: bool = False
+    allows_chat_fallback: bool = True
+
+    def matches_model(self, model: str) -> bool:
+        """Return whether *model* is explicitly routed through Responses."""
+        model_name = model.lower()
+        return any(
+            model_name == supported.lower()
+            or model_name.endswith(f"/{supported.lower()}")
+            for supported in self.models
+        )
+
+
 @dataclass(frozen=True)
 class ProviderSpec:
     """One LLM provider's metadata. See PROVIDERS below for real examples.
@@ -111,10 +137,8 @@ class ProviderSpec:
     # Substring match against the wire model name (lowercased).
     implicit_reasoning_models: tuple[str, ...] = ()
 
-    # Models that expose the OpenAI Responses wire format.  This is model-level
+    # Capabilities for providers/models served through the shared Responses path.
-    # because providers may add Responses support incrementally (DeepSeek V4
+    responses: ResponsesCapabilities | None = None
-    # Flash is supported before V4 Pro).
-    responses_models: tuple[str, ...] = ()
 
     # Provider-hosted Responses tools sent unless extraBody.tools explicitly
     # supplies the hosted-tool selection. Values are raw Responses tool types.
@@ -389,6 +413,13 @@ PROVIDERS: tuple[ProviderSpec, ...] = (
         display_name="OpenAI",
         backend="openai_compat",
         supports_max_completion_tokens=True,
+        responses=ResponsesCapabilities(
+            auto_route=True,
+            requires_direct_openai_base=True,
+            allows_api_type_override=True,
+            reasoning_replay="encrypted",
+            supports_native_compaction=True,
+        ),
     ),
     # OpenAI Codex: OAuth-based, dedicated provider
     ProviderSpec(
@@ -472,6 +503,11 @@ PROVIDERS: tuple[ProviderSpec, ...] = (
         strip_model_prefix=True,
         is_oauth=True,
         supports_max_completion_tokens=True,
+        responses=ResponsesCapabilities(
+            auto_route=True,
+            reasoning_replay="encrypted",
+            allows_chat_fallback=False,
+        ),
     ),
     # DeepSeek: OpenAI-compatible at api.deepseek.com
     ProviderSpec(
@@ -482,7 +518,10 @@ PROVIDERS: tuple[ProviderSpec, ...] = (
         backend="openai_compat",
         default_api_base="https://api.deepseek.com",
         thinking_style="thinking_type",
-        responses_models=("deepseek-v4-flash",),
+        responses=ResponsesCapabilities(
+            models=("deepseek-v4-flash",),
+            reasoning_replay="plaintext",
+        ),
         responses_default_tools=("web_search",),
     ),
     # Gemini: Google's OpenAI-compatible endpoint

nanobot/session/manager.py

@@ -11,7 +11,7 @@ from copy import deepcopy
 from dataclasses import dataclass, field
 from datetime import datetime
 from pathlib import Path
-from typing import Any, Callable, Protocol, TypedDict, cast
+from typing import Any, Callable, Collection, Protocol, TypedDict, cast
 from weakref import WeakValueDictionary
 
 from loguru import logger
@@ -36,6 +36,7 @@ from nanobot.utils.subagent_channel_display import scrub_subagent_announce_body
 FILE_MAX_MESSAGES = 2000
 SESSION_CACHE_MAX_SIZE = 128
 MIN_REPLAY_MAX_MESSAGES = 120
+MIN_COMPACTED_REPLAY_MESSAGES = 8
 REPLAY_TOKENS_PER_MESSAGE = 100
 _MESSAGE_TIME_PREFIX_RE = re.compile(r"^\[Message Time: [^\]]+\]\n?")
 _LOCAL_IMAGE_BREADCRUMB_RE = re.compile(r"^\[image: (?:/|~)[^\]]+\]\s*$")
@@ -146,6 +147,15 @@ class RetentionResult:
     already_consolidated_count: int
 
 
+@dataclass(frozen=True)
+class SessionPolicy:
+    """Runtime rules that do not belong in durable session data."""
+
+    persist: bool = True
+    log_content: bool = True
+    disabled_tools: frozenset[str] = frozenset()
+
+
 @dataclass
 class Session:
     """A conversation session."""
@@ -157,6 +167,7 @@ class Session:
     metadata: dict[str, Any] = field(default_factory=dict)
     last_consolidated: int = 0  # Number of messages already consolidated to files
     provider_state: ProviderConversationState | None = field(default=None, repr=False)
+    policy: SessionPolicy = field(default_factory=SessionPolicy, repr=False, compare=False)
 
     def __post_init__(self) -> None:
         if not isinstance(cast(object, self.metadata), dict):
@@ -191,19 +202,37 @@ class Session:
         extend_to_user: bool = False,
         include_runtime_context: bool = True,
     ) -> list[dict[str, Any]]:
-        """Return unconsolidated messages for LLM input.
+        """Return recent replayable messages for LLM input.
 
         History is sliced by message count first (``max_messages``), then by
         token budget from the tail (``max_tokens``) when provided.
         """
-        unconsolidated = self.messages[self.last_consolidated:]
+        replay_start = self.last_consolidated
+        if replay_start:
+            # ``last_consolidated`` is archive progress, not a replay boundary.
+            # Keep a small raw suffix for continuity, extending back to the user
+            # that started an assistant/tool sequence when necessary.
+            recent_start = recent_message_start_index(
+                self.messages,
+                MIN_COMPACTED_REPLAY_MESSAGES,
+                extend_to_user=True,
+            )
+            replay_start = min(replay_start, recent_start)
+
+        replayable = self.messages[replay_start:]
         max_messages = max_messages if max_messages > 0 else FILE_MAX_MESSAGES
+        unarchived_count = len(self.messages) - self.last_consolidated
+        if replay_start < self.last_consolidated and unarchived_count < max_messages:
+            # The archived replay suffix can exceed the nominal count when one
+            # tool-heavy turn spans the boundary. Preserve that complete turn.
+            start_idx = 0
+        else:
             start_idx = recent_message_start_index(
-            unconsolidated,
+                replayable,
                 max_messages,
                 extend_to_user=extend_to_user,
             )
-        sliced = unconsolidated[start_idx:]
+        sliced = replayable[start_idx:]
 
         # Avoid starting mid-turn when possible, except for proactive
         # assistant deliveries that the user may be replying to.
@@ -352,16 +381,23 @@ class Session:
 
         start_idx = max(0, len(self.messages) - max_messages)
         if extend_to_user:
-            start_idx = next(
+            recovered_user = next(
                 (i for i in range(start_idx, -1, -1) if self.messages[i].get("role") == "user"),
-                start_idx,
+                None,
             )
+            if recovered_user is not None:
+                start_idx = recovered_user
+                if start_idx > 0 and self.messages[start_idx - 1].get("_channel_delivery"):
+                    start_idx -= 1
 
         retained = self.messages[start_idx:]
 
-        # Prefer starting at a user turn when one exists within the retained window.
+        # Prefer starting at a user turn (or its preceding _channel_delivery) when one exists within the retained window.
         first_user = next((i for i, m in enumerate(retained) if m.get("role") == "user"), None)
         if first_user is not None:
+            if first_user > 0 and retained[first_user - 1].get("_channel_delivery"):
+                retained = retained[first_user - 1:]
+            else:
                 retained = retained[first_user:]
         elif not extend_to_user:
             # If the hard-capped tail is assistant/tool-only, anchor to the
@@ -1053,6 +1089,24 @@ class SessionManager:
         self._remember(session)
         return session
 
+    def get_or_create_transient(
+        self,
+        key: str,
+        *,
+        disabled_tools: Collection[str] = (),
+    ) -> Session:
+        """Return a fresh, non-persistent session without loading history."""
+        policy = SessionPolicy(
+            persist=False,
+            log_content=False,
+            disabled_tools=frozenset(disabled_tools),
+        )
+        session = self.get_cached(key)
+        if session is None or session.policy != policy:
+            session = Session(key=key, policy=policy)
+            self._remember(session)
+        return session
+
     def _load(self, key: str) -> Session | None:
         return self._store.load(key)
 
@@ -1060,12 +1114,11 @@ class SessionManager:
         """Attempt to recover a session from a corrupt JSONL file."""
         return self._jsonl_store.repair(key, path=path)
 
-    @staticmethod
-    def _session_payload(session: Session) -> SessionPayload:
-        return JsonlSessionStore.session_payload(session)
-
     def save(self, session: Session, *, fsync: bool = False) -> None:
         """Persist a session and retain it in the cache."""
+        if not session.policy.persist:
+            return
+
         archiver = self._file_cap_archiver
         if archiver is not None:
             session.enforce_file_cap(

nanobot/session/webui_turns.py

@@ -334,6 +334,12 @@ def clear_websocket_turn_if_current(
     return False
 
 
+def clear_websocket_turns(chat_id: str) -> None:
+    """Forget every in-process turn projection for a discarded chat."""
+    _WEBSOCKET_ACTIVE_TURNS.pop(chat_id, None)
+    _sync_websocket_turn_projection(chat_id)
+
+
 def build_bus_progress_callback(
     bus: MessageBus,
     msg: InboundMessage,

nanobot/triggers/local_session_turns.py

@@ -6,7 +6,6 @@ from typing import Any, Mapping
 
 from nanobot.session.automation_turns import (
     AutomationTurnSpec,
-    automation_history_overrides_for_spec,
     automation_trigger,
 )
 
@@ -50,13 +49,3 @@ def local_trigger_delivery_id(metadata: Mapping[str, Any] | None) -> str | None:
         return None
     value = trigger.get("delivery_id")
     return value if isinstance(value, str) and value else None
-
-
-def local_trigger_history_overrides(
-    metadata: Mapping[str, Any] | None,
-) -> tuple[str | None, dict[str, Any]]:
-    """Return session-history text/metadata overrides for a local trigger turn."""
-    return automation_history_overrides_for_spec(
-        metadata,
-        LOCAL_TRIGGER_AUTOMATION_SPEC,
-    )

nanobot/utils/document.py

@@ -11,35 +11,6 @@ from loguru import logger
 
 from nanobot.utils.helpers import detect_image_mime
 
-# Supported file extensions for text extraction
-SUPPORTED_EXTENSIONS: set[str] = {
-    # Document formats
-    ".pdf",
-    ".docx",
-    ".xlsx",
-    ".pptx",
-    # Text formats
-    ".txt",
-    ".md",
-    ".csv",
-    ".json",
-    ".xml",
-    ".html",
-    ".htm",
-    ".log",
-    ".yaml",
-    ".yml",
-    ".toml",
-    ".ini",
-    ".cfg",
-    # Image formats (for future OCR support)
-    ".png",
-    ".jpg",
-    ".jpeg",
-    ".gif",
-    ".webp",
-}
-
 _MAX_TEXT_LENGTH = 200_000
 _MAX_EXTRACT_FILE_SIZE = 50 * 1024 * 1024  # 50 MB
 _MAX_OFFICE_ARCHIVE_MEMBERS = 10_000

nanobot/utils/file_edit_events.py

@@ -274,24 +274,6 @@ def _text_line_count(text: str) -> int:
     return line_count if last_was_newline else line_count + 1
 
 
-def prepare_file_edit_tracker(
-    *,
-    call_id: str,
-    tool_name: str,
-    tool: Any,
-    workspace: Path | None,
-    params: dict[str, Any] | None,
-) -> FileEditTracker | None:
-    trackers = prepare_file_edit_trackers(
-        call_id=call_id,
-        tool_name=tool_name,
-        tool=tool,
-        workspace=workspace,
-        params=params,
-    )
-    return trackers[0] if trackers else None
-
-
 def prepare_file_edit_trackers(
     *,
     call_id: str,

nanobot/utils/gitstore.py

@@ -5,14 +5,13 @@ from __future__ import annotations
 import io
 import time
 from dataclasses import dataclass
-from datetime import datetime, timezone
 from pathlib import Path
-from typing import TYPE_CHECKING, Iterable, cast
+from typing import TYPE_CHECKING, cast
 
 from loguru import logger
 
 if TYPE_CHECKING:
-    from dulwich.objects import Blob, Commit, ObjectID, Tree, TreeEntry
+    from dulwich.objects import Blob, Commit, ObjectID, Tree
     from dulwich.refs import Ref
     from dulwich.repo import Repo
 
@@ -45,25 +44,6 @@ class CommitInfo:
         return f"{header}\n(no file changes)"
 
 
-@dataclass
-class LineAge:
-    """Age of a single line based on git blame."""
-
-    age_days: int  # days since last modification
-
-
-def _compute_line_ages(
-    annotated: Iterable[tuple[tuple["Commit", "TreeEntry"], bytes]],
-) -> list[LineAge]:
-    """Convert annotate results to per-line ages."""
-    now = datetime.now(tz=timezone.utc).date()
-    ages: list[LineAge] = []
-    for (commit, _tree_entry), _line_bytes in annotated:
-        dt = datetime.fromtimestamp(commit.commit_time, tz=timezone.utc).date()
-        ages.append(LineAge(age_days=(now - dt).days))
-    return ages
-
-
 class GitStore:
     """Git-backed version control for memory files."""
 
@@ -293,33 +273,6 @@ class GitStore:
         except Exception as exc:
             raise GitStoreError("Git log failed") from exc
 
-    def line_ages(self, file_path: str) -> list[LineAge]:
-        """Compute the age of each line in a tracked file via git blame.
-
-        Returns one LineAge per line, in order.
-        Returns an empty list if the repo is not initialized or the file is
-        empty. Annotation failures raise :class:`GitStoreError`.
-        """
-
-        if not self.is_initialized():
-            return []
-
-        target = self._workspace / file_path
-        if not target.exists() or target.stat().st_size == 0:
-            return []
-
-        try:
-            from dulwich import porcelain
-
-            annotated = porcelain.annotate(str(self._workspace), file_path)
-        except Exception as exc:
-            raise GitStoreError(f"Git line annotation failed for {file_path}") from exc
-
-        if not annotated:
-            return []
-
-        return _compute_line_ages(annotated)
-
     def diff_commits(self, sha1: str, sha2: str) -> str:
         """Show diff between two commits."""
         if not self.is_initialized():
@@ -461,13 +414,6 @@ class GitStore:
         commit = cast("Commit", commit_obj)
         return cast("Tree", repo[commit.tree])
 
-    def find_commit(self, short_sha: str, max_entries: int = 20) -> CommitInfo | None:
-        """Find a commit by short SHA prefix match."""
-        for c in self.log(max_entries=max_entries):
-            if c.sha.startswith(short_sha):
-                return c
-        return None
-
     def show_commit_diff(
         self,
         short_sha: str,

nanobot/utils/helpers.py

@@ -351,18 +351,6 @@ def timestamp() -> str:
     return datetime.now().isoformat()
 
 
-def current_time_str(timezone: str | None = None) -> str:
-    """Return the current time string."""
-    from zoneinfo import ZoneInfo
-
-    tz = ZoneInfo(timezone) if timezone else None
-    now = datetime.now(tz=tz) if tz else datetime.now().astimezone()
-    offset = now.strftime("%z")
-    offset_fmt = f"{offset[:3]}:{offset[3:]}" if len(offset) == 5 else offset
-    tz_name = timezone or (time.strftime("%Z") or "UTC")
-    return f"{now.strftime('%Y-%m-%d %H:%M (%A)')} ({tz_name}, UTC{offset_fmt})"
-
-
 _UNSAFE_CHARS = re.compile(r'[<>:"/\\|?*]')
 _TOOL_RESULT_PREVIEW_CHARS = 1200
 _TOOL_RESULTS_DIR = ".nanobot/tool-results"

nanobot/utils/subagent_channel_display.py

@@ -3,14 +3,13 @@
 Persisted subagent announcements mirror ``agent/subagent_announce.md``: header,
 full ``Task:`` assignment (model context), ``Result:``, and a trailing model-only
 ``Summarize…`` instruction. External channels (embedded WebUI, session previews)
-should show only the header plus a truncated result body."""
+should show only the header plus a truncated result body.
+"""
 
 from __future__ import annotations
 
-from typing import Any, cast
+# Cap the Result section so session previews stay readable; full text remains on
-
+# disk for LLM replay.
-# Cap Result section length so WebSocket session replay stays readable; full text
-# remains on disk for LLM replay (we only mutate outgoing API copies in websocket).
 _SUBAGENT_CHANNEL_RESULT_MAX_CHARS = 800
 
 
@@ -44,16 +43,3 @@ def scrub_subagent_announce_body(content: str) -> str:
     if header and body:
         return f"{header}\n\n{body}"
     return header or body or stripped
-
-
-def scrub_subagent_messages_for_channel(messages: list[dict[str, Any]]) -> None:
-    """Mutate message dicts in place when they carry ``subagent_result`` inject."""
-    for msg in messages:
-        if not isinstance(cast(object, msg), dict):
-            continue
-        if msg.get("injected_event") != "subagent_result":
-            continue
-        raw = msg.get("content")
-        if not isinstance(raw, str) or not raw.strip():
-            continue
-        msg["content"] = scrub_subagent_announce_body(raw)

nanobot/webui/gateway_services.py

@@ -11,6 +11,7 @@ from loguru import logger as default_logger
 from nanobot.webui.gateway_tokens import GatewayTokenStore
 from nanobot.webui.ingress_policy import DEFAULT_WEBUI_INGRESS_POLICY, WebUIIngressPolicy
 from nanobot.webui.media_gateway import WebUIMediaGateway
+from nanobot.webui.temporary_chats import WebUITemporaryChats
 from nanobot.webui.transcript import WebUITranscriptRecorder
 from nanobot.webui.workspaces import WebUIWorkspaceController
 from nanobot.webui.ws_http import GatewayHTTPHandler
@@ -33,6 +34,7 @@ class GatewayServices:
     ingress: WebUIIngressPolicy
     transcripts: WebUITranscriptRecorder
     workspaces: WebUIWorkspaceController
+    temporary_chats: WebUITemporaryChats
     session_manager: SessionManager | None
     cron_service: CronService | None
     local_trigger_store: LocalTriggerStore | None
@@ -82,6 +84,12 @@ def build_gateway_services(
         default_workspace=workspace_path,
         default_restrict_to_workspace=default_restrict_to_workspace,
     )
+    temporary_chats = WebUITemporaryChats(
+        bus=bus,
+        session_manager=session_manager,
+        workspaces=workspaces,
+        logger=logger,
+    )
     http = GatewayHTTPHandler(
         config=config,
         session_manager=session_manager,
@@ -112,6 +120,7 @@ def build_gateway_services(
         ingress=ingress,
         transcripts=transcripts,
         workspaces=workspaces,
+        temporary_chats=temporary_chats,
         session_manager=session_manager,
         cron_service=cron_service,
         local_trigger_store=local_trigger_store,

nanobot/webui/media_api.py

@@ -13,7 +13,7 @@ import shutil
 import uuid
 from collections.abc import Callable
 from pathlib import Path
-from typing import Any, cast
+from typing import Any
 
 from websockets.http11 import Request as WsRequest
 from websockets.http11 import Response
@@ -32,7 +32,6 @@ from nanobot.webui.http_utils import (
 
 MediaDirProvider = Callable[[str | None], Path]
 SignedMediaPath = Callable[[Path], dict[str, str] | None]
-SignedMediaUrl = Callable[[Path], str | None]
 
 
 def b64url_encode(data: bytes) -> str:
@@ -190,37 +189,6 @@ def signed_media_attachments(
     return out
 
 
-def attach_signed_media_urls(
-    payload: dict[str, Any],
-    *,
-    sign_path: SignedMediaUrl,
-) -> None:
-    """Replace raw media path lists in a WebUI session payload with signed URLs."""
-    messages = payload.get("messages")
-    if not isinstance(messages, list):
-        return
-    raw_messages = cast(list[Any], messages)
-    for msg in raw_messages:
-        if not isinstance(msg, dict):
-            continue
-        message = cast(dict[str, Any], msg)
-        media = message.get("media")
-        if not isinstance(media, list) or not media:
-            continue
-        media_entries = cast(list[Any], media)
-        urls: list[dict[str, str]] = []
-        for entry in media_entries:
-            if not isinstance(entry, str) or not entry:
-                continue
-            signed = sign_path(Path(entry))
-            if signed is None:
-                continue
-            urls.append({"url": signed, "name": Path(entry).name})
-        if urls:
-            message["media_urls"] = urls
-        message.pop("media", None)
-
-
 def serve_signed_media(
     sig: str,
     payload: str,

nanobot/webui/media_gateway.py

@@ -17,9 +17,7 @@ from nanobot.webui.attachment_ingress import (
 )
 from nanobot.webui.ingress_policy import AttachmentIngressLimits
 from nanobot.webui.media_api import (
-    attach_signed_media_urls,
     serve_signed_media,
-    sign_media_path,
     sign_or_stage_media_path,
     signed_media_attachments,
 )
@@ -72,13 +70,6 @@ class WebUIMediaGateway:
             media_dir=self._media_dir,
         )
 
-    def sign_media_path(self, abs_path: Path) -> str | None:
-        return sign_media_path(
-            abs_path,
-            secret=self.secret,
-            media_dir=self._media_dir,
-        )
-
     def sign_or_stage_media_path(self, path: Path) -> dict[str, str] | None:
         return sign_or_stage_media_path(
             path,
@@ -99,9 +90,6 @@ class WebUIMediaGateway:
             sign_path=self.sign_or_stage_media_path,
         )
 
-    def augment_media_urls(self, payload: dict[str, Any]) -> None:
-        attach_signed_media_urls(payload, sign_path=self.sign_media_path)
-
     def augment_transcript_media(self, paths: list[str]) -> list[dict[str, Any]]:
         return signed_media_attachments(
             paths,

nanobot/webui/settings_api.py

@@ -4,7 +4,7 @@ The WebSocket channel owns transport/authentication. This module owns the
 settings payload shape and the allowlisted config mutations exposed to WebUI.
 """
 
-# oauth-cli-kit is an optional dependency and does not publish type stubs.
+# oauth-cli-kit does not publish type stubs.
 # pyright: reportMissingTypeStubs=false
 
 from __future__ import annotations
@@ -36,6 +36,7 @@ from nanobot.providers.image_generation import (
     get_image_gen_provider,
     image_gen_provider_names,
 )
+from nanobot.providers.oauth_guidance import OAUTH_CLI_KIT_MISSING_MESSAGE
 from nanobot.providers.registry import PROVIDERS, create_dynamic_spec, find_by_name
 from nanobot.security.network import is_loopback_host
 from nanobot.security.workspace_access import workspace_sandbox_status
@@ -1794,9 +1795,7 @@ def login_oauth_provider(query: QueryParams) -> dict[str, Any]:
         try:
             from nanobot.providers.openai_codex_oauth import start_openai_codex_oauth_login
         except ImportError:
-            raise WebUISettingsError(
+            raise WebUISettingsError(OAUTH_CLI_KIT_MISSING_MESSAGE, status=500) from None
-                "oauth_cli_kit not installed. Run: pip install oauth-cli-kit", status=500
-            ) from None
 
         try:
             proxy = resolve_config_env_vars(load_config()).providers.openai_codex.proxy or None
@@ -1834,9 +1833,7 @@ def login_oauth_provider(query: QueryParams) -> dict[str, Any]:
                 login_github_copilot,
             )
         except ImportError:
-            raise WebUISettingsError(
+            raise WebUISettingsError(OAUTH_CLI_KIT_MISSING_MESSAGE, status=500) from None
-                "oauth_cli_kit not installed. Run: pip install oauth-cli-kit", status=500
-            ) from None
 
         token = get_github_copilot_login_status()
         if not token:
@@ -1934,18 +1931,14 @@ def logout_oauth_provider(query: QueryParams) -> dict[str, Any]:
             from oauth_cli_kit.providers import OPENAI_CODEX_PROVIDER
             from oauth_cli_kit.storage import FileTokenStorage
         except ImportError:
-            raise WebUISettingsError(
+            raise WebUISettingsError(OAUTH_CLI_KIT_MISSING_MESSAGE, status=500) from None
-                "oauth_cli_kit not installed. Run: pip install oauth-cli-kit", status=500
-            ) from None
         _clear_webui_oauth_flows(spec.name)
         token_path = FileTokenStorage(token_filename=OPENAI_CODEX_PROVIDER.token_filename).get_token_path()
     elif spec.name == "github_copilot":
         try:
             from nanobot.providers.github_copilot_provider import get_storage
         except ImportError:
-            raise WebUISettingsError(
+            raise WebUISettingsError(OAUTH_CLI_KIT_MISSING_MESSAGE, status=500) from None
-                "oauth_cli_kit not installed. Run: pip install oauth-cli-kit", status=500
-            ) from None
         token_path = get_storage().get_token_path()
     elif spec.name == "xai_grok":
         from nanobot.providers.xai_oauth import logout_xai_oauth

nanobot/webui/sidebar_state.py

@@ -25,7 +25,7 @@ _MAX_KEY_LEN = 512
 _MAX_TITLE_LEN = 160
 _MAX_TAG_LEN = 40
 _ALLOWED_DENSITIES = {"comfortable", "compact"}
-_ALLOWED_SORTS = {"updated_desc", "created_desc", "title_asc"}
+_ALLOWED_SORTS = {"updated_desc", "created_desc", "title_asc", "manual"}
 
 
 def webui_sidebar_state_path() -> Path:
@@ -37,6 +37,7 @@ def default_webui_sidebar_state() -> dict[str, Any]:
         "schema_version": WEBUI_SIDEBAR_STATE_SCHEMA_VERSION,
         "pinned_keys": [],
         "archived_keys": [],
+        "session_order": [],
         "title_overrides": {},
         "project_name_overrides": {},
         "tags_by_key": {},
@@ -138,6 +139,7 @@ def normalize_webui_sidebar_state(raw: Any) -> dict[str, Any]:
     state = default_webui_sidebar_state()
     state["pinned_keys"] = _clean_string_list(raw.get("pinned_keys"))
     state["archived_keys"] = _clean_string_list(raw.get("archived_keys"))
+    state["session_order"] = _clean_string_list(raw.get("session_order"))
     state["title_overrides"] = _clean_title_overrides(raw.get("title_overrides"))
     state["project_name_overrides"] = _clean_title_overrides(
         raw.get("project_name_overrides")

nanobot/webui/temporary_chats.py

@@ -0,0 +1,218 @@
+"""Connection-owned Temporary Chat behavior for the WebUI."""
+
+from __future__ import annotations
+
+import uuid
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+from nanobot.bus.events import (
+    INBOUND_META_RUNTIME_CONTROL,
+    RUNTIME_CONTROL_SESSION_DISCARD,
+    InboundMessage,
+)
+from nanobot.bus.queue import MessageBus
+from nanobot.security.workspace_access import WorkspaceScope
+from nanobot.session.manager import Session, SessionManager
+from nanobot.webui.workspaces import WebUIWorkspaceController
+
+_TEMPORARY_CHAT_DISABLED_TOOLS = frozenset({
+    "create_goal",
+    "update_goal",
+    "spawn",
+    "cron",
+})
+_TEMPORARY_CHAT_COMMANDS = frozenset({"/model", "/stop"})
+
+
+class TemporaryChatError(ValueError):
+    """A stable WebUI protocol error for a Temporary Chat operation."""
+
+    def __init__(self, detail: str) -> None:
+        super().__init__(detail)
+        self.detail = detail
+
+
+@dataclass(frozen=True)
+class TemporaryChatMessagePolicy:
+    """Server-owned message rules for one active Temporary Chat."""
+
+    session_key: str
+    workspace_scope: WorkspaceScope
+    require_existing_session: bool = True
+    hydrate_transcript: bool = False
+    persist_transcript: bool = False
+
+
+class WebUITemporaryChats:
+    """Own Temporary Chat creation, policy, attachments, and disposal."""
+
+    def __init__(
+        self,
+        *,
+        bus: MessageBus,
+        session_manager: SessionManager | None,
+        workspaces: WebUIWorkspaceController,
+        logger: Any,
+        channel_name: str = "websocket",
+    ) -> None:
+        self._bus = bus
+        self._sessions = session_manager
+        self._workspaces = workspaces
+        self._logger = logger
+        self._channel_name = channel_name
+        self._owners: dict[str, object] = {}
+        self._owner_chat_ids: dict[object, set[str]] = {}
+        # Keep active sessions alive if the bounded manager cache evicts them
+        # between WebUI turns. SessionPolicy remains the authority below.
+        self._active_sessions: dict[str, Session] = {}
+        # Retain policy-derived tombstones until shutdown so late outbound
+        # events cannot create a durable transcript after a chat is discarded.
+        self._known_transient_chat_ids: set[str] = set()
+        self._media_paths: dict[str, set[str]] = {}
+
+    def _session_key(self, chat_id: str) -> str:
+        return f"{self._channel_name}:{chat_id}"
+
+    def _cached_session_is_transient(self, chat_id: str) -> bool:
+        if self._sessions is None:
+            return False
+        session = self._sessions.get_cached(self._session_key(chat_id))
+        return session is not None and not session.policy.persist
+
+    def create(self, owner: object, *, trusted_webui: bool) -> str:
+        """Create a server-identified chat owned by one authenticated WebUI connection."""
+        if not trusted_webui:
+            raise TemporaryChatError("access_denied")
+        if self._sessions is None:
+            raise TemporaryChatError("temporary_chat_unavailable")
+
+        chat_id = str(uuid.uuid4())
+        session = self._sessions.get_or_create_transient(
+            self._session_key(chat_id),
+            disabled_tools=_TEMPORARY_CHAT_DISABLED_TOOLS,
+        )
+        if session.policy.persist:
+            raise RuntimeError("Temporary Chat must use a non-persistent session policy")
+        self._owners[chat_id] = owner
+        self._owner_chat_ids.setdefault(owner, set()).add(chat_id)
+        self._active_sessions[chat_id] = session
+        self._known_transient_chat_ids.add(chat_id)
+        return chat_id
+
+    def message_policy(
+        self,
+        owner: object,
+        chat_id: str,
+        content: str,
+    ) -> TemporaryChatMessagePolicy | None:
+        """Return Temporary Chat rules, or ``None`` for an ordinary chat."""
+        if not self._cached_session_is_transient(chat_id):
+            if chat_id in self._known_transient_chat_ids:
+                raise TemporaryChatError("temporary_chat_unavailable")
+            return None
+        if self._owners.get(chat_id) is not owner or self._sessions is None:
+            raise TemporaryChatError("temporary_chat_unavailable")
+
+        session = self._sessions.get_cached(self._session_key(chat_id))
+        if session is None:
+            raise TemporaryChatError("temporary_chat_unavailable")
+
+        command = content.strip().split(maxsplit=1)[0].lower() if content.strip() else ""
+        if command.startswith("/") and command not in _TEMPORARY_CHAT_COMMANDS:
+            raise TemporaryChatError("temporary_chat_command_rejected")
+
+        return TemporaryChatMessagePolicy(
+            session_key=self._session_key(chat_id),
+            workspace_scope=self._workspaces.restricted_default_scope(),
+        )
+
+    def validate_attach(self, chat_id: str) -> None:
+        """Reject attempts to recover a non-persistent session."""
+        if not self._cached_session_is_transient(chat_id):
+            if chat_id in self._known_transient_chat_ids:
+                raise TemporaryChatError("temporary_chat_unavailable")
+            return
+        raise TemporaryChatError("temporary_chat_unavailable")
+
+    def validate_workspace_update(self, chat_id: str) -> None:
+        """Prevent non-persistent sessions from acquiring durable workspace state."""
+        if self._cached_session_is_transient(chat_id):
+            raise TemporaryChatError("temporary_chat_workspace_rejected")
+        if chat_id in self._known_transient_chat_ids:
+            raise TemporaryChatError("temporary_chat_unavailable")
+
+    def register_media(self, owner: object, chat_id: str, paths: list[str]) -> None:
+        if not paths:
+            return
+        if self._owners.get(chat_id) is not owner:
+            raise TemporaryChatError("temporary_chat_unavailable")
+        self._media_paths.setdefault(chat_id, set()).update(paths)
+
+    def chat_ids_for_owner(self, owner: object) -> tuple[str, ...]:
+        return tuple(self._owner_chat_ids.get(owner, ()))
+
+    def owns(self, owner: object, chat_id: str) -> bool:
+        return self._owners.get(chat_id) is owner
+
+    def should_persist_transcript(self, chat_id: str) -> bool:
+        """Apply the session policy and retain it for late events after disposal."""
+        return (
+            not self._cached_session_is_transient(chat_id)
+            and chat_id not in self._known_transient_chat_ids
+        )
+
+    def _discard_media(self, chat_id: str) -> None:
+        for raw_path in self._media_paths.pop(chat_id, set()):
+            try:
+                Path(raw_path).unlink(missing_ok=True)
+            except OSError:
+                self._logger.warning("failed to remove a temporary WebUI attachment")
+
+    def _forget_owner(self, owner: object, chat_id: str) -> None:
+        self._owners.pop(chat_id, None)
+        chat_ids = self._owner_chat_ids.get(owner)
+        if chat_ids is None:
+            return
+        chat_ids.discard(chat_id)
+        if not chat_ids:
+            self._owner_chat_ids.pop(owner, None)
+
+    async def discard(self, owner: object, chat_id: str) -> None:
+        """Forget one owned chat and cancel any active work through the message bus."""
+        if (
+            not self._cached_session_is_transient(chat_id)
+            or self._owners.get(chat_id) is not owner
+        ):
+            raise TemporaryChatError("temporary_chat_unavailable")
+
+        session_key = self._session_key(chat_id)
+        self._forget_owner(owner, chat_id)
+        self._active_sessions.pop(chat_id, None)
+        self._discard_media(chat_id)
+        if self._sessions is not None:
+            self._sessions.invalidate(session_key)
+        await self._bus.publish_inbound(
+            InboundMessage(
+                channel=self._channel_name,
+                sender_id="webui",
+                chat_id=chat_id,
+                content="",
+                metadata={
+                    INBOUND_META_RUNTIME_CONTROL: RUNTIME_CONTROL_SESSION_DISCARD,
+                },
+                session_key_override=session_key,
+            )
+        )
+
+    def close(self) -> None:
+        """Release process-local resources during gateway shutdown."""
+        for chat_id in tuple(self._owners):
+            self._discard_media(chat_id)
+            if self._sessions is not None:
+                self._sessions.invalidate(self._session_key(chat_id))
+        self._owners.clear()
+        self._owner_chat_ids.clear()
+        self._active_sessions.clear()
+        self._known_transient_chat_ids.clear()

nanobot/webui/transcript.py

@@ -1313,21 +1313,6 @@ def _recover_incomplete_turns(
     return recovered
 
 
-def recover_incomplete_turns_from_session(
-    lines: list[dict[str, Any]],
-    session_messages: list[dict[str, Any]] | None,
-    *,
-    session_key: str,
-) -> list[dict[str, Any]]:
-    """Recover marked transcript answers only when one durable session turn matches."""
-    if not lines or not session_messages or not _needs_incomplete_turn_recovery(lines):
-        return lines
-    session_turns = _session_backfill_turns(session_key, session_messages)
-    if not session_turns:
-        return lines
-    return _recover_incomplete_turns(lines, session_turns)
-
-
 def _with_backfilled_user(
     records: list[dict[str, Any]],
     user_event: dict[str, Any],
@@ -1365,20 +1350,6 @@ def _inject_missing_user_events(
     return out
 
 
-def inject_missing_user_events_from_session(
-    session_key: str,
-    lines: list[dict[str, Any]],
-    session_messages: list[dict[str, Any]] | None,
-) -> list[dict[str, Any]]:
-    """Backfill user rows for legacy WebUI transcripts that only stored assistant streams."""
-    if not lines or not session_messages or not _needs_user_event_backfill(lines):
-        return lines
-    session_turns = _session_backfill_turns(session_key, session_messages)
-    if not session_turns:
-        return lines
-    return _inject_missing_user_events(lines, session_turns)
-
-
 def _format_tool_call_trace(call: Any) -> str | None:
     if not call or not isinstance(call, dict):
         return None

nanobot/webui/version_check.py

@@ -10,6 +10,7 @@ import time
 from typing import Any
 
 import httpx
+from packaging.version import InvalidVersion, Version
 
 from nanobot import __version__
 
@@ -42,7 +43,13 @@ def check_for_update() -> dict[str, Any] | None:
             return None
         _cache = (now, latest)
 
-    if not latest or latest == __version__:
+    if not isinstance(latest, str) or not latest:
+        return None
+    try:
+        if Version(latest) <= Version(__version__):
+            return None
+    except InvalidVersion:
+        logger.debug("PyPI returned an invalid nanobot version: %r", latest)
         return None
     return {
         "currentVersion": __version__,

nanobot/webui/workspaces.py

@@ -191,6 +191,14 @@ class WebUIWorkspaceController:
             self._default_restrict_to_workspace,
         )
 
+    def restricted_default_scope(self) -> WorkspaceScope:
+        """Return the default workspace with access restricted for this request."""
+        return build_workspace_scope(
+            self._default_workspace,
+            "restricted",
+            source_channel=_WEBUI_SCOPE_CHANNEL,
+        )
+
     def _scope_from_metadata_value(
         self,
         raw_scope: object,

nanobot/webui/ws_http.py

@@ -26,10 +26,8 @@ from websockets.http11 import Response
 from nanobot.command.builtin import builtin_command_palette
 from nanobot.cron.session_turns import is_bound_cron_job
 from nanobot.cron.types import CronJob, CronSchedule
-from nanobot.runtime_context import public_history_messages
 from nanobot.security.workspace_access import WorkspaceScope
 from nanobot.triggers.local_types import LocalTrigger
-from nanobot.utils.subagent_channel_display import scrub_subagent_messages_for_channel
 from nanobot.webui.file_preview import (
     WebUIFilePreviewError,
     file_preview_availability_payload,
@@ -462,10 +460,6 @@ class GatewayHTTPHandler:
     # -- Session routes -----------------------------------------------------
 
     async def _dispatch_session_routes(self, request: WsRequest, got: str) -> Response | None:
-        m = re.match(r"^/api/sessions/([^/]+)/messages$", got)
-        if m:
-            return self._handle_session_messages(request, m.group(1))
-
         m = re.match(r"^/api/sessions/([^/]+)/webui-thread$", got)
         if m:
             return self._handle_webui_thread_get(request, m.group(1))
@@ -527,34 +521,6 @@ class GatewayHTTPHandler:
             cleaned.append(row)
         return {"sessions": cleaned}
 
-    def _handle_session_messages(self, request: WsRequest, key: str) -> Response:
-        if not self.check_api_token(request):
-            return _http_error(401, "Unauthorized")
-        if self.session_manager is None:
-            return _http_error(503, "session manager unavailable")
-        decoded_key = _decode_api_key(key)
-        if decoded_key is None:
-            return _http_error(400, "invalid session key")
-        if not _is_websocket_channel_session_key(decoded_key):
-            return _http_error(404, "session not found")
-        data = self.session_manager.read_session_file(decoded_key)
-        if data is None:
-            return _http_error(404, "session not found")
-        messages = data.get("messages")
-        if isinstance(messages, list):
-            session_messages = cast(list[dict[str, Any]], messages)
-            scrub_subagent_messages_for_channel(session_messages)
-            raw_session_messages = cast(list[Any], messages)
-            data["messages"] = public_history_messages(
-                [
-                    cast(dict[str, Any], message)
-                    for message in raw_session_messages
-                    if isinstance(message, dict)
-                ]
-            )
-        self.media.augment_media_urls(data)
-        return _http_json_response(data)
-
     def _handle_webui_thread_get(self, request: WsRequest, key: str) -> Response:
         if not self.check_api_token(request):
             return _http_error(401, "Unauthorized")

tests/agent/test_auto_compact.py

@@ -80,8 +80,6 @@ def _make_fake_compact(
     track_archived: list | None = None,
     track_count: bool = False,
 ):
-    from nanobot.session.manager import Session as _Session
-
     state = {"count": 0}
 
     async def _fake_compact(key: str, *, runtime, max_suffix: int = 8) -> str:
@@ -92,25 +90,8 @@ def _make_fake_compact(
         if not tail:
             loop.sessions.save(session)
             return ""
-
+        archive_end = session.last_consolidated + len(tail)
-        probe = _Session(
+        archive_msgs = tail
-            key=session.key,
-            messages=tail.copy(),
-            created_at=session.created_at,
-            updated_at=session.updated_at,
-            metadata={},
-            last_consolidated=0,
-        )
-        result = probe.retain_recent_legal_suffix(
-            max_suffix,
-            extend_to_user=True,
-        )
-        visible_suffix = probe.messages
-        archive_msgs = result.dropped
-
-        if not archive_msgs:
-            loop.sessions.save(session)
-            return ""
 
         last_active = session.updated_at
         s = summary
@@ -126,7 +107,7 @@ def _make_fake_compact(
                 "last_active": last_active.isoformat(),
             }
 
-        session.last_consolidated = len(session.messages) - len(visible_suffix)
+        session.last_consolidated = archive_end
         loop.sessions.save(session)
         return s
 
@@ -365,7 +346,7 @@ class TestAutoCompact:
         await loop.close_mcp()
 
     @pytest.mark.asyncio
-    async def test_auto_compact_archives_prefix_without_deleting_history(self, tmp_path):
+    async def test_auto_compact_archives_full_tail_without_deleting_history(self, tmp_path):
         loop = _make_loop(tmp_path, session_ttl_minutes=15)
         session = loop.sessions.get_or_create("cli:test")
         _add_turns(session, 6)
@@ -378,7 +359,7 @@ class TestAutoCompact:
 
         await loop.auto_compact._archive("cli:test", runtime=loop.llm_runtime())
 
-        assert len(archived_messages) == 4
+        assert len(archived_messages) == 12
         session_after = loop.sessions.get_or_create("cli:test")
         assert len(session_after.messages) == 12
         assert session_after.messages[0]["content"] == "msg user 0"
@@ -473,7 +454,7 @@ class TestAutoCompact:
 
         await loop.auto_compact._archive("cli:test", runtime=loop.llm_runtime())
 
-        assert len(archived_messages) == 2
+        assert len(archived_messages) == 10
         await loop.close_mcp()
 
 
@@ -515,7 +496,7 @@ class TestAutoCompactIdleDetection:
         await loop._process_message(msg)
 
         session_after = loop.sessions.get_or_create("cli:test")
-        assert len(archived_messages) == 4
+        assert len(archived_messages) == 12
         assert any(m["content"] == "old user 0" for m in session_after.messages)
         assert not any(
             m["content"] == "old user 0"
@@ -724,7 +705,7 @@ class TestAutoCompactEdgeCases:
         await loop._process_message(msg)
 
         session_after = loop.sessions.get_or_create("cli:test")
-        assert archived_messages == []
+        assert [message["content"] for message in archived_messages] == ["previous message"]
         assert any(m["content"] == "previous message" for m in session_after.messages)
         assert any(m["content"] == "interrupted response" for m in session_after.messages)
 
@@ -912,7 +893,7 @@ class TestProactiveAutoCompact:
         assert len(session_after.get_history(max_messages=10)) == (
             loop.auto_compact._RECENT_SUFFIX_MESSAGES
         )
-        assert len(archived_messages) == 2
+        assert len(archived_messages) == 10
         entry = loop.auto_compact._summaries.get("cli:test")
         assert entry is not None
         assert entry[0] == "User chatted about old things."

tests/agent/test_autocompact_unit.py

@@ -405,13 +405,37 @@ class TestCheckExpired:
         scheduler.assert_not_called()
         assert "dream:20260602-155256" not in ac._archiving
 
-    def test_already_trimmed_session_skips(self):
+    def test_short_unarchived_session_schedules(self):
-        """Expired session with no removable tail should not be re-scheduled."""
+        """A short idle session still needs an archive entry for Dream."""
+        ac = _make_autocompact(ttl=15)
+        mock_sm = MagicMock(spec=SessionManager)
+        last_active = datetime(2026, 1, 1, 10, 0, 0)
+        session = _make_session("cli:short", updated_at=last_active)
+        _add_turns(session, 2)
+        mock_sm.list_sessions.return_value = [
+            {"key": "cli:short", "updated_at": last_active.isoformat()},
+        ]
+        mock_sm.get_or_create.return_value = session
+        ac.sessions = mock_sm
+
+        scheduled = []
+
+        def scheduler(coro):
+            scheduled.append(coro)
+            coro.close()
+
+        ac.check_expired(scheduler, _runtime)
+
+        assert len(scheduled) == 1
+        assert ac._archiving == {"cli:short"}
+
+    def test_fully_archived_session_skips(self):
         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)
+        session.last_consolidated = len(session.messages)
         mock_sm.list_sessions.return_value = [
             {"key": "cli:done", "updated_at": last_active.isoformat()},
         ]

tests/agent/test_consolidator.py

@@ -391,6 +391,25 @@ class TestConsolidatorTokenBudget:
         assert len(captured["history"]) == 160
         assert captured["history"][0]["content"].endswith("msg-0")
 
+    async def test_estimate_includes_recent_archived_replay(self, consolidator, runtime):
+        session = Session(key="test:archived-replay")
+        for i in range(10):
+            session.add_message("user", f"msg-{i}")
+        session.last_consolidated = len(session.messages)
+
+        captured: dict[str, list[dict]] = {}
+
+        def build_messages(**kwargs):
+            captured["history"] = kwargs["history"]
+            return kwargs["history"]
+
+        consolidator._build_messages = build_messages
+
+        consolidator.estimate_session_prompt_tokens(session, runtime=runtime)
+
+        assert len(captured["history"]) == 8
+        assert captured["history"][0]["content"] == "msg-2"
+
     async def test_replay_window_overflow_is_archived_even_under_token_budget(
         self,
         consolidator,
@@ -620,7 +639,7 @@ class TestCompactIdleSession:
         )
 
     @pytest.mark.asyncio
-    async def test_archives_prefix_preserves_messages_and_hides_prefix(
+    async def test_archives_full_tail_preserves_messages_and_replays_recent_suffix(
         self, real_consolidator, mock_provider, runtime
     ):
         mock_provider.chat_with_retry.return_value = MagicMock(
@@ -645,7 +664,7 @@ class TestCompactIdleSession:
         reloaded = sessions.get_or_create("cli:test")
         assert len(reloaded.messages) == 40
         assert reloaded.messages[0]["content"] == "user msg 0"
-        assert reloaded.last_consolidated == 32
+        assert reloaded.last_consolidated == 40
         assert reloaded.provider_state is None
         visible = reloaded.get_history(max_messages=40)
         assert len(visible) == 8
@@ -657,6 +676,82 @@ class TestCompactIdleSession:
         assert "last_active" in meta
         assert reloaded.updated_at == old_ts
 
+    @pytest.mark.asyncio
+    async def test_short_idle_session_archives_once(
+        self, real_consolidator, mock_provider, store, runtime
+    ):
+        mock_provider.chat_with_retry.return_value = MagicMock(
+            content="Short summary.", finish_reason="stop"
+        )
+        sessions = real_consolidator.sessions
+        session = sessions.get_or_create("cli:short")
+        session.add_message("user", "hello")
+        session.add_message("assistant", "hi")
+        sessions.save(session)
+
+        first = await real_consolidator.compact_idle_session("cli:short", runtime=runtime)
+        second = await real_consolidator.compact_idle_session("cli:short", runtime=runtime)
+
+        assert first == "Short summary."
+        assert second == ""
+        mock_provider.chat_with_retry.assert_awaited_once()
+        assert len(store.read_unprocessed_history(since_cursor=0)) == 1
+        reloaded = sessions.get_or_create("cli:short")
+        assert reloaded.last_consolidated == 2
+        assert [message["content"] for message in reloaded.get_history()] == ["hello", "hi"]
+
+    @pytest.mark.asyncio
+    async def test_new_messages_advance_existing_archive_progress(
+        self, real_consolidator, mock_provider, runtime
+    ):
+        mock_provider.chat_with_retry.return_value = MagicMock(
+            content="Summary.", finish_reason="stop"
+        )
+        sessions = real_consolidator.sessions
+        session = sessions.get_or_create("cli:incremental")
+        session.add_message("user", "first user")
+        session.add_message("assistant", "first assistant")
+        sessions.save(session)
+
+        await real_consolidator.compact_idle_session("cli:incremental", runtime=runtime)
+        current = sessions.get_or_create("cli:incremental")
+        current.add_message("user", "second user")
+        current.add_message("assistant", "second assistant")
+        sessions.save(current)
+        await real_consolidator.compact_idle_session("cli:incremental", runtime=runtime)
+
+        assert mock_provider.chat_with_retry.await_count == 2
+        latest_prompt = mock_provider.chat_with_retry.await_args_list[-1].kwargs["messages"][1][
+            "content"
+        ]
+        assert "second user" in latest_prompt
+        assert "first user" not in latest_prompt
+        assert sessions.get_or_create("cli:incremental").last_consolidated == 4
+
+    @pytest.mark.asyncio
+    async def test_concurrent_append_remains_unarchived(
+        self, real_consolidator, mock_provider, runtime
+    ):
+        sessions = real_consolidator.sessions
+        session = sessions.get_or_create("cli:concurrent")
+        session.add_message("user", "captured user")
+        session.add_message("assistant", "captured assistant")
+        sessions.save(session)
+
+        async def append_during_archive(**_kwargs):
+            current = sessions.get_or_create("cli:concurrent")
+            current.add_message("user", "late user")
+            current.add_message("assistant", "late assistant")
+            return LLMResponse(content="Summary.", finish_reason="stop")
+
+        mock_provider.chat_with_retry.side_effect = append_during_archive
+
+        await real_consolidator.compact_idle_session("cli:concurrent", runtime=runtime)
+
+        reloaded = sessions.get_or_create("cli:concurrent")
+        assert len(reloaded.messages) == 4
+        assert reloaded.last_consolidated == 2
+
     @pytest.mark.asyncio
     async def test_summarizes_retained_suffix_not_just_dropped_prefix(
         self, real_consolidator, mock_provider, runtime
@@ -686,10 +781,10 @@ class TestCompactIdleSession:
         assert "CORRECTED_FINAL_RESULT_alpha" in summarized
 
     @pytest.mark.asyncio
-    async def test_raw_dumps_only_dropped_messages_on_llm_failure(
+    async def test_raw_dumps_full_archive_batch_on_llm_failure(
         self, real_consolidator, mock_provider, store, runtime
     ):
-        """Extra summary context must not enter raw fallback. Regression for #4264."""
+        """The fallback covers the same full range as successful idle archival."""
         mock_provider.chat_with_retry.side_effect = RuntimeError("LLM unavailable")
         sessions = real_consolidator.sessions
         session = sessions.get_or_create("cli:rawdrop")
@@ -707,7 +802,7 @@ class TestCompactIdleSession:
         raw = "\n".join(e["content"] for e in store.read_unprocessed_history(since_cursor=0))
         assert "[RAW]" in raw
         assert "user msg 0" in raw
-        assert "RETAINED_SUFFIX_marker" not in raw
+        assert "RETAINED_SUFFIX_marker" in raw
         reloaded = sessions.get_or_create("cli:rawdrop")
         assert len(reloaded.messages) == 38
         assert reloaded.messages[-1]["content"] == "RETAINED_SUFFIX_marker"
@@ -805,8 +900,12 @@ class TestCompactIdleSession:
         reloaded = sessions.get_or_create("cli:fail")
         assert len(reloaded.messages) == 20
         assert reloaded.messages[0]["content"] == "u0"
-        assert reloaded.last_consolidated == 16
+        assert reloaded.last_consolidated == 20
         assert [m["content"] for m in reloaded.get_history(max_messages=20)] == [
+            "u6",
+            "a6",
+            "u7",
+            "a7",
             "u8",
             "a8",
             "u9",
@@ -835,10 +934,10 @@ class TestCompactIdleSession:
         assert result == "Tail summary."
         reloaded = sessions.get_or_create("cli:offset")
         assert len(reloaded.messages) == 60
-        assert reloaded.last_consolidated == 56
+        assert reloaded.last_consolidated == 60
 
         # Verify only the unconsolidated tail was processed:
-        # 10 unconsolidated messages (50-59), keep suffix of 4 → archive 6
+        # All 10 unconsolidated messages (50-59) are archived exactly once.
         archived_call = mock_provider.chat_with_retry.call_args
         user_content = archived_call.kwargs["messages"][1]["content"]
         # Should contain only tail messages, not early ones
@@ -846,7 +945,7 @@ class TestCompactIdleSession:
         assert "u25" in user_content or "a25" in user_content
 
     @pytest.mark.asyncio
-    async def test_extended_suffix_archives_only_hidden_prefix(
+    async def test_full_archive_keeps_extended_legal_replay_suffix(
         self,
         real_consolidator,
         mock_provider,
@@ -870,7 +969,7 @@ class TestCompactIdleSession:
 
         reloaded = sessions.get_or_create("cli:noncontiguous")
         assert len(reloaded.messages) == 25
-        assert reloaded.last_consolidated == 14
+        assert reloaded.last_consolidated == 25
         assert [m["content"] for m in reloaded.get_history(max_messages=25)] == [
             "user-14",
             "assistant-00",
@@ -1034,7 +1133,7 @@ class TestConsolidatorSessionRefresh:
 
         session_after = sessions.get_or_create("cli:test")
         assert len(session_after.messages) == 40
-        assert session_after.last_consolidated == 32
+        assert session_after.last_consolidated == 40
         assert len(session_after.get_history(max_messages=40)) == 8
 
 

tests/agent/test_git_store.py

@@ -169,19 +169,6 @@ class TestDiffCommits:
         assert git_ready.diff_commits("deadbeef", "cafebabe") == ""
 
 
-class TestFindCommit:
-    def test_finds_by_prefix(self, git_ready):
-        ws = git_ready._workspace
-        (ws / "SOUL.md").write_text("v2", encoding="utf-8")
-        sha = git_ready.auto_commit("v2")
-        found = git_ready.find_commit(sha[:4])
-        assert found is not None
-        assert found.sha == sha
-
-    def test_returns_none_for_unknown(self, git_ready):
-        assert git_ready.find_commit("deadbeef") is None
-
-
 class TestShowCommitDiff:
     def test_returns_commit_with_diff(self, git_ready):
         ws = git_ready._workspace

tests/agent/test_loop_save_turn.py

@@ -1074,9 +1074,8 @@ async def test_process_message_persists_media_paths_on_user_turn(tmp_path: Path)
     """User turns that attach images must record the media paths alongside
     the text so the webui can rehydrate previews on session replay.
 
-    This is the producer half of the signed-media-URL round-trip: paths are
+    The WebUI transcript replay can use these paths to restore attachment
-    stored here, then :meth:`WebSocketChannel._augment_media_urls` maps them
+    previews when it backfills from canonical session history.
-    onto signed URLs on the way out.
     """
     img_a = tmp_path / "uuid-1.png"
     img_a.write_bytes(_PNG_1X1)

tests/agent/test_loop_session_policy.py

@@ -0,0 +1,165 @@
+import asyncio
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from nanobot.agent.loop import AgentLoop
+from nanobot.bus.events import (
+    INBOUND_META_RUNTIME_CONTROL,
+    RUNTIME_CONTROL_SESSION_DISCARD,
+    InboundMessage,
+)
+from nanobot.bus.queue import MessageBus
+from nanobot.providers.base import GenerationSettings, LLMResponse
+from nanobot.session.keys import UNIFIED_SESSION_KEY
+
+
+def _message(key: str, content: str) -> InboundMessage:
+    return InboundMessage(
+        channel="websocket",
+        sender_id="user",
+        chat_id=key.removeprefix("websocket:"),
+        content=content,
+        session_key_override=key,
+        require_existing_session=True,
+    )
+
+
+def _loop(tmp_path, responses: list[str], **kwargs) -> AgentLoop:
+    provider = MagicMock()
+    provider.get_default_model.return_value = "test-model"
+    provider.generation = GenerationSettings()
+    provider.chat_with_retry = AsyncMock(
+        side_effect=[LLMResponse(content=response, usage={}) for response in responses]
+    )
+    return AgentLoop(
+        bus=MessageBus(),
+        provider=provider,
+        workspace=tmp_path,
+        model="test-model",
+        cron_service=MagicMock(),
+        **kwargs,
+    )
+
+
+@pytest.mark.asyncio
+async def test_transient_session_keeps_history_without_persisting_or_durable_tools(tmp_path) -> None:
+    loop = _loop(tmp_path, ["first answer", "second answer"])
+    loop.context.memory.write_memory("private durable memory")
+    loop.consolidator.maybe_consolidate_by_tokens = AsyncMock()
+    key = "websocket:transient-test"
+    loop.sessions.get_or_create_transient(
+        key,
+        disabled_tools={"create_goal", "update_goal", "spawn", "cron"},
+    )
+
+    await loop._process_message(_message(key, "first question"))
+    await loop._process_message(_message(key, "second question"))
+
+    calls = loop.provider.chat_with_retry.await_args_list
+    assert "private durable memory" not in str(calls[0].kwargs["messages"])
+    tool_names = {item["function"]["name"] for item in calls[0].kwargs["tools"]}
+    assert "read_session" in tool_names
+    assert {"create_goal", "update_goal", "spawn", "cron"}.isdisjoint(tool_names)
+    assert "first answer" in str(calls[1].kwargs["messages"])
+    session = loop.sessions.get_cached(key)
+    assert session is not None
+    assert [message["role"] for message in session.messages] == [
+        "user",
+        "assistant",
+        "user",
+        "assistant",
+    ]
+    assert loop.sessions.read_session_file(key) is None
+    loop.consolidator.maybe_consolidate_by_tokens.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_transient_session_stays_outside_unified_session(tmp_path) -> None:
+    loop = _loop(tmp_path, ["private answer"], unified_session=True)
+    durable = loop.sessions.get_or_create(UNIFIED_SESSION_KEY)
+    durable.add_message("user", "durable question")
+    loop.sessions.save(durable)
+    key = "websocket:transient-unified"
+    transient = loop.sessions.get_or_create_transient(key)
+
+    await loop._dispatch(_message(key, "private question"))
+
+    assert [message["content"] for message in transient.messages] == [
+        "private question",
+        "private answer",
+    ]
+    assert [message["content"] for message in durable.messages] == ["durable question"]
+    assert loop.sessions.read_session_file(key) is None
+
+
+@pytest.mark.asyncio
+async def test_missing_required_session_cannot_fall_back_to_disk(tmp_path) -> None:
+    loop = _loop(tmp_path, [])
+    key = "websocket:transient-stale"
+    loop.sessions.get_or_create_transient(key)
+    loop.sessions.invalidate(key)
+
+    with pytest.raises(RuntimeError, match="required session is not active"):
+        await loop._process_message(_message(key, "stale private message"))
+
+    loop.provider.chat_with_retry.assert_not_awaited()
+    assert loop.sessions.read_session_file(key) is None
+
+
+@pytest.mark.asyncio
+async def test_session_discard_control_cancels_active_turn(tmp_path, monkeypatch) -> None:
+    provider_started = asyncio.Event()
+
+    async def block_provider(**_kwargs: object) -> LLMResponse:
+        provider_started.set()
+        await asyncio.Event().wait()
+        raise AssertionError("provider blocker unexpectedly released")
+
+    loop = _loop(tmp_path, [])
+
+    async def wait_for_discard(key: str) -> None:
+        while loop.sessions.get_cached(key) is not None or key in loop._discarding_sessions:
+            await asyncio.sleep(0)
+
+    loop.provider.chat_with_retry = AsyncMock(side_effect=block_provider)
+    monkeypatch.setattr(loop, "_connect_mcp", AsyncMock())
+    monkeypatch.setattr(loop, "close_mcp", AsyncMock())
+    terminate_exec_sessions = AsyncMock(return_value=1)
+    monkeypatch.setattr(
+        loop._exec_session_manager,
+        "terminate_by_owner",
+        terminate_exec_sessions,
+    )
+    key = "websocket:transient-cancelled"
+    loop.sessions.get_or_create_transient(
+        key,
+        disabled_tools={"create_goal", "update_goal", "spawn", "cron"},
+    )
+    run_task = asyncio.create_task(loop.run())
+    await loop.bus.publish_inbound(_message(key, "private"))
+    await asyncio.wait_for(provider_started.wait(), timeout=2)
+    active_task = next(iter(loop._active_tasks[key]))
+
+    await loop.bus.publish_inbound(
+        InboundMessage(
+            channel="websocket",
+            sender_id="webui",
+            chat_id="transient-cancelled",
+            content="",
+            metadata={
+                INBOUND_META_RUNTIME_CONTROL: RUNTIME_CONTROL_SESSION_DISCARD,
+            },
+            session_key_override=key,
+        )
+    )
+
+    with pytest.raises(asyncio.CancelledError):
+        await asyncio.wait_for(active_task, timeout=2)
+    await asyncio.wait_for(wait_for_discard(key), timeout=2)
+    assert loop.sessions.get_cached(key) is None
+    terminate_exec_sessions.assert_awaited_once_with(key)
+
+    loop.stop()
+    await loop.bus.publish_inbound(_message(key, "wake"))
+    await asyncio.wait_for(run_task, timeout=2)

tests/agent/test_onboard_logic.py

@@ -1092,6 +1092,23 @@ class TestMainMenuUpdate:
         assert config.providers.openai.api_key == "${UNRELATED_MISSING_KEY}"
         assert config.providers.openai_codex.proxy == "${CODEX_PROXY}"
 
+    def test_quick_start_openai_codex_reports_incomplete_installation(self, monkeypatch):
+        import oauth_cli_kit
+
+        messages: list[str] = []
+        monkeypatch.delattr(oauth_cli_kit, "get_token")
+        monkeypatch.setattr(
+            onboard_wizard.console,
+            "print",
+            lambda message, *args, **kwargs: messages.append(str(message)),
+        )
+
+        assert onboard_wizard._quick_start_oauth_login(Config(), "openai_codex") is False
+        assert messages == [
+            "[red]This nanobot installation is missing the required oauth-cli-kit package. "
+            "Reinstall or upgrade nanobot-ai using the same installation method.[/red]"
+        ]
+
     def test_quick_start_openai_codex_runs_interactive_login_for_bad_cached_token(
         self, monkeypatch
     ):

tests/agent/test_runner_injections.py

@@ -9,6 +9,7 @@ from unittest.mock import AsyncMock, MagicMock, patch
 import pytest
 
 from agent.runner_helpers import make_run_spec
+from nanobot.agent.automation_turns import publish_next_deferred_turn
 from nanobot.config.schema import AgentDefaults
 from nanobot.providers.base import LLMResponse, ToolCallRequest
 
@@ -1047,7 +1048,11 @@ async def test_cron_turn_deferred_while_session_active(tmp_path):
     assert loop._cron_turns.deferred_queues[session_key] == [msg]
     assert loop.pending_cron_job_ids_for_session(session_key) == {"job-1"}
 
-    await loop._cron_turns.publish_next_deferred(session_key)
+    await publish_next_deferred_turn(
+        deferred_queues=loop._cron_turns.deferred_queues,
+        publish_inbound=loop.bus.publish_inbound,
+        session_key=session_key,
+    )
     queued = await asyncio.wait_for(loop.bus.consume_inbound(), timeout=0.5)
     assert queued is msg
     assert session_key not in loop._cron_turns.deferred_queues
@@ -1097,7 +1102,11 @@ async def test_local_trigger_turn_deferred_while_session_active(tmp_path):
     assert loop._local_trigger_turns.deferred_queues[session_key] == [msg]
     assert loop.pending_local_trigger_ids_for_session(session_key) == {"trg_123"}
 
-    assert await loop._local_trigger_turns.publish_next_deferred(session_key) is True
+    assert await publish_next_deferred_turn(
+        deferred_queues=loop._local_trigger_turns.deferred_queues,
+        publish_inbound=loop.bus.publish_inbound,
+        session_key=session_key,
+    ) is True
     queued = await asyncio.wait_for(loop.bus.consume_inbound(), timeout=0.5)
     assert queued is msg
     assert session_key not in loop._local_trigger_turns.deferred_queues

tests/agent/test_session_manager_history.py

@@ -208,6 +208,71 @@ def test_orphan_trim_with_last_consolidated():
     assert all(m.get("role") != "tool" or m["tool_call_id"].startswith("new_") for m in history)
 
 
+def test_get_history_replays_recent_messages_after_full_archive():
+    session = Session(key="test:fully-archived")
+    for i in range(10):
+        session.messages.append({"role": "user", "content": f"u{i}"})
+        session.messages.append({"role": "assistant", "content": f"a{i}"})
+    session.last_consolidated = len(session.messages)
+
+    history = session.get_history(max_messages=100)
+
+    assert [message["content"] for message in history] == [
+        "u6",
+        "a6",
+        "u7",
+        "a7",
+        "u8",
+        "a8",
+        "u9",
+        "a9",
+    ]
+
+
+def test_get_history_extends_compacted_replay_to_preceding_user():
+    session = Session(key="test:compacted-tool-turn")
+    session.messages.extend(
+        [
+            {"role": "user", "content": "old"},
+            {"role": "assistant", "content": "old answer"},
+            {"role": "user", "content": "run tools"},
+            *_tool_turn("keep", 0),
+            *_tool_turn("keep", 1),
+            *_tool_turn("keep", 2),
+            {"role": "assistant", "content": "done"},
+        ]
+    )
+    session.last_consolidated = len(session.messages)
+
+    history = session.get_history(max_messages=100)
+
+    assert history[0]["content"] == "run tools"
+    assert history[-1]["content"] == "done"
+    _assert_no_orphans(history)
+
+
+def test_compacted_tool_turn_can_extend_past_message_cap():
+    session = Session(key="test:long-compacted-tool-turn")
+    session.messages.extend(
+        [
+            {"role": "user", "content": "old"},
+            {"role": "assistant", "content": "old answer"},
+            {"role": "user", "content": "run many tools"},
+        ]
+    )
+    for i in range(50):
+        session.messages.extend(_tool_turn("keep", i))
+    session.messages.append({"role": "assistant", "content": "done"})
+    session.last_consolidated = len(session.messages)
+
+    history = session.get_history(max_messages=120)
+
+    assert len(history) > 120
+    assert history[0]["content"] == "run many tools"
+    assert history[-1]["content"] == "done"
+    _assert_no_orphans(history)
+
+
 # --- Edge: no tool messages at all ---
 
 def test_no_tool_messages_unchanged():

tests/agent/test_session_retention.py

@@ -0,0 +1,259 @@
+from nanobot.session.manager import Session
+
+
+def _assert_no_orphans(history: list[dict]) -> None:
+    declared = {
+        tc["id"]
+        for m in history
+        if m.get("role") == "assistant"
+        for tc in (m.get("tool_calls") or [])
+    }
+    orphans = [
+        m.get("tool_call_id")
+        for m in history
+        if m.get("role") == "tool" and m.get("tool_call_id") not in declared
+    ]
+    assert orphans == [], f"orphan tool_call_ids: {orphans}"
+
+
+def _delivery(content: str) -> dict:
+    return {"role": "assistant", "content": content, "_channel_delivery": True}
+
+
+def _tool_turn(prefix: str, idx: int) -> list[dict]:
+    return [
+        {
+            "role": "assistant",
+            "content": None,
+            "tool_calls": [
+                {
+                    "id": f"{prefix}_{idx}_a",
+                    "type": "function",
+                    "function": {"name": "x", "arguments": "{}"},
+                },
+                {
+                    "id": f"{prefix}_{idx}_b",
+                    "type": "function",
+                    "function": {"name": "y", "arguments": "{}"},
+                },
+            ],
+        },
+        {"role": "tool", "tool_call_id": f"{prefix}_{idx}_a", "name": "x", "content": "ok"},
+        {"role": "tool", "tool_call_id": f"{prefix}_{idx}_b", "name": "y", "content": "ok"},
+    ]
+
+
+def _contents(messages: list[dict]) -> list[str]:
+    return [m.get("content") for m in messages]
+
+
+def _has_delivery(messages: list[dict]) -> bool:
+    return any(m.get("_channel_delivery") for m in messages)
+
+
+# --- Hard-cap trimming must preserve a proactive delivery the user replied to ---
+
+
+def test_retain_hard_cap_keeps_delivery_before_user():
+    session = Session(key="test:cap-delivery")
+    session.messages.append({"role": "user", "content": "setup"})
+    session.messages.append(_delivery("Remember to drink water"))
+    session.messages.append({"role": "user", "content": "ok"})
+    session.messages.append({"role": "assistant", "content": "great"})
+
+    session.retain_recent_legal_suffix(3)
+
+    assert _has_delivery(session.messages), "delivery dropped by hard-cap trim"
+    assert _contents(session.messages) == [
+        "Remember to drink water",
+        "ok",
+        "great",
+    ]
+
+
+def test_retain_hard_cap_matches_get_history_boundary():
+    """The trimmed suffix must start on the same message as get_history()."""
+    session = Session(key="test:cap-boundary")
+    session.messages.append({"role": "user", "content": "setup"})
+    session.messages.append(_delivery("You have 3 pending tasks"))
+    session.messages.append({"role": "user", "content": "show them"})
+    session.messages.append({"role": "assistant", "content": "done"})
+
+    expected = session.get_history(max_messages=3)
+
+    session.retain_recent_legal_suffix(3)
+
+    assert _contents(session.messages) == _contents(expected)
+
+
+def test_retain_extend_to_user_keeps_delivery_before_recovered_user():
+    session = Session(key="test:extend-delivery")
+    session.messages.append({"role": "user", "content": "setup"})
+    session.messages.append({"role": "assistant", "content": "work"})
+    session.messages.append(_delivery("Reminder: deploy at 17:00"))
+    session.messages.append({"role": "user", "content": "ok"})
+    session.messages.append({"role": "assistant", "content": "a1"})
+    session.messages.append({"role": "assistant", "content": "a2"})
+    session.messages.append({"role": "assistant", "content": "a3"})
+
+    session.retain_recent_legal_suffix(3, extend_to_user=True)
+
+    assert _has_delivery(session.messages), "delivery dropped by extend_to_user trim"
+    assert session.messages[0]["content"] == "Reminder: deploy at 17:00"
+    assert session.messages[-1]["content"] == "a3"
+
+
+def test_retain_extend_to_user_matches_get_history_boundary():
+    session = Session(key="test:extend-boundary")
+    session.messages.append({"role": "user", "content": "setup"})
+    session.messages.append({"role": "assistant", "content": "work"})
+    session.messages.append(_delivery("Reminder: review the draft"))
+    session.messages.append({"role": "user", "content": "ok"})
+    session.messages.append({"role": "assistant", "content": "a1"})
+    session.messages.append({"role": "assistant", "content": "a2"})
+    session.messages.append({"role": "assistant", "content": "a3"})
+
+    expected = session.get_history(max_messages=3, extend_to_user=True)
+
+    session.retain_recent_legal_suffix(3, extend_to_user=True)
+
+    assert _contents(session.messages) == _contents(expected)
+
+
+def test_retain_extend_to_user_does_not_extend_delivery_only_tail():
+    session = Session(key="test:extend-no-user")
+    for i in range(4):
+        session.messages.append(_delivery(f"notification {i}"))
+
+    session.retain_recent_legal_suffix(3, extend_to_user=True)
+
+    assert _contents(session.messages) == [
+        "notification 1",
+        "notification 2",
+        "notification 3",
+    ]
+
+
+# --- Only the immediately-preceding delivery is part of the anchor ---
+
+
+def test_retain_keeps_only_immediate_delivery():
+    session = Session(key="test:multi-delivery")
+    session.messages.append({"role": "user", "content": "setup"})
+    session.messages.append(_delivery("old scheduled note"))
+    session.messages.append(_delivery("new scheduled note"))
+    session.messages.append({"role": "user", "content": "ok"})
+    session.messages.append({"role": "assistant", "content": "great"})
+
+    session.retain_recent_legal_suffix(3)
+
+    kept = _contents(session.messages)
+    assert kept == ["new scheduled note", "ok", "great"], kept
+
+
+def test_retain_drops_delivery_not_adjacent_to_anchor_user():
+    """A delivery that does not immediately precede the retained user turn is
+    not part of the anchor and should not be force-retained."""
+    session = Session(key="test:nonadjacent")
+    session.messages.append({"role": "user", "content": "setup"})
+    session.messages.append(_delivery("unrelated scheduled note"))
+    session.messages.append({"role": "assistant", "content": "reply"})
+    session.messages.append({"role": "user", "content": "ok"})
+    session.messages.append({"role": "assistant", "content": "great"})
+
+    session.retain_recent_legal_suffix(2)
+
+    assert not _has_delivery(session.messages)
+    assert _contents(session.messages) == ["ok", "great"]
+
+
+# --- Delivery preservation through the production entry points ---
+
+
+def test_enforce_file_cap_keeps_delivery_in_session():
+    session = Session(key="test:cap-delivery")
+    session.messages.append({"role": "user", "content": "setup"})
+    session.messages.append(_delivery("Remember to drink water"))
+    session.messages.append({"role": "user", "content": "ok"})
+    session.messages.append({"role": "assistant", "content": "great"})
+
+    archived: list[list[dict]] = []
+    session.enforce_file_cap(on_archive=archived.append, limit=3)
+
+    archived_flat = [m for chunk in archived for m in chunk]
+    assert _has_delivery(session.messages)
+    assert not any(m.get("_channel_delivery") for m in archived_flat)
+
+
+def test_enforce_file_cap_archives_only_prefix():
+    session = Session(key="test:cap-prefix")
+    session.messages.append({"role": "user", "content": "setup"})
+    session.messages.append({"role": "assistant", "content": "first reply"})
+    session.messages.append(_delivery("Remember to drink water"))
+    session.messages.append({"role": "user", "content": "ok"})
+    session.messages.append({"role": "assistant", "content": "great"})
+
+    archived: list[list[dict]] = []
+    session.enforce_file_cap(on_archive=archived.append, limit=3)
+
+    archived_flat = [m for chunk in archived for m in chunk]
+    assert _has_delivery(session.messages)
+    assert _contents(archived_flat) == ["setup", "first reply"]
+
+
+def test_compact_probe_keeps_delivery_in_visible_suffix():
+    """compact_idle_session() trims a probe copy with extend_to_user=True; the
+    visible suffix it keeps must still contain the delivery message."""
+    tail = [
+        {"role": "user", "content": "setup"},
+        {"role": "assistant", "content": "work"},
+        _delivery("Reminder: deploy at 17:00"),
+        {"role": "user", "content": "ok"},
+        {"role": "assistant", "content": "a1"},
+        {"role": "assistant", "content": "a2"},
+        {"role": "assistant", "content": "a3"},
+    ]
+    probe = Session(key="test:probe", messages=tail, last_consolidated=0)
+
+    probe.retain_recent_legal_suffix(3, extend_to_user=True)
+
+    assert _has_delivery(probe.messages)
+    assert probe.messages[0]["content"] == "Reminder: deploy at 17:00"
+
+
+# --- Trimming must stay coherent with the rest of replay ---
+
+
+def test_retain_then_replay_keeps_delivery_and_no_orphans():
+    session = Session(key="test:replay-after-trim")
+    session.messages.append({"role": "user", "content": "setup"})
+    session.messages.append(_delivery("You have 3 pending tasks"))
+    session.messages.append({"role": "user", "content": "show them"})
+    session.messages.extend(_tool_turn("cur", 0))
+    session.messages.append({"role": "assistant", "content": "done"})
+
+    session.retain_recent_legal_suffix(6)
+
+    assert _has_delivery(session.messages)
+    history = session.get_history(max_messages=500)
+    _assert_no_orphans(history)
+    assert any(m.get("content") == "You have 3 pending tasks" for m in history)
+
+
+def test_retain_keeps_delivery_when_user_inside_window():
+    """When the capped window already contains a user, its immediately
+    preceding delivery must stay attached to it."""
+    session = Session(key="test:window-user")
+    session.messages.append({"role": "user", "content": "setup"})
+    session.messages.append({"role": "assistant", "content": "a0"})
+    session.messages.append(_delivery("Reminder"))
+    session.messages.append({"role": "user", "content": "ok"})
+    session.messages.append({"role": "assistant", "content": "a1"})
+    session.messages.append({"role": "assistant", "content": "a2"})
+
+    expected = session.get_history(max_messages=4)
+
+    session.retain_recent_legal_suffix(4)
+
+    assert _has_delivery(session.messages)
+    assert _contents(session.messages) == _contents(expected)

tests/agent/test_tool_hint.py

@@ -84,10 +84,6 @@ class TestToolHintKnownTools:
         assert '"C:/Program Files/Git/project"' not in result
         assert '"' in result
 
-    def test_exec_short_command_unchanged(self):
-        result = _hint([_tc("exec", {"command": "npm install typescript"})])
-        assert result == "$ npm install typescript"
-
     def test_exec_chained_commands_truncated_not_mid_path(self):
         """Long chained commands should truncate preserving abbreviated paths."""
         cmd = "cd D:\\Documents\\GitHub\\project && npm run build && npm test"

tests/channels/test_channel_manager_delta_coalescing.py

@@ -17,6 +17,7 @@ from nanobot.bus.outbound_events import (
 from nanobot.bus.queue import MessageBus
 from nanobot.channels.base import BaseChannel
 from nanobot.channels.manager import ChannelManager
+from nanobot.channels.mattermost.runtime import MattermostChannel
 from nanobot.config.schema import Config
 
 
@@ -311,6 +312,38 @@ class TestProgressFiltering:
         assert manager._should_send_progress("mock", tool_hint=False) is False
         assert manager._should_send_progress("mock", tool_hint=True) is False
 
+    def test_channel_config_defaults_do_not_override_global_policy(self, bus):
+        manager = ChannelManager.__new__(ChannelManager)
+        manager.config = Config.model_validate({
+            "channels": {
+                "sendProgress": False,
+                "sendToolHints": False,
+            },
+        })
+        manager.bus = bus
+
+        channel = manager._build_channel(
+            "mattermost",
+            MattermostChannel,
+            {"enabled": True},
+        )
+
+        assert channel.send_progress is False
+        assert channel.send_tool_hints is False
+
+        opted_in = manager._build_channel(
+            "mattermost",
+            MattermostChannel,
+            {
+                "enabled": True,
+                "sendProgress": True,
+                "sendToolHints": True,
+            },
+        )
+
+        assert opted_in.send_progress is True
+        assert opted_in.send_tool_hints is True
+
     def test_progress_visibility_returns_false_for_missing_channel(self, manager):
         assert manager._should_send_progress("nonexistent", tool_hint=False) is False
         assert manager._should_send_progress("nonexistent", tool_hint=True) is False

tests/channels/test_channel_manager_reasoning.py

@@ -241,24 +241,6 @@ async def test_file_edit_events_route_to_channel_capability(manager):
     channel._send_mock.assert_not_awaited()
 
 
-@pytest.mark.asyncio
-async def test_typed_file_edit_event_routes_to_channel_capability(manager):
-    channel = manager.channels["mock"]
-    edits = [{"version": 1, "phase": "start", "path": "src/app.py"}]
-    msg = outbound_message_for_event(
-        channel="mock",
-        chat_id="c1",
-        event=ProgressEvent(file_edit_events=edits),
-    )
-
-    await manager._send_once(channel, msg)
-
-    channel._file_edit_mock.assert_awaited_once_with(
-        "c1", edits, msg.metadata
-    )
-    channel._send_mock.assert_not_awaited()
-
-
 @pytest.mark.asyncio
 async def test_base_channel_file_edit_events_are_noop_safe():
     class _Plain(BaseChannel):

tests/cli/test_agent_interactive.py

@@ -0,0 +1,155 @@
+from __future__ import annotations
+
+import asyncio
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+from typer.testing import CliRunner
+
+from nanobot.bus.events import InboundMessage, OutboundMessage
+from nanobot.bus.outbound_events import (
+    StreamDeltaEvent,
+    StreamedResponseEvent,
+    StreamEndEvent,
+    outbound_message_for_event,
+)
+from nanobot.cli.commands import app
+from nanobot.config.schema import Config
+
+runner = CliRunner()
+
+
+@pytest.mark.parametrize("streamed", [False, True])
+def test_interactive_agent_routes_a_complete_user_turn(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path,
+    streamed: bool,
+) -> None:
+    config = Config()
+    config.agents.defaults.workspace = str(tmp_path / "workspace")
+    seen: dict[str, object] = {}
+    renderers: list[object] = []
+
+    class _Renderer:
+        def __init__(self, **_kwargs: object) -> None:
+            self.streamed = False
+            self.header_printed = False
+            self.deltas: list[str] = []
+            self.ends: list[bool] = []
+            self.closed = 0
+            renderers.append(self)
+
+        async def on_delta(self, content: str) -> None:
+            self.streamed = True
+            self.deltas.append(content)
+
+        async def on_end(self, *, resuming: bool = False) -> None:
+            self.ends.append(resuming)
+
+        async def close(self) -> None:
+            self.closed += 1
+
+        def stop_for_input(self) -> None:
+            return None
+
+    class _AgentLoop:
+        channels_config = None
+
+        @classmethod
+        def from_config(cls, _config, bus, **_kwargs):
+            instance = cls(bus)
+            seen["loop"] = instance
+            return instance
+
+        def __init__(self, bus) -> None:
+            self.bus = bus
+            self.stopped = asyncio.Event()
+            self.close_mcp_calls = 0
+
+        async def run(self) -> None:
+            message = await self.bus.consume_inbound()
+            seen["inbound"] = message
+            if streamed:
+                for event in (
+                    StreamDeltaEvent(content="hello "),
+                    StreamDeltaEvent(content="world"),
+                    StreamEndEvent(),
+                    StreamedResponseEvent(),
+                ):
+                    await self.bus.publish_outbound(
+                        outbound_message_for_event(
+                            channel=message.channel,
+                            chat_id=message.chat_id,
+                            event=event,
+                            content="hello world" if isinstance(event, StreamedResponseEvent) else None,
+                        )
+                    )
+            else:
+                await self.bus.publish_outbound(
+                    OutboundMessage(
+                        channel=message.channel,
+                        chat_id=message.chat_id,
+                        content="hello world",
+                    )
+                )
+            await self.stopped.wait()
+
+        def stop(self) -> None:
+            self.stopped.set()
+
+        async def close_mcp(self) -> None:
+            self.close_mcp_calls += 1
+
+    read_input = AsyncMock(side_effect=["hello nanobot", "exit"])
+    print_response = MagicMock()
+    monkeypatch.setattr("nanobot.cli.agent._load_runtime_config", lambda *_args: config)
+    monkeypatch.setattr("nanobot.cli.agent.sync_workspace_templates", lambda *_args: None)
+    monkeypatch.setattr("nanobot.cli.agent.is_default_workspace", lambda *_args: False)
+    monkeypatch.setattr("nanobot.cli.agent._set_nanobot_logs", lambda *_args: None)
+    monkeypatch.setattr("nanobot.cli.agent._model_display", lambda *_args: ("test-model", ""))
+    monkeypatch.setattr("nanobot.cli.agent.consume_restart_notice_from_env", lambda: None)
+    monkeypatch.setattr("nanobot.cli.agent.AgentLoop", _AgentLoop)
+    monkeypatch.setattr("nanobot.cli.agent.StreamRenderer", _Renderer)
+    monkeypatch.setattr("nanobot.providers.factory.make_provider", lambda *_args: object())
+    monkeypatch.setattr(
+        "nanobot.providers.image_generation.image_gen_provider_configs",
+        lambda *_args: [],
+    )
+    monkeypatch.setattr("nanobot.cron.service.CronService", lambda *_args: object())
+    monkeypatch.setattr("nanobot.cli.agent.signal.signal", lambda *_args: None)
+    monkeypatch.setattr("nanobot.cli.terminal._init_prompt_session", lambda: None)
+    monkeypatch.setattr("nanobot.cli.terminal._flush_pending_tty_input", lambda: None)
+    monkeypatch.setattr("nanobot.cli.terminal._restore_terminal", lambda: None)
+    monkeypatch.setattr("nanobot.cli.terminal._read_interactive_input_async", read_input)
+    monkeypatch.setattr("nanobot.cli.terminal._print_agent_response", print_response)
+
+    result = runner.invoke(app, ["agent", "--session", "cli:journey"])
+
+    assert result.exit_code == 0, result.output
+    inbound = seen["inbound"]
+    assert isinstance(inbound, InboundMessage)
+    assert (inbound.channel, inbound.chat_id, inbound.content) == (
+        "cli",
+        "journey",
+        "hello nanobot",
+    )
+    assert inbound.metadata == {"_wants_stream": True}
+    loop = seen["loop"]
+    assert isinstance(loop, _AgentLoop)
+    assert loop.close_mcp_calls == 1
+    assert len(renderers) == 1
+    renderer = renderers[0]
+    assert isinstance(renderer, _Renderer)
+    if streamed:
+        assert renderer.deltas == ["hello ", "world"]
+        assert renderer.ends == [False]
+        assert renderer.closed == 0
+        print_response.assert_not_called()
+    else:
+        assert renderer.deltas == []
+        assert renderer.closed == 1
+        print_response.assert_called_once_with(
+            "hello world",
+            render_markdown=True,
+            metadata={},
+        )

tests/cli/test_commands.py

@@ -686,7 +686,10 @@ def test_provider_login_openai_codex_handles_missing_oauth_symbol(monkeypatch):
     result = runner.invoke(app, ["provider", "login", "openai-codex"])
 
     assert result.exit_code == 1
-    assert "oauth_cli_kit not installed" in result.stdout
+    assert (
+        "This nanobot installation is missing the required oauth-cli-kit package. "
+        "Reinstall or upgrade nanobot-ai using the same installation method."
+    ) in re.sub(r"\s+", " ", result.stdout)
     assert result.exception is not None
 
 
@@ -1228,27 +1231,6 @@ def test_openai_compat_provider_passes_model_through():
     assert provider.get_default_model() == "github-copilot/gpt-5.3-codex"
 
 
-def test_make_provider_uses_github_copilot_backend():
-    from nanobot.config.schema import Config
-    from nanobot.providers.factory import make_provider
-
-    config = Config.model_validate(
-        {
-            "agents": {
-                "defaults": {
-                    "provider": "github-copilot",
-                    "model": "github-copilot/gpt-4.1",
-                }
-            }
-        }
-    )
-
-    with patch("nanobot.providers.openai_compat_provider.AsyncOpenAI"):
-        provider = make_provider(config)
-
-    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(

tests/cli_apps/test_utils.py

@@ -1,9 +1,7 @@
 """Tests for CLI Apps loop helpers."""
 
-from types import SimpleNamespace
-
 from nanobot.apps.cli.service import CliAppManager
-from nanobot.apps.cli.utils import runtime_lines, session_extra
+from nanobot.apps.cli.utils import runtime_lines_for_request, session_extra
 
 
 def test_session_extra_returns_cli_apps_only_when_present() -> None:
@@ -30,8 +28,9 @@ def test_cli_app_mentions_inject_runtime_metadata(tmp_path, monkeypatch):
         }
     )
 
-    lines = runtime_lines(
+    lines = runtime_lines_for_request(
-        SimpleNamespace(content="please use @zoom tonight; ignore @krita?", metadata={}),
+        "please use @zoom tonight; ignore @krita?",
+        {},
         tmp_path,
     )
 
@@ -43,17 +42,15 @@ def test_cli_app_mentions_inject_runtime_metadata(tmp_path, monkeypatch):
 
 
 def test_structured_cli_app_attachment_injects_runtime_metadata(tmp_path):
-    lines = runtime_lines(
+    lines = runtime_lines_for_request(
-        SimpleNamespace(
+        "please use @zoom tonight",
-            content="please use @zoom tonight",
+        {
-            metadata={
             "cli_apps": [{
                 "name": "zoom",
                 "entry_point": "cli-anything-zoom",
                 "display_name": "Zoom",
             }],
         },
-        ),
         tmp_path,
     )
 

tests/providers/test_github_copilot_routing.py

@@ -48,6 +48,7 @@ def test_build_responses_body_strips_github_copilot_prefix():
         provider_context=ProviderCallContext(context_window_tokens=128_000),
     )
     assert body["model"] == "gpt-5.4-mini"
+    assert body["include"] == ["reasoning.encrypted_content"]
     assert "context_management" not in body
 
 

tests/providers/test_openai_compat_timeout.py

@@ -1,5 +1,6 @@
 from unittest.mock import patch, sentinel
 
+from nanobot.providers import openai_compat_provider
 from nanobot.providers.openai_compat_provider import OpenAICompatProvider
 from nanobot.providers.registry import ProviderSpec
 
@@ -59,3 +60,22 @@ async def test_openai_compat_provider_timeout_can_be_overridden_by_env(monkeypat
         await provider._ensure_client()
 
     assert mock_async_openai.call_args.kwargs["timeout"] == 45.0
+
+
+async def test_missing_langfuse_warning_recommends_plugin_command(monkeypatch) -> None:
+    monkeypatch.setenv("LANGFUSE_SECRET_KEY", "secret")
+    monkeypatch.setattr(openai_compat_provider, "AsyncOpenAI", None)
+
+    with (
+        patch("importlib.util.find_spec", return_value=None),
+        patch("openai.AsyncOpenAI") as mock_async_openai,
+        patch("nanobot.providers.openai_compat_provider.logger.warning") as mock_warning,
+    ):
+        provider = OpenAICompatProvider(api_key="test-key", api_base="https://example.com/v1")
+        await provider._ensure_client()
+
+    mock_warning.assert_called_once_with(
+        "LANGFUSE_SECRET_KEY is set but langfuse is not installed; "
+        "run `nanobot plugins enable langfuse` to enable tracing"
+    )
+    mock_async_openai.assert_called_once()

tests/providers/test_responses_circuit_breaker.py

@@ -11,6 +11,11 @@ from nanobot.providers.openai_compat_provider import (
     OpenAICompatProvider,
 )
 from nanobot.providers.openai_responses.state import build_responses_state
+from nanobot.providers.registry import (
+    ProviderSpec,
+    ResponsesCapabilities,
+    find_by_name,
+)
 
 
 @pytest.fixture()
@@ -18,7 +23,7 @@ def provider():
     """A direct-OpenAI provider with Responses API support."""
     p = OpenAICompatProvider.__new__(OpenAICompatProvider)
     p.default_model = "gpt-5"
-    p._spec = type("Spec", (), {"name": "openai"})()
+    p._spec = find_by_name("openai")
     p._effective_base = "https://api.openai.com/v1"
     p._api_type = "auto"
     p._responses_failures = {}
@@ -31,12 +36,7 @@ def test_responses_api_available_by_default(provider):
 
 
 def test_deepseek_v4_flash_uses_responses_by_model(provider):
-    provider._spec = type("Spec", (), {
+    provider._spec = find_by_name("deepseek")
-        "name": "deepseek",
-        "responses_models": ("deepseek-v4-flash",),
-        "strip_model_prefix": False,
-        "strip_model_prefixes": (),
-    })()
     provider._effective_base = "https://api.deepseek.com"
     provider.default_model = "deepseek-v4-flash"
 
@@ -45,17 +45,51 @@ def test_deepseek_v4_flash_uses_responses_by_model(provider):
 
 
 def test_deepseek_v4_flash_matches_provider_prefixed_model(provider):
-    provider._spec = type("Spec", (), {
+    provider._spec = find_by_name("deepseek")
-        "name": "deepseek",
-        "responses_models": ("deepseek-v4-flash",),
-        "strip_model_prefix": False,
-        "strip_model_prefixes": (),
-    })()
     provider._effective_base = "https://api.deepseek.com"
 
     assert provider._should_use_responses_api("deepseek/deepseek-v4-flash", None) is True
 
 
+def test_responses_behavior_is_declared_by_capabilities(provider):
+    provider._spec = ProviderSpec(
+        name="example",
+        keywords=("example",),
+        env_key="EXAMPLE_API_KEY",
+        responses=ResponsesCapabilities(
+            models=("example-o3",),
+            reasoning_replay="plaintext",
+        ),
+    )
+    provider._effective_base = "https://example.test"
+
+    assert provider._should_use_responses_api("example-o3", None) is True
+
+    body = provider._build_responses_body(
+        messages=[
+            {"role": "user", "content": "question"},
+            {
+                "role": "assistant",
+                "reasoning_content": "think first",
+                "content": "answer",
+            },
+            {"role": "user", "content": "follow-up"},
+        ],
+        tools=None,
+        model="example-o3",
+        max_tokens=100,
+        temperature=0.1,
+        reasoning_effort="high",
+        tool_choice=None,
+    )
+
+    assert {
+        "type": "reasoning",
+        "content": [{"type": "output_text", "text": "think first"}],
+    } in body["input"]
+    assert "include" not in body
+
+
 def test_direct_openai_enables_server_compaction(provider):
     provider._extra_body = {}
 
@@ -74,6 +108,7 @@ def test_direct_openai_enables_server_compaction(provider):
         "type": "compaction",
         "compact_threshold": 70_000,
     }]
+    assert body["include"] == ["reasoning.encrypted_content"]
 
 
 def test_api_type_chat_completions_disables_responses(provider):
@@ -97,7 +132,7 @@ def test_api_type_responses_ignores_circuit_breaker(provider):
 
 
 def test_api_type_responses_does_not_force_non_openai(provider):
-    provider._spec = type("Spec", (), {"name": "custom"})()
+    provider._spec = find_by_name("custom")
     provider._api_type = "responses"
 
     assert provider._should_use_responses_api("gpt-4o", None) is False
@@ -192,12 +227,15 @@ def test_legacy_compatibility_markers_still_trigger_fallback():
 
 
 def _deepseek_provider(provider):
-    provider._spec = type("Spec", (), {
+    provider._spec = ProviderSpec(
-        "name": "deepseek",
+        name="deepseek",
-        "responses_models": ("deepseek-v4-flash",),
+        keywords=("deepseek",),
-        "strip_model_prefix": False,
+        env_key="DEEPSEEK_API_KEY",
-        "strip_model_prefixes": (),
+        responses=ResponsesCapabilities(
-    })()
+            models=("deepseek-v4-flash",),
+            reasoning_replay="plaintext",
+        ),
+    )
     provider._effective_base = "https://api.deepseek.com"
     provider.default_model = "deepseek-v4-flash"
     provider._extra_body = {}

tests/session/test_consolidated_offset_clamp.py

@@ -57,7 +57,7 @@ def test_loaded_corrupt_offset_keeps_messages(tmp_path: Path):
 def test_valid_offset_is_preserved():
     session = _session(10, 4)
     assert session.last_consolidated == 4
-    assert len(session.get_history()) == 6
+    assert len(session.get_history()) == 8
 
 
 def test_loaded_null_metadata_becomes_empty_dict(tmp_path: Path):

tests/session/test_session_cache.py

@@ -73,3 +73,17 @@ def test_flush_all_includes_live_sessions_outside_strong_cache(tmp_path, monkeyp
 
     assert manager.flush_all() == 2
     assert set(saved) == {("test:active", True), ("test:other", True)}
+
+
+def test_transient_session_never_reaches_storage(tmp_path) -> None:
+    manager = SessionManager(tmp_path)
+    session = manager.get_or_create_transient("websocket:temporary-test")
+    session.add_message("user", "secret")
+
+    manager.save(session, fsync=True)
+
+    assert manager.get_cached(session.key) is session
+    assert manager.read_session_file(session.key) is None
+    assert list(manager.sessions_dir.glob("*.jsonl")) == []
+    manager.invalidate(session.key)
+    assert manager.get_cached(session.key) is None

tests/test_document_parsing.py

@@ -6,7 +6,6 @@ from zipfile import ZipFile
 import pytest
 
 from nanobot.utils.document import (
-    SUPPORTED_EXTENSIONS,
     PdfSafetyError,
     _is_text_extension,
     extract_pdf_pages,
@@ -14,31 +13,6 @@ from nanobot.utils.document import (
 )
 
 
-class TestSupportedExtensions:
-    """Test the SUPPORTED_EXTENSIONS constant."""
-
-    def test_supported_extensions_include_common_formats(self):
-        """Test that common document formats are included."""
-        # Document formats
-        assert ".pdf" in SUPPORTED_EXTENSIONS
-        assert ".docx" in SUPPORTED_EXTENSIONS
-        assert ".xlsx" in SUPPORTED_EXTENSIONS
-        assert ".pptx" in SUPPORTED_EXTENSIONS
-
-        # Text formats
-        assert ".txt" in SUPPORTED_EXTENSIONS
-        assert ".md" in SUPPORTED_EXTENSIONS
-        assert ".csv" in SUPPORTED_EXTENSIONS
-        assert ".json" in SUPPORTED_EXTENSIONS
-        assert ".yaml" in SUPPORTED_EXTENSIONS
-        assert ".yml" in SUPPORTED_EXTENSIONS
-
-        # Image formats
-        assert ".png" in SUPPORTED_EXTENSIONS
-        assert ".jpg" in SUPPORTED_EXTENSIONS
-        assert ".jpeg" in SUPPORTED_EXTENSIONS
-
-
 class TestExtractText:
     """Test the extract_text function."""
 

tests/test_nanobot_facade.py

@@ -243,27 +243,6 @@ def test_workspace_override(tmp_path):
     assert bot._loop.workspace == custom_ws
 
 
-def test_sdk_make_provider_uses_github_copilot_backend():
-    from nanobot.config.schema import Config
-    from nanobot.providers.factory import make_provider
-
-    config = Config.model_validate(
-        {
-            "agents": {
-                "defaults": {
-                    "provider": "github-copilot",
-                    "model": "github-copilot/gpt-4.1",
-                }
-            }
-        }
-    )
-
-    with patch("nanobot.providers.openai_compat_provider.AsyncOpenAI"):
-        provider = make_provider(config)
-
-    assert provider.__class__.__name__ == "GitHubCopilotProvider"
-
-
 @pytest.mark.asyncio
 async def test_run_custom_session_key(tmp_path):
     from nanobot.bus.events import OutboundMessage

tests/tools/test_edit_advanced.py

@@ -13,7 +13,7 @@ import os
 import pytest
 
 from nanobot.agent.tools import file_state
-from nanobot.agent.tools.filesystem import EditFileTool, ReadFileTool, _find_match
+from nanobot.agent.tools.filesystem import EditFileTool, ReadFileTool
 
 
 @pytest.fixture(autouse=True)
@@ -68,41 +68,6 @@ class TestDeleteLineCleanup:
 # ---------------------------------------------------------------------------
 
 
-class TestSmartQuoteNormalization:
-    """_find_match should handle curly ↔ straight quote fallback."""
-
-    def test_curly_double_quotes_match_straight(self):
-        content = 'She said \u201chello\u201d to him'
-        old_text = 'She said "hello" to him'
-        match, count = _find_match(content, old_text)
-        assert match is not None
-        assert count == 1
-        # Returned match should be the ORIGINAL content with curly quotes
-        assert "\u201c" in match
-
-    def test_curly_single_quotes_match_straight(self):
-        content = "it\u2019s a test"
-        old_text = "it's a test"
-        match, count = _find_match(content, old_text)
-        assert match is not None
-        assert count == 1
-        assert "\u2019" in match
-
-    def test_straight_matches_curly_in_old_text(self):
-        content = 'x = "hello"'
-        old_text = 'x = \u201chello\u201d'
-        match, count = _find_match(content, old_text)
-        assert match is not None
-        assert count == 1
-
-    def test_exact_match_still_preferred_over_quote_normalization(self):
-        content = 'x = "hello"'
-        old_text = 'x = "hello"'
-        match, count = _find_match(content, old_text)
-        assert match == old_text
-        assert count == 1
-
-
 class TestQuoteStylePreservation:
     """When quote-normalized matching occurs, replacement should preserve actual quote style."""
 

tests/tools/test_filesystem_tools.py

@@ -7,7 +7,6 @@ from nanobot.agent.tools.filesystem import (
     ListDirTool,
     ReadFileTool,
     WriteFileTool,
-    _find_match,
 )
 
 # ---------------------------------------------------------------------------
@@ -116,52 +115,6 @@ class TestReadFileTool:
         assert "Maximum is 100 MiB" in result
 
 
-# ---------------------------------------------------------------------------
-# _find_match  (unit tests for the helper)
-# ---------------------------------------------------------------------------
-
-class TestFindMatch:
-
-    def test_exact_match(self):
-        match, count = _find_match("hello world", "world")
-        assert match == "world"
-        assert count == 1
-
-    def test_exact_no_match(self):
-        match, count = _find_match("hello world", "xyz")
-        assert match is None
-        assert count == 0
-
-    def test_crlf_normalisation(self):
-        # Caller normalises CRLF before calling _find_match, so test with
-        # pre-normalised content to verify exact match still works.
-        content = "line1\nline2\nline3"
-        old_text = "line1\nline2\nline3"
-        match, count = _find_match(content, old_text)
-        assert match is not None
-        assert count == 1
-
-    def test_line_trim_fallback(self):
-        content = "    def foo():\n        pass\n"
-        old_text = "def foo():\n    pass"
-        match, count = _find_match(content, old_text)
-        assert match is not None
-        assert count == 1
-        # The returned match should be the *original* indented text
-        assert "    def foo():" in match
-
-    def test_line_trim_multiple_candidates(self):
-        content = "  a\n  b\n  a\n  b\n"
-        old_text = "a\nb"
-        match, count = _find_match(content, old_text)
-        assert count == 2
-
-    def test_empty_old_text(self):
-        match, count = _find_match("hello", "")
-        # Empty string is always "in" any string via exact match
-        assert match == ""
-
-
 # ---------------------------------------------------------------------------
 # EditFileTool
 # ---------------------------------------------------------------------------

tests/tools/test_web_search_tool.py

@@ -820,4 +820,6 @@ async def test_olostep_package_missing_returns_install_hint(monkeypatch):
     tool = _tool(provider="olostep", api_key="olostep-key")
     result = await tool.execute(query="test query")
 
-    assert result == "Error: olostep package not installed. Run: pip install olostep"
+    assert result == (
+        "Error: Olostep support is not installed. Run `nanobot plugins enable olostep`."
+    )

tests/utils/test_file_edit_events.py

@@ -9,7 +9,6 @@ from nanobot.utils.file_edit_events import (
     build_file_edit_start_event,
     build_unified_diff_payload,
     line_diff_stats,
-    prepare_file_edit_tracker,
     prepare_file_edit_trackers,
     read_file_snapshot,
 )
@@ -44,15 +43,14 @@ def test_write_file_start_tracks_snapshot_and_end_emits_exact_diff(tmp_path: Pat
     target = tmp_path / "notes.txt"
     target.write_text("old\nkeep\n", encoding="utf-8")
     params = {"path": "notes.txt", "content": "new\nkeep\nextra\n"}
-    tracker = prepare_file_edit_tracker(
+    trackers = prepare_file_edit_trackers(
         call_id="call-write",
         tool_name="write_file",
         tool=_write_tool(tmp_path),
         workspace=tmp_path,
         params=params,
     )
-
+    [tracker] = trackers
-    assert tracker is not None
     start = build_file_edit_start_event(tracker)
     assert start == {
         "version": 1,
@@ -103,15 +101,14 @@ def test_unified_diff_payload_truncates_large_diffs() -> None:
 def test_binary_file_is_reported_but_not_counted(tmp_path: Path) -> None:
     target = tmp_path / "data.bin"
     target.write_bytes(b"\x00\x01before")
-    tracker = prepare_file_edit_tracker(
+    trackers = prepare_file_edit_trackers(
         call_id="call-bin",
         tool_name="edit_file",
         tool=_edit_tool(tmp_path),
         workspace=tmp_path,
         params={"path": "data.bin", "old_text": "before", "new_text": "after"},
     )
-
+    [tracker] = trackers
-    assert tracker is not None
     assert not read_file_snapshot(target).countable
     target.write_bytes(b"\x00\x01after")
     event = build_file_edit_end_event(tracker)
@@ -123,15 +120,14 @@ def test_binary_file_is_reported_but_not_counted(tmp_path: Path) -> None:
 def test_binary_before_file_is_reported_but_not_counted(tmp_path: Path) -> None:
     target = tmp_path / "data.bin"
     target.write_bytes(b"\x00\x01before")
-    tracker = prepare_file_edit_tracker(
+    trackers = prepare_file_edit_trackers(
         call_id="call-bin",
         tool_name="write_file",
         tool=_write_tool(tmp_path),
         workspace=tmp_path,
         params={"path": "data.bin", "content": "after\n"},
     )
-
+    [tracker] = trackers
-    assert tracker is not None
     target.write_text("after\n", encoding="utf-8")
     event = build_file_edit_end_event(tracker)
     assert event["binary"] is True
@@ -215,15 +211,14 @@ def test_apply_patch_dry_run_does_not_prepare_file_edit_trackers(tmp_path: Path)
 def test_oversized_file_is_reported_but_not_counted(tmp_path: Path) -> None:
     target = tmp_path / "large.txt"
     params = {"path": "large.txt", "content": "x"}
-    tracker = prepare_file_edit_tracker(
+    trackers = prepare_file_edit_trackers(
         call_id="call-large",
         tool_name="write_file",
         tool=_write_tool(tmp_path),
         workspace=tmp_path,
         params=params,
     )
-
+    [tracker] = trackers
-    assert tracker is not None
     target.write_text("x" * (2 * 1024 * 1024 + 1), encoding="utf-8")
     event = build_file_edit_end_event(tracker)
     assert event["binary"] is True
@@ -232,11 +227,11 @@ def test_oversized_file_is_reported_but_not_counted(tmp_path: Path) -> None:
     assert "diff" not in event
 
 
-def test_untracked_tools_do_not_prepare_file_edit_tracker(tmp_path: Path) -> None:
+def test_untracked_tools_do_not_prepare_file_edit_trackers(tmp_path: Path) -> None:
-    assert prepare_file_edit_tracker(
+    assert prepare_file_edit_trackers(
         call_id="call-exec",
         tool_name="exec",
         tool=None,
         workspace=tmp_path,
         params={"path": "created-by-shell.txt"},
-    ) is None
+    ) == []