Grayscale/nanobot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(webui): add temporary chat mode

Browse Source
This commit is contained in:
Xubin Ren
2026-08-08 23:20:59 +08:00
parent 113e8d67ad
commit c9a6145878
41 changed files with 1500 additions and 119 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nanobot/channels/websocket/runtime.py
+135 -21
View File
@@ -20,7 +20,11 @@ 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 (
INBOUND_META_TRANSIENT_SESSION,
OUTBOUND_META_AGENT_UI,
OutboundMessage,
)
from nanobot.bus.outbound_events import (
GoalStateSyncEvent,
GoalStatusEvent,
@@ -33,6 +37,11 @@ from nanobot.bus.outbound_events import (
)
from nanobot.bus.queue import MessageBus
from nanobot.channels.base import BaseChannel
from nanobot.channels.websocket.temporary_chat import (
TEMPORARY_COMMANDS,
TemporaryChats,
has_temporary_chat_prefix,
)
from nanobot.command.builtin import builtin_command_starts_agent_turn
from nanobot.config.schema import Base
from nanobot.runtime_context import (
@@ -48,6 +57,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,
@@ -309,6 +319,10 @@ def _is_valid_chat_id(value: Any) -> TypeGuard[str]:
return isinstance(value, str) and _CHAT_ID_RE.match(value) is not None
def _is_temporary_chat_id(value: Any) -> TypeGuard[str]:
return _is_valid_chat_id(value) and has_temporary_chat_prefix(value)
def _parse_envelope(raw: str) -> dict[str, Any] | None:
"""Return a typed envelope dict if the frame is a new-style JSON envelope, else None.
@@ -383,6 +397,7 @@ class WebSocketChannel(BaseChannel):
if gateway.session_manager is not None
else None
)
self._temporary_chats = TemporaryChats(gateway.session_manager, self._media, bus)
self._stream_text_buffers: dict[tuple[str, str], list[str]] = {}
@@ -396,6 +411,49 @@ 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)
def _claim_temporary_chat(
self,
connection: ServerConnection,
chat_id: str,
) -> str | None:
"""Create the connection's single in-memory chat on first use."""
if connection not in self._webui_connections:
return "temporary_chat_unavailable"
if detail := self._temporary_chats.claim(connection, chat_id):
return detail
self._attach(connection, chat_id)
return None
async def _discard_temporary_chat(
self,
connection: ServerConnection,
chat_id: str,
) -> str | None:
detail = await self._temporary_chats.discard(connection, chat_id)
if detail is not None:
return detail
self._detach(connection, chat_id)
clear_websocket_turns(chat_id)
self._clear_stream_buffers(chat_id)
return None
async def send_webui_protocol_error(
self,
connection: ServerConnection,
@@ -424,18 +482,17 @@ 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())
for cid in chat_ids:
subs = self._subs.get(cid)
if subs is None:
continue
subs.discard(connection)
if not subs:
self._subs.pop(cid, None)
self._conn_default.pop(connection, None)
self._webui_connections.discard(connection)
try:
temporary_chat_id = self._temporary_chats.chat_id_for(connection)
if temporary_chat_id is not None:
await self._discard_temporary_chat(connection, temporary_chat_id)
finally:
for chat_id in tuple(self._conn_chats.get(connection, ())):
self._detach(connection, chat_id)
self._conn_default.pop(connection, None)
self._webui_connections.discard(connection)
async def _maybe_push_active_goal_state(self, chat_id: str) -> None:
"""Replay an active sustained goal from session metadata after *chat_id* is subscribed.
@@ -486,7 +543,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)
@@ -713,7 +770,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 ---------------------------------------
@@ -751,11 +808,27 @@ class WebSocketChannel(BaseChannel):
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_temporary_chat_id(cid):
await self._send_event(connection, "error", detail="invalid temporary chat_id")
return
if detail := await self._discard_temporary_chat(connection, cid):
await self._send_event(connection, "error", detail=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
if _is_temporary_chat_id(cid):
await self._send_event(
connection,
"error",
detail="temporary_chat_cannot_attach",
chat_id=cid,
)
return
self._attach(connection, cid)
await self._send_event(connection, "attached", chat_id=cid)
await self._hydrate_after_subscribe(cid)
@@ -789,6 +862,14 @@ class WebSocketChannel(BaseChannel):
if not _is_valid_chat_id(cid):
await self._send_event(connection, "error", detail="invalid chat_id")
return
if _is_temporary_chat_id(cid):
await self._send_event(
connection,
"error",
detail="temporary_chat_scope_is_per_message",
chat_id=cid,
)
return
scope = await self._workspace_scope_or_error(
connection,
lambda: self._workspaces.scope_for_set_request(
@@ -820,6 +901,7 @@ class WebSocketChannel(BaseChannel):
if not _is_valid_chat_id(cid):
await self._send_event(connection, "error", detail="invalid chat_id")
return
temporary = _is_temporary_chat_id(cid)
raw_turn_id = envelope.get("turn_id")
turn_id = raw_turn_id if isinstance(raw_turn_id, str) and raw_turn_id else None
rejection_fields = {
@@ -857,6 +939,25 @@ class WebSocketChannel(BaseChannel):
)
return
if temporary:
command = content.strip().partition(" ")[0].lower()
if command.startswith("/") and command not in TEMPORARY_COMMANDS:
await self._send_event(
connection,
"error",
detail="temporary_chat_command_rejected",
**rejection_fields,
)
return
if detail := self._claim_temporary_chat(connection, cid):
await self._send_event(
connection,
"error",
detail=detail,
**rejection_fields,
)
return
raw_media = envelope.get("media")
media_paths: list[str] = []
if raw_media is not None:
@@ -869,7 +970,12 @@ class WebSocketChannel(BaseChannel):
**rejection_fields,
)
return
media_paths, reason = self._media.store_inbound_attachments(cast(list[Any], raw_media))
store_attachments = (
self._media.store_temporary_attachments
if temporary
else self._media.store_inbound_attachments
)
media_paths, reason = store_attachments(cast(list[Any], raw_media))
if reason is not None:
await self._send_event(
connection,
@@ -879,7 +985,8 @@ class WebSocketChannel(BaseChannel):
**rejection_fields,
)
return
if temporary:
self._temporary_chats.remember_attachments(cid, media_paths)
# Allow media-only turns (content may be empty when attachments are present).
if not content.strip() and not media_paths:
await self._send_event(
@@ -889,9 +996,10 @@ class WebSocketChannel(BaseChannel):
**rejection_fields,
)
return
# Auto-attach on first use so clients can one-shot without a separate attach.
self._attach(connection, cid)
await self._hydrate_after_subscribe(cid)
if not temporary:
# Auto-attach on first use so clients can one-shot without a separate attach.
self._attach(connection, cid)
await self._hydrate_after_subscribe(cid)
# Resolve after hydration so a concurrent downgrade cannot be overwritten.
scope = await self._workspace_scope_or_error(
@@ -921,6 +1029,8 @@ class WebSocketChannel(BaseChannel):
return
metadata: dict[str, Any] = {"remote": getattr(connection, "remote_address", None)}
if temporary:
metadata[INBOUND_META_TRANSIENT_SESSION] = True
if envelope.get("webui") is True:
metadata["webui"] = True
metadata.update(self._transcripts.client_turn_metadata(envelope.get("turn_id")))
@@ -953,7 +1063,7 @@ class WebSocketChannel(BaseChannel):
metadata[WEBSOCKET_TURN_OWNER_METADATA_KEY] = queued_owner
accepted = False
try:
if is_webui:
if is_webui and not temporary:
self._transcripts.append_user_message(
cid,
content,
@@ -1037,6 +1147,8 @@ class WebSocketChannel(BaseChannel):
except Exception as e:
self.logger.warning("server task error during shutdown: {}", e)
self._server_task = None
for connection in tuple(self._conn_chats):
await self._cleanup_connection(connection)
self._subs.clear()
self._conn_chats.clear()
self._conn_default.clear()
@@ -1054,7 +1166,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)
@@ -1071,6 +1183,8 @@ class WebSocketChannel(BaseChannel):
transcript_overrides: dict[str, Any] | None = None,
) -> bool:
"""Persist one canonical turn event and retain unsafe owners on failure."""
if _is_temporary_chat_id(chat_id):
return True
persisted = self._transcripts.prepare_and_append(
chat_id,
event,