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

fix(webui): prevent redundant thread and media reloads (#5164)

Browse Source
This commit is contained in:
chengyongru
2026-07-30 10:25:22 +08:00
committed by GitHub
parent fc73d5ff39
commit 11fcd9cc5f
29 changed files with 1465 additions and 247 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nanobot/channels/websocket/runtime.py
+9 -1
View File
@@ -67,6 +67,7 @@ from nanobot.webui.http_utils import (
from nanobot.webui.mcp_presets_api import normalize_mcp_preset_mentions
from nanobot.webui.metadata import (
WEBSOCKET_TURN_OWNER_METADATA_KEY,
WEBUI_SYSTEM_COMMAND_TURN_PREFIX,
WEBUI_TURN_METADATA_KEY,
)
from nanobot.webui.transcript import WEBUI_TRANSCRIPT_INCOMPLETE_KEY
@@ -1003,6 +1004,13 @@ class WebSocketChannel(BaseChannel):
return
# Signal that the agent has fully finished processing the current turn.
if isinstance(event, TurnEndEvent):
turn_id = (msg.metadata or {}).get(WEBUI_TURN_METADATA_KEY)
session_update_scope = (
"metadata"
if isinstance(turn_id, str)
and turn_id.startswith(WEBUI_SYSTEM_COMMAND_TURN_PREFIX)
else "thread"
)
turn_owner = (msg.metadata or {}).get(WEBSOCKET_TURN_OWNER_METADATA_KEY)
await self.send_turn_end(
msg.chat_id,
@@ -1011,7 +1019,7 @@ class WebSocketChannel(BaseChannel):
metadata=msg.metadata,
turn_owner=turn_owner if isinstance(turn_owner, str) else None,
)
await self.send_session_updated(msg.chat_id, scope="thread")
await self.send_session_updated(msg.chat_id, scope=session_update_scope)
return
if isinstance(event, SessionUpdatedEvent):
if conns:
nanobot/channels/websocket/tests/test_websocket_channel.py
+42 -1
View File
@@ -49,7 +49,11 @@ from nanobot.webui.http_utils import (
from nanobot.webui.http_utils import (
parse_request_path as _parse_request_path,
)
from nanobot.webui.metadata import WEBSOCKET_TURN_OWNER_METADATA_KEY
from nanobot.webui.metadata import (
WEBSOCKET_TURN_OWNER_METADATA_KEY,
WEBUI_SYSTEM_COMMAND_TURN_PREFIX,
WEBUI_TURN_METADATA_KEY,
)
from nanobot.webui.settings_api import settings_payload, update_provider_settings
from nanobot.webui.transcript import (
append_transcript_object,
@@ -1618,6 +1622,43 @@ async def test_send_turn_end_emits_turn_end_event() -> None:
]
@pytest.mark.asyncio
async def test_system_command_turn_end_only_refreshes_session_metadata() -> None:
bus = MagicMock()
channel = WebSocketChannel(
{"enabled": True, "allowFrom": ["*"]},
bus,
gateway=_basic_handler(bus),
)
mock_ws = AsyncMock()
channel._attach(mock_ws, "chat-model")
await channel.send(OutboundMessage(
channel="websocket",
chat_id="chat-model",
content="",
metadata={
WEBUI_TURN_METADATA_KEY: f"{WEBUI_SYSTEM_COMMAND_TURN_PREFIX}model-switch",
},
event=TurnEndEvent(),
))
assert _sent_ws_payloads(mock_ws) == [
{
"event": "turn_end",
"chat_id": "chat-model",
"turn_id": f"{WEBUI_SYSTEM_COMMAND_TURN_PREFIX}model-switch",
"turn_phase": "complete",
"turn_seq": 1,
},
{
"event": "session_updated",
"chat_id": "chat-model",
"scope": "metadata",
},
]
@pytest.mark.asyncio
@pytest.mark.parametrize(
("active_owner", "event_owner", "expected_cleared"),
nanobot/channels/websocket/tests/test_websocket_http_routes.py
+11
View File
@@ -2937,6 +2937,17 @@ async def test_webui_thread_resigns_assistant_media_urls(
assert media[0]["url"].startswith("/api/media/")
assert media[0]["url"] != "/api/media/old-sig/old-payload"
repeated = await _http_get(
"http://127.0.0.1:29914/api/sessions/websocket:video-replay/webui-thread",
headers=auth,
)
repeated_assistant = next(
m for m in repeated.json()["messages"] if m["role"] == "assistant"
)
assert repeated_assistant["id"] == assistant["id"]
assert repeated_assistant["media"][0]["url"] == media[0]["url"]
assert len(list(websocket_media.iterdir())) == 1
fetched = await _http_get(f"http://127.0.0.1:29914{media[0]['url']}")
assert fetched.status_code == 200
assert fetched.content == b"video"
nanobot/channels/websocket/tests/test_websocket_media_route.py
+27 -2
View File
@@ -146,16 +146,41 @@ def test_local_markdown_image_is_staged_and_rewritten(
channel = _ch(bus, workspace_path=workspace, port=0)
with patch("nanobot.webui.media_gateway.get_media_dir", side_effect=_fake_media_dir(media)):
rewritten = channel.gateway.media.rewrite_local_markdown_images(
first = channel.gateway.media.rewrite_local_markdown_images(
"The result:\n![Cloud Architecture Diagram](demo_arch.png)"
)
second = channel.gateway.media.rewrite_local_markdown_images(
"The result:\n![Cloud Architecture Diagram](demo_arch.png)"
)
assert "![Cloud Architecture Diagram](/api/media/" in rewritten
assert "![Cloud Architecture Diagram](/api/media/" in first
assert second == first
staged = list((media / "websocket").iterdir())
assert len(staged) == 1
assert staged[0].read_bytes() == _PNG_BYTES
def test_modified_local_markdown_image_gets_a_new_immutable_url(
bus: MagicMock,
tmp_path: Path,
) -> None:
workspace = tmp_path / "workspace"
workspace.mkdir()
source = workspace / "demo_arch.png"
source.write_bytes(_PNG_BYTES)
media = tmp_path / "media"
channel = _ch(bus, workspace_path=workspace, port=0)
markdown = "![Cloud Architecture Diagram](demo_arch.png)"
with patch("nanobot.webui.media_gateway.get_media_dir", side_effect=_fake_media_dir(media)):
first = channel.gateway.media.rewrite_local_markdown_images(markdown)
source.write_bytes(_PNG_BYTES + b"updated")
second = channel.gateway.media.rewrite_local_markdown_images(markdown)
assert second != first
assert len(list((media / "websocket").iterdir())) == 2
def test_local_markdown_video_is_staged_and_rewritten(
bus: MagicMock,
tmp_path: Path,
nanobot/webui/media_api.py
+20 -3
View File
@@ -7,6 +7,7 @@ import binascii
import hashlib
import hmac
import mimetypes
import os
import re
import shutil
import uuid
@@ -126,17 +127,33 @@ def sign_or_stage_media_path(
signed = sign_media_path(path, secret=secret, media_dir=media_dir)
if signed is not None:
return {"url": signed, "name": path.name}
staged_tmp: Path | None = None
try:
if not path.is_file():
resolved = path.resolve(strict=True)
if not resolved.is_file():
return None
source_stat = resolved.stat()
target_dir = media_dir("websocket")
safe_name = safe_filename(path.name) or "attachment"
staged = target_dir / f"{uuid.uuid4().hex[:12]}-{safe_name}"
shutil.copyfile(path, staged)
source_version = "\0".join((
os.path.normcase(str(resolved)),
str(source_stat.st_size),
str(source_stat.st_mtime_ns),
str(source_stat.st_ctime_ns),
))
source_digest = hashlib.sha256(source_version.encode("utf-8")).hexdigest()[:20]
staged = target_dir / f"{source_digest}-{safe_name}"
if not staged.is_file() or staged.stat().st_size != source_stat.st_size:
staged_tmp = target_dir / f".{source_digest}-{uuid.uuid4().hex}.tmp"
shutil.copyfile(resolved, staged_tmp)
staged_tmp.replace(staged)
except OSError as exc:
if logger is not None:
logger.warning("failed to stage outbound media {}: {}", path, exc)
return None
finally:
if staged_tmp is not None:
staged_tmp.unlink(missing_ok=True)
signed = sign_media_path(staged, secret=secret, media_dir=media_dir)
if signed is None:
return None
nanobot/webui/metadata.py
+1
View File
@@ -1,5 +1,6 @@
"""Shared WebUI metadata keys."""
WEBUI_TURN_METADATA_KEY = "webui_turn_id"
WEBUI_SYSTEM_COMMAND_TURN_PREFIX = "webui-system:"
WEBSOCKET_TURN_OWNER_METADATA_KEY = "_websocket_turn_owner"
WEBUI_MESSAGE_SOURCE_METADATA_KEY = "_webui_message_source"
nanobot/webui/transcript.py
+104 -3
View File
@@ -4,6 +4,7 @@ from __future__ import annotations
import base64
import binascii
import hashlib
import json
import os
import re
@@ -34,6 +35,7 @@ _TRANSCRIPT_SEGMENT_RE = re.compile(r"^\d{6}\.jsonl$")
_DEFAULT_TRANSCRIPT_PAGE_LIMIT = 160
_MAX_TRANSCRIPT_PAGE_LIMIT = 1000
_WEBUI_TURN_ID_RE = re.compile(r"^[A-Za-z0-9._:-]{1,128}$")
_WEBUI_REPLAY_IDENTITY_KEY = "_webui_replay_identity"
_MARKDOWN_LOCAL_IMAGE_RE = re.compile(
r"!\[([^\]]*)\]\((<[^>]+>|[^)\s]+)(\s+(?:\"[^\"]*\"|'[^']*'))?\)"
)
@@ -194,6 +196,20 @@ def _flatten_turns(turns: list[list[dict[str, Any]]]) -> list[dict[str, Any]]:
return [record for turn in turns for record in turn]
def _records_with_replay_identity(
records: list[dict[str, Any]],
*,
turn_ordinal: int,
) -> list[dict[str, Any]]:
return [
{
**record,
_WEBUI_REPLAY_IDENTITY_KEY: f"turn:{turn_ordinal}:record:{record_index}",
}
for record_index, record in enumerate(records)
]
def _write_records_to_path(path: Path, rows: list[dict[str, Any]]) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
tmp_path = path.with_suffix(path.suffix + ".tmp")
@@ -543,7 +559,14 @@ def _select_transcript_page(
break
selected_chronological = list(reversed(selected))
lines = [record for ref in selected_chronological for record in ref.records]
lines = [
record
for ref in selected_chronological
for record in _records_with_replay_identity(
ref.records,
turn_ordinal=ref.ordinal,
)
]
if not selected_chronological:
return [], {
"before_cursor": None,
@@ -1030,6 +1053,74 @@ def _split_transcript_turns(lines: list[dict[str, Any]]) -> list[list[dict[str,
return turns
def _annotate_replay_identities(lines: list[dict[str, Any]]) -> list[dict[str, Any]]:
return [
record
for turn_ordinal, turn in enumerate(_split_transcript_turns(lines))
for record in _records_with_replay_identity(
turn,
turn_ordinal=turn_ordinal,
)
]
def _stable_record_digest(record: dict[str, Any]) -> str:
persisted = {
key: value
for key, value in record.items()
if key != _WEBUI_REPLAY_IDENTITY_KEY
}
raw = json.dumps(
persisted,
ensure_ascii=False,
separators=(",", ":"),
sort_keys=True,
default=str,
)
return hashlib.sha256(raw.encode("utf-8")).hexdigest()[:16]
def _ensure_replay_identities(lines: list[dict[str, Any]]) -> list[dict[str, Any]]:
"""Give backfilled/recovered rows a stable identity beside persisted rows."""
annotated: list[dict[str, Any]] = []
for fallback_turn_index, turn in enumerate(_split_transcript_turns(lines)):
anchor = next(
(
value
for record in turn
if isinstance(
value := record.get(_WEBUI_REPLAY_IDENTITY_KEY),
str,
)
and value
),
None,
)
if anchor and ":record:" in anchor:
turn_identity = anchor.rsplit(":record:", 1)[0]
else:
turn_digest = hashlib.sha256(
"\n".join(_stable_record_digest(record) for record in turn).encode("ascii")
).hexdigest()[:16]
turn_identity = f"legacy:{fallback_turn_index}:{turn_digest}"
synthetic_occurrences: dict[str, int] = {}
for record in turn:
identity = record.get(_WEBUI_REPLAY_IDENTITY_KEY)
if isinstance(identity, str) and identity:
annotated.append(record)
continue
digest = _stable_record_digest(record)
occurrence = synthetic_occurrences.get(digest, 0)
synthetic_occurrences[digest] = occurrence + 1
annotated.append({
**record,
_WEBUI_REPLAY_IDENTITY_KEY: (
f"{turn_identity}:synthetic:{digest}:{occurrence}"
),
})
return annotated
def _transcript_turn_signature(records: list[dict[str, Any]]) -> tuple[str, ...]:
texts: list[str] = []
for message in replay_transcript_to_ui_messages(records):
@@ -1464,9 +1555,18 @@ def replay_transcript_to_ui_messages(
_ts_base = _now_ms()
closed_turn_ids: set[str] = set()
replay_turn_aliases: dict[str, str] = {}
generated_id_occurrences: dict[str, int] = {}
def _new_id(prefix: str, idx: int) -> str:
return f"{prefix}-{idx}-{uuid.uuid4().hex[:8]}"
record = lines[idx] if 0 <= idx < len(lines) else {}
identity = record.get(_WEBUI_REPLAY_IDENTITY_KEY)
if not isinstance(identity, str) or not identity:
identity = f"direct:{idx}:{_stable_record_digest(record)}"
digest = hashlib.sha256(f"{prefix}\0{identity}".encode("utf-8")).hexdigest()[:16]
base = f"{prefix}-{digest}"
occurrence = generated_id_occurrences.get(base, 0)
generated_id_occurrences[base] = occurrence + 1
return base if occurrence == 0 else f"{base}-{occurrence}"
def _created_at_ms(rec: dict[str, Any], idx: int) -> int:
created_at_ms = _valid_created_at_ms(rec.get("created_at_ms"))
@@ -2255,7 +2355,7 @@ def build_webui_thread_response(
if paginated:
lines, page = _select_transcript_page(session_key, limit=limit, before=before)
else:
lines = read_transcript_lines(session_key)
lines = _annotate_replay_identities(read_transcript_lines(session_key))
if not lines and active_turn_started_at is None:
return None
lines = inject_missing_user_events_from_session(session_key, lines, session_messages)
@@ -2264,6 +2364,7 @@ def build_webui_thread_response(
session_messages,
session_key=session_key,
)
lines = _ensure_replay_identities(lines)
fork_boundary = fork_boundary_message_count(lines)
msgs = replay_transcript_to_ui_messages(
lines,