local triggers: support mid-turn interruption via origin_metadata {"interrupt": true}

POST /admin/inject?chat_id=...&content=... (auth: token_issue_secret,
same as /webui/bootstrap) publishes an InboundMessage into the target
session's queue, triggering a user turn in that chat. First piece of
the cross-session admin project.

nanobot/agent/loop.py

@@ -378,6 +378,7 @@ class AgentLoop:
         )
         self._unified_session = unified_session
         self._running = False
+        self._shutting_down = False
         self._mcp_servers = mcp_servers or {}
         self._mcp_stacks: dict[str, MCPConnection] = {}
         self._mcp_connecting = False
@@ -1224,11 +1225,11 @@ class AgentLoop:
                     # it into session history now makes it visible in the
                     # next conversation turn.
                     #
-                    # During gateway shutdown (self._running is False), skip
+                    # During gateway shutdown (self._shutting_down is True),
-                    # this so the pending markers survive and
+                    # skip this so the pending markers survive and
                     # recover_stale_sessions() can notify the user and
                     # re-trigger the interrupted turn on next startup.
-                    if self._running:
+                    if not self._shutting_down:
                         try:
                             key = self._effective_session_key(msg)
                             session = self.sessions.get_or_create(key)
@@ -1318,6 +1319,7 @@ class AgentLoop:
     def stop(self) -> None:
         """Stop the agent loop."""
         self._running = False
