diff --git a/CHANGELOG.md b/CHANGELOG.md index 85a3f9437e..cedb3358dd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,18 @@ between releases do not reach them. Remote-PC runtime bundles are versioned sepa ## [Unreleased] +### Changed +- Chat updates follow native transcript changes pushed through the existing WebSocket. + Only visible Chat views subscribe; Terminal views and hidden tabs release conversation + monitoring. Continuous pushes coalesce into at most one newest-page refresh every 2 seconds, + with a 10-second fallback; without pushes (an older remote PC, or a tab that let go of its pane) + Chat reads every 2 seconds as before. Initial display, sends, reconnect and returning to the tab refresh + immediately without overlapping reads or dropping held history. A Chat that shows the + terminal's text because no native transcript exists yet rereads on the pane's output instead. + Conversation responses use negotiated gzip locally, and through a remote PC once its bridge + runs a bundle with this change, while retaining conditional reads and pagination. + Approval-card polling remains independent. + ### Fixed - A request that carries a session cookie but states no `Origin` is now read as cross-site rather than same-origin, so a header that went missing is no longer trusted as though a browser had diff --git a/docs/chat-mode-audit.md b/docs/chat-mode-audit.md index b350c9d221..876b22b842 100644 --- a/docs/chat-mode-audit.md +++ b/docs/chat-mode-audit.md @@ -34,8 +34,8 @@ changes, and Claude assistant-message models. Missing effort stays `null`; it is never inferred from reasoning text. Model and effort are cached with the turns. The composer status line displays the model and `Reasoning ` on desktop -and mobile, or `Reasoning —` when unavailable. Existing conversation polling -refreshes these labels even without a new message. Metadata is tagged with its +and mobile, or `Reasoning —` when unavailable. A conversation read after each native +transcript change refreshes these labels even without a new message. Metadata is tagged with its pane so a pane switch cannot display another conversation's settings; scrollback fallback clears it. The live Codex API returned the configured model and `xhigh` effort after the change. Refresh the page to load the rebuilt client. diff --git a/docs/development.md b/docs/development.md index 6a1b0a6ca8..bfba80d65b 100644 --- a/docs/development.md +++ b/docs/development.md @@ -89,6 +89,10 @@ Tests run against a herdr session of their own, `herdr-web-ui-test`. The first r Browser checks look for Chrome at `/opt/google/chrome/chrome`; set `CHROME_PATH` otherwise. After a herdr upgrade, refresh the generated wire types with `bun run generate:types --refresh` (and `--check` to verify). +`HEADED=1 bun scripts/chat-browser-qa.ts` opens its owned native-transcript fixture in a +visible Chrome window. Its detected Codex fixture process keeps the rollout open; no +real model is invoked by this script. Set `CHROME_PATH` to the macOS Chrome binary there. + For isolated phone viewer and keyboard layout regressions (no herdr session; the demo runner builds the real client locally): ```bash @@ -106,6 +110,118 @@ bun scripts/font-swap-demo-regression.ts # the app's faces arriving late on `FILE_VIEWER_CASE=landscape-notch` selects a viewer case; `FILE_VIEWER_CSS=/path/to/before.css` compares another stylesheet. These checks use Chromium mobile emulation and synthetic safe-area/keyboard geometry; they cannot verify actual iOS Safari keyboard dismissal or notch insets. The existing `bun scripts/file-viewer-regression.ts` separately checks history with an owned herdr pane. The original `scripts/mobile-viewport-regression.ts` exports `checkMobileViewport` for the real-app `bun run test:ui` suite; it also checks the command palette and xterm focus transitions. The demo runners build the real client into a temporary directory, inject the committed fictional-session transport and serve it only on loopback; they do not use a live herdr session or download website media. Run on its own, each builds the client itself (`scripts/demo-build.ts`); the browser lane builds it once and names the directory in `HERDR_DEMO_BUILD`, and each runner copies that instead. They exercise real-app viewport and alert geometry, but not live herdr connectivity. +## Conversation refresh + +The chat reads `/api/pane/conversation` after a `conversation-changed` WebSocket frame. +The frame contains only `pane_id` and an opaque `signature`; REST remains the authoritative +paged conversation, including metadata and history identity. + +A mounted, visible Chat enables `conversation-watch {pane_id, enabled:true}` on its existing +machine-bound WebSocket, after attaching. Bridges advertise the `"conversation-watch"` feature; +older bridges receive no unknown control frame, and their visible Chat reads every 2 s as before. +The client retains current interest through reconnect, replaying it after attachment and +capability discovery. Hiding the page, leaving Chat or unmounting disables interest. The bridge +drops interest with the attachment, and the client asks again with the pane's next attach: a +tab that let go of its pane while out of use (Settings → Use alongside herdr's own window) has +no pushes while it only watches, so it reads every 2 s, and gets pushes back when it attaches. +Observers may subscribe without gaining input or resize authority. + +The bridge shares one native-file monitor per interested attached pane and one 250 ms stat +timer. Terminal-only and hidden clients do not keep a conversation monitor alive. The monitor +resolves through existing provider/process evidence, checks file identity, size and timestamps +without parsing turns, and rechecks session resolution every 5 s (2 s while a transcript is +missing). The last interested client, failed/ended attachment and server stop release it. + +Newest-page refreshes are single-flight. Push-triggered reads start at least 2 s apart using +monotonic elapsed time, with one trailing refresh at a fixed deadline: continuous writes cannot +keep postponing it. A push after idle reads immediately. Initial display, explicit sends, +status/history repairs, reconnect and visibility return bypass that delay but not serialization. +Gap fills run inside the newest read. An older page starts once the newest read in flight has +settled, so a history change found there drops it first, but it never holds the next newest read +back: a clear or a new turn shows while a slow older page loads. A visible Chat with pushes retains a 10 s +backstop for lost notifications; without pushes (an older bridge, a released pane) the backstop +is the former 2 s poll. Hidden Chat starts no new automatic reads or queued +pagination work; returning preserves held history and scroll. Pending work never follows a +pane or machine switch. + +A Chat showing the scrollback stand-in (no native transcript yet, or none recognized) has no +file for the monitor to watch. While it is visible, the pane's own `pty-data` (a mirror's screen +frames on Windows) invalidates it through the same coordinator, so continuous output rereads +at most every 2 s and a quiet pane is not read. A Chat showing a native transcript ignores +terminal output. + +Conversation responses negotiate gzip, including through the remote-PC HTTP proxy. +Encoding-independent weak ETags preserve conditional reads; 304 has no body and responses +vary on `Accept-Encoding`. No other endpoint is compressed by this change. +Prompt polling remains separate at 2 s: approval cards and suggestions can change on the +terminal screen without a transcript write or agent-status transition. + +### Initial push + gzip measurement + +On macOS, headed Chrome read an owned native omp fixture with 22 prompt/answer pairs +(initial JSON about 580.7 KB). Five assistant appends ran in a 30 s interval with the agent +status held constant. Initial loading was excluded. The same loopback fixture was measured +before and after; this is not a bandwidth estimate for arbitrary conversations. + +These measurements predate explicit visible-Chat subscriptions and the 2 s push-read bound. +They are evidence for the initial push/gzip implementation, not a remeasurement of the current +refresh policy. + +| Metric | Before: 2 s polling | After: push + gzip | +| --- | ---: | ---: | +| Write-to-visible latency (ms) | 1659, 666, 684, 697, 710 | 233, 292, 55, 102, 137 | +| Mean latency | 883.2 ms | 163.8 ms | +| Conversation requests / 30 s | 15 (5 × 200, 10 × 304) | 5 (200) | +| Encoded HTTP body bytes | 2,903,810 | 324,544 | +| New conversation WS payload bytes | 0 | 430 | +| Body + invalidation payload bytes | 2,903,810 | 324,974 | + +Mean latency fell 81.5% and the measured payload fell 88.8%. HTTP/TLS headers, terminal +traffic and prompt requests are excluded. The deliberately repetitive fixture compressed +to about 11.2% of its identity response; real tool output and conversations will differ. + +### Visible-Chat subscription and bounded-refresh resource measurement + +The current policy was compared with upstream 2 s polling on macOS in headed Chrome, +using owned native Codex fixtures initially about 60 KB each. Each row is a paired +20 s sample, excluding initial loading, with fresh bridge processes and the same +16 owned workspaces present for both versions. Both versions retain the existing +WebSocket connections; "push" below means WS invalidation followed by REST reads, +not conversation bodies delivered over WS. + +CPU is cumulative CPU time of the Bun bridge during the sample; RSS is its sampled +mean. Chrome, herdr, PTY and resolver subprocess CPU is excluded. Hidden visibility +was simulated; these are not physical-phone or long-running leak measurements. +Quota requests were blocked equally in both versions. + +| Scenario | Conversation REST reads: polling → push | Bun CPU ms: polling → push | Mean Bun RSS MiB: polling → push | +| --- | ---: | ---: | ---: | +| One idle Chat | 9 → 2 | 119 → 135 | 54.4 → 60.6 | +| Eight idle Chats, same pane | 78 → 15 | 381 → 275 | 58.2 → 68.4 | +| Eight idle Chats, distinct panes | 78 → 16 | 351 → 343 | 70.4 → 71.4 | +| Eight distinct Chats, writes every 4 s | 78 → 39 | 370 → 333 | 70.4 → 73.2 | +| Eight distinct Terminal views | 0 → 0 | 108 → 112 | 58.7 → 58.3 | +| Eight hidden Chats | 0 → 0 | 82 → 100 | 67.6 → 67.8 | +| Eight distinct Chats, writes about every 100 ms | 78 → 80 | 416 → 485 | 73.3 → 72.7 | +| Eight ordinary-shell Terminal views | 0 → 0 | 143 → 150 | 57.2 → 57.1 | + +The high-frequency run wrote 197 updates per pane: 656 WS invalidations produced +80 REST reads across eight views, and every view eventually rendered its final +update. Its bridge CPU remained about 16.5% above polling; bounded reads do not +eliminate server monitor work. Resolver subprocess launches were 78 → 110 in that +sample, so bridge CPU alone must not be presented as total system CPU. + +Terminal-only, hidden and ordinary-shell Terminal samples had zero conversation +monitors, file-stat checks, transcript resolutions, invalidations and REST reads. +Eight views kept eight live sockets in both versions; same-pane Chats shared one +monitor. Leaving Chat released interest, reconnect restored it, and returning from +hidden rendered an appended marker in under 2 s. Closing seven same-pane views +retained the monitor; closing the last view removed all monitors, the monitor timer +and owned PTY children. No browser errors occurred in these samples. + +These resource results do not remeasure the latency or compressed payload figures +in the historical table above. + ## README media `bun run build && bun scripts/readme-media/capture.ts` regenerates the stills and demos in `docs/screenshots/` from a staged, fictional session in its own herdr session (`herdr-web-ui-demo`). Pass `shots` or `video` to redo only one of them. It needs ffmpeg. diff --git a/scripts/chat-browser-qa.ts b/scripts/chat-browser-qa.ts index aa5efe60a0..d580080b0d 100644 --- a/scripts/chat-browser-qa.ts +++ b/scripts/chat-browser-qa.ts @@ -3,12 +3,12 @@ import "./test-herdr.ts"; // a herdr session of its own: nothing shows in the user's import assert from "node:assert/strict"; import { Database } from "bun:sqlite"; -import { copyFileSync, mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { chmodSync, copyFileSync, mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { chromium } from "playwright-core"; import { createServer } from "../server/index.ts"; -import { herdrRpc, workspaceClose, workspaceCreate } from "../server/herdr/client.ts"; +import { herdrRpc, sessionSnapshot, workspaceClose, workspaceCreate } from "../server/herdr/client.ts"; const root = mkdtempSync(join(tmpdir(), "herdr-web-ui-chat-browser-")); const codexHome = join(root, "codex"); @@ -41,19 +41,34 @@ try { add({ type: "function_call_output", call_id: "c2", output: "Process exited with code 1\n1 fail" }, 5); add(message("assistant", answer, "final_answer"), 8); persist(); + // A detected fixture process keeps native descriptor evidence for the rollout; + // the ignored path field of a manual lifecycle report is not a session binding. + mkdirSync(join(root, "bin")); + const executable = join(root, "bin", "codex"); + writeFileSync(executable, `#!/bin/sh\nexec 3<"$1"\nprintf '%s\\n' '${answer}'\nwhile IFS= read -r line; do :; done\n`); + chmodSync(executable, 0o755); const created = await workspaceCreate({ cwd: root, label: "herdr-web-ui-test-chat-browser" }); workspaceId = created.workspace.workspace_id; const paneId = created.root_pane.pane_id; - await herdrRpc("pane.report_agent", { pane_id: paneId, source: "manual", agent: "codex", state: "idle", agent_session_path: rollout }); - await herdrRpc("pane.send_text", { pane_id: paneId, text: `printf '%s\\n' '${answer}'\n` }); + await herdrRpc("pane.send_text", { pane_id: paneId, text: `${executable} ${rollout}\n` }); + // Real herdr process detection runs outside this browser check's clock. + const deadline = Date.now() + 10_000; + while ((await sessionSnapshot()).panes.find((pane) => pane.pane_id === paneId)?.agent !== "codex") { + if (Date.now() >= deadline) throw new Error("owned Codex fixture process was not detected"); + await Bun.sleep(50); + } + await herdrRpc("pane.report_agent", { pane_id: paneId, source: "manual", agent: "codex", state: "idle" }); // tailscaleOwner null: the plain-HTTP page below reaches the server under a name that is not // loopback, which a PC running Tailscale would otherwise send to pairing server = createServer({ port: 0, hostname: "127.0.0.1", token: "", stateDir: join(root, "push"), codexHome, tailscaleOwner: null }); - browser = await chromium.launch({ executablePath: process.env["CHROME_PATH"] ?? "/opt/google/chrome/chrome", headless: true, args: ["--no-sandbox", "--host-resolver-rules=MAP clipboard.test 127.0.0.1", "--no-proxy-server"] }); + browser = await chromium.launch({ executablePath: process.env["CHROME_PATH"] ?? "/opt/google/chrome/chrome", headless: process.env["HEADED"] !== "1", args: ["--no-sandbox", "--host-resolver-rules=MAP clipboard.test 127.0.0.1", "--no-proxy-server"] }); const context = await browser.newContext({ viewport: { width: 1280, height: 800 }, locale: "en-US" }); const page = await context.newPage(); // a pane opens its terminal the first time; this QA is about the chat lens - await page.addInitScript((id) => localStorage.setItem(`herdr-web-ui:view:${id}`, "chat"), paneId); + await page.addInitScript((id) => { + localStorage.setItem("herdr-web-ui:settings", JSON.stringify({ language: "en" })); + localStorage.setItem(`herdr-web-ui:view:${id}`, "chat"); + }, paneId); const errors: string[] = []; page.on("pageerror", (error) => errors.push(error.message)); page.setDefaultTimeout(10_000); @@ -85,7 +100,7 @@ try { return hidden.every((item) => item !== null && item.textContent !== "" && item.getBoundingClientRect().width <= 1); }), true, "the agent's name, every pane status word and the reasoning sentence are read, not drawn"); const work = log.locator(".work-block-head"); - const report = (state: "working" | "idle") => herdrRpc("pane.report_agent", { pane_id: paneId, source: "manual", agent: "codex", state, agent_session_path: rollout }); + const report = (state: "working" | "idle") => herdrRpc("pane.report_agent", { pane_id: paneId, source: "manual", agent: "codex", state }); const head = (expanded: boolean, title: string) => log.locator(`.work-block-head[aria-expanded="${expanded}"]`).filter({ hasText: title }); await head(false, "Worked for 7s").waitFor(); await report("working"); @@ -160,19 +175,26 @@ try { assert.equal(await nativeImage.count(), 1); console.log("PASS native Codex image-only turn loads its pane-scoped thumbnail"); - const gate = new Promise((resolve) => { release = resolve; }); + const { promise: gate, resolve } = Promise.withResolvers(); + release = resolve; let requests = 0; await page.route("**/api/pane/conversation?*", async (route) => { requests += 1; await gate; await route.continue(); }); await page.getByRole("textbox", { name: "Message", exact: true }).fill("true"); await page.getByRole("button", { name: "Send message", exact: true }).click(); - await page.waitForTimeout(250); + // Observe the real browser's send and timer path; this bounded negative window + // catches the old overlapping 2-second refresh without faking herdr/Chromium clocks. + const readDeadline = Date.now() + 10_000; + while (requests === 0) { + if (Date.now() >= readDeadline) throw new Error("send refresh did not begin"); + await Bun.sleep(25); + } assert.equal(await log.getByText(answer, { exact: true }).count(), 1, "send refresh must not blank existing history"); const beforeWait = requests; await page.waitForTimeout(2500); - assert.equal(requests, beforeWait, "slow polls must not overlap"); + assert.equal(requests, beforeWait, "slow refreshes must not overlap"); release(); await page.unrouteAll({ behavior: "wait" }); - console.log("PASS send refresh preserves history and slow polling does not overlap"); + console.log("PASS send refresh preserves history and slow refreshes do not overlap"); await page.setViewportSize({ width: 390, height: 844 }); await page.locator("#workspace-drawer").waitFor({ state: "hidden" }); diff --git a/server/AGENTS.md b/server/AGENTS.md index cf8a1a5f06..2817e941e1 100644 --- a/server/AGENTS.md +++ b/server/AGENTS.md @@ -32,7 +32,8 @@ Bun backend: HTTP API + WebSocket fan-out over herdr's unix-socket API, plus the - The `attach` handler records the pane in `client.data.attached` BEFORE awaiting the creation, so a `detach` (the client's pane switch sends detach-old then attach-new) or `close` landing mid-creation cancels it; on resume a cancelled attach calls `releaseUnclaimed`, which closes the record only if it has no clients AND no other client still has the pane in `attached` (a joiner of the same creation that has not resumed yet). A failed creation removes the pane from `attached` again. - `closeAttachment` removes the pane from every member's `attached`: after a pty exit the members sit on "terminal ended" and a stale entry would read as a live claim in `releaseUnclaimed`. - An attach exit (not a read race, `attach_held` or a retired refusal) is not yet the pane's end: `ended` looks the pane up every 100 ms for `ATTACH_RELOOKUP_FOR_MS` (5 s). A new `terminal_id` (herdr live handoff) starts a new attach for the same clients; a gone pane gets `pty-exit` at once; the same terminal or an unreachable herdr gets it at the deadline. The timer is `attachment.relookup`, cleared in `closeAttachment` (#370). -- SocketData holds `attached`, `mode`, per-pane `output` windows and a `closing` flag. An observe connection's `input`/`keys`/`resize` answer `{type:"error", code:"read_only"}`, and its `attach` never resizes the pty — it receives the current `pane-geometry` and adopts it. Observers can acknowledge output. +- SocketData holds `attached`, explicit `conversationInterests`, `mode`, per-pane `output` windows and a `closing` flag. Conversation monitoring requires an attached claim and enabled interest; the last interested client releases it even while terminal-only clients remain attached. An observe connection's `input`/`keys`/`resize` answer `{type:"error", code:"read_only"}`, and its `attach` never resizes the pty — it receives the current `pane-geometry` and adopts it. Observers can acknowledge output and subscribe to conversation invalidations. +- Native conversation invalidations stat the resolved transcript, or Devin/OpenCode's SQLite database and its `-wal` when present; session changes invalidate even when the database path stays the same, and REST remains authoritative. - WS `keys` messages go through herdr RPC (`pane.send_keys`), not the pty; `input` goes straight into the pty, or to `pane.send_text` on a mirrored pane. A key for an attached pane waits `KEY_SETTLE_MS` after the pane's last pty keystroke, and `TYPED_SETTLE_MS` when that keystroke ended in an ESC (herdr holds a lone ESC on the pty ~150ms), so it never overtakes typing; typing that arrives while a key waits queues behind it. - `SocketData.attached` maps pane IDs to opaque claims that last until detach or terminal end; repeated attaches preserve the current claim. Queued terminal `keys` retain that sender claim, attachment and pty. Recheck their identity, sender membership and readiness before sending, so detach (even followed by reattach) or terminal replacement cancels stale chords. A cancelled chord or text is answered `input_failed`, never `input_not_ready`: the client drops the pane's readiness on that code, and the attach it holds by now was already told `input-ready`. Unattached RPC keys keep their existing path. - Pending chat messages are bridge-owned input intents, never native Tab copies. `pending-input.ts` gives automatic delivery and an explicit Send now action the same ID claim; only the originating live interact connection and pane lease can deliver it. Status comes through the existing collector and CompletionTracker. A committing key is not a new-turn receipt: further automatic sends require a later working/ready cycle. Unknown delivery outcomes never retry. Its checks read `source: "detection"` (a scrolled viewport hides a new menu) and refuse on `modelListWaits`; before the paste, a Claude input box that is not known to be empty (`claudeInputDraft`: typed text, bash mode, a clipped box; the box comes from the detection read, and the viewport's ANSI read, the only one herdr colors, tells grey from typed only when it shows the same box) refuses with `input_draft` and the message is held, never retried; a status read from a snapshot goes in through `mark`/`observe`, so it never overwrites a newer collector event. diff --git a/server/api.contract.test.ts b/server/api.contract.test.ts index 2d831ebe8d..f44b1a9a9b 100644 --- a/server/api.contract.test.ts +++ b/server/api.contract.test.ts @@ -1,8 +1,11 @@ import { describe, expect, it, beforeAll, afterAll } from "bun:test"; import { Database } from "bun:sqlite"; -import { chmodSync, mkdtempSync, existsSync, readFileSync, realpathSync, rmSync, statSync, writeFileSync } from "node:fs"; -import { tmpdir } from "node:os"; +import { chmodSync, mkdirSync, mkdtempSync, existsSync, readFileSync, realpathSync, rmSync, statSync, writeFileSync } from "node:fs"; +import { get, type IncomingHttpHeaders } from "node:http"; +import { homedir, tmpdir } from "node:os"; import { basename, join } from "node:path"; +import { gunzipSync } from "node:zlib"; +import { forgetTranscriptState } from "./conversation.ts"; import { createServer } from "./index.ts"; import type { AgentKind, AgentStatus, ApiError, HealthAuth, PushKey, RemoteAccess, SessionSnapshot, PaneReadResult, UsageReport, WorkspaceCreated, WorktreeListing, WorktreeOpened, WorktreeRemoved } from "../shared/protocol.ts"; import { HerdrUpdater } from "./herdr-update.ts"; @@ -122,6 +125,125 @@ describe("usage API", () => { }); }); +describe("conversation response encoding", () => { + type WireResponse = { status: number; headers: IncomingHttpHeaders; bytes: Buffer }; + + const root = mkdtempSync(join(tmpdir(), "herdr-conversation-encoding-")); + const sessionStore = join(homedir(), ".omp", "agent", "sessions"); + mkdirSync(sessionStore, { recursive: true, mode: 0o700 }); + const store = mkdtempSync(join(sessionStore, "herdr-conversation-encoding-")); + const path = join(store, "rollout.jsonl"); + let bridge: { port: number; stop: () => void }; + let workspaceId: string; + let paneId: string; + + beforeAll(async () => { + forgetTranscriptState(); + writeFileSync(path, [ + { type: "session", version: 3, id: "01a0c7a1-56d9-7e20-9f08-f7a2d973bcff", cwd: root }, + { type: "model_change", modelId: "contract-model" }, + { type: "thinking_level_change", thinkingLevel: "high" }, + { type: "message", message: { role: "user", content: [{ type: "text", text: "Read the native transcript" }] } }, + { type: "message", message: { role: "assistant", content: [{ type: "text", text: "A compressible native transcript response. ".repeat(100) }] } }, + ].map((entry) => JSON.stringify(entry)).join("\n")); + mkdirSync(join(root, "bin")); + const executable = join(root, "bin", "omp"); + writeFileSync(executable, "#!/bin/sh\nsleep 600\n"); + chmodSync(executable, 0o755); + const created = await workspaceCreate({ cwd: root, label: "herdr-web-ui-test-conversation-encoding" }); + workspaceId = created.workspace.workspace_id; + paneId = created.root_pane.pane_id; + await herdrRpc("pane.send_text", { pane_id: paneId, text: `${executable}\n` }); + // Real herdr process detection runs outside this test's clock. + const deadline = Date.now() + 10_000; + while ((await sessionSnapshot()).panes.find((pane) => pane.pane_id === paneId)?.agent !== "omp") { + if (Date.now() >= deadline) throw new Error("owned omp process was not detected"); + await Bun.sleep(50); + } + const seq = Date.now() * 1000; + await herdrRpc("pane.report_agent", { pane_id: paneId, source: "herdr:omp", agent: "omp", state: "idle", seq }); + await herdrRpc("pane.report_agent_session", { pane_id: paneId, source: "herdr:omp", agent: "omp", seq: seq + 1, agent_session_path: path, session_start_source: "startup" }); + bridge = createServer({ port: 0, hostname: "127.0.0.1", token: "", stateDir: join(root, "state") }); + }); + + afterAll(async () => { + bridge?.stop(); + if (workspaceId) await workspaceClose(workspaceId); + forgetTranscriptState(); + rmSync(root, { recursive: true, force: true }); + rmSync(store, { recursive: true, force: true }); + }); + + // node:http retains the wire bytes; fetch transparently decompresses them. + function raw(headers: Record, route = `/api/pane/conversation?pane_id=${encodeURIComponent(paneId)}`): Promise { + const { promise, resolve, reject } = Promise.withResolvers(); + const request = get(`http://127.0.0.1:${bridge.port}${route}`, { headers }, (response) => { + const chunks: Buffer[] = []; + response.on("data", (chunk: Buffer) => chunks.push(chunk)); + response.on("error", reject); + response.on("end", () => resolve({ status: response.statusCode!, headers: response.headers, bytes: Buffer.concat(chunks) })); + }); + request.on("error", reject); + request.setTimeout(5000, () => request.destroy(new Error("conversation response timed out"))); + return promise; + } + + it("gzips equivalent JSON, retains the semantic ETag across encodings, and keeps 304 empty", async () => { + const identity = await raw({ "accept-encoding": "identity" }); + const gzip = await raw({ "accept-encoding": "gzip" }); + expect(identity.status).toBe(200); + expect(identity.headers["content-encoding"]).toBeUndefined(); + expect(gzip.status).toBe(200); + expect(gzip.headers["content-encoding"]).toBe("gzip"); + expect(gzip.headers["content-type"]).toBe("application/json; charset=utf-8"); + expect(gzip.headers.vary).toBe("Accept-Encoding"); + expect(identity.headers.vary).toBe("Accept-Encoding"); + expect(gzip.headers.etag).toBe(identity.headers.etag); + expect(gzip.headers["cache-control"]).toBe("no-store"); + expect(JSON.parse(gunzipSync(gzip.bytes).toString())).toEqual(JSON.parse(identity.bytes.toString())); + expect(gzip.bytes.length).toBeLessThan(identity.bytes.length); + const unchanged = await raw({ "accept-encoding": "gzip", "if-none-match": gzip.headers.etag! }); + expect(unchanged.status).toBe(304); + expect(unchanged.headers.etag).toBe(gzip.headers.etag); + expect(unchanged.headers.vary).toBe("Accept-Encoding"); + expect(unchanged.headers["content-encoding"]).toBeUndefined(); + expect(unchanged.bytes.length).toBe(0); + }); + + it("honours explicit exclusions, wildcards, identity preferences and unacceptable encodings", async () => { + for (const encoding of ["identity", "gzip;q=0", "gzip;q=0, *;q=1", "gzip;q=0.2, identity;q=0.8", "gzip;q=0, gzip;q=1", "gzip;q=0;q=1", "gzip ; q = 0", "gzip;q=invalid", "gzip;q=2"]) { + const response = await raw({ "accept-encoding": encoding }); + expect(response.status).toBe(200); + expect(response.headers["content-encoding"]).toBeUndefined(); + expect(JSON.parse(response.bytes.toString()).source).toBe("omp-transcript"); + } + for (const encoding of ["*", "br, *;q=0.5", "gzip;q=0.5, identity;q=0", "GZip;Q=1", "*;q=0, gzip;q=1"]) { + const response = await raw({ "accept-encoding": encoding }); + expect(response.status).toBe(200); + expect(response.headers["content-encoding"]).toBe("gzip"); + expect(JSON.parse(gunzipSync(response.bytes).toString()).source).toBe("omp-transcript"); + } + const absent = await raw({}); + expect(absent.headers["content-encoding"]).toBeUndefined(); + for (const encoding of ["*;q=0", "identity;q=0, gzip;q=0", "br, identity;q=0"]) { + const response = await raw({ "accept-encoding": encoding }); + expect(response.status).toBe(406); + expect(JSON.parse(response.bytes.toString())).toMatchObject({ error: { code: "not_acceptable" } }); + } + }); + + it("compresses the scrollback JSON fallback but leaves other routes and errors uncompressed", async () => { + const fallback = await raw({ "accept-encoding": "gzip" }, "/api/pane/conversation?pane_id=w999999:p999999"); + expect(fallback.status).toBe(200); + expect(fallback.headers.vary).toBe("Accept-Encoding"); + expect(JSON.parse(gunzipSync(fallback.bytes).toString())).toEqual({ source: "scrollback", turns: [] }); + expect((await raw({ "accept-encoding": "gzip" }, "/api/session")).headers["content-encoding"]).toBeUndefined(); + const invalid = await raw({ "accept-encoding": "gzip" }, "/api/pane/conversation"); + expect(invalid.status).toBe(400); + expect(invalid.headers["content-encoding"]).toBeUndefined(); + }); +}); + describe("voice API", () => { it("keeps the key on the server, refuses cross-site writes and streams a transcript", async () => { const voiceState = mkdtempSync(join(tmpdir(), "herdr-voice-contract-")); diff --git a/server/codex.contract.test.ts b/server/codex.contract.test.ts index 813295c4e1..fb6f88e993 100644 --- a/server/codex.contract.test.ts +++ b/server/codex.contract.test.ts @@ -75,7 +75,6 @@ it("answers an unchanged conversation with a bodyless 304 and a changed one in f const first = await fetch(url); const etag = first.headers.get("etag"); expect(first.status).toBe(200); - expect(etag).toMatch(/^"[\w-]+"$/); expect(first.headers.get("cache-control")).toBe("no-store"); await first.json(); const unchanged = await fetch(url, { headers: { "if-none-match": etag! } }); diff --git a/server/codex.ts b/server/codex.ts index c7b62d8c90..31295c8c21 100644 --- a/server/codex.ts +++ b/server/codex.ts @@ -344,10 +344,11 @@ export const defaultCodexHome = (): string => process.env["CODEX_HOME"] || join( /** What processCodexHome found, by pid and argv, and when (PROCESS_HOME_TTL_MS). */ const processHomes = new Map(); /** - * A chat polls every 2 s and macOS reads the environment with a `ps` spawn: a process's - * environment does not change, so its answer is kept. A pid reused by another process - * carries other arguments or comes after this; ponytail: a reused pid with the same argv - * inside it reads the old store until it runs out. + * A visible chat reads its conversation up to every 2 s while the transcript changes, the + * conversation monitor re-resolves its file every 5 s, and macOS reads the environment with + * a `ps` spawn: a process's environment does not change, so its answer is kept. A pid + * reused by another process carries other arguments or comes after this; ponytail: a + * reused pid with the same argv inside it reads the old store until it runs out. */ const PROCESS_HOME_TTL_MS = 30_000; @@ -1243,8 +1244,9 @@ export async function codexTranscriptPath(paneId: string, cwd: string, home = de // lsof is available on macOS, where /proc does not exist. Keep the same // canonical-store and unambiguous-open-file checks as the Linux path. One run // for every Codex process of the pane (a wrapper and the binary are two), and - // -b keeps lsof off the stat calls it does not need for a name: a chat polls this - // every 2 s, and each run costs about 20 ms of process start and kernel walk + // -b keeps lsof off the stat calls it does not need for a name: a chat reads this up + // to every 2 s while its transcript changes, the conversation monitor every 5 s, and + // each run costs about 20 ms of process start and kernel walk const child = Bun.spawn(["/usr/sbin/lsof", "-nPbw", "-a", "-p", codexProcesses.map((process) => process.pid).join(","), "-Fn"], { stdout: "pipe", stderr: "ignore" }); const timer = setTimeout(() => child.kill(), 3000); try { diff --git a/server/conversation-monitor.ts b/server/conversation-monitor.ts new file mode 100644 index 0000000000..15d774a53b --- /dev/null +++ b/server/conversation-monitor.ts @@ -0,0 +1,125 @@ +import { createHash, randomUUID } from "node:crypto"; +import { statSync } from "node:fs"; +import type { SessionSnapshot } from "../shared/protocol.ts"; +import { ConversationNotStarted, ConversationUnavailable, paneTranscript, type NativeTranscript } from "./conversation.ts"; +import { sessionSnapshot } from "./herdr/client.ts"; + +const STAT_INTERVAL_MS = 250; +const RESOLVE_INTERVAL_MS = 5000; +const MISSING_RESOLVE_INTERVAL_MS = 2000; + +interface WatchedConversation { + transcript: NativeTranscript | null; + observed?: string; + nextResolve: number; + resolving: boolean; +} + +function fileIdentity(path: string): string { + try { + const stat = statSync(path); + return `${stat.dev}:${stat.ino}:${stat.size}:${stat.mtimeMs}:${stat.ctimeMs}`; + } catch { return "missing"; } +} + +/** One server-local stat loop, one resolution per interested pane, never a turn parser. SQLite includes its WAL. */ +export class ConversationMonitor { + private readonly panes = new Map(); + private readonly salt = randomUUID(); + private timer?: ReturnType; + private snapshot: Promise | null = null; + private stopped = false; + + constructor( + private readonly changed: (paneId: string, signature: string) => void, + private readonly codexHome?: string, + private readonly devinDbPath?: string, + private readonly opencodeDb?: string, + ) {} + + get size(): number { return this.panes.size; } + + watch(paneId: string): void { + if (this.stopped || this.panes.has(paneId)) return; + this.panes.set(paneId, { transcript: null, nextResolve: 0, resolving: false }); + if (!this.timer) { + this.timer = setInterval(() => this.poll(), STAT_INTERVAL_MS); + this.timer.unref(); + } + this.poll(); + } + + unwatch(paneId: string): void { + this.panes.delete(paneId); + if (this.panes.size === 0) { + clearInterval(this.timer); + this.timer = undefined; + } + } + + stop(): void { + this.stopped = true; + this.panes.clear(); + clearInterval(this.timer); + this.timer = undefined; + } + + private publish(paneId: string, state: WatchedConversation, observed: string): void { + if (this.panes.get(paneId) !== state || observed === state.observed) return; + state.observed = observed; + const signature = createHash("sha256").update(`${this.salt}\0${paneId}\0${observed}`).digest("base64url").slice(0, 22); + this.changed(paneId, signature); + } + + private read(paneId: string, state: WatchedConversation): boolean { + const transcript = state.transcript; + if (!transcript) return false; + const identity = fileIdentity(transcript.path); + const database = transcript.source === "devin-transcript" || transcript.source === "opencode-transcript"; + const wal = database ? fileIdentity(`${transcript.path}-wal`) : ""; + const session = database ? transcript.session : ""; + const cwd = transcript.source === "devin-transcript" ? transcript.cwd : ""; + const observed = `${transcript.source}\0${transcript.path}\0${session}\0${cwd}\0${identity}\0${wal}`; + // Resolve once immediately on disappearance, not on every missing-file stat. + if (identity === "missing" && observed !== state.observed) state.nextResolve = 0; + this.publish(paneId, state, observed); + return identity !== "missing"; + } + + private poll(): void { + const now = Date.now(); + for (const [paneId, state] of this.panes) { + this.read(paneId, state); + if (!state.resolving && state.nextResolve <= now) void this.resolve(paneId, state); + } + } + + private async resolve(paneId: string, state: WatchedConversation): Promise { + state.resolving = true; + let present = false; + try { + // Concurrent panes share the fresh snapshot. Existing native resolvers retain their + // process/session evidence; no store is searched by transcript modification time. + this.snapshot ??= sessionSnapshot().finally(() => { this.snapshot = null; }); + const snapshot = await this.snapshot; + if (this.panes.get(paneId) !== state) return; + const transcript = await paneTranscript(paneId, this.codexHome, snapshot, this.devinDbPath, this.opencodeDb); + if (this.panes.get(paneId) !== state) return; + state.transcript = transcript; + present = this.read(paneId, state); + } catch (error) { + if (this.panes.get(paneId) !== state) return; + if (error instanceof ConversationUnavailable) { + state.transcript = null; + const pending = error instanceof ConversationNotStarted ? `:${error.sessionId}` : ""; + this.publish(paneId, state, `unavailable:${error.message}${pending}`); + } else { + // A transient herdr failure must not replace a known native file with scrollback. + present = state.transcript !== null; + } + } finally { + state.resolving = false; + state.nextResolve = Date.now() + (present ? RESOLVE_INTERVAL_MS : MISSING_RESOLVE_INTERVAL_MS); + } + } +} diff --git a/server/conversation-push.contract.test.ts b/server/conversation-push.contract.test.ts new file mode 100644 index 0000000000..e20697067d --- /dev/null +++ b/server/conversation-push.contract.test.ts @@ -0,0 +1,470 @@ +import { afterAll, beforeAll, describe, expect, it, spyOn } from "bun:test"; +import { appendFileSync, chmodSync, mkdirSync, mkdtempSync, renameSync, rmSync, writeFileSync } from "node:fs"; +import { homedir, tmpdir } from "node:os"; +import { join } from "node:path"; +import type { ClientMessage, ConversationResponse, ServerMessage } from "../shared/protocol.ts"; +import { ConversationMonitor } from "./conversation-monitor.ts"; +import { forgetTranscriptState } from "./conversation.ts"; +import { createServer } from "./index.ts"; +import { herdrRpc, sessionSnapshot, workspaceClose, workspaceCreate } from "./herdr/client.ts"; + +type TestBridge = { port: number; stop: () => void }; +type ConversationChanged = Extract; + +// These integration tests exercise real native-file stat ticks and asynchronous herdr/WS +// lifecycles. Platform timers cannot be faked; all waits name a condition and a deadline. +async function until(check: () => boolean | Promise, label: string): Promise { + const deadline = Date.now() + 12_000; + while (!(await check())) { + if (Date.now() >= deadline) throw new Error(`Timed out: ${label}`); + await Bun.sleep(10); + } +} + +class TranscriptSocket { + readonly frames: ServerMessage[] = []; + readonly ws: WebSocket; + + constructor(bridge: TestBridge) { + this.ws = new WebSocket(`ws://127.0.0.1:${bridge.port}/ws`); + this.ws.addEventListener("message", (event) => this.frames.push(JSON.parse(String(event.data)) as ServerMessage)); + } + + async open(): Promise { + await until(() => this.frames.some((frame) => frame.type === "snapshot"), "socket snapshot"); + } + + send(message: ClientMessage): void { this.ws.send(JSON.stringify(message)); } + + async waitFor(predicate: (frame: ServerMessage) => boolean, after = 0): Promise { + let found: ServerMessage | undefined; + await until(() => { + found = this.frames.find((frame, index) => index >= after && predicate(frame)); + return found !== undefined; + }, "expected websocket frame"); + return found!; + } + + async changed(paneId: string, after = 0): Promise { + return await this.waitFor((frame) => frame.type === "conversation-changed" && frame.pane_id === paneId, after) as ConversationChanged; + } +} + +describe("native conversation invalidation", () => { + const root = mkdtempSync(join(tmpdir(), "herdr-conversation-push-")); + const sessionStore = join(homedir(), ".omp", "agent", "sessions"); + mkdirSync(sessionStore, { recursive: true, mode: 0o700 }); + const store = mkdtempSync(join(sessionStore, "herdr-conversation-push-")); + const path = join(store, "first.jsonl"); + const replacement = join(store, "second.jsonl"); + const bridges: TestBridge[] = []; + const sockets: TranscriptSocket[] = []; + const monitors = new Map(); + let bridge: TestBridge; + let workspaceId: string; + let paneId: string; + let reportSeq = Date.now() * 1000; + const originalWatch = ConversationMonitor.prototype.watch; + const watched = spyOn(ConversationMonitor.prototype, "watch").mockImplementation(function (this: ConversationMonitor, id: string) { + monitors.set(id, this); + originalWatch.call(this, id); + }); + + function transcript(answer = "first answer", id = "01a0c7a1-56d9-7e20-9f08-f7a2d973bcfe"): string { + return `${[ + { type: "session", version: 3, id, cwd: root }, + { type: "model_change", modelId: "initial-model" }, + { type: "thinking_level_change", thinkingLevel: "high" }, + { type: "message", message: { role: "user", content: [{ type: "text", text: "Watch native changes" }] } }, + { type: "message", message: { role: "assistant", content: [{ type: "text", text: answer }] } }, + ].map((entry) => JSON.stringify(entry)).join("\n")}\n`; + } + + async function report(file = path): Promise { + await herdrRpc("pane.report_agent_session", { pane_id: paneId, source: "herdr:omp", agent: "omp", seq: ++reportSeq, agent_session_path: file, session_start_source: "startup" }); + const binding = await herdrRpc<{ agent: { agent_session?: { kind: string; value: string } } }>("agent.get", { target: paneId }); + expect(binding.agent.agent_session).toMatchObject({ kind: "path", value: file }); + } + + async function connect(target = bridge): Promise { + const socket = new TranscriptSocket(target); + sockets.push(socket); + await socket.open(); + return socket; + } + + async function read(target = bridge): Promise { + const response = await fetch(`http://127.0.0.1:${target.port}/api/pane/conversation?pane_id=${encodeURIComponent(paneId)}`); + expect(response.status).toBe(200); + return await response.json() as ConversationResponse; + } + + function attach(socket: TranscriptSocket): void { + socket.send({ type: "attach", pane_id: paneId, cols: 80, rows: 24, keep_size: true }); + } + + function watch(socket: TranscriptSocket, enabled = true, id = paneId): void { + socket.send({ type: "conversation-watch", pane_id: id, enabled }); + } + + async function barrier(socket: TranscriptSocket, mode: "interact" | "observe" = "interact"): Promise { + const after = socket.frames.length; + socket.send({ type: "role", mode }); + await socket.waitFor((frame) => frame.type === "role-ack" && frame.mode === mode, after); + } + + beforeAll(async () => { + forgetTranscriptState(); + writeFileSync(path, transcript()); + mkdirSync(join(root, "bin")); + const executable = join(root, "bin", "omp"); + writeFileSync(executable, "#!/bin/sh\nsleep 600\n"); + chmodSync(executable, 0o755); + const created = await workspaceCreate({ cwd: root, label: "herdr-web-ui-test-conversation-push" }); + workspaceId = created.workspace.workspace_id; + paneId = created.root_pane.pane_id; + await herdrRpc("pane.send_text", { pane_id: paneId, text: `${executable}\n` }); + await until(async () => (await sessionSnapshot()).panes.find((pane) => pane.pane_id === paneId)?.agent === "omp", "owned omp process detected"); + await herdrRpc("pane.report_agent", { pane_id: paneId, source: "herdr:omp", agent: "omp", state: "idle", seq: ++reportSeq }); + await report(); + bridge = createServer({ port: 0, hostname: "127.0.0.1", token: "", stateDir: join(root, "state"), terminalAttach: false }); + bridges.push(bridge); + expect((await read()).source).toBe("omp-transcript"); + }); + + afterAll(async () => { + for (const socket of sockets) socket.ws.close(); + for (const server of bridges) server.stop(); + watched.mockRestore(); + if (workspaceId) await workspaceClose(workspaceId); + forgetTranscriptState(); + rmSync(root, { recursive: true, force: true }); + rmSync(store, { recursive: true, force: true }); + }); + + it("does not monitor terminal-only attachments and releases disabled interest without detaching", async () => { + const terminal = await connect(); + const reader = await connect(); + const watchCalls = () => { + let count = 0; + for (const [id] of watched.mock.calls) if (id === paneId) count += 1; + return count; + }; + const before = watchCalls(); + attach(terminal); + await terminal.waitFor((frame) => frame.type === "input-ready" && frame.pane_id === paneId); + await barrier(terminal); + expect(watchCalls()).toBe(before); + expect(terminal.frames.some((frame) => frame.type === "conversation-changed")).toBe(false); + + attach(reader); + await reader.waitFor((frame) => frame.type === "pane-geometry" && frame.pane_id === paneId); + watch(reader); + await reader.changed(paneId); + const monitor = monitors.get(paneId)!; + expect(monitor.size).toBe(1); + const afterDisable = reader.frames.length; + watch(reader, false); + await barrier(reader, "observe"); + await reader.waitFor((frame) => frame.type === "pane-geometry" && frame.pane_id === paneId, afterDisable); + expect(monitor.size).toBe(0); + + // Idempotent terminal attachment must not implicitly restore chat interest. + const disabledCalls = watchCalls(); + attach(reader); + await barrier(reader); + expect(watchCalls()).toBe(disabledCalls); + expect(monitor.size).toBe(0); + const afterTerminal = terminal.frames.length; + const afterReader = reader.frames.length; + appendFileSync(path, `${JSON.stringify({ type: "model_change", modelId: "interest-model" })}\n`); + watch(reader); + await reader.changed(paneId, afterReader); + expect((await read()).metadata?.model).toBe("interest-model"); + expect(terminal.frames.slice(afterTerminal).some((frame) => frame.type === "conversation-changed")).toBe(false); + + reader.ws.close(); + await until(() => monitor.size === 0, "closing the only interested client kept no monitor"); + // The uninterested terminal is still attached, and can subscribe without attaching again. + const after = terminal.frames.length; + await barrier(terminal, "observe"); + await terminal.waitFor((frame) => frame.type === "pane-geometry" && frame.pane_id === paneId, after); + watch(terminal); + await terminal.changed(paneId); + watch(terminal, false); + await barrier(terminal, "observe"); + expect(monitor.size).toBe(0); + terminal.ws.close(); + }, 30_000); + + it("rejects malformed interest and unattached enables in-band without closing the socket", async () => { + const socket = await connect(); + const snapshot = socket.frames.find((frame) => frame.type === "snapshot"); + expect(snapshot?.type === "snapshot" && snapshot.features?.includes("conversation-watch")).toBe(true); + for (const fields of [ + { pane_id: paneId }, + { pane_id: paneId, enabled: "true" }, + { pane_id: paneId, enabled: null }, + { enabled: true }, + { pane_id: 7, enabled: true }, + { pane_id: "", enabled: true }, + { pane_id: " ", enabled: false }, + ]) { + const after = socket.frames.length; + socket.ws.send(JSON.stringify({ type: "conversation-watch", ...fields })); + await socket.waitFor((frame) => frame.type === "error" && frame.code === "invalid_conversation_watch", after); + } + const after = socket.frames.length; + watch(socket); + await socket.waitFor((frame) => frame.type === "error" && frame.code === "not_attached", after); + // Disabling an already released subscription is harmless. + watch(socket, false); + await barrier(socket, "observe"); + expect(socket.ws.readyState).toBe(WebSocket.OPEN); + socket.ws.close(); + }); + + it("shares interested pane monitors, notices metadata/tear/replacement/missing/new-session writes and releases detach/close claims", async () => { + const first = await connect(); + const second = await connect(); + const third = await connect(); + const stranger = await connect(); + attach(first); + watch(first); + const initial = await first.changed(paneId); + expect(JSON.stringify(initial)).not.toContain(store); + attach(second); + await barrier(second, "observe"); + watch(second); + await barrier(second, "observe"); + attach(third); + watch(third); + await barrier(third, "observe"); + attach(stranger); + await stranger.waitFor((frame) => frame.type === "pane-geometry" && frame.pane_id === paneId); + await second.waitFor((frame) => frame.type === "pane-geometry" && frame.pane_id === paneId); + const monitor = monitors.get(paneId)!; + expect(monitor.size).toBe(1); + const original = await read(); + expect(original.source).toBe("omp-transcript"); + + let afterFirst = first.frames.length; + let afterSecond = second.frames.length; + appendFileSync(path, `${JSON.stringify({ type: "model_change", modelId: "changed-model" })}\n${JSON.stringify({ type: "thinking_level_change", thinkingLevel: "low" })}\n`); + const metadataChange = await first.changed(paneId, afterFirst); + expect(metadataChange.signature).not.toBe(initial.signature); + expect((await second.changed(paneId, afterSecond)).signature).toBe(metadataChange.signature); + const metadata = await read(); + expect(metadata.metadata).toEqual({ model: "changed-model", reasoning_effort: "low" }); + expect(metadata.turns).toEqual(original.turns); + + afterFirst = first.frames.length; + appendFileSync(path, `${JSON.stringify({ type: "message", message: { role: "assistant", content: [{ type: "text", text: "appended native answer" }] } })}\n`); + await first.changed(paneId, afterFirst); + expect(JSON.stringify((await read()).turns)).toContain("appended native answer"); + + afterFirst = first.frames.length; + appendFileSync(path, '{"type":"model_change","modelId":"torn-model"'); + await first.changed(paneId, afterFirst); + expect((await read()).metadata?.model).toBe("changed-model"); + afterFirst = first.frames.length; + appendFileSync(path, `}\n${JSON.stringify({ type: "thinking_level_change", thinkingLevel: "medium" })}\n`); + await first.changed(paneId, afterFirst); + expect((await read()).metadata).toEqual({ model: "torn-model", reasoning_effort: "medium" }); + + afterFirst = first.frames.length; + writeFileSync(path, transcript()); + await first.changed(paneId, afterFirst); + const truncated = await read(); + expect(truncated.history_id).not.toBe(original.history_id); + afterFirst = first.frames.length; + expect(Buffer.byteLength(transcript("other answer"))).toBe(Buffer.byteLength(transcript())); + writeFileSync(`${path}.new`, transcript("other answer")); + renameSync(`${path}.new`, path); + await first.changed(paneId, afterFirst); + const replaced = await read(); + expect(replaced.history_id).not.toBe(truncated.history_id); + expect(JSON.stringify(replaced.turns)).toContain("other answer"); + + afterFirst = first.frames.length; + rmSync(path); + await first.changed(paneId, afterFirst); + expect(await read()).toEqual({ source: "scrollback", turns: [] }); + afterFirst = first.frames.length; + writeFileSync(path, transcript()); + await first.changed(paneId, afterFirst); + await until(async () => (await read()).source === "omp-transcript", "the missing transcript reappeared"); + + afterFirst = first.frames.length; + writeFileSync(replacement, transcript("new session answer", "01a0c7a1-56d9-7e20-9f08-f7a2d973bcfd")); + await report(replacement); + await first.changed(paneId, afterFirst); + expect(JSON.stringify((await read()).turns)).toContain("new session answer"); + + // The role ACK is a frame-order barrier: detach was processed before the write. + first.send({ type: "detach", pane_id: paneId }); + await barrier(first, "observe"); + expect(monitor.size).toBe(1); + afterFirst = first.frames.length; + afterSecond = second.frames.length; + const afterThird = third.frames.length; + appendFileSync(replacement, `${JSON.stringify({ type: "model_change", modelId: "last-model" })}\n`); + const finalChange = await second.changed(paneId, afterSecond); + expect((await third.changed(paneId, afterThird)).signature).toBe(finalChange.signature); + expect(first.frames.slice(afterFirst).some((frame) => frame.type === "conversation-changed")).toBe(false); + expect(stranger.frames.some((frame) => frame.type === "conversation-changed")).toBe(false); + second.ws.close(); + await until(() => second.ws.readyState === WebSocket.CLOSED, "one interested client closed"); + expect(monitor.size).toBe(1); + third.send({ type: "detach", pane_id: paneId }); + await barrier(third, "observe"); + await until(() => monitor.size === 0, "the last interest detached while terminal-only clients stayed"); + // A detached client reattaching later has no interest until it opts in again. + const beforeReattach = first.frames.length; + attach(first); + await first.waitFor((frame) => frame.type === "pane-geometry" && frame.pane_id === paneId, beforeReattach); + expect(monitor.size).toBe(0); + first.ws.close(); + stranger.ws.close(); + third.ws.close(); + }, 60_000); + + it("discovers the first native file written after a pane was attached", async () => { + const unwritten = join(store, "not-yet-written.jsonl"); + await report(unwritten); + const socket = await connect(); + attach(socket); + watch(socket); + await socket.changed(paneId); + const initialConversation = await read(); + expect(initialConversation.turns).toEqual([]); + const after = socket.frames.length; + writeFileSync(unwritten, transcript("the first newly written answer")); + await socket.changed(paneId, after); + const writtenConversation = await read(); + expect(JSON.stringify(writtenConversation.turns)).toContain("the first newly written answer"); + expect(writtenConversation.history_id).not.toBe(initialConversation.history_id); + const monitor = monitors.get(paneId)!; + socket.ws.close(); + await until(() => monitor.size === 0, "new-file monitor released after close"); + await report(); + }, 25_000); + + it("releases failed attaches, disabled/detached/closed pending interest, and stop without reviving a monitor", async () => { + writeFileSync(path, transcript()); + await report(); + const gate = Promise.withResolvers(); + const pending = createServer({ port: 0, hostname: "127.0.0.1", token: "", stateDir: join(root, "pending"), sidecar: false, terminalAttach: () => gate.promise }); + bridges.push(pending); + const socket = await connect(pending); + attach(socket); + watch(socket); + await socket.changed(paneId); + const detachedMonitor = monitors.get(paneId)!; + socket.send({ type: "detach", pane_id: paneId }); + await barrier(socket, "observe"); + expect(detachedMonitor.size).toBe(0); + const after = socket.frames.length; + attach(socket); + watch(socket); + await socket.changed(paneId, after); + socket.ws.close(); + await until(() => detachedMonitor.size === 0, "close cancelled the pending native monitor"); + const survivor = await connect(pending); + attach(survivor); + watch(survivor); + await survivor.changed(paneId); + watch(survivor, false); + await barrier(survivor); + expect(detachedMonitor.size).toBe(0); + const beforeCompletion = survivor.frames.length; + gate.resolve(false); + await survivor.waitFor((frame) => frame.type === "input-ready" && frame.pane_id === paneId); + expect(detachedMonitor.size).toBe(0); + expect(survivor.frames.slice(beforeCompletion).some((frame) => frame.type === "conversation-changed")).toBe(false); + watch(survivor); + await survivor.changed(paneId, beforeCompletion); + watch(survivor, false); + await barrier(survivor); + expect(detachedMonitor.size).toBe(0); + survivor.ws.close(); + pending.stop(); + + const failedGate = Promise.withResolvers(); + const failing = createServer({ port: 0, hostname: "127.0.0.1", token: "", stateDir: join(root, "failing"), sidecar: false, terminalAttach: () => failedGate.promise }); + bridges.push(failing); + const failed = await connect(failing); + const missingPane = "w999999:p999999"; + failed.send({ type: "attach", pane_id: missingPane, cols: 80, rows: 24 }); + watch(failed, true, missingPane); + await barrier(failed); + const failedMonitor = monitors.get(missingPane)!; + expect(failedMonitor.size).toBe(1); + failedGate.resolve(false); + await failed.waitFor((frame) => frame.type === "error" && frame.code === "pane_not_found"); + expect(failedMonitor.size).toBe(0); + const afterFailure = failed.frames.length; + watch(failed, true, missingPane); + await failed.waitFor((frame) => frame.type === "error" && frame.code === "not_attached", afterFailure); + failing.stop(); + failed.ws.close(); + + const stoppedGate = Promise.withResolvers(); + const stopping = createServer({ port: 0, hostname: "127.0.0.1", token: "", stateDir: join(root, "stopping"), sidecar: false, terminalAttach: () => stoppedGate.promise }); + bridges.push(stopping); + const stoppedSocket = await connect(stopping); + attach(stoppedSocket); + watch(stoppedSocket); + await stoppedSocket.changed(paneId); + const stoppedMonitor = monitors.get(paneId)!; + stopping.stop(); + expect(stoppedMonitor.size).toBe(0); + stoppedGate.resolve(false); + stoppedMonitor.watch(paneId); + expect(stoppedMonitor.size).toBe(0); + }, 45_000); + + it("continues native invalidation while another bridge holds the terminal attach slot", async () => { + writeFileSync(path, transcript()); + await report(); + const owner = createServer({ port: 0, hostname: "127.0.0.1", token: "", stateDir: join(root, "owner"), terminalAttach: true }); + const held = createServer({ port: 0, hostname: "127.0.0.1", token: "", stateDir: join(root, "held"), terminalAttach: true, attachHeldRetryMs: 200 }); + bridges.push(owner, held); + const a = await connect(owner); + const b = await connect(held); + attach(a); + await a.waitFor((frame) => frame.type === "input-ready" && frame.pane_id === paneId && frame.ready !== false); + attach(b); + watch(b); + await b.waitFor((frame) => frame.type === "error" && frame.code === "attach_held"); + await b.changed(paneId); + const heldMonitor = monitors.get(paneId)!; + const after = b.frames.length; + appendFileSync(path, `${JSON.stringify({ type: "model_change", modelId: "held-model" })}\n`); + await b.changed(paneId, after); + expect((await read(held)).metadata?.model).toBe("held-model"); + expect(b.frames.some((frame) => frame.type === "pty-exit")).toBe(false); + held.stop(); + expect(heldMonitor.size).toBe(0); + owner.stop(); + a.ws.close(); + b.ws.close(); + }, 45_000); + + it("releases native interest when the owned pane ends and refuses a stale enable", async () => { + const socket = await connect(); + attach(socket); + watch(socket); + await socket.changed(paneId); + const monitor = monitors.get(paneId)!; + await workspaceClose(workspaceId); + workspaceId = ""; + await socket.waitFor((frame) => frame.type === "pty-exit" && frame.pane_id === paneId); + expect(monitor.size).toBe(0); + const after = socket.frames.length; + watch(socket); + await socket.waitFor((frame) => frame.type === "error" && frame.code === "not_attached", after); + expect(monitor.size).toBe(0); + socket.ws.close(); + }, 25_000); +}); diff --git a/server/conversation.ts b/server/conversation.ts index 166d11f3f8..3de6f9f0e8 100644 --- a/server/conversation.ts +++ b/server/conversation.ts @@ -30,7 +30,7 @@ * This module turns those files into the conversation the chat lens renders; * the pty stays the input path. Pure parsing lives in parseClaudeTranscript / * parseOmpTranscript (unit-tested); pane/session/file resolution is - * integration and lives in paneConversation. + * integration and lives in paneTranscript. */ import { createHash, randomUUID } from "node:crypto"; @@ -679,9 +679,10 @@ function pageBefore(stream: TranscriptStream, source: StreamSource, to: number, } /** - * The newest page is asked for on every poll while an agent works, and between polls its - * file only grows. Rescanning its whole window (16 MB on a long session) and reparsing - * the page each time held the event loop 50-110 ms every 2 s, so per live file: + * The newest page is read again after each transcript change while an agent works (at most + * every 2 s per chat), and between reads its file only grows. Rescanning its whole window + * (16 MB on a long session) and reparsing the page each time held the event loop 50-110 ms + * per read, so per live file: * - the turn starts found so far are kept, and only the bytes appended since are scanned; * - the turns before the page's last turn start are kept (a later append cannot change a * turn that another has followed), and only the last turn is parsed again. @@ -1104,6 +1105,11 @@ type ResolvedTranscript = | { source: StreamSource; path: string; codexHome?: string } | { source: "opencode-transcript"; path: string; session: string }; +/** A native file or database session, resolved without parsing conversation turns. */ +export type NativeTranscript = + | ResolvedTranscript + | { source: "devin-transcript"; path: string; session: string; cwd: string }; + /** * The store a pane's transcript lives in. herdr's agent label follows the * pane's foreground processes, so an omo pane reads as `pi` while it waits and @@ -1117,7 +1123,7 @@ async function resolveTranscript(pane: HerdrPane, cwd: string, codexHome?: strin let agent = pane.agent ?? pane.agent_session?.agent ?? ""; // herdr names no agent for this pane: a session report an earlier agent left behind says // nothing about what runs now, so the pane's processes are asked before it is followed. - // A pane herdr does label is not asked: the lookup would cost every chat poll an RPC (and + // A pane herdr does label is not asked: the lookup would cost every transcript lookup an RPC (and // a process-table read on Windows), and a gjc below another agent would take its chat. if (!pane.agent) { const info = await herdrRpc<{ process_info?: { shell_pid?: number; foreground_processes?: { argv?: unknown }[] } }>( @@ -1216,22 +1222,10 @@ export function devinSessionForPane(pane: HerdrPane, panes: HerdrPane[], argv: s return explicit; } -/** - * pane -> agent session -> transcript turns. Read-only, same-user files only. - * Claude sessions are looked up by id under ~/.claude/projects; omp sessions - * come from herdr or the pane's omp process (ompTranscriptPath); omo sessions are resolved from its own store by - * process/session evidence (omoTranscriptForPane). Throws ConversationUnavailable when the pane has - * no recognized agent store (the caller falls back to the scrollback - * transcript view, like chatmux). - * - * Without `page` this is the newest page. `before` is the page ending at a - * returned cursor; `from` is every turn after one, for a chat that already - * shows the pages before it. A cursor from another file throws HistoryChanged. - * `opencodeDb` is OpenCode's store, unset where OpenCode itself would find it. - */ -export async function paneConversation(paneId: string, codexHome?: string, page: ConversationPage = {}, devinDbPath?: string, opencodeDb?: string): Promise { - const snapshot = await sessionSnapshot(); - const pane = snapshot.panes.find((candidate) => candidate.pane_id === paneId); +/** Resolve the native live file or database session without reading conversation turns. */ +export async function paneTranscript(paneId: string, codexHome?: string, snapshot?: SessionSnapshot, devinDbPath?: string, opencodeDb?: string): Promise { + const panes = (snapshot ?? await sessionSnapshot()).panes; + const pane = panes.find((candidate) => candidate.pane_id === paneId); if (pane === undefined) throw new ConversationUnavailable("pane_not_found"); if (typeof pane.cwd !== "string" || pane.cwd.length === 0) throw new ConversationUnavailable("no_recognized_transcript"); @@ -1249,21 +1243,33 @@ export async function paneConversation(paneId: string, codexHome?: string, page: // No Devin runs under the label: what does run is resolved below, as it was before this // reader (an omo is found by its process tree, whatever herdr calls the pane). if (processes.length === 1) { - const sessionId = devinSessionForPane(pane, snapshot.panes, processes[0]!.argv as string[]); - try { - const answer = devinConversation(sessionId, cwd, page, storePath); - rememberPaneDevin(paneId, sessionId, cwd, storePath); - return answer; - } catch (error) { - if (error instanceof DevinHistoryUnavailable) throw new ConversationUnavailable("transcript_missing"); - throw error; - } + const session = devinSessionForPane(pane, panes, processes[0]!.argv as string[]); + return { source: "devin-transcript", path: storePath, session, cwd }; } } - let resolved: ResolvedTranscript; + return resolveTranscript(pane, pane.cwd, codexHome, panes, opencodeDb); +} + +/** + * pane -> agent session -> transcript turns. Read-only, same-user files only. + * Claude sessions are looked up by id under ~/.claude/projects; omp sessions + * come from herdr or the pane's omp process (ompTranscriptPath); omo sessions are resolved from its own store by + * process/session evidence (omoTranscriptForPane). Throws ConversationUnavailable when the pane has + * no recognized agent store (the caller falls back to the scrollback + * transcript view, like chatmux). + * + * Without `page` this is the newest page. `before` is the page ending at a + * returned cursor; `from` is every turn after one, for a chat that already + * shows the pages before it. A cursor from another file throws HistoryChanged. + * `opencodeDb` is OpenCode's store, unset where OpenCode itself would find it. + */ +export async function paneConversation(paneId: string, codexHome?: string, page: ConversationPage = {}, devinDbPath?: string, opencodeDb?: string): Promise { + const snapshot = await sessionSnapshot(); + let resolved: NativeTranscript; + try { - resolved = await resolveTranscript(pane, pane.cwd, codexHome, snapshot.panes, opencodeDb); + resolved = await paneTranscript(paneId, codexHome, snapshot, devinDbPath, opencodeDb); } catch (error) { if (!(error instanceof ConversationNotStarted)) throw error; // a file read before and gone since is no conversation not begun: the terminal stands in for it @@ -1271,12 +1277,23 @@ export async function paneConversation(paneId: string, codexHome?: string, page: // nothing comes before a conversation not begun: a cursor into it is another one's if (page.before !== undefined || page.since !== undefined || page.from !== undefined) throw new HistoryChanged(); // an agent at work on its first turn has written nothing yet, but its terminal shows the turn - if (pane.agent_status === "working" || pane.agent_status === "blocked") throw error; + const status = snapshot.panes.find((pane) => pane.pane_id === paneId)?.agent_status; + if (status === "working" || status === "blocked") throw error; // the chat says there is nothing yet; the session's first prompt writes the file and - // the next poll's history_id differs, so the chat takes it whole + // the next refresh's history_id differs, so the chat takes it whole const id = `unwritten:${error.sessionId}`; return { source: error.source, turns: [], metadata: { model: null, reasoning_effort: null }, cursor: null, history_id: id, version: answerVersion(id, "") }; } + if (resolved.source === "devin-transcript") { + try { + const answer = devinConversation(resolved.session, resolved.cwd, page, resolved.path); + rememberPaneDevin(paneId, resolved.session, resolved.cwd, resolved.path); + return answer; + } catch (error) { + if (error instanceof DevinHistoryUnavailable) throw new ConversationUnavailable("transcript_missing"); + throw error; + } + } if (resolved.source === "opencode-transcript") { // its caches are keyed by store and session, which is what the pane is remembered against rememberPaneRead(paneId, opencodeReadKey(resolved.path, resolved.session)); diff --git a/server/index.ts b/server/index.ts index 0eed3dc4aa..0b14418c45 100644 --- a/server/index.ts +++ b/server/index.ts @@ -19,6 +19,7 @@ import { startStatusCollector } from "./collector.ts"; import { claudePanePid, claudePaneSession, conversationImage, ConversationUnavailable, forgetPaneTranscriptState, HistoryChanged, paneConversation, paneRunsOmo, toolOutput } from "./conversation.ts"; import { OPENCODE_TOOL_REF } from "./opencode.ts"; import { DevinHistoryChanged } from "./devin.ts"; +import { ConversationMonitor } from "./conversation-monitor.ts"; import { omoPanes } from "./omo.ts"; import { OMO_ALIASES, OmoStatus, processAlive } from "./omo-status.ts"; import { omoRuns, omoTasks } from "./omo-tasks.ts"; @@ -160,7 +161,7 @@ const MAX_WAITING_KEYS = 256; */ export const SUBMIT_DEADLINE_MS = 45_000; const CLAUDE_INPUT_DRAFT_MESSAGE = "Claude Code's input box is not empty (a draft, bash mode, or a box that could not be read); send or clear it in the terminal, then send this message"; -const SERVER_FEATURES: ServerFeature[] = ["submit", "pending-input", "secret-input", "input-ready", "take-over", "watch"]; +const SERVER_FEATURES: ServerFeature[] = ["submit", "pending-input", "secret-input", "input-ready", "take-over", "watch", "conversation-watch"]; /** Bind addresses only this machine can reach, so an unset token is nobody else's business. */ const LOOPBACK_HOSTNAMES = new Set(["127.0.0.1", "localhost", "::1"]); @@ -269,6 +270,7 @@ interface SocketData { relay?: MachineRelay; /** A fresh claim per attach lifetime; deleting it invalidates queued terminal keys. */ attached: Map; + conversationInterests: Set; watches: Map; output: Map; closing: boolean; @@ -329,6 +331,38 @@ function send(client: Client, message: ServerMessage): number { } } +/** The conversation's encoding is negotiated separately from its semantic version/ETag. */ +function conversationResponse(request: Request, body: unknown, headers: Record = {}): Response { + const qualities: Partial> = {}; + for (const entry of (request.headers.get("accept-encoding") ?? "").split(",")) { + const [token, ...parameters] = entry.toLowerCase().split(";"); + const name = token?.trim(); + if (name !== "gzip" && name !== "identity" && name !== "*") continue; + let quality = 1; + for (const parameter of parameters) { + const match = /^\s*q\s*=\s*(.*?)\s*$/.exec(parameter); + if (!match) continue; + const value = match[1]; + if (value === undefined || !/^(?:0(?:\.\d{0,3})?|1(?:\.0{0,3})?)$/.test(value)) { + quality = 0; + break; + } + quality = Math.min(quality, Number(value)); + } + qualities[name] = Math.min(qualities[name] ?? 1, quality); + } + const gzipQuality = qualities.gzip ?? qualities["*"] ?? 0; + const identityQuality = qualities.identity ?? (qualities["*"] === 0 ? 0 : 1); + const responseHeaders = { ...headers, vary: "Accept-Encoding" }; + if (gzipQuality > 0 && (qualities.identity === undefined || gzipQuality >= identityQuality)) { + return new Response(Bun.gzipSync(JSON.stringify(body)), { + headers: { "content-type": "application/json; charset=utf-8", ...responseHeaders, "content-encoding": "gzip" }, + }); + } + if (identityQuality > 0) return jsonResponse(body, 200, responseHeaders); + return jsonResponse({ error: { code: "not_acceptable", message: "no acceptable conversation content encoding" } }, 406, responseHeaders); +} + export interface ServerInstance { port: number; hostname: string; @@ -424,6 +458,21 @@ export function createServer( const pendingDrains = new Set(); type SubmitReply = { ok: boolean; pending?: PendingMessage; code?: string; message?: string }; const pendingRequests = new WeakMap>(); + let stopped = false; + const conversations = new ConversationMonitor((paneId, signature) => { + for (const client of clients) { + if (!client.data.closing && client.readyState === WebSocket.OPEN && client.data.attached.has(paneId) && client.data.conversationInterests.has(paneId)) { + send(client, { type: "conversation-changed", pane_id: paneId, signature }); + } + } + }, options.codexHome, options.devinDbPath, options.opencodeDb); + + function releaseConversation(paneId: string): void { + for (const client of clients) { + if (!client.data.closing && client.readyState === WebSocket.OPEN && client.data.attached.has(paneId) && client.data.conversationInterests.has(paneId)) return; + } + conversations.unwatch(paneId); + } /** each pane's input while a composer message is in flight, one step after another */ const paneQueues = new Map>(); /** when each pane last got keystrokes through its attach pty */ @@ -723,6 +772,7 @@ export function createServer( clients.delete(client); for (const paneId of client.data.attached.keys()) detach(paneId, client); client.data.attached.clear(); + client.data.conversationInterests.clear(); client.data.output.clear(); client.close(OUTPUT_STALLED_CLOSE_CODE, "terminal output consumer stalled"); } @@ -867,7 +917,9 @@ export function createServer( // the "terminal ended" screen): a stale entry would read as a live claim in // releaseUnclaimed and keep a later, empty pty on this pane running for (const member of attachment.clients) member.data.attached.delete(paneId); + for (const member of attachment.clients) member.data.conversationInterests.delete(paneId); for (const member of attachment.clients) member.data.output.delete(paneId); + releaseConversation(paneId); const retired = attachment.pty.exited.finally(() => { if (retiringAttachments.get(paneId) === retired) retiringAttachments.delete(paneId); }); @@ -912,6 +964,7 @@ export function createServer( await retiringAttachments.get(paneId); const mirrored = !(await terminalAttach()); const { terminalId, rect } = await terminalInfoFor(paneId); + if (stopped) throw new HerdrError("server_stopped", "the web bridge stopped during attachment creation"); // an observer-first attachment spawns at the pane's own grid (fallback 80x24 when // the layout has no rect for it): the attach must not seed the shared pty with a // watching phone's viewport @@ -1222,7 +1275,9 @@ export function createServer( function detach(paneId: string, client: Client): void { holdPending(client, paneId); + client.data.conversationInterests.delete(paneId); client.data.output.delete(paneId); + releaseConversation(paneId); const attachment = attachments.get(paneId); if (!attachment) return; attachment.clients.delete(client); @@ -1475,13 +1530,13 @@ export function createServer( try { relay = new MachineRelay(machines, machineId, readOnly); await relay.ready; - const upgraded = bunServer.upgrade(request, { data: { attached: new Map(), watches: new Map(), mode: readOnly ? "observe" : "interact", output: new Map(), closing: false, roles: 0, relay, deviceId, readOnly } }); + const upgraded = bunServer.upgrade(request, { data: { attached: new Map(), watches: new Map(), conversationInterests: new Set(), mode: readOnly ? "observe" : "interact", output: new Map(), closing: false, roles: 0, relay, deviceId, readOnly } }); if (upgraded) return undefined as unknown as Response; relay.close(); } catch { relay?.close(); return new Response("remote websocket unavailable", { status: 502 }); } return new Response("websocket upgrade required", { status: 426 }); } - const upgraded = bunServer.upgrade(request, { data: { attached: new Map(), watches: new Map(), mode: readOnly ? "observe" : "interact", output: new Map(), closing: false, roles: 0, deviceId, readOnly } }); + const upgraded = bunServer.upgrade(request, { data: { attached: new Map(), watches: new Map(), conversationInterests: new Set(), mode: readOnly ? "observe" : "interact", output: new Map(), closing: false, roles: 0, deviceId, readOnly } }); if (upgraded) return undefined as unknown as Response; return new Response("websocket upgrade required", { status: 426 }); } @@ -1977,17 +2032,16 @@ export function createServer( }; try { const { version, ...conversation } = await paneConversation(paneId, options.codexHome, page, options.devinDbPath, options.opencodeDb); - // The chat polls every 2s: an unchanged conversation answers 304 with no body. // no-store keeps the browser's own cache out of it, so the chat sees the 304. - const etag = `"${version}"`; - const headers = { etag, "cache-control": "no-store" }; + const etag = `W/"${version}"`; + const headers = { etag, "cache-control": "no-store", vary: "Accept-Encoding" }; if (request.headers.get("if-none-match") === etag) return new Response(null, { status: 304, headers }); - return jsonResponse(conversation, 200, headers); + return conversationResponse(request, conversation, headers); } catch (error) { if (error instanceof HistoryChanged || error instanceof DevinHistoryChanged) return jsonResponse({ error: { code: "history_changed", message: error.message } }, 409); // an unrecognized pane is not an error: the client falls back to the // scrollback transcript, exactly like chatmux's terminal fallback - if (error instanceof ConversationUnavailable) return jsonResponse({ source: "scrollback", turns: [] }); + if (error instanceof ConversationUnavailable) return conversationResponse(request, { source: "scrollback", turns: [] }); return errorResponse(error); } } @@ -2102,6 +2156,7 @@ export function createServer( clients.delete(client); for (const paneId of client.data.attached.keys()) detach(paneId, client); client.data.attached.clear(); + client.data.conversationInterests.clear(); client.data.output.clear(); client.data.relay?.close(1008, "Device access revoked"); client.close(1008, "Device access revoked"); @@ -2119,7 +2174,7 @@ export function createServer( }, async message(client, raw) { - if (client.data.closing) return; + if (stopped || client.data.closing) return; if (client.data.relay) { client.data.relay.message(raw); return; } let message: ClientMessage; try { @@ -2183,17 +2238,22 @@ export function createServer( // a covered grid (keep_size) creates the pty at the pane's own size, as an observer does attachment = await ensureAttachment(message.pane_id, geometry.cols, geometry.rows, client.data.mode === "observe" || message.keep_size === true); } catch (error) { - if (client.data.attached.get(message.pane_id) === claim) client.data.attached.delete(message.pane_id); + if (client.data.attached.get(message.pane_id) === claim) { + client.data.attached.delete(message.pane_id); + client.data.conversationInterests.delete(message.pane_id); + releaseConversation(message.pane_id); + } throw error; } - if (client.data.attached.get(message.pane_id) !== claim) { + if (stopped || client.data.closing || client.data.attached.get(message.pane_id) !== claim) { releaseUnclaimed(message.pane_id, attachment); break; } - // a mirror whose pane went away during its first read has already ended: joining - // it would leave this client on a terminal that never says anything again - if (attachment.mirror && attachments.get(message.pane_id) !== attachment) { + // A pane that ended during creation must not gain clients or revive interest. + if (attachments.get(message.pane_id) !== attachment) { client.data.attached.delete(message.pane_id); + client.data.conversationInterests.delete(message.pane_id); + releaseConversation(message.pane_id); send(client, { type: "pty-exit", pane_id: message.pane_id, code: null }); break; } @@ -2242,6 +2302,24 @@ export function createServer( detach(message.pane_id, client); break; } + case "conversation-watch": { + if (typeof message.pane_id !== "string" || message.pane_id.trim() === "" || typeof message.enabled !== "boolean") { + send(client, { type: "error", code: "invalid_conversation_watch", message: "pane_id must be a nonempty string and enabled must be a boolean" }); + break; + } + if (message.enabled) { + if (!client.data.attached.has(message.pane_id)) { + send(client, { type: "error", code: "not_attached", message: "conversation interest requires an attached pane", pane_id: message.pane_id }); + break; + } + client.data.conversationInterests.add(message.pane_id); + conversations.watch(message.pane_id); + } else { + client.data.conversationInterests.delete(message.pane_id); + releaseConversation(message.pane_id); + } + break; + } case "pty-ack": { // Observers may acknowledge output, but never input or resize it. const window = client.data.output.get(message.pane_id); @@ -2623,12 +2701,14 @@ export function createServer( close(client) { client.data.unwatchDevice?.(); + client.data.closing = true; if (client.data.relay) { client.data.relay.close(); return; } killWatches(client); pending.close(client); clients.delete(client); for (const paneId of client.data.attached.keys()) detach(paneId, client); client.data.attached.clear(); + client.data.conversationInterests.clear(); client.data.output.clear(); }, }, @@ -2649,6 +2729,8 @@ export function createServer( statusReady: collector.ready, alertsSettled: () => push.settled(), stop: () => { + stopped = true; + conversations.stop(); clearInterval(outputTimer); collector.stop(); omo.stop(); diff --git a/server/machine-api.ts b/server/machine-api.ts index 9869c6107e..d2b0e11258 100644 --- a/server/machine-api.ts +++ b/server/machine-api.ts @@ -103,6 +103,8 @@ export async function handleMachineRequest(request: Request, manager: MachineMan // a remote conversation answers 304 when unchanged, as a local one does const ifNoneMatch = request.headers.get("if-none-match"); if (ifNoneMatch) headers.set("if-none-match", ifNoneMatch); + const conversation = path === "pane/conversation"; + if (conversation) headers.set("accept-encoding", request.headers.get("accept-encoding") ?? "identity"); const abort = new AbortController(); if (path === "fs/file") { // a file (a video, say) streams through, with its ranges, never held here; it can @@ -121,11 +123,16 @@ export async function handleMachineRequest(request: Request, manager: MachineMan } const untrack = manager.trackTerminal(id, () => abort.abort()); try { - const response = await fetch(`${endpoint.url}/api/${path}${url.search}`, { method: request.method, headers, body: ["GET", "HEAD"].includes(request.method) ? undefined : request.body, redirect: "error", signal: AbortSignal.any([request.signal, abort.signal, AbortSignal.timeout(path === "worktree/create" ? 165_000 : 75_000)]) }); - const etag = response.headers.get("etag"); - return new Response(response.status === 304 ? null : await response.arrayBuffer(), { status: response.status, headers: { - "content-type": response.headers.get("content-type") ?? "application/json", "cache-control": "no-store", ...(etag ? { etag } : {}), - } }); + // Conversation bytes retain the browser's negotiated encoding through the relay. + const response = await fetch(`${endpoint.url}/api/${path}${url.search}`, { method: request.method, headers, body: ["GET", "HEAD"].includes(request.method) ? undefined : request.body, redirect: "error", decompress: !conversation, signal: AbortSignal.any([request.signal, abort.signal, AbortSignal.timeout(path === "worktree/create" ? 165_000 : 75_000)]) }); + const passed = new Headers({ + "content-type": response.headers.get("content-type") ?? "application/json", "cache-control": "no-store", + }); + for (const name of conversation ? ["etag", "content-encoding", "vary"] : ["etag"]) { + const value = response.headers.get(name); + if (value) passed.set(name, value); + } + return new Response(response.status === 304 ? null : await response.arrayBuffer(), { status: response.status, headers: passed }); } catch (error) { return transportFailure(id, path, error); } finally { untrack(); } } diff --git a/server/machine-proxy.test.ts b/server/machine-proxy.test.ts index d4ed07b143..83f0400464 100644 --- a/server/machine-proxy.test.ts +++ b/server/machine-proxy.test.ts @@ -1,4 +1,5 @@ import { afterAll, describe, expect, it } from "bun:test"; +import { gunzipSync } from "node:zlib"; import { handleMachineRequest } from "./machine-api.ts"; import type { MachineManager } from "./machines.ts"; @@ -12,6 +13,14 @@ const remote = Bun.serve({ if (path === "/api/pane/conversation/image") return new Response(new Uint8Array([137, 80, 78, 71]), { headers: { "content-type": "image/png" } }); if (path === "/api/pane/conversation/tool-output") return new Response("complete remote output", { headers: { "content-type": "text/plain; charset=utf-8" } }); if (path === "/api/fs/file") return new Response("%PDF-1.7", { headers: { "content-type": "application/pdf" } }); + if (path === "/api/pane/conversation" && new URL(request.url).searchParams.has("compressed")) { + const headers = { etag: "\"gzip-v1\"", vary: "Accept-Encoding", "content-type": "application/json" }; + if (request.headers.get("if-none-match") === headers.etag) return new Response(null, { status: 304, headers }); + const body = JSON.stringify({ source: "omp-transcript", turns: [{ role: "assistant", ts: null, parts: [{ kind: "text", text: "remote native answer" }] }] }); + return request.headers.get("accept-encoding") === "gzip" + ? new Response(Bun.gzipSync(body), { headers: { ...headers, "content-encoding": "gzip" } }) + : new Response(body, { headers }); + } const ifNoneMatch = request.headers.get("if-none-match"); asked.push(ifNoneMatch); if (ifNoneMatch === "\"v1\"") return new Response(null, { status: 304, headers: { etag: "\"v1\"" } }); @@ -40,6 +49,22 @@ describe("PC proxy", () => { }); }); +it("preserves negotiated remote conversation bytes without decompressing the phone response", async () => { + const url = "http://127.0.0.1/api/machines/pc1/pane/conversation?pane_id=w1:p1&compressed=1"; + const compressed = await handleMachineRequest(new Request(url, { headers: { "accept-encoding": "gzip" } }), manager); + expect(compressed.headers.get("content-encoding")).toBe("gzip"); + expect(compressed.headers.get("vary")).toBe("Accept-Encoding"); + const decoded = JSON.parse(gunzipSync(await compressed.arrayBuffer()).toString()); + const identity = await handleMachineRequest(new Request(url, { headers: { "accept-encoding": "identity" } }), manager); + expect(identity.headers.get("content-encoding")).toBeNull(); + expect(await identity.json()).toEqual(decoded); + const unchanged = await handleMachineRequest(new Request(url, { headers: { "accept-encoding": "gzip", "if-none-match": "\"gzip-v1\"" } }), manager); + expect(unchanged.status).toBe(304); + expect(unchanged.headers.get("etag")).toBe("\"gzip-v1\""); + expect(unchanged.headers.get("vary")).toBe("Accept-Encoding"); + expect(await unchanged.text()).toBe(""); +}); + it("forwards conversation images and complete output, while rejecting arbitrary nested paths", async () => { const base = "http://127.0.0.1/api/machines/pc1/pane/conversation"; const image = await handleMachineRequest(new Request(`${base}/image?pane_id=w1:p1&ref=asset`), manager); diff --git a/shared/protocol.ts b/shared/protocol.ts index 11adf7236c..2849e08b31 100644 --- a/shared/protocol.ts +++ b/shared/protocol.ts @@ -610,9 +610,10 @@ export interface PushPayload { * Client -> server frames: attach {pane_id, cols, rows} | detach {pane_id} | input {pane_id, text} * | keys {pane_id, keys} | resize {pane_id, cols, rows} | role {mode} * | pty-ack {pane_id, stream_id, offset} | secret {id, pane_id, prompt, secret} - * | watch {pane_id, cols, rows} | unwatch {pane_id} + * | watch {pane_id, cols, rows} | unwatch {pane_id} | conversation-watch {pane_id, enabled} * Server -> client frames: snapshot | pty-data | pty-exit | pane-geometry | role-ack - * | pane-status | pane-exited | session-changed | secret-result | watch-data | watch-end | error + * | pane-status | pane-exited | session-changed | conversation-changed | secret-result + * | watch-data | watch-end | error * * attach {flow_control:"ack"} opts into per-subscription output credit. * pty-data.flow carries a stream_id and cumulative UTF-8 payload offset; @@ -651,6 +652,9 @@ export type ClientMessage = /** keep_size: the grid is covered (the chat lens), so the attach leaves the shared pty's size as it is */ | { type: "attach"; pane_id: string; cols: number; rows: number; flow_control?: "ack"; keep_size?: boolean } | { type: "detach"; pane_id: string } + /** Native invalidations require "conversation-watch" and an attachment; observers may subscribe. + * Detaching clears interest, so reconnect/reattach must enable it again. REST remains authoritative. */ + | { type: "conversation-watch"; pane_id: string; enabled: boolean } /** a pane another web bridge holds (`attach_held`): take herdr's attach slot from it, here, now */ | { type: "take-over"; pane_id: string } | { type: "input"; pane_id: string; text: string } @@ -679,7 +683,7 @@ export type ClientMessage = | { type: "unwatch"; pane_id: string }; /** What a server supports beyond the base protocol, listed in its first snapshot; older bridges list nothing. */ -export type ServerFeature = "submit" | "pending-input" | "secret-input" | "input-ready" | "take-over" | "watch"; +export type ServerFeature = "submit" | "pending-input" | "secret-input" | "input-ready" | "take-over" | "watch" | "conversation-watch"; export type ServerMessage = | { type: "snapshot"; snapshot: SessionSnapshot; features?: ServerFeature[] } @@ -701,6 +705,8 @@ export type ServerMessage = | { type: "secret-result"; id: number; pane_id: string; ok: boolean; code?: string } /** agent-status push for ANY pane, attached or not (server-side status collector) */ | { type: "pane-status"; pane_id: string; agent_status: AgentStatus; /** a pane's running background tasks (OmO's, or a Claude Code pane's subagents and commands), when the frame is about them: it changes no status */ background_tasks?: number; /** every frame says it: absent, the pane does not wait on its turn's background work */ background_wait?: true } + /** Native transcript invalidation for an interested attached pane; signature is opaque, not a REST ETag. Refetch the existing conversation endpoint. */ + | { type: "conversation-changed"; pane_id: string; signature: string } /** a pane's process exited (pushed even when nobody is attached to it) */ | { type: "pane-exited"; pane_id: string } /** session structure changed (pane created/closed): refetch /api/session */ diff --git a/site/demo/transport.ts b/site/demo/transport.ts index 85a9a44d8f..c56e1c1132 100644 --- a/site/demo/transport.ts +++ b/site/demo/transport.ts @@ -162,6 +162,7 @@ function createDemoWorkspace(cwd: string, label: string, agent: string | null, w if (agent) { keyOfPane.set(pane.pane_id, pane.pane_id); chats.set(pane.pane_id, { turns: [], metadata: { model: agent === "codex" ? "gpt-5.6-sol" : "claude-opus-5-5", reasoning_effort: "medium" } }); + conversationChanged(pane.pane_id); setTimeout(() => { if (!replying.has(pane.pane_id)) finishDemoTurn(pane.pane_id, "idle"); }, 1500); } return { workspace_id: id, pane_id: pane.pane_id, agent_started: agent !== null }; @@ -190,6 +191,7 @@ function closeDemoWorkspaces(ids: Set): void { const sseListeners = new Set(); const sockets = new Set(); +const conversationVersions = new Map(); function emitSse(event: MachineEvent): void { const message = new MessageEvent("message", { data: JSON.stringify(event) }); @@ -213,6 +215,12 @@ function setStatus(paneId: string, status: AgentStatus): void { for (const socket of sockets) socket.push({ type: "pane-status", pane_id: paneId, agent_status: status }); } +function conversationChanged(paneId: string): void { + const revision = (conversationVersions.get(paneId) ?? 0) + 1; + conversationVersions.set(paneId, revision); + for (const socket of sockets) socket.push({ type: "conversation-changed", pane_id: paneId, signature: `demo:${revision}` }); +} + function structureChanged(): void { emitSse({ type: "machines", machines }); emitSse({ type: "machine-message", machine_id: local.id, message: { type: "session-changed" } }); @@ -242,12 +250,14 @@ function submitToChat(paneId: string, text: string): void { const chat = key ? chats.get(key) : undefined; if (!chat) return; chat.turns.push({ role: "user", ts: now(), parts: [{ kind: "text", text }] }); + conversationChanged(paneId); const agent = agentOf(paneId); if (replying.has(paneId)) return; replying.add(paneId); setStatus(paneId, "working"); setTimeout(() => { chat.turns.push({ role: "assistant", ts: now(), end_ts: now(), parts: [{ kind: "text", text: `This is the demo, so nothing ran. In the real app that message went to ${agent} in this pane, and its answer would be written here as it arrives, with its commands and edits folded above it.` }] }); + conversationChanged(paneId); finishDemoTurn(paneId); }, CHAT_ANSWER_MS); } @@ -266,6 +276,7 @@ function answerPrompt(paneId: string, optionIndex: number | undefined): void { : [{ kind: "tool", name: "exec", summary: "git push origin feat/export-guard", input: JSON.stringify({ cmd: "git push origin feat/export-guard" }, null, 2), output: "To github.com:acme/web-dashboard.git\n * [new branch] feat/export-guard -> feat/export-guard" }, { kind: "text", text: "Pushed `feat/export-guard`. The export button is guarded and the branch is ready for a pull request." }], }); + if (chat) conversationChanged(paneId); finishDemoTurn(paneId); }, PROMPT_ANSWER_TURN_MS); } @@ -583,6 +594,7 @@ async function route(url: URL, method: string, init: RequestInit | undefined, in if (agent) { keyOfPane.set(pane.pane_id, pane.pane_id); chats.set(pane.pane_id, { turns: [], metadata: { model: agent === "codex" ? "gpt-5.6-sol" : "claude-opus-5-5", reasoning_effort: "medium" } }); + conversationChanged(pane.pane_id); setTimeout(() => { if (!replying.has(pane.pane_id)) finishDemoTurn(pane.pane_id, "idle"); }, 1500); } structureChanged(); @@ -690,6 +702,7 @@ class DemoSocket extends EventTarget { private readonly pending = new Map(); private readonly outcomes = new Map(); private readonly requests = new Map(); + private readonly conversationInterests = new Set(); /** the pretend shell's current line, per shell pane */ private readonly lines = new Map(); @@ -705,7 +718,7 @@ class DemoSocket extends EventTarget { const open = new Event("open"); this.onopen?.(open); this.dispatchEvent(open); - this.push({ type: "snapshot", snapshot: snapshot(), features: ["submit", "pending-input", "secret-input", "input-ready"] }); + this.push({ type: "snapshot", snapshot: snapshot(), features: ["submit", "pending-input", "secret-input", "input-ready", "conversation-watch"] }); }, 20); this.timers.add(opening); } @@ -718,6 +731,19 @@ class DemoSocket extends EventTarget { push(message: ServerMessage): void { if (this.readyState !== 1) return; + if (message.type === "conversation-changed" && (!this.attached.has(message.pane_id) || !this.conversationInterests.has(message.pane_id))) return; + if (message.type === "pty-exit") { + this.attached.delete(message.pane_id); + this.conversationInterests.delete(message.pane_id); + } + if (message.type === "session-changed") { + for (const paneId of this.attached) { + if (!paneOf(paneId)) { + this.attached.delete(paneId); + this.conversationInterests.delete(paneId); + } + } + } const event = new MessageEvent("message", { data: JSON.stringify(message) }); this.onmessage?.(event); this.dispatchEvent(event); @@ -725,7 +751,7 @@ class DemoSocket extends EventTarget { send(raw: string): void { if (this.readyState !== 1) return; - let message: { type: string; pane_id?: string; text?: string; payload?: string; keys?: string[]; id?: number; mode?: string; typed?: unknown; delivery?: unknown; pending_id?: string; action?: unknown }; + let message: { type: string; pane_id?: string; text?: string; payload?: string; keys?: string[]; id?: number; mode?: string; typed?: unknown; delivery?: unknown; pending_id?: string; action?: unknown; enabled?: boolean }; try { message = JSON.parse(raw); } catch { return; } if ((message.type === "submit" || message.type === "pending-action") && Number.isSafeInteger(message.id)) { const signature = JSON.stringify({ type: message.type, pane_id: message.pane_id, text: message.text, payload: message.payload, typed: message.typed, delivery: message.delivery, pending_id: message.pending_id, action: message.action }); @@ -773,6 +799,20 @@ class DemoSocket extends EventTarget { if (message.pane_id) { this.attached.delete(message.pane_id); this.holdPending(message.pane_id, "not_attached", "This pane was detached. Pending messages will not be sent automatically."); + this.conversationInterests.delete(message.pane_id); + } + break; + case "conversation-watch": + if (typeof message.pane_id !== "string" || message.pane_id.trim() === "" || typeof message.enabled !== "boolean") { + this.push({ type: "error", code: "invalid_conversation_watch", message: "pane_id must be a nonempty string and enabled must be a boolean" }); + } else if (message.enabled) { + if (!this.attached.has(message.pane_id)) { + this.push({ type: "error", code: "not_attached", message: "conversation interest requires an attached pane", pane_id: message.pane_id }); + } else { + this.conversationInterests.add(message.pane_id); + } + } else { + this.conversationInterests.delete(message.pane_id); } break; case "input": if (message.pane_id && message.text !== undefined) this.typed(message.pane_id, message.text); break; @@ -919,11 +959,17 @@ class DemoSocket extends EventTarget { } private attach(paneId: string): void { + const pane = paneOf(paneId); + if (!pane) { + this.attached.delete(paneId); + this.conversationInterests.delete(paneId); + this.push({ type: "error", code: "pane_not_found", message: "no such pane", pane_id: paneId }); + return; + } this.attached.add(paneId); this.publishPending(paneId); this.push({ type: "input-ready", pane_id: paneId }); const key = keyOfPane.get(paneId); - const pane = paneOf(paneId); if (key === "shell") { let at = 0; let previous = 0; @@ -973,6 +1019,8 @@ class DemoSocket extends EventTarget { this.holdPending(); for (const timer of this.timers) clearTimeout(timer); sockets.delete(this); + this.attached.clear(); + this.conversationInterests.clear(); const event = new CloseEvent("close", { code, reason, wasClean: true }); this.onclose?.(event); this.dispatchEvent(event); @@ -1000,6 +1048,7 @@ setTimeout(() => { { kind: "text", text: "Added `payments_idempotent_replays_total`, incremented where a stored response is replayed, and exposed with the other counters on `/metrics`. 6 tests pass." }, ); turn.end_ts = now(); + conversationChanged(paneId); finishDemoTurn(paneId); }, 4500); diff --git a/src/AGENTS.md b/src/AGENTS.md index df9681a8df..877186c9b5 100644 --- a/src/AGENTS.md +++ b/src/AGENTS.md @@ -49,7 +49,7 @@ React 18 + xterm.js browser client. The render-state machine, xterm lifecycle an | Geometry/fit | components/PaneTerminal.tsx | fit on attach, ResizeObserver on the host, `focus`/`visibilitychange` refit — all SKIPPED while observing; only a tab in use (`inUse`: visible AND `document.hasFocus()`) resizes the pty from the observer, the refit, `role-ack`, a font load or the lens return; out of use, a pane attach (a reload, the next pane after one closed) goes with `keep_size`, and blur or hidden marks `keepSize` so a background reconnect adopts the grid: a window left behind another app must not take the pane from a phone or herdr's own TUI; with `settings.releasePaneAway` on (off by default), after `RELEASE_AFTER_MS` out of use (from a blur, hidden, or a pane attach made out of use) it also detaches (`released`), because herdr holds a pane at an attach's size until that attach ends and its TUI cannot take it back, and watches the pane instead (`socket.watch`, feature "watch": a read-only view at this grid that leaves the pane's size alone, written as it comes; a grid change starts it again; banner "View only…"); a server without the feature, or a `watch-end`, leaves it paused on its last screen. It attaches again on focus or a pointerdown (unwatch first); a released tab moving to another pane watches that one; it keeps the pane while a message it queued waits (`queueWaits`), and never releases a mirrored pane or an embedded page (the site's demo). `HerdrSocket.resize` sends nothing for a pane it is not attached to: the server would resize the holder's attach; single-finger touch drag becomes synthetic wheel events (listeners `passive: false` only there) | | Key bar | components/KeyBar.tsx + KeyBarSettings.tsx + lib/keyBar.ts + lib/keys.ts | `settings.keyBarItems` owns the complete ordered row; absent legacy records migrate from `keyBarExtras`, explicit empty lists stay empty. Ordinary keys inherit held modifiers; configured combinations are exact. Taps go through `term.input()` into the guarded onData path. PaneTerminal owns sticky Ctrl/Alt/Shift, clearing state on pane/lens change, disconnect or modifier removal; hidden while observing | | Chat composer | components/Composer.tsx + lib/compose.ts + lib/mentions.ts + lib/modelName.ts + PendingMessages.tsx | the model pill shows a name only for an id `modelLabel` matches whole, else the id as received in mono: never add a guessed name; Enter or Mod+Enter sends per `settings.enterSends`, IME Enter never sends; `/` commands cache 60s per pane, `@` files debounce 150ms; one Send/Stop control for every agent, no Queue button/menu; working Send uses `socket.submit` queue delivery only with `pending-input`, otherwise keeps the draft; pending rows offer an explicit Send now action on server IDs, retained uncertain text never retries; Stop is Escape via `onAbort`, also from the box's own Escape while a draft shows Send in its place | -| Chat lens | components/ChatView.tsx + Markdown.tsx (lib/markdown.ts) + lib/workBlocks.ts + lib/toolVerbs.ts + PromptCard.tsx | `splitTurn` folds tools/thinking/narration into one `.work-block` and honors explicit Codex commentary/final phases; a row is verb + object by exact tool id (`toolVerb`), the id kept in its title and opened detail; duration = `end_ts − ts`; conversation polls run serially, a send refresh preserves history/scroll, a pane change resets; ChatView polls `/api/pane/prompt` every 2s whatever the pane's status (a menu can wait while herdr reports idle: Codex's trust check, pi's dialogs, Claude Code's `/model` list) and owns the prompt; PromptCard answers through `/api/pane/prompt/answer` (409 → re-read) and is drawn through a portal into PaneTerminal's `.prompt-dock`, over the input card and outside the transcript | +| Chat lens | components/ChatView.tsx + lib/conversationRefresh.ts + Markdown.tsx (lib/markdown.ts) + lib/workBlocks.ts + lib/toolVerbs.ts + PromptCard.tsx | `splitTurn` folds tools/thinking/narration into one `.work-block` and honors explicit Codex commentary/final phases; a row is verb + object by exact tool id (`toolVerb`), the id kept in its title and opened detail; duration = `end_ts − ts`; only mounted visible Chat subscribes to `conversation-watch` on its attached machine-bound socket; `conversation-changed` uses a serial coordinator with push reads at least 2s apart, a visible-only 10s backstop (2s while `conversationPushes` is false: an older bridge or a released pane) and immediate initial/send/visibility/reconnect refresh; a visible scrollback stand-in is invalidated by the pane's `pty-data` through a ref listener (`onTerminalOutput`), never by React state per frame; newest and older pages share the coordinator; a send refresh preserves history/scroll, a pane change resets; ChatView polls `/api/pane/prompt` every 2s whatever the pane's status (a menu can wait while herdr reports idle: Codex's trust check, pi's dialogs, Claude Code's `/model` list) and owns the prompt; PromptCard answers through `/api/pane/prompt/answer` (409 → re-read) and is drawn through a portal into PaneTerminal's `.prompt-dock`, over the input card and outside the transcript | | Sidebar roster | components/Sidebar.tsx + AgentSidebar.tsx + TabStrip.tsx; lib/useWorktreeBranches.ts + worktreeBranches.ts | workspace rows render inside `MachineSidebar`, one roster per PC; parent names stay `workspace.label`, opened linked worktree rows show branches from existing `GET /api/worktree/list` matched by repo key + open workspace ID + normalized checkout path, never labels or directory names; `Detached HEAD` requires the listing's flag; custom workspace names are secondary, while the default slash-to-dash name is suppressed, with full names/path in tooltips; create/open responses seed branch metadata and expand the parent; structural changes and visibility return refresh inventory, with a serial 30 s visible+online backstop and stale-request guards; status rolls up each workspace's panes in herdr's order, blocked > done > working > idle (`rollupStatus`, lib/status.ts), a folded worktree parent rolls up its whole group, and the row selects its current/last-viewed/focused pane; only linked worktree groups fold (`herdr-web-ui:worktree-group-collapsed::`), preserving the selected child and deliberate folds across status updates; a separate Agents list joins `snapshot.agents` to panes across the combined machine roster, with a recognized OmO pane fallback and pane status authoritative; every row sits on the `--side-*` grid declared on `.sidebar-shell` (leading glyph cell, title, trailing status cell); a workspace row leads with the mark of the agent in the pane it opens (`paneMark`, lib/sidebarAgents.ts), never another pane's, else the terminal or branch glyph; a top-level row starts with a folder, which is the fold (`.workspace-toggle`) when linked worktrees sit under it; the roster has no label or fold of its own, the caret beside the PC's `+` folds it; an agent row's line-two rules live in `agentContext` (lib/sidebarAgents.ts); the row menu button takes no width at rest but stays focusable (RowMenu and ConfirmDialog return focus to it); both lists use existing SSE, never another status subscription or server contract; both read the status they draw, and the Agents list its Activity order, through `useSidebarActivity` (lib/sidebarActivity.tsx, provided by MachineSidebar; logic in lib/sidebarOrder.ts); workspace rows have no separate reorder grip; dragging the row or Alt+Up/Down on its selector uses `workspace.move`, optimistic and preserving herdr's repository/worktree hierarchy; workspace menus act on workspace scope except Rename pane for the current representative; tab/pane navigation stays in TabStrip and the palette; no search or filter here on purpose, the command palette is the search; Add PC lives in Settings → Remote PCs and the palette (`actions.openAddPc`) | | New workspace / New tab | components/NewSessionDialog.tsx | `createWorkspace`, or `createTab` (POST /api/tab/create) when given a `tab` prop; agent start may take 60s and the dialog stays open while pending; `agent_started:false` + error offers to open the already-created workspace | | Command palette | components/CommandPalette.tsx + lib/paletteSearch.ts | Mod+Shift+K; panes (recent first, fuzzy ranked) + actions with `.kbd` hints; Escape closes | diff --git a/src/components/ChatView.tsx b/src/components/ChatView.tsx index 81b14bd60f..f8f48c514d 100644 --- a/src/components/ChatView.tsx +++ b/src/components/ChatView.tsx @@ -24,6 +24,7 @@ import { isTodoTool, parseTodoAnswer, todoCallSummary, type TodoItem, type TodoS import { formatGoalTime, turnGoal, type GoalState, type GoalStatus } from "../lib/goals.ts"; import { useSettings } from "../lib/settings.ts"; import { statusEdgeRead } from "../lib/status.ts"; +import { ConversationRefresh } from "../lib/conversationRefresh.ts"; import { usePageVisible } from "../lib/visibility.ts"; import { dismissKeyboardOn } from "../lib/keyboard.ts"; import { useFacesArrived } from "../lib/fontFaces.ts"; @@ -46,13 +47,21 @@ import { chatIsBlank, type ChatRead } from "../lib/greeting.ts"; import { currentLocale, useT } from "../lib/i18n.ts"; const TRANSCRIPT_LINES = 400; -const POLL_MS = 2000; +const PROMPT_POLL_MS = 2000; /** Scrolling this close to the top asks for the page before it. */ const LOAD_OLDER_PX = 400; export interface ChatViewProps { paneId: string; refreshKey: number; + /** bumped by a conversation push; coalesced at the former 2s polling cadence */ + invalidationKey?: number; + /** only a mounted, visible chat asks the bridge to watch its native conversation */ + onConversationWatch?: (paneId: string, enabled: boolean) => void; + /** the bridge pushes this pane's transcript changes; without, the chat reads every 2s as before */ + conversationPushes?: boolean; + /** while the chat shows herdr's scrollback, the pane's terminal output invalidates it */ + onTerminalOutput?: (listener: (() => void) | null) => void; /** bumped when a composer message goes out, before the transcript holds it */ sentKey?: number; connected: boolean; @@ -526,7 +535,7 @@ function FallbackTurn({ paneId, message }: { paneId: string; message: Transcript } // the app re-renders on every pane-status and poll; an unchanged transcript sits those out -export const ChatView = memo(function ChatView({ paneId, refreshKey, sentKey = 0, connected, ended, agent, agentStatus, onMetadata, onRead, greeted = false, onPrompt, onSuggestion, promptRefreshKey = 0, pendingAnswer = null, onPendingAnswerDone, promptDock = null, onPromptAnswered }: ChatViewProps) { +export const ChatView = memo(function ChatView({ paneId, refreshKey, invalidationKey = 0, onConversationWatch, conversationPushes = false, onTerminalOutput, sentKey = 0, connected, ended, agent, agentStatus, onMetadata, onRead, greeted = false, onPrompt, onSuggestion, promptRefreshKey = 0, pendingAnswer = null, onPendingAnswerDone, promptDock = null, onPromptAnswered }: ChatViewProps) { const t = useT(); const { fetchPaneConversation, fetchPanePromptState, fetchPaneTranscript } = useMachineApi(); const { settings } = useSettings(); @@ -534,6 +543,9 @@ export const ChatView = memo(function ChatView({ paneId, refreshKey, sentKey = 0 const visible = usePageVisible(); /** the answer last laid out: an unchanged poll (a 304) hands back this very object */ const lastAnswer = useRef(null); + const [conversationRefresh] = useState(() => new ConversationRefresh()); + const onMetadataRef = useRef(onMetadata); + onMetadataRef.current = onMetadata; const [state, setState] = useState(EMPTY_STATE); const [error, setError] = useState(null); const [errorStatus, setErrorStatus] = useState(null); @@ -587,6 +599,8 @@ export const ChatView = memo(function ChatView({ paneId, refreshKey, sentKey = 0 const heldPage = useRef([]); const seenSent = useRef(sentKey); const seenStatus = useRef<{ pane: string; status: AgentStatus | undefined }>({ pane: paneId, status: agentStatus }); + const seenConnected = useRef(connected); + const seenInvalidation = useRef({ pane: paneId, key: invalidationKey }); const dropOlder = (): void => { olderGeneration.current += 1; loadingOlder.current = false; @@ -611,9 +625,8 @@ export const ChatView = memo(function ChatView({ paneId, refreshKey, sentKey = 0 setSentOver(last?.role === "assistant" && agentStatus !== "working" && agentStatus !== "blocked" ? { turn: last, page } : null); }, [sentKey, agentStatus]); - // A turn starts or ends when the status enters or leaves `working`: read now, not at the next - // poll. Bumping pollKey re-runs the read effect, whose cleanup cancels the loop that was - // running (its answer is dropped, its timer cleared) before the single new loop starts. + // Status edges still read immediately, including on a bridge without conversation pushes. + // They share the same serialized reader as pushes, sends and the periodic backstop. useEffect(() => { const seen = seenStatus.current; seenStatus.current = { pane: paneId, status: agentStatus }; @@ -623,13 +636,13 @@ export const ChatView = memo(function ChatView({ paneId, refreshKey, sentKey = 0 useEffect(() => { if (!visible) return; let cancelled = false; - let timer: number | undefined; /** The turns between the held start and where the newest page now starts, a page at a time; null when they cannot be joined. */ const turnsBetween = async (held: string, start: string): Promise => { const pages: ConversationTurn[][] = []; let before = start; try { for (let step = 0; step < 32 && before !== held; step++) { + if (cancelled) return null; const page = await fetchPaneConversation(paneId, { before, since: held }); if (typeof page.cursor !== "string") return null; pages.unshift(page.turns); @@ -677,13 +690,13 @@ export const ChatView = memo(function ChatView({ paneId, refreshKey, sentKey = 0 } if (moved.length > 0) setOlder((turns) => [...turns, ...moved]); if (heldFrom.current === null) setOlderCursor(conversation.source === "scrollback" ? undefined : conversation.cursor); - onMetadata?.(paneId, conversation.source === "scrollback" ? null : conversation.metadata ?? null); + onMetadataRef.current?.(paneId, conversation.source === "scrollback" ? null : conversation.metadata ?? null); setAbandoned(conversation.abandoned ?? null); let next: ChatState; if (conversation.source !== "scrollback") next = { source: "conversation", turns: conversation.turns, messages: [], truncated: false }; else { const result = await fetchPaneTranscript(paneId, TRANSCRIPT_LINES); - if (cancelled) return; + if (cancelled || generation !== olderGeneration.current) return; next = { source: "scrollback", turns: [], messages: toTranscriptMessages(result.text).filter((message) => message.role !== "status"), truncated: result.truncated === true }; } const nextSignature = JSON.stringify(next); @@ -701,13 +714,49 @@ export const ChatView = memo(function ChatView({ paneId, refreshKey, sentKey = 0 setLoaded(true); setError(cause instanceof Error ? cause.message : String(cause)); setErrorStatus(cause instanceof ApiError ? cause.status : null); - } finally { - if (!cancelled) timer = window.setTimeout(() => void read(), POLL_MS); } }; - void read(); - return () => { cancelled = true; window.clearTimeout(timer); }; - }, [paneId, refreshKey, onMetadata, pollKey, visible]); + conversationRefresh.setRead(read); + onConversationWatch?.(paneId, true); + return () => { + cancelled = true; + // Older pages queued behind this reader must not start for a hidden/replaced screen. + olderGeneration.current += 1; + loadingOlder.current = false; setOlderState("idle"); + onConversationWatch?.(paneId, false); + conversationRefresh.stop(); + }; + }, [paneId, fetchPaneConversation, fetchPaneTranscript, conversationRefresh, visible, onConversationWatch]); + + // Sends, status/history repairs and visibility return read immediately without cancelling + // the answer in flight. The reader is installed before this effect requests its first read. + useEffect(() => { + if (visible) conversationRefresh.refresh(); + }, [paneId, refreshKey, pollKey, visible, fetchPaneConversation, fetchPaneTranscript, conversationRefresh, onConversationWatch]); + + // Mount/visibility return already read immediately; only a changed push key invalidates. + useEffect(() => { + const seen = seenInvalidation.current; + seenInvalidation.current = { pane: paneId, key: invalidationKey }; + if (seen.pane === paneId && seen.key !== invalidationKey && visible) conversationRefresh.invalidate(); + }, [paneId, invalidationKey, visible, conversationRefresh]); + + useEffect(() => { conversationRefresh.setPushes(conversationPushes); }, [conversationPushes, conversationRefresh]); + + // A scrollback stand-in has no native file for the server to watch. Its source is the + // pane's own screen, so terminal output is the invalidation, under the same 2s bound. + useEffect(() => { + if (state.source !== "scrollback" || !visible || onTerminalOutput === undefined) return; + onTerminalOutput(() => conversationRefresh.invalidate()); + return () => onTerminalOutput(null); + }, [state.source, visible, onTerminalOutput, conversationRefresh]); + + // Catch writes made offline as soon as the existing socket is connected again. + useEffect(() => { + const before = seenConnected.current; + seenConnected.current = connected; + if (connected && !before && visible) conversationRefresh.refresh(); + }, [connected, visible, conversationRefresh]); const loadOlder = useCallback(async (): Promise => { const node = scroller.current; @@ -716,34 +765,37 @@ export const ChatView = memo(function ChatView({ paneId, refreshKey, sentKey = 0 const generation = olderGeneration.current; loadingOlder.current = true; setOlderState("loading"); - try { - const page = await fetchPaneConversation(paneId, { before }); - if (shownPane.current !== paneId || olderGeneration.current !== generation) return; - if (page.history_id !== history.current) { - dropOlder(); setState(EMPTY_STATE); setLoaded(false); signature.current = ""; lastAnswer.current = null; - setPollKey((key) => key + 1); return; - } - // a bridge without pages answers with its newest turns: nothing older to add - if (page.source === "scrollback" || page.cursor === undefined) { setOlderCursor(undefined); setOlderState("idle"); return; } - const first = heldFrom.current === null; - heldFrom.current ??= before; - prepended.current = { top: node.scrollTop, height: node.scrollHeight }; - setOlder((turns) => [...page.turns, ...turns]); - setOlderCursor(page.cursor); - setOlderState("idle"); - // the newest page may have slid since it was read: poll it from the held start now - if (first) setPollKey((key) => key + 1); - } catch (cause) { - if (shownPane.current !== paneId || olderGeneration.current !== generation) return; - if (cause instanceof ApiError && cause.status === 409) { - dropOlder(); setState(EMPTY_STATE); setLoaded(false); signature.current = ""; lastAnswer.current = null; - setPollKey((key) => key + 1); + await conversationRefresh.page(async () => { + try { + if (shownPane.current !== paneId || olderGeneration.current !== generation) return; + const page = await fetchPaneConversation(paneId, { before }); + if (shownPane.current !== paneId || olderGeneration.current !== generation) return; + if (page.history_id !== history.current) { + dropOlder(); setState(EMPTY_STATE); setLoaded(false); signature.current = ""; lastAnswer.current = null; + setPollKey((key) => key + 1); return; + } + // a bridge without pages answers with its newest turns: nothing older to add + if (page.source === "scrollback" || page.cursor === undefined) { setOlderCursor(undefined); setOlderState("idle"); return; } + const first = heldFrom.current === null; + heldFrom.current ??= before; + prepended.current = { top: node.scrollTop, height: node.scrollHeight }; + setOlder((turns) => [...page.turns, ...turns]); + setOlderCursor(page.cursor); + setOlderState("idle"); + // the newest page may have slid since it was read: poll it from the held start now + if (first) setPollKey((key) => key + 1); + } catch (cause) { + if (shownPane.current !== paneId || olderGeneration.current !== generation) return; + if (cause instanceof ApiError && cause.status === 409) { + dropOlder(); setState(EMPTY_STATE); setLoaded(false); signature.current = ""; lastAnswer.current = null; + setPollKey((key) => key + 1); + } + else setOlderState("failed"); + } finally { + if (olderGeneration.current === generation) loadingOlder.current = false; } - else setOlderState("failed"); - } finally { - if (olderGeneration.current === generation) loadingOlder.current = false; - } - }, [fetchPaneConversation, olderCursor, paneId]); + }); + }, [fetchPaneConversation, olderCursor, paneId, conversationRefresh]); // Older turns went in above the reader: keep the same turns under their eyes. useLayoutEffect(() => { @@ -754,8 +806,8 @@ export const ChatView = memo(function ChatView({ paneId, refreshKey, sentKey = 0 node.scrollTop = anchor.top + (node.scrollHeight - anchor.height); }, [older]); - // A new Codex TUI can show its directory-trust menu while herdr still reports - // idle. The visible prompt, not the status badge, decides whether to offer answers. + // Prompts belong to the visible screen, not the native transcript: a Codex directory-trust + // menu can appear while idle without any file mutation. Keep their independent 2s poll. const pollPrompt = connected && !ended && agent !== null; useEffect(() => { if (!pollPrompt) { setPrompt(null); onSuggestionRef.current?.(paneId, null); return; } @@ -773,7 +825,7 @@ export const ChatView = memo(function ChatView({ paneId, refreshKey, sentKey = 0 } } catch { if (!cancelled) { setPrompt(null); onSuggestionRef.current?.(paneId, null); } } - finally { if (!cancelled) timer = window.setTimeout(() => void readPrompt(), POLL_MS); } + finally { if (!cancelled) timer = window.setTimeout(() => void readPrompt(), PROMPT_POLL_MS); } }; void readPrompt(); return () => { cancelled = true; window.clearTimeout(timer); }; diff --git a/src/components/PaneTerminal.tsx b/src/components/PaneTerminal.tsx index 1071c302a6..d17bef539e 100644 --- a/src/components/PaneTerminal.tsx +++ b/src/components/PaneTerminal.tsx @@ -280,8 +280,19 @@ export function PaneTerminal({ // transient OSC 52 feedback ("copied") — a pill in the banner column const [clipboardNote, setClipboardNote] = useState(null); const clipboardTimerRef = useRef(null); - // the composer's send bumps this so the chat lens refetches without waiting a poll beat + // Explicit sends refresh immediately; continuous transcript invalidations are rate-limited. const [chatRefresh, setChatRefresh] = useState(0); + const [chatInvalidation, setChatInvalidation] = useState(0); + // an older bridge pushes no transcript changes: the chat reads at the former 2s cadence + const [conversationPushes, setConversationPushes] = useState(false); + const onConversationWatch = useCallback((pane: string, enabled: boolean) => { + socketRef.current?.watchConversation(pane, enabled); + }, []); + /** A scrollback stand-in's listener; a ref, so terminal output never re-renders the pane. */ + const chatOutputRef = useRef<(() => void) | null>(null); + const onTerminalOutput = useCallback((listener: (() => void) | null) => { + chatOutputRef.current = listener; + }, []); // bumped as a composer message goes out: the chat must not title the turn before it as running const [chatSent, setChatSent] = useState(0); const [chatMetadata, setChatMetadata] = useState<{ pane: string; value: ConversationMetadata | null } | null>(null); @@ -339,7 +350,7 @@ export function PaneTerminal({ if (agentStatus === "working") setPendingAnswer(null); }, [agentStatus]); const onChatMetadata = useCallback((pane: string, value: ConversationMetadata | null) => { - // the same settings keep the same object: every 2 s poll would otherwise re-render the composer + // the same settings keep the same object: every conversation read would otherwise re-render the composer setChatMetadata((previous) => previous?.pane === pane && previous.value?.model === value?.model && previous.value?.reasoning_effort === value?.reasoning_effort && previous.value?.context?.used === value?.context?.used @@ -867,9 +878,16 @@ export function PaneTerminal({ } } } + // Ignore late invalidations for a detached pane or a chat no longer on screen. + if (message.type === "conversation-changed" && message.pane_id === paneRef.current + && chatViewRef.current && document.visibilityState !== "hidden") { + setChatInvalidation((current) => current + 1); + } if (paneRef.current) setInputReady(socket.canInput(paneRef.current)); if (message.type === "pty-data") { if (message.pane_id !== paneRef.current || releasedRef.current) return; + // A chat showing herdr's scrollback has no native file to watch: output invalidates it. + if (chatViewRef.current) chatOutputRef.current?.(); // raw pty bytes: append, never repaint, so xterm keeps the screen and selection const acknowledge = socket.outputAcknowledgement(message); const owner = message.pane_id; @@ -978,6 +996,7 @@ export function PaneTerminal({ term.writeln(`\r\n\u001b[31m[herdr-web-ui] ${message.code}: ${message.message}\u001b[0m`); } setConnected(socket.connected); + setConversationPushes(socket.conversationPushes()); }); const offDisconnect = socket.onDisconnect(() => { clearModifiers(); @@ -988,6 +1007,7 @@ export function PaneTerminal({ setOutputReady(false); setInputReady(false); setConnected(false); + setConversationPushes(false); // the reconnect attaches afresh: it says attach_held again if the other bridge still has // the pane, and a pane it gets straight away sends no attach-resumed to clear this setHeld(false); @@ -996,7 +1016,7 @@ export function PaneTerminal({ }); socket.connect(); - const poll = window.setInterval(() => setConnected(socket.connected), 1000); + const poll = window.setInterval(() => { setConnected(socket.connected); setConversationPushes(socket.conversationPushes()); }, 1000); // onKey runs after xterm drains a pending IME commit, immediately before the // key's onData. Remap only that CR, preserving composition text and its order. @@ -1978,6 +1998,10 @@ export function PaneTerminal({ key={paneId} paneId={paneId} refreshKey={chatRefresh} + invalidationKey={chatInvalidation} + onConversationWatch={onConversationWatch} + conversationPushes={conversationPushes && !released} + onTerminalOutput={onTerminalOutput} sentKey={chatSent} connected={connected} ended={ended} diff --git a/src/lib/api.ts b/src/lib/api.ts index 305ccb7c0d..1ef94264e3 100644 --- a/src/lib/api.ts +++ b/src/lib/api.ts @@ -137,7 +137,7 @@ export async function fetchSession(machineId = "local"): Promise(); /** - * GET /api/pane/read as the chat view polls it: herdr's own scrollback (up to + * GET /api/pane/read as the chat view reads it: herdr's own scrollback (up to * `lines`), ANSI-stripped text. herdr owns scrollback — the attach stream cannot * serve history, so the transcript reads it back instead. `recent_unwrapped` * rejoins the rows the terminal soft-wrapped, so a line reflows to the chat's @@ -162,7 +162,9 @@ export async function fetchPaneTranscript(paneId: string, lines: number, machine export type ConversationPageQuery = { before?: string; since?: string; from?: string }; /** - * The last answer per polled conversation URL and its ETag. The chat polls every 2s + * The last answer per newest-page conversation URL and its ETag. The chat reads it again on + * every transcript push (at most every 2s), every send and status edge, and a 10s backstop + * (2s without pushes), * and a newest page can be megabytes: an unchanged one comes back as a bodyless 304, * and the chat gets the very same object back, which tells it nothing changed. An * older page (`before`) is asked for once, so it keeps no ETag and takes no slot. diff --git a/src/lib/conversationFetch.test.ts b/src/lib/conversationFetch.test.ts index 1a131d6263..bfd4d6d976 100644 --- a/src/lib/conversationFetch.test.ts +++ b/src/lib/conversationFetch.test.ts @@ -1,9 +1,19 @@ -import { afterEach, describe, expect, it } from "bun:test"; +import { afterEach, describe, expect, it, spyOn } from "bun:test"; import { fetchPaneConversation, conversationAnswerBytes } from "./api.ts"; +import { ConversationRefresh } from "./conversationRefresh.ts"; +import type { ConversationResponse } from "../../shared/protocol.ts"; const realFetch = globalThis.fetch; -afterEach(() => { globalThis.fetch = realFetch; }); +const globals = globalThis as unknown as { window?: unknown }; +const realWindow = globals.window; +let restoreClock: (() => void) | undefined; +afterEach(() => { + globalThis.fetch = realFetch; + globals.window = realWindow; + restoreClock?.(); + restoreClock = undefined; +}); describe("conversation polling", () => { it("asks with the last ETag and reuses the very same answer on a 304", async () => { @@ -73,3 +83,413 @@ it("counts a compressed answer by its parsed body, not the bytes on the wire", ( expect(conversationAnswerBytes(body, "300", "identity")).toBe(300); expect(conversationAnswerBytes(body, "300", "gzip")).toBe(JSON.stringify(body).length); }); + +/** The browser timer the refresher uses, advanced without sleeping or a DOM. */ +function refreshClock(): { advance(ms: number): void; pending(): number } { + let now = 0; + let next = 1; + const timers = new Map void }>(); + const clock = spyOn(performance, "now").mockImplementation(() => now); + restoreClock = () => { clock.mockRestore(); }; + globals.window = { + setTimeout(callback: () => void, delay: number): number { + const id = next++; + timers.set(id, { at: now + delay, callback }); + return id; + }, + clearTimeout(id: number | undefined): void { if (id !== undefined) timers.delete(id); }, + }; + return { + advance(ms: number): void { + now += ms; + for (const [id, timer] of [...timers]) if (timer.at <= now) { + timers.delete(id); + timer.callback(); + } + }, + pending(): number { return timers.size; }, + }; +} + +async function settleRefresh(): Promise { + for (let step = 0; step < 20; step++) await Promise.resolve(); +} + +/** A chat on a bridge that pushes its pane's transcript changes. */ +function pushedRefresh(): ConversationRefresh { + const refresh = new ConversationRefresh(); + refresh.setPushes(true); + return refresh; +} + +describe("conversation invalidation refresh", () => { + it("coalesces pushes at exactly 2000ms and keeps the 10s backstop, ETag and 304 identity", async () => { + const clock = refreshClock(); + const refresh = pushedRefresh(); + const requests: { resolve: (response: Response) => void }[] = []; + const etags: (string | null)[] = []; + const answers: unknown[] = []; + let active = 0; + let maxActive = 0; + globalThis.fetch = (async (_url: string, init?: RequestInit) => { + etags.push(new Headers(init?.headers).get("if-none-match")); + const { promise, resolve } = Promise.withResolvers(); + requests.push({ resolve }); + return promise; + }) as typeof fetch; + refresh.setRead(async () => { + active += 1; maxActive = Math.max(maxActive, active); + try { answers.push(await fetchPaneConversation("refresh-burst:p1")); } + finally { active -= 1; } + }); + refresh.refresh(); + clock.advance(100); + for (let push = 0; push < 8; push++) refresh.invalidate(); + requests[0]!.resolve(new Response(JSON.stringify({ source: "claude-transcript", turns: [], cursor: null }), { headers: { etag: "\"push-v1\"" } })); + await settleRefresh(); + expect(requests).toHaveLength(1); + clock.advance(1899); + refresh.invalidate(); + expect(requests).toHaveLength(1); + clock.advance(1); + expect(requests).toHaveLength(2); + expect(etags).toEqual([null, "\"push-v1\""]); + requests[1]!.resolve(new Response(null, { status: 304, headers: { etag: "\"push-v1\"" } })); + await settleRefresh(); + expect(answers[1]).toBe(answers[0]); + expect(maxActive).toBe(1); + clock.advance(9999); + expect(requests).toHaveLength(2); + clock.advance(1); + expect(requests).toHaveLength(3); + refresh.stop(); + requests[2]!.resolve(new Response(null, { status: 304, headers: { etag: "\"push-v1\"" } })); + await settleRefresh(); + expect(clock.pending()).toBe(0); + }); + + it("bounds sustained pushes without extending the deadline and eventually shows the latest REST output", async () => { + const clock = refreshClock(); + const refresh = pushedRefresh(); + const starts: number[] = []; + const answers: ConversationResponse[] = []; + let output = "initial"; + let active = 0; + let maxActive = 0; + globalThis.fetch = (async (_url: string) => new Response(JSON.stringify({ + source: "claude-transcript", + turns: [{ role: "assistant", ts: null, parts: [{ kind: "text", text: output }] }], + cursor: null, + }), { headers: { etag: `"${output}"` } })) as typeof fetch; + refresh.setRead(async () => { + starts.push(performance.now()); + active += 1; maxActive = Math.max(maxActive, active); + try { answers.push(await fetchPaneConversation("refresh-sustained:p1")); } + finally { active -= 1; } + }); + refresh.refresh(); + await settleRefresh(); + for (let push = 1; push <= 53; push++) { + clock.advance(100); + output = `output-${push}`; + refresh.invalidate(); + await settleRefresh(); + } + expect(starts).toEqual([0, 2000, 4000]); + clock.advance(699); + expect(starts).toHaveLength(3); + clock.advance(1); + await settleRefresh(); + expect(starts).toEqual([0, 2000, 4000, 6000]); + expect(answers.at(-1)?.turns[0]?.parts[0]).toEqual({ kind: "text", text: "output-53" }); + expect(maxActive).toBe(1); + // A push after idle is not delayed until the periodic backstop. + clock.advance(2500); + output = "after-idle"; + refresh.invalidate(); + await settleRefresh(); + expect(starts).toEqual([0, 2000, 4000, 6000, 8500]); + expect(answers.at(-1)?.turns[0]?.parts[0]).toEqual({ kind: "text", text: "after-idle" }); + refresh.stop(); + }); + + it("lets an explicit refresh bypass and replace a delayed push", async () => { + const clock = refreshClock(); + const refresh = pushedRefresh(); + const starts: number[] = []; + refresh.setRead(async () => { starts.push(performance.now()); }); + refresh.refresh(); + await settleRefresh(); + clock.advance(100); + refresh.invalidate(); + clock.advance(400); + refresh.refresh(); + await settleRefresh(); + expect(starts).toEqual([0, 500]); + clock.advance(1500); + expect(starts).toEqual([0, 500]); + // The explicit read also establishes the next push deadline. + refresh.invalidate(); + clock.advance(499); + expect(starts).toEqual([0, 500]); + clock.advance(1); + await settleRefresh(); + expect(starts).toEqual([0, 500, 2500]); + refresh.stop(); + }); + + it("runs an immediate refresh after the in-flight read, without waiting for a push deadline", async () => { + const clock = refreshClock(); + const refresh = pushedRefresh(); + const first = Promise.withResolvers(); + const starts: number[] = []; + refresh.setRead(async () => { + starts.push(performance.now()); + if (starts.length === 1) await first.promise; + }); + refresh.refresh(); + clock.advance(100); + refresh.invalidate(); + clock.advance(100); + refresh.refresh(); + expect(starts).toEqual([0]); + first.resolve(); + await settleRefresh(); + expect(starts).toEqual([0, 200]); + clock.advance(1800); + expect(starts).toEqual([0, 200]); + refresh.stop(); + }); + + it("keeps long in-flight reads serial and trails at the later of completion and the fixed deadline", async () => { + const clock = refreshClock(); + const refresh = pushedRefresh(); + const starts: number[] = []; + const releases: (() => void)[] = []; + let output = "initial"; + let shown = ""; + let active = 0; + let maxActive = 0; + refresh.setRead(async () => { + const captured = output; + const { promise, resolve } = Promise.withResolvers(); + releases.push(resolve); + starts.push(performance.now()); + active += 1; maxActive = Math.max(maxActive, active); + try { await promise; shown = captured; } + finally { active -= 1; } + }); + refresh.invalidate(); + clock.advance(100); + output = "during-first"; + refresh.invalidate(); + clock.advance(2400); + output = "latest-first"; + refresh.invalidate(); + expect(starts).toEqual([0]); + releases[0]!(); + await settleRefresh(); + expect(starts).toEqual([0, 2500]); + expect(shown).toBe("initial"); + clock.advance(100); + output = "during-second"; + refresh.invalidate(); + clock.advance(1700); + output = "latest-second"; + refresh.invalidate(); + releases[1]!(); + await settleRefresh(); + expect(shown).toBe("latest-first"); + expect(starts).toEqual([0, 2500]); + clock.advance(199); + expect(starts).toEqual([0, 2500]); + clock.advance(1); + expect(starts).toEqual([0, 2500, 4500]); + releases[2]!(); + await settleRefresh(); + expect(shown).toBe("latest-second"); + expect(maxActive).toBe(1); + refresh.stop(); + }); + + it("starts an older page after the newest read and its gap fill, and never holds the next newest read behind it", async () => { + const clock = refreshClock(); + const refresh = pushedRefresh(); + const first = Promise.withResolvers(); + const gap = Promise.withResolvers(); + const page = Promise.withResolvers(); + const reads: string[] = []; + let active = 0; + let maxActive = 0; + let newest = 0; + refresh.setRead(async () => { + active += 1; maxActive = Math.max(maxActive, active); + reads.push("latest"); + newest += 1; + try { + if (newest === 1) { + await first.promise; + reads.push("gap"); + await gap.promise; + } + } finally { active -= 1; } + }); + refresh.refresh(); + const older = refresh.page(async () => { + reads.push("older"); + await page.promise; + return "page"; + }); + clock.advance(100); + for (let push = 0; push < 4; push++) refresh.invalidate(); + first.resolve(); + await settleRefresh(); + expect(reads).toEqual(["latest", "gap"]); + gap.resolve(); + await settleRefresh(); + expect(reads).toEqual(["latest", "gap", "older"]); + // the older page is still loading: the pushed newest read keeps its own 2 s deadline + clock.advance(1900); + await settleRefresh(); + expect(reads).toEqual(["latest", "gap", "older", "latest"]); + // and an explicit refresh (a clear, a send) does not wait for it either + refresh.refresh(); + await settleRefresh(); + expect(reads).toEqual(["latest", "gap", "older", "latest", "latest"]); + page.resolve(); + expect(await older).toBe("page"); + expect(maxActive).toBe(1); + refresh.stop(); + }); + + it("resumes visibility with the current reader only after the cancelled request finishes", async () => { + const clock = refreshClock(); + const refresh = pushedRefresh(); + const old = Promise.withResolvers(); + const current = Promise.withResolvers(); + const reads: string[] = []; + refresh.setRead(async () => { reads.push("old"); await old.promise; }); + refresh.refresh(); + refresh.invalidate(); + refresh.stop(); + refresh.setRead(async () => { reads.push("current"); await current.promise; }); + refresh.refresh(); + refresh.refresh(); + expect(reads).toEqual(["old"]); + old.resolve(); + await settleRefresh(); + expect(reads).toEqual(["old", "current"]); + current.resolve(); + await settleRefresh(); + clock.advance(2000); + expect(reads).toEqual(["old", "current"]); + refresh.stop(); + clock.advance(30_000); + expect(reads).toEqual(["old", "current"]); + expect(clock.pending()).toBe(0); + }); + + it("drops an old reader's in-flight trailing work on replacement without starting the new reader", async () => { + const clock = refreshClock(); + const refresh = pushedRefresh(); + const old = Promise.withResolvers(); + const reads: string[] = []; + refresh.setRead(async () => { reads.push("old"); await old.promise; }); + refresh.refresh(); + refresh.invalidate(); + refresh.setRead(async () => { reads.push("current"); }); + old.resolve(); + await settleRefresh(); + clock.advance(1999); + expect(reads).toEqual(["old"]); + refresh.refresh(); + await settleRefresh(); + expect(reads).toEqual(["old", "current"]); + refresh.stop(); + }); + + it("drops delayed push work on reader replacement even without stop", async () => { + const clock = refreshClock(); + const refresh = pushedRefresh(); + const reads: string[] = []; + refresh.setRead(async () => { reads.push("old"); }); + refresh.refresh(); + await settleRefresh(); + clock.advance(100); + refresh.invalidate(); + refresh.setRead(async () => { reads.push("current"); }); + clock.advance(30_000); + await settleRefresh(); + expect(reads).toEqual(["old"]); + expect(clock.pending()).toBe(0); + refresh.invalidate(); + await settleRefresh(); + expect(reads).toEqual(["old", "current"]); + refresh.stop(); + }); + + it("cancels the delayed trailing read and its timer when the chat becomes hidden", async () => { + const clock = refreshClock(); + const refresh = pushedRefresh(); + let reads = 0; + refresh.setRead(async () => { reads += 1; }); + refresh.refresh(); + await settleRefresh(); + clock.advance(100); + refresh.invalidate(); + refresh.stop(); + expect(clock.pending()).toBe(0); + clock.advance(30_000); + await settleRefresh(); + expect(reads).toBe(1); + expect(clock.pending()).toBe(0); + }); + + it("drops pending pushes and timers while hidden, even when a read is still in flight", async () => { + const clock = refreshClock(); + const refresh = pushedRefresh(); + const result = Promise.withResolvers(); + let reads = 0; + refresh.setRead(async () => { reads += 1; await result.promise; }); + refresh.refresh(); + refresh.invalidate(); + refresh.stop(); + refresh.invalidate(); + refresh.refresh(); + expect(clock.pending()).toBe(0); + result.resolve(); + await settleRefresh(); + clock.advance(30_000); + expect(reads).toBe(1); + expect(clock.pending()).toBe(0); + }); + + it("reads every 2s without pushes and moves a waiting backstop when pushes come and go", async () => { + const clock = refreshClock(); + const refresh = new ConversationRefresh(); + let reads = 0; + refresh.setRead(async () => { reads += 1; }); + refresh.refresh(); + await settleRefresh(); + clock.advance(1999); + expect(reads).toBe(1); + clock.advance(1); + await settleRefresh(); + expect(reads).toBe(2); + // the bridge starts pushing: the waiting poll becomes the 10s backstop from now + clock.advance(1000); + refresh.setPushes(true); + clock.advance(9999); + expect(reads).toBe(2); + clock.advance(1); + await settleRefresh(); + expect(reads).toBe(3); + // the tab lets go of its pane: no pushes, so the 2s poll is back without waiting out 10s + refresh.setPushes(false); + clock.advance(2000); + await settleRefresh(); + expect(reads).toBe(4); + refresh.stop(); + expect(clock.pending()).toBe(0); + }); +}); diff --git a/src/lib/conversationRefresh.ts b/src/lib/conversationRefresh.ts new file mode 100644 index 0000000000..60500d92d1 --- /dev/null +++ b/src/lib/conversationRefresh.ts @@ -0,0 +1,117 @@ +/** With pushes, a lost one still gets a full conversation read. */ +const BACKSTOP_MS = 10_000; + +/** Pushes never start newest-page reads faster than the former 2s polling cadence, and a chat + * without them (an older bridge, a tab that let go of its pane) reads at that cadence. */ +const INVALIDATION_MS = 2000; + +/** One serial lane for newest pages and their gap fills; an older page starts in it, after the + * newest read before it, but never holds the next newest read back. */ +export class ConversationRefresh { + private read: (() => Promise) | null = null; + private pending = false; + private invalidated = false; + private running = false; + private timer: number | undefined; + private newestAt = -Infinity; + private pushes = false; + private readonly pages: (() => Promise)[] = []; + + /** Replace a cancelled reader without letting its still-pending REST request overlap this one. */ + setRead(read: () => Promise): void { + this.clearTimer(); + this.pending = false; + this.invalidated = false; + this.newestAt = -Infinity; + this.read = read; + } + + /** Whether the bridge pushes this pane's transcript changes; without, the backstop is the poll. */ + setPushes(available: boolean): void { + if (this.pushes === available) return; + this.pushes = available; + // a waiting backstop takes the new cadence; a read in flight schedules it when it ends + if (this.read !== null && !this.running && !this.invalidated && this.timer !== undefined) { + this.clearTimer(); + this.timer = window.setTimeout(() => this.refresh(), this.backstop()); + } + } + + refresh(): void { + if (this.read === null) return; + this.clearTimer(); + this.pending = true; + this.invalidated = false; + void this.drain(); + } + + /** The first push after idle reads immediately; further pushes share a fixed trailing deadline. */ + invalidate(): void { + if (this.read === null || this.invalidated) return; + this.invalidated = true; + this.clearTimer(); + void this.drain(); + } + + /** An explicitly requested older page starts once the newest read in flight (and its gap fill) + * has settled, so a history change found there drops it first. A slow older page never delays + * the newest page: a clear or a new turn still shows while it loads, and its caller discards it. */ + page(read: () => Promise): Promise { + const { promise, resolve, reject } = Promise.withResolvers(); + this.pages.push(async () => { + try { resolve(await read()); } + catch (cause) { reject(cause); } + }); + this.clearTimer(); + void this.drain(); + return promise; + } + + /** Hidden/unmounted readers leave no timer or pending refresh; an in-flight read finishes first. */ + stop(): void { + this.read = null; + this.pending = false; + this.invalidated = false; + this.clearTimer(); + } + + private backstop(): number { + return this.pushes ? BACKSTOP_MS : INVALIDATION_MS; + } + + private clearTimer(): void { + window.clearTimeout(this.timer); + this.timer = undefined; + } + + private async drain(): Promise { + if (this.running) return; + this.running = true; + try { + while (true) { + const page = this.pages.shift(); + if (page !== undefined) { + void page(); + continue; + } + if (this.read === null || (!this.pending && !this.invalidated)) break; + if (!this.pending && performance.now() < this.newestAt + INVALIDATION_MS) break; + this.pending = false; + this.invalidated = false; + // Monotonic elapsed time keeps wall-clock corrections out of the rate bound. + this.newestAt = performance.now(); + await this.read(); + } + } finally { + this.running = false; + if (this.read !== null) { + this.timer = this.invalidated + ? window.setTimeout(() => { + this.timer = undefined; + void this.drain(); + }, Math.max(0, this.newestAt + INVALIDATION_MS - performance.now())) + : window.setTimeout(() => this.refresh(), this.backstop()); + } + } + } +} diff --git a/src/lib/status.test.ts b/src/lib/status.test.ts index f2b057176b..e21712f68a 100644 --- a/src/lib/status.test.ts +++ b/src/lib/status.test.ts @@ -44,7 +44,7 @@ describe("statusEdgeRead", () => { expect(statusEdgeRead(undefined, "working")).toBe(true); }); - it("leaves an unchanged status and changes that neither start nor end a turn to the poll", () => { + it("leaves an unchanged status and changes that neither start nor end a turn to the regular reads", () => { expect(statusEdgeRead("working", "working")).toBe(false); expect(statusEdgeRead("idle", "idle")).toBe(false); expect(statusEdgeRead("idle", "done")).toBe(false); diff --git a/src/lib/status.ts b/src/lib/status.ts index 248afef2dd..fb116db3fd 100644 --- a/src/lib/status.ts +++ b/src/lib/status.ts @@ -47,9 +47,10 @@ export function rollupStatus(statuses: ReadonlyArray): } /** - * Whether a pushed status change should read the conversation now instead of at the next poll: - * a turn starts or ends when the pane enters or leaves `working`, and the chat's last block - * follows the status while the transcript it holds is up to POLL_MS old. + * Whether a pushed status change should read the conversation now instead of waiting for a + * transcript push or the backstop: a turn starts or ends when the pane enters or leaves + * `working`, and the chat's last block follows the status while the transcript it holds is + * older than that status (a chat without pushes reads only every 2 s). */ export function statusEdgeRead(previous: AgentStatus | undefined, next: AgentStatus | undefined): boolean { return previous !== next && (previous === "working" || next === "working"); diff --git a/src/lib/ws.test.ts b/src/lib/ws.test.ts index cafecc4556..878aae0219 100644 --- a/src/lib/ws.test.ts +++ b/src/lib/ws.test.ts @@ -467,3 +467,29 @@ it("waits for capabilities when output precedes snapshot, and supports old bridg client.close(); } }); + +it("asks again for a visible chat's conversation pushes when its pane is attached again", () => { + const client = new HerdrSocket("ws://test/ws"); client.connect(); + const socket = FakeSocket.last; socket.open(); + socket.receive(snapshot(["conversation-watch"])); + const watches = () => socket.sent.filter((frame) => frame.type === "conversation-watch"); + // the chat can ask before its pane's attach goes out; the attach carries the interest + client.watchConversation("w1:p1", true); + expect(watches()).toEqual([]); + client.attach("w1:p1", 80, 24, true); + expect(watches()).toEqual([{ type: "conversation-watch", pane_id: "w1:p1", enabled: true }]); + // a tab out of use lets go of its pane: the bridge drops the interest, the visible chat keeps it + client.detach("w1:p1"); + client.attach("w1:p1", 80, 24, true); + expect(watches()).toHaveLength(2); + // a chat that took its interest back is not watched by the next attach + client.watchConversation("w1:p1", false); + client.detach("w1:p1"); + client.attach("w1:p1", 80, 24, true); + expect(watches()).toEqual([ + { type: "conversation-watch", pane_id: "w1:p1", enabled: true }, + { type: "conversation-watch", pane_id: "w1:p1", enabled: true }, + { type: "conversation-watch", pane_id: "w1:p1", enabled: false }, + ]); + client.close(); +}); diff --git a/src/lib/ws.ts b/src/lib/ws.ts index abda6e5b18..41232e34fa 100644 --- a/src/lib/ws.ts +++ b/src/lib/ws.ts @@ -53,6 +53,10 @@ export class HerdrSocket { private readonly attached = new Map(); /** panes this tab views read-only while out of use (feature "watch"), at the grid it asked for; replayed after each snapshot */ private readonly watched = new Map(); + /** Panes whose mounted, visible chat asks for transcript invalidations. The server drops interest + * with the attachment, so each attach of a pane listed here (a reconnect's, or a tab back in use + * after letting go of it) asks again; only the chat taking its interest back removes it. */ + private readonly conversationWanted = new Set(); private retries = 0; private reconnectTimer: number | null = null; private disposed = false; @@ -120,6 +124,13 @@ export class HerdrSocket { this.features = new Set(message.features ?? []); this.snapshotKnown = true; if (!this.features.has("input-ready")) for (const pane of this.outputSeen) if (this.attached.has(pane)) this.inputReady.add(pane); + // Capabilities arrive after open's role/attach replay. Restore only current interest, + // never send an unknown frame to an older bridge or replay a hidden/detached chat. + if (this.features.has("conversation-watch")) { + for (const paneId of this.attached.keys()) { + if (this.conversationWanted.has(paneId)) this.rawSend({ type: "conversation-watch", pane_id: paneId, enabled: true }); + } + } if (this.features.has("watch")) { for (const [paneId, grid] of this.watched) this.rawSend({ type: "watch", pane_id: paneId, cols: grid.cols, rows: grid.rows }); } else { @@ -217,12 +228,31 @@ export class HerdrSocket { this.inputReady.delete(paneId); this.attached.set(paneId, { cols, rows, keepSize }); this.send({ type: "attach", pane_id: paneId, cols, rows, flow_control: "ack", ...(keepSize ? { keep_size: true } : {}) }); + if (this.conversationWanted.has(paneId) && this.features.has("conversation-watch")) { + this.send({ type: "conversation-watch", pane_id: paneId, enabled: true }); + } if (this.outputStopped) { this.outputStopped = false; this.connect(); } } + /** Retained control state, not queued input: kept across detach, asked again by the next attach. */ + watchConversation(paneId: string, enabled: boolean): void { + if (this.conversationWanted.has(paneId) === enabled) return; + if (enabled) this.conversationWanted.add(paneId); + else this.conversationWanted.delete(paneId); + // a detached pane has no interest on the server: its next attach asks + if (this.attached.has(paneId) && this.features.has("conversation-watch")) { + this.send({ type: "conversation-watch", pane_id: paneId, enabled }); + } + } + + /** Whether this connection's bridge pushes transcript changes to a watching chat. */ + conversationPushes(): boolean { + return this.connected && this.features.has("conversation-watch"); + } + /** Capture connection AND subscription before xterm's asynchronous write. Never queue ACKs. */ outputAcknowledgement(message: Extract): (() => void) | undefined { const flow = message.flow;