Compare commits
6
Commits
main
..
known-good
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
50e67a621a | ||
|
|
c5637b8cff | ||
|
|
ded5d4d8ed | ||
|
|
d5d6c76a93 | ||
|
|
79a0fd8ed0 | ||
|
|
206ab0943c |
+3
-11
@@ -378,7 +378,6 @@ class AgentLoop:
|
|||||||
)
|
)
|
||||||
self._unified_session = unified_session
|
self._unified_session = unified_session
|
||||||
self._running = False
|
self._running = False
|
||||||
self._shutting_down = False
|
|
||||||
self._mcp_servers = mcp_servers or {}
|
self._mcp_servers = mcp_servers or {}
|
||||||
self._mcp_stacks: dict[str, MCPConnection] = {}
|
self._mcp_stacks: dict[str, MCPConnection] = {}
|
||||||
self._mcp_connecting = False
|
self._mcp_connecting = False
|
||||||
@@ -1225,11 +1224,11 @@ class AgentLoop:
|
|||||||
# it into session history now makes it visible in the
|
# it into session history now makes it visible in the
|
||||||
# next conversation turn.
|
# next conversation turn.
|
||||||
#
|
#
|
||||||
# During gateway shutdown (self._shutting_down is True),
|
# During gateway shutdown (self._running is False), skip
|
||||||
# skip this so the pending markers survive and
|
# this so the pending markers survive and
|
||||||
# recover_stale_sessions() can notify the user and
|
# recover_stale_sessions() can notify the user and
|
||||||
# re-trigger the interrupted turn on next startup.
|
# re-trigger the interrupted turn on next startup.
|
||||||
if not self._shutting_down:
|
if self._running:
|
||||||
try:
|
try:
|
||||||
key = self._effective_session_key(msg)
|
key = self._effective_session_key(msg)
|
||||||
session = self.sessions.get_or_create(key)
|
session = self.sessions.get_or_create(key)
|
||||||
@@ -1319,7 +1318,6 @@ class AgentLoop:
|
|||||||
def stop(self) -> None:
|
def stop(self) -> None:
|
||||||
"""Stop the agent loop."""
|
"""Stop the agent loop."""
|
||||||
self._running = False
|
self._running = False
|
||||||
self._shutting_down = True
|
|
||||||
logger.info("Agent loop stopping")
|
logger.info("Agent loop stopping")
|
||||||
|
|
||||||
async def _process_message(
|
async def _process_message(
|
||||||
@@ -2153,12 +2151,6 @@ class AgentLoop:
|
|||||||
sender_id=chat_id,
|
sender_id=chat_id,
|
||||||
chat_id=chat_id,
|
chat_id=chat_id,
|
||||||
content=last_user_msg["content"],
|
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(
|
logger.info(
|
||||||
|
|||||||
@@ -2838,48 +2838,6 @@ def test_bootstrap_accepts_static_token_as_secret(bus: MagicMock) -> None:
|
|||||||
assert body["api_token"] != body["token"]
|
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:
|
def test_bootstrap_ws_url_uses_forwarded_https_host(bus: MagicMock) -> None:
|
||||||
channel = _ch(bus, host="127.0.0.1", port=29931, tokenIssueSecret="s3cret")
|
channel = _ch(bus, host="127.0.0.1", port=29931, tokenIssueSecret="s3cret")
|
||||||
resp = channel.gateway.http._handle_bootstrap(
|
resp = channel.gateway.http._handle_bootstrap(
|
||||||
|
|||||||
@@ -44,15 +44,7 @@ def _should_defer_local_trigger_turn(
|
|||||||
session_key: str,
|
session_key: str,
|
||||||
active_session_keys: Iterable[str],
|
active_session_keys: Iterable[str],
|
||||||
) -> bool:
|
) -> bool:
|
||||||
if not (local_trigger(msg.metadata) is not None and session_key in active_session_keys):
|
return 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:
|
def _local_trigger_id(msg: InboundMessage) -> str | None:
|
||||||
|
|||||||
@@ -101,8 +101,6 @@ if TYPE_CHECKING:
|
|||||||
from nanobot.session.manager import SessionManager
|
from nanobot.session.manager import SessionManager
|
||||||
from nanobot.triggers.local_store import LocalTriggerStore
|
from nanobot.triggers.local_store import LocalTriggerStore
|
||||||
|
|
||||||
from nanobot.bus.events import InboundMessage
|
|
||||||
|
|
||||||
|
|
||||||
def _decode_api_key(raw_key: str) -> str | None:
|
def _decode_api_key(raw_key: str) -> str | None:
|
||||||
key = unquote(raw_key)
|
key = unquote(raw_key)
|
||||||
@@ -247,10 +245,6 @@ class GatewayHTTPHandler:
|
|||||||
if got == "/webui/bootstrap":
|
if got == "/webui/bootstrap":
|
||||||
return self._handle_bootstrap(connection, request)
|
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)
|
# Settings routes (delegated)
|
||||||
response = await self.settings_routes.dispatch(connection, request, got)
|
response = await self.settings_routes.dispatch(connection, request, got)
|
||||||
if response is not None:
|
if response is not None:
|
||||||
@@ -378,48 +372,6 @@ class GatewayHTTPHandler:
|
|||||||
expected_path = _normalize_config_path(self.config.path)
|
expected_path = _normalize_config_path(self.config.path)
|
||||||
return f"{scheme}://{host}{expected_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 -----------------------------------------------------
|
# -- Session routes -----------------------------------------------------
|
||||||
|
|
||||||
async def _dispatch_session_routes(self, request: WsRequest, got: str) -> Response | None:
|
async def _dispatch_session_routes(self, request: WsRequest, got: str) -> Response | None:
|
||||||
|
|||||||
Reference in New Issue
Block a user