+        self._shutting_down = True
         logger.info("Agent loop stopping")
 
     async def _process_message(
@@ -2151,6 +2153,12 @@ class AgentLoop:
                             sender_id=chat_id,
                             chat_id=chat_id,
                             content=last_user_msg["content"],
+                            # The message already exists in session history
+                            # (persisted before the crash); skip persisting it
+                            # again so recovery does not duplicate it.
+                            metadata={
+                                turn_continuation.SKIP_USER_PERSIST_META: True
+                            },
                         )
                     )
                     logger.info(

nanobot/channels/websocket/tests/test_websocket_http_routes.py

@@ -2838,6 +2838,48 @@ def test_bootstrap_accepts_static_token_as_secret(bus: MagicMock) -> None:
     assert body["api_token"] != body["token"]
 
 
+async def test_admin_inject_requires_secret(bus: MagicMock) -> None:
+    """Admin inject rejects requests without the token_issue_secret."""
+    channel = _ch(bus, tokenIssueSecret="s3cret")
+    resp = await channel.gateway.http._handle_admin_inject(
+        _FakeReq({}, path="/admin/inject")
+    )
+    assert resp.status_code == 401
+    bus.publish_inbound.assert_not_called()
+
+
+async def test_admin_inject_publishes_inbound_message(bus: MagicMock) -> None:
+    """Admin inject enqueues an InboundMessage for the target chat."""
+    channel = _ch(bus, tokenIssueSecret="s3cret")
+    path = (
+        "/admin/inject?chat_id=16765ad8-7222-488f-be1c-44680f4c06e2"
+        "&content=screen%20is%20black&sender_id=tester"
+    )
+    resp = await channel.gateway.http._handle_admin_inject(
+        _FakeReq({"Authorization": "Bearer s3cret"}, path=path)
+    )
+    assert resp.status_code == 200
+    body = json.loads(resp.body)
+    assert body["ok"] is True
+    assert body["chat_id"] == "16765ad8-7222-488f-be1c-44680f4c06e2"
+    bus.publish_inbound.assert_awaited_once()
+    msg = bus.publish_inbound.await_args.args[0]
+    assert msg.channel == "websocket"
+    assert msg.chat_id == "16765ad8-7222-488f-be1c-44680f4c06e2"
+    assert msg.content == "screen is black"
+    assert msg.sender_id == "tester"
+
+
+async def test_admin_inject_requires_chat_id_and_content(bus: MagicMock) -> None:
+    """Admin inject rejects missing chat_id or empty content."""
+    channel = _ch(bus, tokenIssueSecret="s3cret")
+    resp = await channel.gateway.http._handle_admin_inject(
+        _FakeReq({"Authorization": "Bearer s3cret"}, path="/admin/inject?chat_id=x")
+    )
+    assert resp.status_code == 400
+    bus.publish_inbound.assert_not_called()
+
+
 def test_bootstrap_ws_url_uses_forwarded_https_host(bus: MagicMock) -> None:
     channel = _ch(bus, host="127.0.0.1", port=29931, tokenIssueSecret="s3cret")
     resp = channel.gateway.http._handle_bootstrap(

nanobot/triggers/local_turns.py

@@ -44,7 +44,15 @@ def _should_defer_local_trigger_turn(
     session_key: str,
     active_session_keys: Iterable[str],
 ) -> bool:
-    return local_trigger(msg.metadata) is not None and session_key in active_session_keys
+    if not (local_trigger(msg.metadata) is not None and session_key in active_session_keys):
+        return False
+    # Mid-turn interruption requested: the trigger was created with
+    # origin_metadata {"interrupt": true} (copied verbatim into msg.metadata
+    # by the local runner). Skip deferral so the message routes into the
+    # session's pending queue and is injected at the next tool-call boundary.
+    if msg.metadata.get("interrupt"):
+        return False
+    return True
 
 
 def _local_trigger_id(msg: InboundMessage) -> str | None:

nanobot/webui/ws_http.py

@@ -101,6 +101,8 @@ if TYPE_CHECKING:
     from nanobot.session.manager import SessionManager
     from nanobot.triggers.local_store import LocalTriggerStore
 
+from nanobot.bus.events import InboundMessage
+
 
 def _decode_api_key(raw_key: str) -> str | None:
     key = unquote(raw_key)
@@ -245,6 +247,10 @@ class GatewayHTTPHandler:
         if got == "/webui/bootstrap":
             return self._handle_bootstrap(connection, request)
 
+        # Admin: cross-session message injection
+        if got == "/admin/inject":
+            return await self._handle_admin_inject(request)
+
         # Settings routes (delegated)
         response = await self.settings_routes.dispatch(connection, request, got)
         if response is not None:
@@ -372,6 +378,48 @@ class GatewayHTTPHandler:
         expected_path = _normalize_config_path(self.config.path)
         return f"{scheme}://{host}{expected_path}"
 
+    # -- Admin routes --------------------------------------------------------
+
+    async def _handle_admin_inject(self, request: WsRequest) -> Response:
+        """Inject a user message into another session's turn queue (admin).
+
+        Auth: ``Authorization: Bearer <secret>`` or ``X-Nanobot-Auth: <secret>``
+        with the websocket ``token_issue_secret`` (or static token), same as
+        ``/webui/bootstrap``. Query params: ``chat_id`` (required), ``content``
+        (required), ``sender_id`` (optional, default ``admin-inject``),
+        ``channel`` (optional, default ``websocket``).
+        """
+        secret = self.config.token_issue_secret.strip() or self.config.token.strip()
+        if not secret or not _issue_route_secret_matches(request.headers, secret):
+            return _http_error(401, "Unauthorized")
+        query = _parse_query(request.path)
+        chat_id = (_query_first(query, "chat_id") or "").strip()
+        content = _query_first(query, "content")
+        if not chat_id or not content or not content.strip():
+            return _http_error(400, "chat_id and non-empty content are required")
+        if len(content) > 4000:
+            return _http_error(400, "content too long (max 4000 chars)")
+        sender_id = (_query_first(query, "sender_id") or "").strip() or "admin-inject"
+        channel = (_query_first(query, "channel") or "").strip() or "websocket"
+        msg = InboundMessage(
+            channel=channel,
+            sender_id=sender_id,
+            chat_id=chat_id,
+            content=content,
+            metadata={"admin_inject": True},
+        )
+        await self.bus.publish_inbound(msg)
+        self._log.info(
+            "Admin inject: {}:{} <- {} ({})",
+            channel,
+            chat_id,
+            sender_id,
+            content[:80],
+        )
+        return _http_json_response(
+            {"ok": True, "channel": channel, "chat_id": chat_id, "sender_id": sender_id}
+        )
+
     # -- Session routes -----------------------------------------------------
 
     async def _dispatch_session_routes(self, request: WsRequest, got: str) -> Response | None: