diff --git a/CLAUDE.md b/CLAUDE.md index 7280acf..9ef6e8b 100755 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -108,7 +108,7 @@ On startup, `load_system_context()` (`app/infra/startup.py`) loads `app/core/sys **Channel types** are defined in `ChannelType` enum (`app/channels/channel.py`): `CLI`, `TELEGRAM`, `DISCORD`, `WEB`. Each channel implements the `Channel` ABC and owns a `MessageQueue` instance. `bg_server.py` wires up enabled channels — each gets its own `MessageQueue`, `BackgroundAgent`, and set of coroutines (`run_polling`, `process_incoming`, `process_outgoing`) gathered into the event loop. -**WebChannel** (`app/channels/web_channel.py`) is a plain Starlette app served by uvicorn: a JSON HTTP API plus a JSON WebSocket endpoint (`WS /ws`) on the same port. The browser UI is static HTML/CSS/JS in `app/channels/frontend/` (`index.html`, `app.js`, `app.css`, `login.html/js/css`, `vendor/`; no build step), located via `FRONTEND_DIR` and served by a catch-all `Mount("/", StaticFiles(html=True))` registered after the API routes; every frontend response carries `Cache-Control: no-cache` (revalidated via ETag) so edits show up on reload without version query strings. The frontend talks to the server only through HTTP JSON and WebSocket frames — no server-side templating. Multiple concurrent browser tabs are supported — each connection gets a UUID tracked in `_connections`. A per-connection `asyncio.Lock` in `_send_locks` serializes writes to each WebSocket. The channel exposes `GET /api/conversations`, `GET /api/messages` (`load_messages(..., with_meta=True)`: each message also has its row `id` and `has_trajectory`), `GET /api/messages/{id}/trajectory` (a turn's trajectory as a JSON download, 404 if none), `GET /api/status` (`{"model", "auth"}` — the header shows `model` as a button that pre-fills `/model `, refreshed on connect, after system messages and at each turn start since the model is shared across channels; `auth` tells the UI whether to show the logout button), `GET /api/commands` (the agent's `CommandRegistry` — `bg_server` sets `web_channel.commands = web_agent.registry` — plus the channel-handled `/whoami`, sorted; drives the input's slash-command autocomplete, which after `/load|fork|rename|export|delete ` suggests conversations from the last `/api/conversations` list), `POST /api/upload`, and the sidebar's per-conversation `⋯` actions: `POST /api/conversations/{id}/rename` (JSON `{"name"}`; done over HTTP rather than `/rename` because slash commands queue behind a running agent turn) and `GET /api/conversations/{id}/export` (a JSON download built by `ConversationStore.export_payload()`; nothing is written server-side). Fork has no endpoint — it needs the agent to switch conversations, so the sidebar sends `/fork ` over the WebSocket like `/load`. `ConversationStore.fork()` names the copy `" (fork)"`, then `(fork 2)`, … (forking a fork bumps the number instead of stacking suffixes); an unnamed `"New Conversation"` keeps its name so auto-naming still applies. Delete likewise goes over the WebSocket as `/delete ` (two-click confirm in the menu; disabled for the active conversation while a turn is running, since the command would queue behind it): `delete_conversation_cmd` refuses a conversation that another channel has active (`conversation_id:` runtime key — that agent would keep writing to the deleted ID; `MessageHistory` writes don't enforce FKs), and deleting the agent's own active conversation switches it to the channel's most recently updated remaining one (`get_last`, as on startup) or a new one if none is left; `app.js` (`handleDeleteReply`) reads the switched-to ID from that reply, loads its messages and shows the reply below them. `ConversationStore.delete()` removes the messages and row in one transaction and nulls forks' `parent_id` first (FKs are enforced). Rename goes through `rename_conversation()` in `commands.py` (shared with `/rename`), which also refreshes the cached `conversation_name` runtime key of every channel that has that conversation active. `app.js` parks sidebar refreshes while an actions menu or rename box is open (`sidebarBusy()`), since re-rendering would destroy them. The web UI is an admin interface. Auth lives in `app/channels/web_auth.py`: when `WEB_PASSWORD` is set, `WebChannel.start()` wraps the Starlette app (`_app`) in `WebAuthMiddleware` (pure ASGI, so it also guards the `/ws` handshake) and uvicorn serves `_asgi_app`. The middleware handles `POST /api/login` (JSON `{"password"}` → `{"ok": true}` or 401 `{"error"}`) and `POST /api/logout` itself, issues a signed `HttpOnly`/`SameSite=Lax` cookie (itsdangerous `TimestampSigner`, key derived from the password, 30-day max age), lets `GET /login`, `/login.css`, `/login.js` through unauthenticated (a logged-in `/login` redirects to `/`), redirects `GET /` to `/login`, returns 401 for every other HTTP path (API and frontend files), rejects unauthenticated WebSocket handshakes, and rejects cross-origin WebSocket handshakes and non-GET requests (`Origin` host must equal `Host`). `/login` is 404 when no password is set. Without a password, `start()` raises `RuntimeError` unless the host is loopback; `bg_server` catches it, logs, and skips only the web channel. Tests use `TestClient(ch._asgi_app)`. Because it is an admin view, `/api/conversations` calls `ConversationStore.list()` with no channel and returns conversations from **all** channels (each row includes `channel`, rendered as a badge in the sidebar), and `/api/messages` serves any existing conversation (404 only if the ID does not exist). `POST /api/upload` accepts multipart `files` (one or more), writes each under `$ANOTHERBOT_HOME/uploads` with a UUID prefix, and returns the stored basenames; the browser (paperclip button next to the input, or pasting files into the message box — skipped when the clipboard also has plain text, e.g. spreadsheet cells; generic `image.png` screenshots are renamed `pasted-.png`) sends those basenames back in the WebSocket `message` frame's `files` field. `_resolve_upload_paths()` joins each basename to the upload dir (basename-only, so client paths can't traverse out) and the resolved absolute paths reach the agent via `metadata["files"]`, which `Agent._build_user_message()` encodes as image/file parts. History keeps them as `[Attachment: ]` markers (`_build_placeholder_content`); `app.js` parses those out of user messages (live-sent ones are rendered from the same markers) and shows raster images as thumbnails that open a lightbox, other files as download chips with the UUID prefix stripped. `GET /api/uploads/{name}` serves them: basename-only via `_resolve_upload_paths`, png/jpeg/gif/webp inline, everything else (SVG/HTML included) as an `application/octet-stream` attachment, always with `nosniff` + `Content-Security-Policy: default-src 'none'; sandbox` because uploads share the UI's origin and session cookie. Markers whose file isn't in the upload dir (e.g. CLI attachments) fall back to a plain, unlinked chip. Bot replies render as markdown by default via vendored `app/channels/frontend/vendor/marked.min.js` + `purify.min.js` (DOMPurify sanitizes the HTML; no CDN); a header **Markdown | Raw** switch re-renders every reply from the original text kept in `bubble.dataset.raw` and is remembered in `localStorage` (`ab-render`). Each reply row also gets a **Download trajectory** button (stacked under the copy button in `.msg-actions`) when it has a stored trajectory: history replies use `has_trajectory`, and for live turns `WebChannel.on_trajectory_saved` sends a `{"type":"trajectory","message_id","conversation_id"}` frame (through the outgoing queue, to the requesting tab) after the turn's last reply, which `app.js` attaches to that turn's last reply row (`turnReplyRow`, reset on each `busy: true`). Each reply row also gets a copy button (appended to the row, not the bubble, since re-rendering replaces the bubble's HTML) that copies `dataset.raw`; it falls back to a hidden textarea + `execCommand('copy')` because the Clipboard API only exists in secure contexts and the UI is often opened as plain `http://`. `/whoami` (needs the per-connection client ID) and `/stop` are handled inline. `/stop` must not go through the message queue: `BackgroundAgent.process_incoming()` handles one item at a time, so a queued `/stop` would only run after the turn it should stop (and then poison the next turn). It sets `stopped` only while a turn is running ("Nothing to stop." otherwise); the loop checks it between steps. Turn state comes from the `Channel.on_agent_turn(running)` hook (no-op by default), which `process_incoming()` awaits around each `agent_loop()` (after `clear_stopped()`, so a stale idle `/stop` never cuts a turn short); `WebChannel` overrides it to enqueue `{"type":"busy","busy","stopped"}` frames on the outgoing queue (addressed to the channel object, which is what `MessageQueue` routes by) so they stay ordered after the turn's replies, broadcasts them to every tab, and sends the current state on connect. The UI drives the thinking dots and a Stop button ("Stopping…" until the turn ends, then "Stopped.") from those frames. All other slash commands — including `/help`, `/status` — are forwarded to `BackgroundAgent`'s `CommandRegistry` via the message queue with `is_command=True` in metadata so `send_message()` emits `{"type":"system"}` responses, enabling the sidebar to refresh after conversation-mutating commands. +**WebChannel** (`app/channels/web_channel.py`) is a plain Starlette app served by uvicorn: a JSON HTTP API plus a JSON WebSocket endpoint (`WS /ws`) on the same port. The browser UI is static HTML/CSS/JS in `app/channels/frontend/` (`index.html`, `app.js`, `app.css`, `login.html/js/css`, `vendor/`; no build step), located via `FRONTEND_DIR` and served by a catch-all `Mount("/", StaticFiles(html=True))` registered after the API routes; every frontend response carries `Cache-Control: no-cache` (revalidated via ETag) so edits show up on reload without version query strings. The frontend talks to the server only through HTTP JSON and WebSocket frames — no server-side templating. Multiple concurrent browser tabs are supported — each connection gets a UUID tracked in `_connections`. A per-connection `asyncio.Lock` in `_send_locks` serializes writes to each WebSocket. The channel exposes `GET /api/conversations`, `GET /api/messages` (`load_messages(..., with_meta=True)`: each message also has its row `id` and `has_trajectory`), `GET /api/messages/{id}/trajectory` (a turn's trajectory as a JSON download, 404 if none), `GET /api/status` (`{"model", "auth"}` — the header shows `model` as a button that pre-fills `/model `, refreshed on connect, after system messages and at each turn start since the model is shared across channels; `auth` tells the UI whether to show the logout button), `GET /api/commands` (the agent's `CommandRegistry` — `bg_server` sets `web_channel.commands = web_agent.registry` — plus the channel-handled `/whoami`, sorted; drives the input's slash-command autocomplete, which after `/load|fork|rename|export|delete ` suggests conversations from the last `/api/conversations` list), `POST /api/upload`, and the sidebar's per-conversation `⋯` actions: `POST /api/conversations/{id}/rename` (JSON `{"name"}`; done over HTTP rather than `/rename` because slash commands queue behind a running agent turn) and `GET /api/conversations/{id}/export` (a JSON download built by `ConversationStore.export_payload()`; nothing is written server-side). Fork has no endpoint — it needs the agent to switch conversations, so the sidebar sends `/fork ` over the WebSocket like `/load`. `ConversationStore.fork()` names the copy `" (fork)"`, then `(fork 2)`, … (forking a fork bumps the number instead of stacking suffixes); an unnamed `"New Conversation"` keeps its name so auto-naming still applies. Delete likewise goes over the WebSocket as `/delete ` (two-click confirm in the menu; disabled for the active conversation while a turn is running, since the command would queue behind it): `delete_conversation_cmd` refuses a conversation that another channel has active (`conversation_id:` runtime key — that agent would keep writing to the deleted ID; `MessageHistory` writes don't enforce FKs), and deleting the agent's own active conversation switches it to the channel's most recently updated remaining one (`get_last`, as on startup) or a new one if none is left; `app.js` (`handleDeleteReply`) reads the switched-to ID from that reply, loads its messages and shows the reply below them. `ConversationStore.delete()` removes the messages and row in one transaction and nulls forks' `parent_id` first (FKs are enforced). Rename goes through `rename_conversation()` in `commands.py` (shared with `/rename`), which also refreshes the cached `conversation_name` runtime key of every channel that has that conversation active. `app.js` parks sidebar refreshes while an actions menu or rename box is open (`sidebarBusy()`), since re-rendering would destroy them. The web UI is an admin interface. Auth lives in `app/channels/web_auth.py`: when `WEB_PASSWORD` is set, `WebChannel.start()` wraps the Starlette app (`_app`) in `WebAuthMiddleware` (pure ASGI, so it also guards the `/ws` handshake) and uvicorn serves `_asgi_app`. The middleware handles `POST /api/login` (JSON `{"password"}` → `{"ok": true}` or 401 `{"error"}`) and `POST /api/logout` itself, issues a signed `HttpOnly`/`SameSite=Lax` cookie (itsdangerous `TimestampSigner`, key derived from the password, 30-day max age), lets `GET /login`, `/login.css`, `/login.js` through unauthenticated (a logged-in `/login` redirects to `/`), redirects `GET /` to `/login`, returns 401 for every other HTTP path (API and frontend files), rejects unauthenticated WebSocket handshakes, and rejects cross-origin WebSocket handshakes and non-GET requests (`Origin` host must equal `Host`). `/login` is 404 when no password is set. Without a password, `start()` raises `RuntimeError` unless the host is loopback; `bg_server` catches it, logs, and skips only the web channel. Tests use `TestClient(ch._asgi_app)`. Because it is an admin view, `/api/conversations` calls `ConversationStore.list()` with no channel and returns conversations from **all** channels (each row includes `channel`, rendered as a badge in the sidebar), and `/api/messages` serves any existing conversation (404 only if the ID does not exist). `POST /api/upload` accepts multipart `files` (one or more), writes each under `$ANOTHERBOT_HOME/uploads` with a UUID prefix, and returns the stored basenames; the browser (paperclip button next to the input, or pasting files into the message box — skipped when the clipboard also has plain text, e.g. spreadsheet cells; generic `image.png` screenshots are renamed `pasted-.png`) sends those basenames back in the WebSocket `message` frame's `files` field. `_resolve_upload_paths()` joins each basename to the upload dir (basename-only, so client paths can't traverse out) and the resolved absolute paths reach the agent via `metadata["files"]`, which `Agent._build_user_message()` encodes as image/file parts. History keeps them as `[Attachment: ]` markers (`_build_placeholder_content`); `app.js` parses those out of user messages (live-sent ones are rendered from the same markers) and shows raster images as thumbnails that open a lightbox, other files as download chips with the UUID prefix stripped. `GET /api/uploads/{name}` serves them: basename-only via `_resolve_upload_paths`, png/jpeg/gif/webp inline, everything else (SVG/HTML included) as an `application/octet-stream` attachment, always with `nosniff` + `Content-Security-Policy: default-src 'none'; sandbox` because uploads share the UI's origin and session cookie. Markers whose file isn't in the upload dir (e.g. CLI attachments) fall back to a plain, unlinked chip. Bot replies render as markdown by default via vendored `app/channels/frontend/vendor/marked.min.js` + `purify.min.js` (DOMPurify sanitizes the HTML; no CDN); a header **Markdown | Raw** switch re-renders every reply from the original text kept in `bubble.dataset.raw` and is remembered in `localStorage` (`ab-render`). Each reply row also gets a **Download trajectory** button (stacked under the copy button in `.msg-actions`) when it has a stored trajectory: history replies use `has_trajectory`, and for live turns `WebChannel.on_trajectory_saved` sends a `{"type":"trajectory","message_id","conversation_id"}` frame (through the outgoing queue, like every turn frame) after the turn's last reply, which `app.js` attaches to that turn's last reply row (`turnReplyRow`, reset on each `busy: true`). Each reply row also gets a copy button (appended to the row, not the bubble, since re-rendering replaces the bubble's HTML) that copies `dataset.raw`; it falls back to a hidden textarea + `execCommand('copy')` because the Clipboard API only exists in secure contexts and the UI is often opened as plain `http://`. `/whoami` (needs the per-connection client ID) and `/stop` are handled inline. `/stop` must not go through the message queue: `BackgroundAgent.process_incoming()` handles one item at a time, so a queued `/stop` would only run after the turn it should stop (and then poison the next turn). It sets `stopped` only while a turn is running ("Nothing to stop." otherwise); the loop checks it between steps. Turn state comes from the `Channel.on_agent_turn(running)` hook (no-op by default), which `process_incoming()` awaits around each `agent_loop()` (after `clear_stopped()`, so a stale idle `/stop` never cuts a turn short); `WebChannel` overrides it to enqueue `{"type":"busy","busy","stopped"}` frames on the outgoing queue (addressed to the channel object, which is what `MessageQueue` routes by) so they stay ordered after the turn's replies, broadcasts them to every tab, and sends the current state on connect. The UI drives the thinking dots and a Stop button ("Stopping…" until the turn ends, then "Stopped.") from those frames. **Turn frames carry the conversation.** `process_incoming()` adds `conversation_id` to a turn's metadata (thinking text, tool status, reply, error, trajectory), and busy frames carry the turn's conversation (the `conversation_id:web` runtime key at turn start). `WebChannel.send_message()` broadcasts every turn frame to all tabs (they share one agent; the tab that sent the message may have been closed) — only command replies (`is_command`) stay with the requesting tab. It also records, as frames are *delivered*, the last busy frame (`_shown_busy`) and the running turn's frames (`_turn_frames`, capped at `_MAX_TURN_FRAMES`); a connecting tab gets those first, under its send lock so live frames queue behind the replay. In `app.js`, frames that arrive while the chat is loading from the server wait in `holdFrames()` and are handled on top of the history (a replayed final reply already in the history isn't shown twice); turn frames for a conversation the tab isn't showing are dropped; a `busy: true` for a turn this tab didn't start (or in another conversation) reloads that conversation from the history. While a turn runs or is pending (`switchLocked()`), conversation switching is blocked: sidebar clicks, **+ New**, the **Fork** menu item and typed `/new`, `/load`, `/fork` — those would queue behind the turn and its replies would land in the wrong conversation. All other slash commands — including `/help`, `/status` — are forwarded to `BackgroundAgent`'s `CommandRegistry` via the message queue with `is_command=True` in metadata so `send_message()` emits `{"type":"system"}` responses, enabling the sidebar to refresh after conversation-mutating commands. **Slash commands** are handled entirely by `BackgroundAgent`. Channel handlers (`command_handler` in Telegram, `on_message` in Discord) intercept only `/whoami` (resolved inline using the platform user object) and `/stop` (sets the channel's `stopped` flag immediately, as the web channel does), and enqueue everything else as a plain `IncomingMessage`. `BackgroundAgent.process_incoming()` detects the leading `/` and dispatches via its own `CommandRegistry`. That registry owns the full command set: `/model`, `/status`, `/stop`, `/restart`, `/help`, `/list`, `/new`, `/load`, `/fork`, `/rename`, `/export`, `/delete`, `/mcp`. Both `BackgroundAgent` and the CLI build it with `build_command_registry()` (`app/channels/commands.py`); the bot adds `/stop` via `extra=` — only so `/help` lists it: every channel answers `/stop` itself, and the registry handler (which could only run between turns) just replies "Nothing to stop." without setting the flag. Conversation commands use the agent through the `ConversationAgent` protocol (`store`, `channel_str`, `conversation_id`, `switch_conversation()`). `/list` is scoped to the caller's channel, but `/load`, `/fork`, `/rename` and `/export` accept any conversation ID regardless of which channel created it — `ConversationStore.rename/fork/export` no longer enforce channel ownership (`fork` creates the copy under the caller's channel). diff --git a/app/channels/frontend/app.css b/app/channels/frontend/app.css index 9618b7a..7ff8bc7 100644 --- a/app/channels/frontend/app.css +++ b/app/channels/frontend/app.css @@ -110,6 +110,11 @@ body { transition: background .15s; } #new-conv-btn:hover { background: var(--accent-h); } +#new-conv-btn:disabled { opacity: .5; cursor: not-allowed; background: var(--accent); } + +/* While the agent works, other conversations can't be opened (app.js) */ +#conv-list.locked .conv-item:not(.active) .conv-text { opacity: .55; } +#conv-list.locked .conv-item:not(.active) { cursor: not-allowed; } #conv-list { flex: 1; diff --git a/app/channels/frontend/app.js b/app/channels/frontend/app.js index 2c015a7..7fe28fa 100644 --- a/app/channels/frontend/app.js +++ b/app/channels/frontend/app.js @@ -247,7 +247,9 @@ menu.setAttribute('role', 'menu'); const items = [ { label: 'Rename', action: () => startRename(conv) }, - { label: 'Fork', action: () => forkConv(conv.id) }, + // Fork switches the agent to the copy, like /load. + { label: 'Fork', action: () => forkConv(conv.id), + disabled: switchLocked() ? SWITCH_LOCKED : '' }, { label: 'Export', action: () => exportConv(conv.id) }, // Destructive, so it takes a second click. While the agent is // working on this conversation, /delete would only run after the @@ -380,6 +382,8 @@ appendMessage('Fork failed: disconnected (please retry).', 'system'); return; } + // The menu may have been opened before a turn started. + if (switchLocked()) { appendMessage(SWITCH_LOCKED_MSG, 'system'); return; } closeSidebarOnNarrow(); // The fork is an exact copy, so show the source conversation's messages. // Load them before sending /fork so the reply lands below them. @@ -389,6 +393,13 @@ appendMessage('Fork failed: disconnected (please retry).', 'system'); return; } + if (switchLocked()) { + // a turn started while the history loaded: show its conversation again + appendMessage(SWITCH_LOCKED_MSG, 'system'); + clearMessages(); + await loadMessages(activeConvId); + return; + } ws.send(JSON.stringify({ type: 'message', content: `/fork ${id}` })); } @@ -456,6 +467,12 @@ function loadConv(id) { if (!ws || ws.readyState !== WebSocket.OPEN) return; + if (switchLocked()) { + // Switching mid-turn would queue the /load behind the turn and show + // the turn's replies in the wrong conversation. + if (id !== activeConvId) appendMessage(SWITCH_LOCKED_MSG, 'system'); + return; + } ws.send(JSON.stringify({ type: 'message', content: `/load ${id}` })); activeConvId = id; convListEl.querySelectorAll('.conv-item').forEach(el => { @@ -467,7 +484,7 @@ } newConvBtn.addEventListener('click', () => { - if (!ws || ws.readyState !== WebSocket.OPEN) return; + if (!ws || ws.readyState !== WebSocket.OPEN || switchLocked()) return; ws.send(JSON.stringify({ type: 'message', content: '/new' })); clearMessages(); closeSidebarOnNarrow(); @@ -508,8 +525,13 @@ refreshSendState(); loadStatus(); if (!commands.length) loadCommands(); - clearMessages(); // always reset DOM before reloading — prevents duplicates on reconnect - loadConversations(true); + // The server first replays the agent's state (and a running turn's + // progress); hold those frames until the history is on screen so + // they land below it. + holdFrames(async () => { + clearMessages(); // always reset DOM before reloading — prevents duplicates on reconnect + await loadConversations(true); + }); }; ws.onclose = () => { @@ -527,34 +549,91 @@ ws.onerror = () => setStatus(''); ws.onmessage = (evt) => { - try { - const data = JSON.parse(evt.data); - if (data.type === 'busy') { - onBusy(data); - } else if (data.type === 'trajectory') { - // follows the turn's reply once its trajectory is stored; a turn - // with no reply (stopped, or empty) gets no button anywhere - if (turnReplyRow) addTrajectoryButton(turnReplyRow, data.message_id); - turnReplyRow = null; - } else if (data.type === 'message') { - const row = appendMessage(data.content, 'ai'); - // only the turn's reply, not status lines ("> tool [...]") or thinking text - if (data.final) turnReplyRow = row; - // agent_loop() calls _store.touch() after every response, so the - // sidebar's updated_at and message count change — refresh it - setTimeout(() => loadConversations(false), 300); - } else if (data.type === 'system') { - if (!handleDeleteReply(data.content)) appendMessage(data.content, 'system'); - // always refresh after any system message; conversation-mutating - // commands (new, load, fork, rename) change the sidebar list too, - // and /model changes the model shown in the header - setTimeout(() => { loadConversations(false); loadStatus(); }, 150); + let data; + try { data = JSON.parse(evt.data); } catch (e) { data = { type: 'message', content: evt.data }; } + if (heldFrames) heldFrames.push(data); + else handleFrame(data); + }; + } + + // While the chat is being (re)loaded from the server, incoming frames wait + // here, then are handled in order on top of the loaded history. + let heldFrames = null; + async function holdFrames(load) { + const outer = heldFrames === null; + if (outer) heldFrames = []; + try { await load(); } + finally { + if (outer) { + const frames = markReplyInHistory(heldFrames); + heldFrames = null; + for (let i = 0; i < frames.length; i++) { + // a frame started another reload: the rest wait for that one + if (heldFrames) { heldFrames.push(...frames.slice(i)); break; } + handleFrame(frames[i]); } - } catch (e) { - appendMessage(evt.data, 'ai'); - setTimeout(() => loadConversations(false), 300); } - }; + } + } + + // A turn's reply is stored as it is sent, so the loaded history can already + // end with the reply of a turn whose frames were held. Then that turn's + // held lines are left out (they'd land below its reply) and its reply + // frame only points at the history row, for the trajectory button. + function markReplyInHistory(frames) { + const last = lastAiRow(); + const raw = last && last.querySelector('.bubble').dataset.raw; + const end = frames.findIndex(f => f.type === 'message' && f.final && f.content === raw); + if (end < 0) return frames; + return frames.flatMap((f, i) => { + if (i === end) return [{ type: 'reply-in-history', row: last }]; + const turnLine = f.type === 'message' && f.conversation_id != null; + return i < end && turnLine ? [] : [f]; + }); + } + + // A turn frame for a conversation this tab isn't showing (e.g. a stale + // tab) is left out; the conversation's history has its reply. + const otherConv = data => data.conversation_id != null && activeConvId != null && + data.conversation_id !== activeConvId; + + function handleFrame(data) { + if (data.type === 'busy') { + onBusy(data); + } else if (data.type === 'trajectory') { + if (otherConv(data)) return; + // follows the turn's reply once its trajectory is stored; a turn + // with no reply (stopped, or empty) gets no button anywhere + if (turnReplyRow) addTrajectoryButton(turnReplyRow, data.message_id); + turnReplyRow = null; + } else if (data.type === 'reply-in-history') { + turnReplyRow = data.row; + } else if (data.type === 'message') { + if (data.not_started && !agentBusy && pendingTurn) { + // e.g. "Restarting, please send that again": the turn never started + pendingTurn = false; + renderBusy(); + } + if (otherConv(data)) return; + const row = appendMessage(data.content, 'ai'); + // only the turn's reply, not status lines ("> tool [...]") or thinking text + if (data.final) turnReplyRow = row; + // agent_loop() calls _store.touch() after every response, so the + // sidebar's updated_at and message count change — refresh it + setTimeout(() => loadConversations(false), 300); + } else if (data.type === 'system') { + if (!handleDeleteReply(data.content)) appendMessage(data.content, 'system'); + // always refresh after any system message; conversation-mutating + // commands (new, load, fork, rename) change the sidebar list too, + // and /model changes the model shown in the header + setTimeout(() => { loadConversations(false); loadStatus(); }, 150); + } + } + + function lastAiRow() { + const rows = messagesEl.querySelectorAll('.msg-row'); + const last = rows[rows.length - 1]; + return last && last.classList.contains('ai') ? last : null; } // ---- agent busy state + stop ---- @@ -567,12 +646,34 @@ // The running turn's reply row (a "final" message): gets the trajectory download. let turnReplyRow = null; - function onBusy({ busy, stopped }) { + // All tabs share one agent, and a /load, /new or /fork sent mid-turn + // would only run after the turn (whose replies would then land in the + // wrong conversation), so switching waits for the turn to end. + const switchLocked = () => agentBusy || pendingTurn; + const SWITCH_LOCKED = 'Wait for the agent to finish (or stop it) first'; + const SWITCH_LOCKED_MSG = 'The agent is still working. Wait for it to finish, or stop it, before switching conversations.'; + + function onBusy({ busy, stopped, replay, conversation_id: convId }) { agentBusy = busy; if (busy) { + const mine = pendingTurn; pendingTurn = false; turnReplyRow = null; loadStatus(); // the model is shared: another channel may have changed it + // A turn this tab didn't start (another tab, or a message sent + // before a reload) or one in another conversation: show that + // conversation from the history, which already has its message. + // Only the state replayed on connect is already in the history the + // tab loaded then; a turn that started while it loaded is not. + const elsewhere = convId != null && convId !== activeConvId; + if (convId != null && (elsewhere || (!mine && !replay))) { + activeConvId = convId; + holdFrames(async () => { + clearMessages(); + await loadMessages(convId); + }); + loadConversations(false); + } } else stopping = false; renderBusy(); @@ -587,6 +688,11 @@ if (ws && ws.readyState === WebSocket.OPEN) setStatus(working ? 'thinking' : 'connected'); stopBtn.hidden = !agentBusy; stopBtn.disabled = stopping; + const locked = switchLocked(); + newConvBtn.disabled = locked; + newConvBtn.title = locked ? SWITCH_LOCKED : 'New conversation'; + convListEl.classList.toggle('locked', locked); + convListEl.title = locked ? SWITCH_LOCKED : ''; const label = stopping ? 'Stopping after the current step…' : 'Stop'; stopBtn.title = label; stopBtn.setAttribute('aria-label', label); @@ -935,6 +1041,10 @@ const text = inputEl.value.trim(); const files = selectedFiles.slice(); if (sendBtn.disabled) return; + if (switchLocked() && /^\/(new|load|fork)\b/i.test(text)) { + appendMessage(SWITCH_LOCKED_MSG, 'system'); + return; + } let uploaded = []; if (files.length) { diff --git a/app/channels/web_channel.py b/app/channels/web_channel.py index 5f59142..1f3ec17 100644 --- a/app/channels/web_channel.py +++ b/app/channels/web_channel.py @@ -21,15 +21,23 @@ WebSocket message framing:: → {"type": "message", "content": "...", "files": ["", ...]} - ← {"type": "message" | "system", "content": "...", "final": true?} - ← {"type": "busy", "busy": bool, "stopped": bool} + ← {"type": "message", "content": "...", "conversation_id": int?, "final": true?} + ← {"type": "system", "content": "..."} + ← {"type": "busy", "busy": bool, "stopped": bool, "conversation_id": int?} ← {"type": "trajectory", "message_id": int, "conversation_id": int} Plain text is also accepted as a convenience. ``busy`` frames go to every tab -when an agent turn starts/ends (and once on connect); ``stopped`` is true when -a ``/stop`` cut the turn short. A ``trajectory`` frame follows a turn's last -reply (to the tab that sent the message) once its trajectory is stored; -``final`` marks that reply (status lines and thinking text don't have it). +when an agent turn starts/ends; ``stopped`` is true when a ``/stop`` cut the +turn short. Everything an agent turn sends (thinking text, tool status lines, +the reply, an error, the ``trajectory`` frame) carries the turn's +``conversation_id`` and also goes to every tab, since they all share one +agent; ``final`` marks the turn's reply. Command replies (``system``) go only +to the tab that sent the command, and messages with no tab (scheduled tasks) +to every tab. + +On connect a tab gets the last ``busy`` frame delivered and, if a turn is +running, the turn frames delivered since it started, so a reopened browser +sees the turn's progress so far and its remaining replies. """ from __future__ import annotations @@ -120,6 +128,12 @@ def __init__( # Whether an agent turn is running, and whether /stop was sent during it. self._turn_running = False self._stop_requested = False + self._turn_conv_id: int | None = None + # What the browsers have been sent (frames are delivered after the + # agent queues them): the last busy frame and, while that says busy, + # the turn's frames since. Replayed to a tab when it connects. + self._shown_busy: dict = self._busy_frame(False) + self._turn_frames: list[dict] = [] self._connections: dict[str, WebSocket] = {} self._send_locks: dict[str, asyncio.Lock] = {} self._conn_lock = asyncio.Lock() @@ -136,6 +150,9 @@ def __init__( # agent enforces its own per-message limit when building the LLM payload. _MAX_UPLOAD_BYTES = 5 * 1024 * 1024 # 5 MB + # Turn frames kept for replay on connect (the oldest are dropped first). + _MAX_TURN_FRAMES = 500 + # -- Channel ABC -------------------------------------------------------- @property @@ -403,9 +420,28 @@ async def _ws_endpoint(self, ws: WebSocket) -> None: client_id = str(uuid.uuid4()) log.info("WebSocket client connected: %s", client_id) - async with self._conn_lock: - self._connections[client_id] = ws - await self._safe_send_json(client_id, self._busy_frame(self._turn_running)) + # Hold this tab's send lock while catching it up, so frames delivered + # meanwhile queue behind the replay. Registering the connection and + # copying the replay under _conn_lock (send_message records frames + # under it too) means each frame is either replayed or sent live, + # never both or neither. + lock = asyncio.Lock() + async with lock: + async with self._conn_lock: + self._connections[client_id] = ws + self._send_locks[client_id] = lock + # "stopped" was news for the tabs open when the turn ended; + # "replay" tells the tab this turn began before it connected + # (so its history, loaded after this, has the user message) + replay = [ + {**self._shown_busy, "stopped": False, "replay": True}, + *self._turn_frames, + ] + try: + for frame in replay: + await ws.send_json(frame) + except Exception: # pylint: disable=broad-exception-caught # the receive loop below sees the disconnect + log.info("WebSocket client %s left during catch-up", client_id) try: while True: @@ -496,30 +532,41 @@ def _request_stop(self) -> str: return "Stopping after the current step…" @staticmethod - def _busy_frame(busy: bool, stopped: bool = False) -> dict: - return {"type": "busy", "busy": busy, "stopped": stopped} + def _busy_frame( + busy: bool, stopped: bool = False, conversation_id: int | None = None + ) -> dict: + frame = {"type": "busy", "busy": busy, "stopped": stopped} + if conversation_id is not None: + frame["conversation_id"] = conversation_id + return frame async def on_agent_turn(self, running: bool) -> None: """Track the running turn and tell every tab (they share one agent). The frame goes through the outgoing queue, so it reaches the browser - after the turn's replies rather than overtaking them. + after the turn's replies rather than overtaking them. It names the + turn's conversation: the agent's active one, which can't change + until the turn ends (commands queue behind it). """ stopped = not running and self._stop_requested + if running: + self._turn_conv_id = runtime.get(f"conversation_id:{ChannelType.WEB.value}") self._turn_running = running self._stop_requested = False await self.mq.outgoing_msg( OutgoingMessage( content="", channel=self, # the queue routes by the registered channel object - metadata={"busy_frame": self._busy_frame(running, stopped)}, + metadata={ + "busy_frame": self._busy_frame(running, stopped, self._turn_conv_id) + }, ) ) async def on_trajectory_saved( self, message_id: int, conversation_id: int, metadata: dict ) -> None: - """Tell the tab that sent the message its reply can now be downloaded. + """Tell the tabs the turn's reply can now be downloaded. Sent through the outgoing queue so it arrives after the reply itself. """ @@ -539,24 +586,46 @@ async def on_trajectory_saved( ) async def send_message(self, message: OutgoingMessage) -> None: - client_id = message.metadata.get("websocket_id") - is_command = message.metadata.get("is_command", False) - msg_type = "system" if is_command else "message" - payload = {"type": msg_type, "content": message.content} - if message.metadata.get("final") and not is_command: - payload["final"] = True - if "busy_frame" in message.metadata: - client_id, payload = None, message.metadata["busy_frame"] - elif "trajectory_frame" in message.metadata: - payload = message.metadata["trajectory_frame"] - if client_id: - await self._safe_send_json(client_id, payload) + meta = message.metadata or {} + client_id = meta.get("websocket_id") + is_command = meta.get("is_command", False) + # Part of an agent turn: BackgroundAgent tags those with the conversation. + in_turn = meta.get("conversation_id") is not None and not is_command + if "busy_frame" in meta: + payload = meta["busy_frame"] + elif "trajectory_frame" in meta: + payload = meta["trajectory_frame"] else: - # No specific client (e.g. scheduled task delivery) — broadcast to all connected clients - async with self._conn_lock: - targets = list(self._connections.keys()) - for cid in targets: - await self._safe_send_json(cid, payload) + payload = { + "type": "system" if is_command else "message", + "content": message.content, + } + if in_turn: + payload["conversation_id"] = meta["conversation_id"] + if meta.get("final") and not is_command: + payload["final"] = True + if meta.get("not_started"): + payload["not_started"] = True + + if client_id and not in_turn and "busy_frame" not in meta: + await self._safe_send_json(client_id, payload) + return + # Busy and turn frames go to every tab (they share one agent), and so + # does anything without a tab (e.g. a scheduled task's output). + async with self._conn_lock: + self._record(payload, in_turn) + targets = list(self._connections.keys()) + for cid in targets: + await self._safe_send_json(cid, payload) + + def _record(self, payload: dict, in_turn: bool) -> None: + """Remember what a tab connecting now needs to catch up on.""" + if payload.get("type") == "busy": + self._shown_busy = payload + self._turn_frames = [] + elif in_turn and self._shown_busy.get("busy"): + self._turn_frames.append(payload) + del self._turn_frames[: -self._MAX_TURN_FRAMES] # -- Helpers ------------------------------------------------------------ diff --git a/app/core/background_agent.py b/app/core/background_agent.py index 3046ddb..e0eade1 100755 --- a/app/core/background_agent.py +++ b/app/core/background_agent.py @@ -159,6 +159,7 @@ async def process_incoming(self) -> None: init_task_todos() # per-channel todo list (this coroutine runs as its own task) while True: msg = await self.mq.incoming.get() + reply_metadata = msg.metadata try: if msg.content.startswith("/"): parts = msg.content[1:].split(maxsplit=1) @@ -181,23 +182,30 @@ async def process_incoming(self) -> None: OutgoingMessage( content="Restarting, please send that again in a few seconds.", channel=self.channel, - metadata=msg.metadata, + # tells the web UI this message's turn never started + metadata={**(msg.metadata or {}), "not_started": True}, ) ) else: + # Tag the turn's replies (and an error) with its conversation, + # so the web UI can tell which conversation they belong to. + reply_metadata = { + **(msg.metadata or {}), + "conversation_id": self.conversation_id, + } # A /stop that arrived while idle must not cut this turn short. self.channel.clear_stopped() await self.channel.on_agent_turn(True) try: with restart.turn(): # a restart waits for this turn - await self.agent_loop(msg.content, msg.metadata) + await self.agent_loop(msg.content, reply_metadata) finally: await self.channel.on_agent_turn(False) except Exception as e: # pylint: disable=broad-exception-caught # report it, keep serving log.error("Agent loop error: %s", e) await self.mq.outgoing_msg( OutgoingMessage( - content=str(e), channel=self.channel, metadata=msg.metadata + content=str(e), channel=self.channel, metadata=reply_metadata ) ) diff --git a/tests/test_background_agent.py b/tests/test_background_agent.py index 3a3b256..6c48bfc 100644 --- a/tests/test_background_agent.py +++ b/tests/test_background_agent.py @@ -346,7 +346,31 @@ async def test_process_incoming_sends_error_to_user_on_agent_loop_failure(): assert not mq.outgoing.empty() error_msg = await mq.outgoing.get() assert "model not found" in error_msg.content - assert error_msg.metadata == {"chat_id": 99} + # tagged with the turn's conversation, like the turn's replies + assert error_msg.metadata == { + "chat_id": 99, + "conversation_id": agent.conversation_id, + } + + +@pytest.mark.asyncio +async def test_process_incoming_tags_turn_metadata_with_the_conversation(): + agent, _, mq = make_agent() + agent.conversation_id = 42 + with patch.object(agent, "agent_loop", AsyncMock()) as loop: + await mq.incoming.put( + IncomingMessage( + content="hi", channel=ChannelType.TELEGRAM, metadata={"chat_id": 1} + ) + ) + task = asyncio.create_task(agent.process_incoming()) + await asyncio.sleep(0.05) + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + loop.assert_awaited_once_with("hi", {"chat_id": 1, "conversation_id": 42}) @pytest.mark.asyncio @@ -554,6 +578,7 @@ async def test_no_new_turns_while_restart_is_pending(): loop.assert_not_awaited() reply = mq.outgoing.get_nowait() assert "Restarting" in reply.content + assert reply.metadata["not_started"] is True # the web UI clears its pending dots # --- trajectories --- diff --git a/tests/test_web_channel.py b/tests/test_web_channel.py index e66c877..48fb363 100644 --- a/tests/test_web_channel.py +++ b/tests/test_web_channel.py @@ -485,7 +485,12 @@ async def test_websocket_message_is_enqueued_for_agent(): ch, mq = make_web_channel() ch.start() with TestClient(ch._asgi_app).websocket_connect("/ws") as ws: - assert ws.receive_json() == {"type": "busy", "busy": False, "stopped": False} + assert ws.receive_json() == { + "type": "busy", + "busy": False, + "stopped": False, + "replay": True, + } ws.send_text(json.dumps({"type": "message", "content": "hello"})) ws.send_text(json.dumps({"type": "message", "content": "/help"})) ws.send_text(json.dumps({"type": "message", "content": "/whoami"})) @@ -717,6 +722,9 @@ async def test_stop_during_turn_sets_flag_immediately_and_is_not_queued(): @pytest.mark.asyncio async def test_busy_frames_are_queued_in_order_and_broadcast_with_stopped_flag(): + from app.core import runtime + + runtime.set("conversation_id:web", 5) ch, mq = make_web_channel() ws1, ws2 = AsyncMock(), AsyncMock() ch._connections.update({"a": ws1, "b": ws2}) @@ -735,7 +743,12 @@ async def test_busy_frames_are_queued_in_order_and_broadcast_with_stopped_flag() while not mq.outgoing.empty(): await ch.send_message(mq.outgoing.get_nowait()) - busy = lambda b, s: {"type": "busy", "busy": b, "stopped": s} # noqa: E731 + busy = lambda b, s: { # noqa: E731 + "type": "busy", + "busy": b, + "stopped": s, + "conversation_id": 5, + } assert [c.args[0] for c in ws2.send_json.call_args_list] == [ busy(True, False), busy(False, True), @@ -749,14 +762,176 @@ async def test_busy_frames_are_queued_in_order_and_broadcast_with_stopped_flag() ) -def test_websocket_gets_current_busy_state_on_connect(): +def _turn_msg(content, conv_id=7, **meta): + return OutgoingMessage( + content=content, + channel=ChannelType.WEB, + metadata={"websocket_id": "old-tab", "conversation_id": conv_id, **meta}, + ) + + +def _busy_msg(busy, stopped=False, conv_id=7): + return OutgoingMessage( + content="", + channel=ChannelType.WEB, + metadata={"busy_frame": WebChannel._busy_frame(busy, stopped, conv_id)}, + ) + + +def test_websocket_gets_idle_state_on_connect(): from starlette.testclient import TestClient ch, _ = make_web_channel() ch.start() - ch._turn_running = True with TestClient(ch._asgi_app).websocket_connect("/ws") as ws: - assert ws.receive_json() == {"type": "busy", "busy": True, "stopped": False} + assert ws.receive_json() == { + "type": "busy", + "busy": False, + "stopped": False, + "replay": True, + } + + +@pytest.mark.asyncio +async def test_turn_frames_go_to_every_tab_with_the_conversation_id(): + """All tabs share one agent: a tab opened mid-turn must see its replies, + even though the turn was started from another (maybe closed) tab.""" + ch, _ = make_web_channel() + a, b = AsyncMock(), AsyncMock() + ch._connections.update({"old-tab": a, "new-tab": b}) + await ch.send_message(_turn_msg("> bash [ls]...")) + await ch.send_message(_turn_msg("done", final=True)) + expected = [ + {"type": "message", "content": "> bash [ls]...", "conversation_id": 7}, + {"type": "message", "content": "done", "conversation_id": 7, "final": True}, + ] + assert [c.args[0] for c in a.send_json.await_args_list] == expected + assert [c.args[0] for c in b.send_json.await_args_list] == expected + + +@pytest.mark.asyncio +async def test_refused_message_is_marked_not_started_for_its_tab_only(): + ch, _ = make_web_channel() + a, b = AsyncMock(), AsyncMock() + ch._connections.update({"a": a, "b": b}) + await ch.send_message( + OutgoingMessage( + content="Restarting, please send that again in a few seconds.", + channel=ChannelType.WEB, + metadata={"websocket_id": "a", "not_started": True}, + ) + ) + a.send_json.assert_awaited_once_with( + { + "type": "message", + "content": "Restarting, please send that again in a few seconds.", + "not_started": True, + } + ) + b.send_json.assert_not_awaited() + + +def test_frontend_clears_pending_turn_only_for_refused_messages(): + js = (FRONTEND_DIR / "app.js").read_text() + assert "if (data.not_started && !agentBusy && pendingTurn)" in js + assert "data.conversation_id == null && !agentBusy" not in js + + +@pytest.mark.asyncio +async def test_command_replies_still_go_only_to_the_requesting_tab(): + ch, _ = make_web_channel() + a, b = AsyncMock(), AsyncMock() + ch._connections.update({"a": a, "b": b}) + await ch.send_message( + OutgoingMessage( + content="Loaded", + channel=ChannelType.WEB, + metadata={"websocket_id": "a", "is_command": True, "conversation_id": 7}, + ) + ) + a.send_json.assert_awaited_once_with({"type": "system", "content": "Loaded"}) + b.send_json.assert_not_awaited() + + +def test_tab_opened_mid_turn_gets_busy_state_and_the_turn_so_far(): + import asyncio + + from starlette.testclient import TestClient + + ch, _ = make_web_channel() + ch.start() + + async def deliver(*msgs): + for m in msgs: + await ch.send_message(m) + + asyncio.run(deliver(_busy_msg(True), _turn_msg("thinking..."), _turn_msg("> bash"))) + client = TestClient(ch._asgi_app) + with client.websocket_connect("/ws") as ws: + assert ws.receive_json() == { + "type": "busy", + "busy": True, + "stopped": False, + "conversation_id": 7, + "replay": True, + } + assert ws.receive_json()["content"] == "thinking..." + assert ws.receive_json()["content"] == "> bash" + + # once the turn has ended its frames are in the history, not replayed; + # and "stopped" was news only for the tabs open at the time + asyncio.run(deliver(_turn_msg("done", final=True), _busy_msg(False, stopped=True))) + assert ch._turn_frames == [] + with client.websocket_connect("/ws") as ws: + assert ws.receive_json() == { + "type": "busy", + "busy": False, + "stopped": False, + "conversation_id": 7, + "replay": True, + } + # live busy frames aren't marked: a turn that starts while a tab loads + # its history makes the tab reload it (to show the new user message) + assert "replay" not in ch._shown_busy + + +def test_frontend_reloads_history_for_turns_that_start_while_loading(): + js = (FRONTEND_DIR / "app.js").read_text() + assert "(elsewhere || (!mine && !replay))" in js + + +@pytest.mark.asyncio +async def test_turn_replay_is_capped(): + ch, _ = make_web_channel() + ch._MAX_TURN_FRAMES = 3 + await ch.send_message(_busy_msg(True)) + for i in range(5): + await ch.send_message(_turn_msg(f"line {i}")) + assert [f["content"] for f in ch._turn_frames] == ["line 2", "line 3", "line 4"] + + +@pytest.mark.asyncio +async def test_scheduled_task_output_is_not_replayed_as_part_of_the_turn(): + ch, _ = make_web_channel() + await ch.send_message(_busy_msg(True)) + await ch.send_message( + OutgoingMessage(content="task output", channel=ChannelType.WEB, metadata={}) + ) + assert ch._turn_frames == [] + + +def test_frontend_locks_conversation_switching_during_a_turn(): + js = (FRONTEND_DIR / "app.js").read_text() + for needle in ( + "const switchLocked = () => agentBusy || pendingTurn;", + "if (switchLocked()) {", # sidebar click + "|| switchLocked()) return;", # + New + "/^\\/(new|load|fork)\\b/i", # typed commands + "disabled: switchLocked() ? SWITCH_LOCKED : ''", # Fork menu item + "holdFrames(", # replayed frames wait for the history + "data.conversation_id !== activeConvId", + ): + assert needle in js, needle def test_index_has_hidden_stop_button():