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

feat(webui): persist agent activity events

Browse Source
This commit is contained in:
Xubin Ren
2026-05-17 23:51:52 +08:00
parent 4b5de66c58
commit c8bb04a8fe
13 changed files with 1238 additions and 208 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nanobot/utils/file_edit_events.py
+311
View File
@@ -0,0 +1,311 @@
"""File-edit activity helpers for WebUI progress events."""
from __future__ import annotations
import difflib
import json
from dataclasses import dataclass
from pathlib import Path
from typing import Any
TRACKED_FILE_EDIT_TOOLS = frozenset({"write_file", "edit_file", "notebook_edit"})
_MAX_SNAPSHOT_BYTES = 2 * 1024 * 1024
@dataclass(slots=True)
class FileSnapshot:
path: Path
exists: bool
text: str | None
unreadable: bool = False
binary: bool = False
oversized: bool = False
@property
def countable(self) -> bool:
return (
self.text is not None
and not self.binary
and not self.oversized
and not self.unreadable
)
@dataclass(slots=True)
class FileEditTracker:
call_id: str
tool: str
path: Path
display_path: str
before: FileSnapshot
def is_file_edit_tool(tool_name: str | None) -> bool:
return bool(tool_name) and tool_name in TRACKED_FILE_EDIT_TOOLS
def resolve_file_edit_path(
tool: Any,
workspace: Path | None,
params: dict[str, Any] | None,
) -> Path | None:
"""Resolve the target file path after tool argument preparation."""
if not isinstance(params, dict):
return None
raw_path = params.get("path")
if not isinstance(raw_path, str) or not raw_path.strip():
return None
resolver = getattr(tool, "_resolve", None)
if callable(resolver):
try:
resolved = resolver(raw_path)
if isinstance(resolved, Path):
return resolved
if resolved:
return Path(resolved)
except Exception:
return None
if workspace is None:
return Path(raw_path).expanduser().resolve()
return (workspace / raw_path).expanduser().resolve()
def display_file_edit_path(path: Path, workspace: Path | None) -> str:
if workspace is not None:
try:
return path.resolve().relative_to(workspace.resolve()).as_posix()
except Exception:
pass
return path.as_posix()
def read_file_snapshot(path: Path, *, max_bytes: int = _MAX_SNAPSHOT_BYTES) -> FileSnapshot:
try:
if not path.exists() or not path.is_file():
return FileSnapshot(path=path, exists=False, text="")
size = path.stat().st_size
if size > max_bytes:
return FileSnapshot(path=path, exists=True, text=None, oversized=True)
raw = path.read_bytes()
except OSError:
return FileSnapshot(path=path, exists=path.exists(), text=None, unreadable=True)
if b"\x00" in raw:
return FileSnapshot(path=path, exists=True, text=None, binary=True)
try:
text = raw.decode("utf-8")
except UnicodeDecodeError:
return FileSnapshot(path=path, exists=True, text=None, binary=True)
return FileSnapshot(path=path, exists=True, text=text.replace("\r\n", "\n"))
def line_diff_stats(before: str | None, after: str | None) -> tuple[int, int]:
"""Return ``(added, deleted)`` for a UTF-8 text line-level diff."""
if before is None or after is None:
return 0, 0
before_lines = before.replace("\r\n", "\n").splitlines()
after_lines = after.replace("\r\n", "\n").splitlines()
added = 0
deleted = 0
matcher = difflib.SequenceMatcher(a=before_lines, b=after_lines, autojunk=False)
for tag, i1, i2, j1, j2 in matcher.get_opcodes():
if tag == "equal":
continue
if tag in ("replace", "delete"):
deleted += i2 - i1
if tag in ("replace", "insert"):
added += j2 - j1
return added, deleted
def prepare_file_edit_tracker(
*,
call_id: str,
tool_name: str,
tool: Any,
workspace: Path | None,
params: dict[str, Any] | None,
) -> FileEditTracker | None:
if not is_file_edit_tool(tool_name):
return None
path = resolve_file_edit_path(tool, workspace, params)
if path is None:
return None
before = read_file_snapshot(path)
return FileEditTracker(
call_id=str(call_id or ""),
tool=tool_name,
path=path,
display_path=display_file_edit_path(path, workspace),
before=before,
)
def build_file_edit_start_event(
tracker: FileEditTracker,
params: dict[str, Any] | None,
) -> dict[str, Any]:
predicted_after = _predict_after_text(tracker.tool, params or {}, tracker.before)
if tracker.before.countable and predicted_after is not None:
added, deleted = line_diff_stats(tracker.before.text, predicted_after)
else:
added, deleted = 0, 0
return _event_payload(
tracker,
phase="start",
status="editing",
added=added,
deleted=deleted,
approximate=True,
)
def build_file_edit_end_event(tracker: FileEditTracker) -> dict[str, Any]:
after = read_file_snapshot(tracker.path)
if tracker.before.countable and after.countable:
added, deleted = line_diff_stats(tracker.before.text, after.text)
else:
added, deleted = 0, 0
return _event_payload(
tracker,
phase="end",
status="done",
added=added,
deleted=deleted,
approximate=False,
binary=after.binary or after.oversized or after.unreadable,
)
def build_file_edit_error_event(tracker: FileEditTracker, error: str | None = None) -> dict[str, Any]:
payload = _event_payload(
tracker,
phase="error",
status="error",
added=0,
deleted=0,
approximate=False,
)
if error:
payload["error"] = error.strip()[:240]
return payload
def _event_payload(
tracker: FileEditTracker,
*,
phase: str,
status: str,
added: int,
deleted: int,
approximate: bool,
binary: bool = False,
) -> dict[str, Any]:
payload: dict[str, Any] = {
"version": 1,
"call_id": tracker.call_id,
"tool": tracker.tool,
"path": tracker.display_path,
"phase": phase,
"added": max(0, int(added)),
"deleted": max(0, int(deleted)),
"approximate": bool(approximate),
"status": status,
}
if binary:
payload["binary"] = True
return payload
def _predict_after_text(
tool_name: str,
params: dict[str, Any],
before: FileSnapshot,
) -> str | None:
if not before.countable:
return None
before_text = before.text or ""
if tool_name == "write_file":
content = params.get("content")
return content if isinstance(content, str) else ""
if tool_name == "edit_file":
old_text = params.get("old_text")
new_text = params.get("new_text")
if not isinstance(old_text, str) or not isinstance(new_text, str):
return None
replace_all = bool(params.get("replace_all"))
if old_text == "":
return new_text if not before.exists else before_text
if old_text in before_text:
if replace_all:
return before_text.replace(old_text, new_text)
return before_text.replace(old_text, new_text, 1)
return None
if tool_name == "notebook_edit":
return _predict_notebook_after_text(params, before_text)
return None
def _predict_notebook_after_text(params: dict[str, Any], before_text: str) -> str | None:
try:
nb = json.loads(before_text) if before_text.strip() else _empty_notebook()
except Exception:
return None
cells = nb.get("cells")
if not isinstance(cells, list):
return None
try:
cell_index = int(params.get("cell_index", 0))
except (TypeError, ValueError):
return None
new_source = params.get("new_source")
source = new_source if isinstance(new_source, str) else ""
cell_type = params.get("cell_type") if params.get("cell_type") in ("code", "markdown") else "code"
mode = params.get("edit_mode") if params.get("edit_mode") in ("replace", "insert", "delete") else "replace"
if mode == "delete":
if 0 <= cell_index < len(cells):
cells.pop(cell_index)
else:
return None
elif mode == "insert":
insert_at = min(max(cell_index + 1, 0), len(cells))
cells.insert(insert_at, _new_notebook_cell(source, str(cell_type)))
else:
if not (0 <= cell_index < len(cells)):
return None
cell = cells[cell_index]
if not isinstance(cell, dict):
return None
cell["source"] = source
cell["cell_type"] = cell_type
if cell_type == "code":
cell.setdefault("outputs", [])
cell.setdefault("execution_count", None)
else:
cell.pop("outputs", None)
cell.pop("execution_count", None)
nb["cells"] = cells
try:
return json.dumps(nb, indent=1, ensure_ascii=False)
except Exception:
return None
def _empty_notebook() -> dict[str, Any]:
return {
"nbformat": 4,
"nbformat_minor": 5,
"metadata": {
"kernelspec": {"display_name": "Python 3", "language": "python", "name": "python3"},
"language_info": {"name": "python"},
},
"cells": [],
}
def _new_notebook_cell(source: str, cell_type: str) -> dict[str, Any]:
cell: dict[str, Any] = {"cell_type": cell_type, "source": source, "metadata": {}}
if cell_type == "code":
cell["outputs"] = []
cell["execution_count"] = None
return cell
nanobot/utils/progress_events.py
+18 -1
View File
@@ -10,13 +10,21 @@ from nanobot.agent.hook import AgentHookContext
def on_progress_accepts_tool_events(cb: Callable[..., Any]) -> bool:
return _on_progress_accepts(cb, "tool_events")
def on_progress_accepts_file_edit_events(cb: Callable[..., Any]) -> bool:
return _on_progress_accepts(cb, "file_edit_events")
def _on_progress_accepts(cb: Callable[..., Any], name: str) -> bool:
try:
sig = inspect.signature(cb)
except (TypeError, ValueError):
return False
if any(p.kind == inspect.Parameter.VAR_KEYWORD for p in sig.parameters.values()):
return True
return "tool_events" in sig.parameters
return name in sig.parameters
async def invoke_on_progress(
@@ -32,6 +40,15 @@ async def invoke_on_progress(
await on_progress(content, tool_hint=tool_hint)
async def invoke_file_edit_progress(
on_progress: Callable[..., Awaitable[None]],
file_edit_events: list[dict[str, Any]],
) -> None:
if not file_edit_events or not on_progress_accepts_file_edit_events(on_progress):
return
await on_progress("", file_edit_events=file_edit_events)
def build_tool_event_start_payload(tool_call: Any) -> dict[str, Any]:
return {
"version": 1,
nanobot/utils/webui_titles.py
-138
View File
@@ -1,138 +0,0 @@
"""Helpers for WebUI chat title generation."""
from __future__ import annotations
import re
from typing import Any
from loguru import logger
from nanobot.providers.base import LLMProvider
from nanobot.session.manager import Session, SessionManager
from nanobot.utils.helpers import truncate_text
WEBUI_SESSION_METADATA_KEY = "webui"
WEBUI_TITLE_METADATA_KEY = "title"
WEBUI_TITLE_USER_EDITED_METADATA_KEY = "title_user_edited"
TITLE_MAX_CHARS = 60
def mark_webui_session(session: Session, metadata: dict[str, Any]) -> bool:
"""Persist a WebUI marker only when the inbound websocket frame opted in."""
if metadata.get(WEBUI_SESSION_METADATA_KEY) is not True:
return False
session.metadata[WEBUI_SESSION_METADATA_KEY] = True
return True
def clean_generated_title(raw: str | None) -> str:
text = (raw or "").strip()
if not text:
return ""
text = re.sub(r"^\s*(title|标题)\s*[:：]\s*", "", text, flags=re.IGNORECASE)
text = text.strip().strip("\"'`“”‘’")
text = re.sub(r"\s+", " ", text).strip()
text = text.rstrip("。.!！?？,，;；:")
if len(text) > TITLE_MAX_CHARS:
text = text[: TITLE_MAX_CHARS - 1].rstrip() + "…"
return text
def _title_inputs(session: Session) -> tuple[str, str]:
user_text = ""
assistant_text = ""
for message in session.messages:
role = message.get("role")
content = message.get("content")
if not isinstance(content, str) or not content.strip():
continue
if role == "user" and not user_text:
user_text = content.strip()
elif role == "assistant" and not assistant_text:
assistant_text = content.strip()
if user_text and assistant_text:
break
return user_text, assistant_text
async def maybe_generate_webui_title(
*,
sessions: SessionManager,
session_key: str,
provider: LLMProvider,
model: str,
) -> bool:
"""Generate and persist a short title for WebUI-owned sessions only."""
session = sessions.get_or_create(session_key)
if session.metadata.get(WEBUI_SESSION_METADATA_KEY) is not True:
return False
if session.metadata.get(WEBUI_TITLE_USER_EDITED_METADATA_KEY) is True:
return False
current_title = session.metadata.get(WEBUI_TITLE_METADATA_KEY)
if isinstance(current_title, str) and current_title.strip():
return False
user_text, assistant_text = _title_inputs(session)
if not user_text:
return False
prompt = (
"Generate a concise title for this chat.\n"
"Rules:\n"
"- Use the same language as the user when practical.\n"
"- 3 to 8 words.\n"
"- No quotes.\n"
"- No punctuation at the end.\n"
"- Return only the title.\n\n"
f"User: {truncate_text(user_text, 1_000)}"
)
if assistant_text:
prompt += f"\nAssistant: {truncate_text(assistant_text, 1_000)}"
try:
response = await provider.chat_with_retry(
[
{
"role": "system",
"content": (
"You write short, neutral chat titles. "
"Return only the title text."
),
},
{"role": "user", "content": prompt},
],
tools=None,
model=model,
max_tokens=32,
temperature=0.2,
retry_mode="standard",
)
except Exception:
logger.debug("Failed to generate webui session title for {}", session_key, exc_info=True)
return False
title = clean_generated_title(response.content)
if not title or title.lower().startswith("error"):
return False
session.metadata[WEBUI_TITLE_METADATA_KEY] = title
sessions.save(session)
return True
async def maybe_generate_webui_title_after_turn(
*,
channel: str,
metadata: dict[str, Any],
sessions: SessionManager,
session_key: str,
provider: LLMProvider,
model: str,
) -> bool:
if channel != "websocket" or metadata.get(WEBUI_SESSION_METADATA_KEY) is not True:
return False
return await maybe_generate_webui_title(
sessions=sessions,
session_key=session_key,
provider=provider,
model=model,
)
nanobot/utils/webui_transcript.py
+107 -2
View File
@@ -125,11 +125,25 @@ def replay_transcript_to_ui_messages(
buffer_message_id: str | None = None
buffer_parts: list[str] = []
suppress_until_turn_end = False
active_activity_segment_id: str | None = None
active_file_edit_segment_id: str | None = None
activity_segment_counter = 0
_ts_base = int(time.time() * 1000)
def _new_id(prefix: str, idx: int) -> str:
return f"{prefix}-{idx}-{uuid.uuid4().hex[:8]}"
def _new_activity_segment(*, activate: bool = True) -> str:
nonlocal active_activity_segment_id, activity_segment_counter
activity_segment_counter += 1
segment_id = f"activity-{activity_segment_counter}"
if activate:
active_activity_segment_id = segment_id
return segment_id
def _ensure_activity_segment() -> str:
return active_activity_segment_id or _new_activity_segment()
def attach_reasoning_chunk(prev: list[dict[str, Any]], chunk: str, idx: int) -> None:
for i in range(len(prev) - 1, -1, -1):
candidate = prev[i]
@@ -151,12 +165,19 @@ def replay_transcript_to_ui_messages(
**candidate,
"reasoning": (str(candidate.get("reasoning") or "")) + chunk,
"reasoningStreaming": True,
"activitySegmentId": candidate.get("activitySegmentId") or _ensure_activity_segment(),
}
return
if not has_answer and candidate.get("isStreaming"):
prev[i] = {**candidate, "reasoning": chunk, "reasoningStreaming": True}
prev[i] = {
**candidate,
"reasoning": chunk,
"reasoningStreaming": True,
"activitySegmentId": candidate.get("activitySegmentId") or _ensure_activity_segment(),
}
return
break
segment = _ensure_activity_segment()
prev.append(
{
"id": _new_id("as", idx),
@@ -165,6 +186,7 @@ def replay_transcript_to_ui_messages(
"isStreaming": True,
"reasoning": chunk,
"reasoningStreaming": True,
"activitySegmentId": segment,
"createdAt": _ts_base + idx,
},
)
@@ -221,6 +243,7 @@ def replay_transcript_to_ui_messages(
return
def absorb_complete(extra: dict[str, Any], idx: int) -> None:
nonlocal active_activity_segment_id
last = messages[-1] if messages else None
if last and is_reasoning_only_placeholder(last):
messages[-1] = {
@@ -238,10 +261,76 @@ def replay_transcript_to_ui_messages(
**extra,
},
)
active_activity_segment_id = None
def _file_edit_key(edit: dict[str, Any]) -> str:
return "|".join(
str(edit.get(k) or "")
for k in ("call_id", "tool", "path")
)
def upsert_file_edits(edits: list[dict[str, Any]], idx: int) -> None:
nonlocal active_file_edit_segment_id
if not edits:
return
last = messages[-1] if messages else None
if (
active_file_edit_segment_id
and last
and last.get("kind") == "trace"
and last.get("fileEdits")
):
segment = active_file_edit_segment_id
else:
segment = _new_activity_segment(activate=False)
active_file_edit_segment_id = segment
if not (
last
and last.get("kind") == "trace"
and not last.get("isStreaming")
and last.get("fileEdits")
and last.get("activitySegmentId") == segment
):
messages.append(
{
"id": _new_id("tr", idx),
"role": "tool",
"kind": "trace",
"content": "",
"traces": [],
"fileEdits": [],
"activitySegmentId": segment,
"createdAt": _ts_base + idx,
},
)
last = messages[-1]
existing = list(last.get("fileEdits") or [])
index_by_key = {
_file_edit_key(edit): pos
for pos, edit in enumerate(existing)
if isinstance(edit, dict)
}
for edit in edits:
if not isinstance(edit, dict):
continue
key = _file_edit_key(edit)
if key in index_by_key:
pos = index_by_key[key]
existing[pos] = {**existing[pos], **edit}
else:
index_by_key[key] = len(existing)
existing.append(dict(edit))
messages[-1] = {
**last,
"fileEdits": existing,
"activitySegmentId": last.get("activitySegmentId") or segment,
}
for idx, rec in enumerate(lines):
ev = rec.get("event")
if ev == "user":
active_activity_segment_id = None
active_file_edit_segment_id = None
text = rec.get("text")
text_s = text if isinstance(text, str) else ""
media_paths = rec.get("media_paths")
@@ -264,6 +353,12 @@ def replay_transcript_to_ui_messages(
messages.append(row)
continue
if ev == "file_edit":
raw_edits = rec.get("edits")
if isinstance(raw_edits, list):
upsert_file_edits([e for e in raw_edits if isinstance(e, dict)], idx)
continue
if ev == "delta":
if suppress_until_turn_end:
continue
@@ -338,14 +433,21 @@ def replay_transcript_to_ui_messages(
trace_lines = structured if structured else ([text] if isinstance(text, str) and text else [])
if not trace_lines:
continue
segment = _ensure_activity_segment()
last = messages[-1] if messages else None
if last and last.get("kind") == "trace" and not last.get("isStreaming"):
if (
last
and last.get("kind") == "trace"
and not last.get("isStreaming")
and (last.get("activitySegmentId") in (None, segment))
):
prev_traces = list(last.get("traces") or [last.get("content")])
merged_traces = prev_traces + trace_lines
messages[-1] = {
**last,
"traces": merged_traces,
"content": trace_lines[-1],
"activitySegmentId": last.get("activitySegmentId") or segment,
}
else:
messages.append(
@@ -355,6 +457,7 @@ def replay_transcript_to_ui_messages(
"kind": "trace",
"content": trace_lines[-1],
"traces": trace_lines,
"activitySegmentId": segment,
"createdAt": _ts_base + idx,
},
)
@@ -389,6 +492,8 @@ def replay_transcript_to_ui_messages(
if ev == "turn_end":
suppress_until_turn_end = False
active_activity_segment_id = None
active_file_edit_segment_id = None
for i, m in enumerate(messages):
if m.get("isStreaming"):
messages[i] = {**m, "isStreaming": False}
nanobot/utils/webui_turn_helpers.py
+268
View File
@@ -6,15 +6,161 @@ AgentLoop uses these without importing a concrete channel plugin; only
from __future__ import annotations
import re
import time
from collections.abc import Awaitable, Callable
from dataclasses import dataclass, field
from typing import Any
from loguru import logger
from nanobot.bus.events import InboundMessage, OutboundMessage
from nanobot.bus.queue import MessageBus
from nanobot.providers.base import LLMProvider
from nanobot.session.goal_state import goal_state_ws_blob
from nanobot.session.manager import Session, SessionManager
from nanobot.utils.helpers import truncate_text
WEBUI_SESSION_METADATA_KEY = "webui"
WEBUI_TITLE_METADATA_KEY = "title"
WEBUI_TITLE_USER_EDITED_METADATA_KEY = "title_user_edited"
TITLE_MAX_CHARS = 60
TITLE_GENERATION_MAX_TOKENS = 96
TITLE_GENERATION_REASONING_EFFORT = "none"
# Wall-clock turn start per ``chat_id`` (websocket only). Survives browser refresh while the
# gateway process stays up; cleared on idle/stop and implicitly dropped on restart.
_WEBSOCKET_TURN_WALL_STARTED_AT: dict[str, float] = {}
TitleContext = tuple[LLMProvider, str]
def mark_webui_session(session: Session, metadata: dict[str, Any]) -> bool:
"""Persist a WebUI marker only when the inbound websocket frame opted in."""
if metadata.get(WEBUI_SESSION_METADATA_KEY) is not True:
return False
session.metadata[WEBUI_SESSION_METADATA_KEY] = True
return True
def clean_generated_title(raw: str | None) -> str:
text = (raw or "").strip()
if not text:
return ""
text = re.sub(r"^\s*(title|标题)\s*[:：]\s*", "", text, flags=re.IGNORECASE)
text = text.strip().strip("\"'`“”‘’")
text = re.sub(r"\s+", " ", text).strip()
text = text.rstrip("。.!！?？,，;；:")
if len(text) > TITLE_MAX_CHARS:
text = text[: TITLE_MAX_CHARS - 1].rstrip() + "…"
return text
def _title_inputs(session: Session) -> tuple[str, str]:
user_text = ""
assistant_text = ""
for message in session.messages:
if message.get("_command") is True:
continue
role = message.get("role")
content = message.get("content")
if not isinstance(content, str) or not content.strip():
continue
if role == "user" and not user_text:
user_text = content.strip()
elif role == "assistant" and not assistant_text:
assistant_text = content.strip()
if user_text and assistant_text:
break
return user_text, assistant_text
async def maybe_generate_webui_title(
*,
sessions: SessionManager,
session_key: str,
provider: LLMProvider,
model: str,
) -> bool:
"""Generate and persist a short title for WebUI-owned sessions only."""
session = sessions.get_or_create(session_key)
if session.metadata.get(WEBUI_SESSION_METADATA_KEY) is not True:
return False
if session.metadata.get(WEBUI_TITLE_USER_EDITED_METADATA_KEY) is True:
return False
current_title = session.metadata.get(WEBUI_TITLE_METADATA_KEY)
if isinstance(current_title, str) and current_title.strip():
return False
user_text, assistant_text = _title_inputs(session)
if not user_text:
return False
prompt = (
"Generate a concise title for this chat.\n"
"Rules:\n"
"- Use the same language as the user when practical.\n"
"- 3 to 8 words.\n"
"- No quotes.\n"
"- No punctuation at the end.\n"
"- Return only the title.\n\n"
f"User: {truncate_text(user_text, 1_000)}"
)
if assistant_text:
prompt += f"\nAssistant: {truncate_text(assistant_text, 1_000)}"
try:
response = await provider.chat_with_retry(
[
{
"role": "system",
"content": (
"You write short, neutral chat titles. "
"Return only the title text."
),
},
{"role": "user", "content": prompt},
],
tools=None,
model=model,
max_tokens=TITLE_GENERATION_MAX_TOKENS,
temperature=0.2,
reasoning_effort=TITLE_GENERATION_REASONING_EFFORT,
retry_mode="standard",
)
except Exception:
logger.debug("Failed to generate webui session title for {}", session_key, exc_info=True)
return False
title = clean_generated_title(response.content)
if not title or title.lower().startswith("error"):
logger.debug(
"WebUI title generation returned no usable title for {} (finish_reason={})",
session_key,
response.finish_reason,
)
return False
session.metadata[WEBUI_TITLE_METADATA_KEY] = title
sessions.save(session)
return True
async def maybe_generate_webui_title_after_turn(
*,
channel: str,
metadata: dict[str, Any],
sessions: SessionManager,
session_key: str,
provider: LLMProvider,
model: str,
) -> bool:
if channel != "websocket" or metadata.get(WEBUI_SESSION_METADATA_KEY) is not True:
return False
return await maybe_generate_webui_title(
sessions=sessions,
session_key=session_key,
provider=provider,
model=model,
)
def websocket_turn_wall_started_at(chat_id: str) -> float | None:
@@ -46,3 +192,125 @@ async def publish_turn_run_status(bus: MessageBus, msg: InboundMessage, status:
metadata=meta,
),
)
def build_bus_progress_callback(
bus: MessageBus,
msg: InboundMessage,
) -> Callable[..., Awaitable[None]]:
"""Return the bus progress callback for agent runtime events."""
async def _bus_progress(
content: str,
*,
tool_hint: bool = False,
tool_events: list[dict[str, Any]] | None = None,
file_edit_events: list[dict[str, Any]] | None = None,
reasoning: bool = False,
reasoning_end: bool = False,
) -> None:
if file_edit_events and msg.channel != "websocket":
return
meta = dict(msg.metadata or {})
meta["_progress"] = True
meta["_tool_hint"] = tool_hint
if reasoning:
meta["_reasoning_delta"] = True
if reasoning_end:
meta["_reasoning_end"] = True
if tool_events:
meta["_tool_events"] = tool_events
if file_edit_events:
meta["_file_edit_events"] = file_edit_events
await bus.publish_outbound(
OutboundMessage(
channel=msg.channel,
chat_id=msg.chat_id,
content=content,
metadata=meta,
)
)
return _bus_progress
@dataclass
class WebuiTurnCoordinator:
"""Own the WebUI/WebSocket wire details that hang off AgentLoop turns."""
bus: MessageBus
sessions: SessionManager
schedule_background: Callable[[Awaitable[None]], None]
_title_contexts: dict[str, TitleContext] = field(default_factory=dict)
def capture_title_context(
self,
session_key: str,
msg: InboundMessage,
provider: LLMProvider,
model: str,
) -> None:
if msg.channel == "websocket" and msg.metadata.get("webui") is True:
self._title_contexts[session_key] = (provider, model)
def discard(self, session_key: str) -> None:
self._title_contexts.pop(session_key, None)
async def publish_run_status(self, msg: InboundMessage, status: str) -> None:
await publish_turn_run_status(self.bus, msg, status)
async def handle_turn_end(
self,
msg: InboundMessage,
*,
session_key: str,
latency_ms: int | None,
) -> None:
if msg.channel != "websocket":
return
turn_metadata: dict[str, Any] = {**msg.metadata, "_turn_end": True}
if latency_ms is not None:
turn_metadata["latency_ms"] = int(latency_ms)
session = self.sessions.get_or_create(session_key)
turn_metadata["goal_state"] = goal_state_ws_blob(session.metadata)
await self.bus.publish_outbound(OutboundMessage(
channel=msg.channel,
chat_id=msg.chat_id,
content="",
metadata=turn_metadata,
))
self._schedule_title_update(msg, session_key=session_key)
def _schedule_title_update(self, msg: InboundMessage, *, session_key: str) -> None:
title_context = self._title_contexts.pop(session_key, None)
if msg.metadata.get("webui") is not True or title_context is None:
return
title_provider, title_model = title_context
async def _generate_title_and_notify(
provider: LLMProvider = title_provider,
model: str = title_model,
) -> None:
generated = await maybe_generate_webui_title_after_turn(
channel=msg.channel,
metadata=msg.metadata,
sessions=self.sessions,
session_key=session_key,
provider=provider,
model=model,
)
if generated:
await self.bus.publish_outbound(OutboundMessage(
channel=msg.channel,
chat_id=msg.chat_id,
content="",
metadata={
**msg.metadata,
"_session_updated": True,
"_session_update_scope": "metadata",
},
))
self.schedule_background(_generate_title_and_notify())