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

fix(webui): derive temporary chats from session policy

Browse Source
This commit is contained in:
chengyongru
2026-08-08 23:20:59 +08:00
committed by Xubin Ren
parent a5bc3bfbb9
commit 75e333a3c5
13 changed files with 647 additions and 185 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nanobot/channels/websocket/runtime.py
+77 -77
View File
@@ -21,10 +21,7 @@ from websockets.exceptions import ConnectionClosed
from websockets.http11 import Request as WsRequest
from nanobot.bus.events import (
INBOUND_META_RUNTIME_CONTROL,
OUTBOUND_META_AGENT_UI,
RUNTIME_CONTROL_SESSION_DISCARD,
InboundMessage,
OutboundMessage,
)
from nanobot.bus.outbound_events import (
@@ -88,6 +85,7 @@ from nanobot.webui.session_access import (
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
@@ -310,23 +308,12 @@ def _parse_inbound_payload(raw: str) -> str | None:
# Accept UUIDs and short scoped keys like "unified:default". Keeps the capability
# namespace small enough to rule out path traversal / quote injection tricks.
_CHAT_ID_RE = re.compile(r"^[A-Za-z0-9_:-]{1,64}$")
_TEMPORARY_CHAT_PREFIX = "temporary-"
_TEMPORARY_CHAT_DISABLED_TOOLS = frozenset({
"create_goal",
"update_goal",
"spawn",
"cron",
})
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 value.startswith(_TEMPORARY_CHAT_PREFIX)
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.
@@ -396,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
@@ -403,7 +391,6 @@ class WebSocketChannel(BaseChannel):
)
self._stream_text_buffers: dict[tuple[str, str], list[str]] = {}
self._temporary_media_paths: dict[str, set[str]] = {}
# -- Subscription bookkeeping -------------------------------------------
@@ -432,38 +419,15 @@ class WebSocketChannel(BaseChannel):
if key[0] == chat_id:
self._stream_text_buffers.pop(key, None)
def _discard_temporary_media(self, chat_id: str) -> None:
"""Remove uploads owned by one connection-scoped temporary chat."""
for raw_path in self._temporary_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")
async def _discard_temporary_chat(
async def _discard_connection_owned_chat(
self,
connection: ServerConnection,
chat_id: str,
) -> None:
session_key = f"{self.name}:{chat_id}"
await self._temporary_chats.discard(connection, chat_id)
self._detach(connection, chat_id)
clear_websocket_turns(chat_id)
self._clear_stream_buffers(chat_id)
self._discard_temporary_media(chat_id)
if self.gateway.session_manager is not None:
self.gateway.session_manager.invalidate(session_key)
await self.bus.publish_inbound(
InboundMessage(
channel=self.name,
sender_id="webui",
chat_id=chat_id,
content="",
metadata={
INBOUND_META_RUNTIME_CONTROL: RUNTIME_CONTROL_SESSION_DISCARD,
},
session_key_override=session_key,
)
)
async def send_webui_protocol_error(
self,
@@ -497,10 +461,12 @@ class WebSocketChannel(BaseChannel):
"""Remove *connection* from every subscription set; safe to call multiple times."""
chat_ids = tuple(self._conn_chats.get(connection, ()))
for cid in chat_ids:
if _is_temporary_chat_id(cid):
await self._discard_temporary_chat(connection, cid)
if self._temporary_chats.owns(connection, cid):
await self._discard_connection_owned_chat(connection, cid)
else:
self._detach(connection, cid)
for cid in self._temporary_chats.chat_ids_for_owner(connection):
await self._discard_connection_owned_chat(connection, cid)
self._conn_default.pop(connection, None)
self._webui_connections.discard(connection)
@@ -815,21 +781,46 @@ 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_temporary_chat_id(cid):
if not _is_valid_chat_id(cid):
await self._send_event(connection, "error", detail="invalid temporary chat_id")
return
await self._discard_temporary_chat(connection, cid)
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)
@@ -863,6 +854,11 @@ class WebSocketChannel(BaseChannel):
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(
@@ -894,7 +890,6 @@ 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 = {
@@ -932,28 +927,20 @@ class WebSocketChannel(BaseChannel):
)
return
if temporary:
command = content.strip().partition(" ")[0].lower()
if command.startswith("/") and command not in {"/model", "/stop"}:
await self._send_event(
connection,
"error",
detail="temporary_chat_command_rejected",
**rejection_fields,
)
return
if self.gateway.session_manager is None:
await self._send_event(
connection,
"error",
detail="temporary_chat_unavailable",
**rejection_fields,
)
return
self.gateway.session_manager.get_or_create_transient(
f"{self.name}:{cid}",
disabled_tools=_TEMPORARY_CHAT_DISABLED_TOOLS,
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] = []
@@ -977,8 +964,8 @@ class WebSocketChannel(BaseChannel):
**rejection_fields,
)
return
if temporary and media_paths:
self._temporary_media_paths.setdefault(cid, set()).update(media_paths)
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:
@@ -991,15 +978,15 @@ class WebSocketChannel(BaseChannel):
return
# Auto-attach on first use so clients can one-shot without a separate attach.
self._attach(connection, cid)
if not temporary:
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.restricted_default_scope()
if temporary
temporary_policy.workspace_scope
if temporary_policy is not None
else self._workspaces.scope_for_message(
envelope,
chat_id=cid,
@@ -1058,7 +1045,13 @@ class WebSocketChannel(BaseChannel):
metadata[WEBSOCKET_TURN_OWNER_METADATA_KEY] = queued_owner
accepted = False
try:
if is_webui and not temporary:
if (
is_webui
and (
temporary_policy is None
or temporary_policy.persist_transcript
)
):
self._transcripts.append_user_message(
cid,
content,
@@ -1087,8 +1080,16 @@ class WebSocketChannel(BaseChannel):
media=media_paths or None,
metadata=metadata,
is_dm=False,
session_key=f"{self.name}:{cid}" if temporary else None,
require_existing_session=temporary,
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:
@@ -1149,8 +1150,7 @@ class WebSocketChannel(BaseChannel):
self._conn_default.clear()
self._webui_connections.clear()
self._tokens.clear()
for chat_id in tuple(self._temporary_media_paths):
self._discard_temporary_media(chat_id)
self._temporary_chats.close()
async def _safe_send_to(
self,
@@ -1180,7 +1180,7 @@ 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):
if not self._temporary_chats.should_persist_transcript(chat_id):
return True
persisted = self._transcripts.prepare_and_append(
chat_id,