diff --git a/.github/workflows/studio.yml b/.github/workflows/studio.yml index c3394b835d..62c61d403c 100644 --- a/.github/workflows/studio.yml +++ b/.github/workflows/studio.yml @@ -163,6 +163,7 @@ jobs: node perf/settings-rows.mjs node perf/narrow-steps.mjs node perf/card-shell.mjs + node perf/notice-line.mjs node perf/brought-line.mjs node perf/modal-focus.mjs node perf/focus-ring.mjs diff --git a/desktop/frontend-next/perf/cards.mjs b/desktop/frontend-next/perf/cards.mjs index 72f59d5d94..6ec58b1294 100644 --- a/desktop/frontend-next/perf/cards.mjs +++ b/desktop/frontend-next/perf/cards.mjs @@ -101,6 +101,12 @@ check("回执没有重复根容器", await page.locator(".rc > .rc").count() === // between any two glyphs, so it is the one that must not give up its width. const rc = await browser.newPage({ viewport: { width: 1440, height: 900 }, locale: "zh-CN", colorScheme: "dark" }); rc.on("pageerror", (e) => fails.push("页面异常: " + e.message)); +await rc.addInitScript(() => { + try { + localStorage.setItem("rx-turn-receipt", "on"); + } catch { + } +}); await rc.goto(PAGE, { waitUntil: "networkidle" }); await rc.waitForSelector(".compose"); await rc.evaluate(() => { diff --git a/desktop/frontend-next/perf/harness.tsx b/desktop/frontend-next/perf/harness.tsx index fd73c77b99..0bae6ea784 100644 --- a/desktop/frontend-next/perf/harness.tsx +++ b/desktop/frontend-next/perf/harness.tsx @@ -30,6 +30,7 @@ import "../src/styles/gitgraph.css"; import "../src/styles/tips.css"; import "../src/styles/waitnote.css"; import "../src/styles/bring.css"; +import "../src/styles/notice.css"; import "../src/styles/peers.css"; import "../src/styles/jsontext.css"; import "../src/styles/skillpin.css"; diff --git a/desktop/frontend-next/perf/notice-line.mjs b/desktop/frontend-next/perf/notice-line.mjs new file mode 100644 index 0000000000..8c154f13b0 --- /dev/null +++ b/desktop/frontend-next/perf/notice-line.mjs @@ -0,0 +1,50 @@ +// A host notice is a bar, not a card: one sentence with a glyph, no header strip, no box and no state wash. Severity is +// the glyph's ink, so warn and error differ only there. +import { chromium } from "playwright"; + +const BASE = process.env.PERF_URL ?? "http://localhost:4399/perf.html"; +const fails = []; +const check = (name, ok, detail = "") => { + console.log(`${ok ? " ok" : "FAIL"} ${name}${detail ? " — " + detail : ""}`); + if (!ok) fails.push(name); +}; + +const browser = await chromium.launch(); +for (const scheme of ["light", "dark"]) { + const ctx = await browser.newContext({ locale: "zh-CN", viewport: { width: 1000, height: 760 }, colorScheme: scheme }); + const page = await ctx.newPage(); + page.on("pageerror", (e) => fails.push("page error: " + e.message)); + await page.goto(BASE, { waitUntil: "networkidle" }); + await page.waitForSelector(".app"); + await page.waitForTimeout(600); + await page.evaluate(() => { + window.__feed({ kind: "turn_started", authoredTurn: 1, msgIndex: 0, text: "go" }); + for (const level of ["info", "warn", "error"]) window.__feed({ kind: "notice", level, text: `a ${level} notice`, detail: "the diagnostic" }); + window.__feed({ kind: "turn_done" }); + }); + await page.waitForTimeout(500); + await page.mouse.move(990, 10); + const m = await page.evaluate(() => { + const read = (el) => { + const cs = getComputedStyle(el); + const g = el.querySelector(".nb-ic, .nb-dot"); + return { + border: [cs.borderTopWidth, cs.borderLeftWidth, cs.borderRightWidth, cs.borderBottomWidth].join(","), + bg: cs.backgroundColor, + header: !!el.querySelector(".hl, .tag"), + glyph: g ? getComputedStyle(g).color : "", + }; + }; + return [...document.querySelectorAll(".notice-bar")].map(read); + }); + check(`${scheme}: three notices drawn as bars`, m.length === 3, `${m.length} found`); + check(`${scheme}: no box, wash or header strip`, m.length > 0 && m.every((x) => x.border === "0px,0px,0px,0px" && (x.bg === "rgba(0, 0, 0, 0)" || x.bg === "transparent") && !x.header), JSON.stringify(m[0] ?? {})); + check(`${scheme}: severity is the glyph ink alone`, m.length === 3 && new Set(m.map((x) => x.glyph)).size === 3, m.map((x) => x.glyph).join(" / ")); + await ctx.close(); +} +await browser.close(); +if (fails.length) { + console.log(`\n${fails.length} failed`); + process.exit(1); +} +console.log("\nall ok"); diff --git a/desktop/frontend-next/public/_headers b/desktop/frontend-next/public/_headers index ae0670c12c..26bf62e94d 100644 --- a/desktop/frontend-next/public/_headers +++ b/desktop/frontend-next/public/_headers @@ -1,2 +1,12 @@ +# The shell names the build that is current, and the page's own out-of-date +# check reads this same file over no-store. A cached or revalidated copy makes +# an open page look behind and keeps reloading onto it. Hashed bundles under +# /assets are the opposite: they are named by content, so they never change. +/ + Cache-Control: no-store + +/index.html + Cache-Control: no-store + /assets/* Cache-Control: public, max-age=31536000, immutable diff --git a/desktop/frontend-next/src/actions.ts b/desktop/frontend-next/src/actions.ts index 5daa44cd61..3400c3e9ad 100644 --- a/desktop/frontend-next/src/actions.ts +++ b/desktop/frontend-next/src/actions.ts @@ -63,6 +63,7 @@ export const ACTIONS: UIAction[] = [ { id: "appearance.live-grace", kind: "view", target: "none", proof: "interaction" }, { id: "appearance.legacy-sessions", kind: "view", target: "none", proof: "interaction" }, { id: "appearance.open-session-limit", kind: "view", target: "none", proof: "interaction" }, + { id: "appearance.live-stream-budget", kind: "view", target: "none", proof: "interaction" }, { id: "appearance.session-quick-actions", kind: "view", target: "none", proof: "interaction" }, { id: "appearance.graphics", kind: "view", target: "none", proof: "interaction" }, { id: "appearance.session-tabs", kind: "view", target: "none", proof: "interaction" }, diff --git a/desktop/frontend-next/src/i18n/en_kernel.ts b/desktop/frontend-next/src/i18n/en_kernel.ts index e817d44aca..dc9b347dd0 100644 --- a/desktop/frontend-next/src/i18n/en_kernel.ts +++ b/desktop/frontend-next/src/i18n/en_kernel.ts @@ -331,7 +331,7 @@ export const EN_KERNEL: Record = { "这个会话不支持辅助提问": "Side questions are not available in this session", "这个辅助提问已经关闭，请重新打开": "That side question is closed; open it again", "上一个问题还在回答，请稍候再问": "The previous question is still being answered; wait a moment", - "同时打开的辅助提问太多，请先关闭几个": "Too many side questions are open; close some first", + "同时打开的辅助提问已到你设置的上限（agent.max_side_sessions），请先关闭几个": "The side questions open at once have reached your limit (agent.max_side_sessions); close some first", "会话间消息未开启": "Messages between sessions are not turned on", "这个客户端不能收发会话间消息": "This client cannot send or decide messages between sessions", "本机没有叫这个名字的会话": "No session on this machine has that name", diff --git a/desktop/frontend-next/src/i18n/en_settings.ts b/desktop/frontend-next/src/i18n/en_settings.ts index 5072ba29be..c05dbe7ede 100644 --- a/desktop/frontend-next/src/i18n/en_settings.ts +++ b/desktop/frontend-next/src/i18n/en_settings.ts @@ -388,6 +388,10 @@ export const EN_SETTINGS: Record = { "启用内置浏览器": "Enable the built-in browser", "乐观模式（效率优先）": "Optimistic mode (efficiency first)", "同时保留的会话标签数": "Session tabs kept open", + "保持实时连接的会话标签数": "Session tabs kept live", + "默认留空：自动。浏览器对同一个地址只开约 6 个连接，会话列表占一个，所以只让少数标签保持实时连接，其余的标签在显示时从断点接着读，运行中的后台标签每 5 秒读一次状态。调高会让更多后台标签实时更新，但开太多会让停止、发送排不上队。": "Empty means automatic. A browser opens only about six connections to one address and the session list takes one, so only a few tabs stay live; the others resume from where they stopped when shown, and a running background tab has its status read every 5 seconds. A higher number updates more background tabs live, but too many makes Stop and Send queue behind them.", + "实时连接的标签数": "Tabs with a live connection", + "留空表示自动；可填 1 到 {max}": "Empty is automatic; 1 to {max}", "默认 0：不自动收起，打开的会话都保留。设成一个数字后，打开的会话超过它时，会收起最久没动的空闲会话；对话仍在列表里，随时可以重新打开。保留得越多，占用的内存越多。": "Default 0: nothing is folded away and every open session stays. Set a number and, once more sessions than that are open, the idle one untouched longest is folded away; its conversation stays in the list and can be opened again. The more you keep, the more memory it uses.", "会话标签数上限": "Limit on session tabs", "0 表示不限制": "0 means no limit", diff --git a/desktop/frontend-next/src/i18n/kernel.ts b/desktop/frontend-next/src/i18n/kernel.ts index abaca560bd..02a83effae 100644 --- a/desktop/frontend-next/src/i18n/kernel.ts +++ b/desktop/frontend-next/src/i18n/kernel.ts @@ -262,7 +262,7 @@ const SAID: Record = { "side.unavailable": "这个会话不支持辅助提问", "side.unknown": "这个辅助提问已经关闭，请重新打开", "side.busy": "上一个问题还在回答，请稍候再问", - "side.limit": "同时打开的辅助提问太多，请先关闭几个", + "side.limit": "同时打开的辅助提问已到你设置的上限（agent.max_side_sessions），请先关闭几个", "side.empty": "请先写下要问的内容", "side.failed": "辅助提问没能完成，请重试", diff --git a/desktop/frontend-next/src/main.tsx b/desktop/frontend-next/src/main.tsx index edfbf132b9..43a3bdb725 100644 --- a/desktop/frontend-next/src/main.tsx +++ b/desktop/frontend-next/src/main.tsx @@ -14,6 +14,7 @@ import "./styles/tips.css"; import "./styles/skillpin.css"; import "./styles/waitnote.css"; import "./styles/bring.css"; +import "./styles/notice.css"; import "./styles/peers.css"; import "./styles/contentmatches.css"; import "./styles/jsontext.css"; diff --git a/desktop/frontend-next/src/port/cloud_remote.test.ts b/desktop/frontend-next/src/port/cloud_remote.test.ts index d4af3df080..faf0503e83 100644 --- a/desktop/frontend-next/src/port/cloud_remote.test.ts +++ b/desktop/frontend-next/src/port/cloud_remote.test.ts @@ -66,7 +66,9 @@ describe("remote Studio binary framing", () => { source.onmessage = (event) => seen.push(JSON.parse(event.data) as Record); await vi.advanceTimersByTimeAsync(0); - expect(String(fetchMock.mock.calls[0][0])).toBe(`/rt/one/events/replay?lastEventId=${Number.MAX_SAFE_INTEGER}`); + // No cursor is an attach, and an attach is answered with whatever a run is + // stopped on rather than with the stream's history. + expect(String(fetchMock.mock.calls[0][0])).toBe("/rt/one/events/replay"); expect(seen).toEqual([{ kind: "stream_watermark", seq: 7 }]); await vi.advanceTimersByTimeAsync(1500); @@ -78,6 +80,33 @@ describe("remote Studio binary framing", () => { vi.useRealTimers(); }); + it("hands an attaching page the question a run is waiting on", async () => { + vi.useFakeTimers(); + const fetchMock = vi.spyOn(globalThis, "fetch") + .mockResolvedValueOnce(new Response(JSON.stringify({ + frames: [{ kind: "ask_request", ask: { id: "ask-1" } }], + complete: true, + watermark: 3, + }), { status: 200 })) + .mockResolvedValue(new Response(JSON.stringify({ frames: [], complete: true, watermark: 3 }), { status: 200 })); + const seen: Array> = []; + const source = new RemoteEventSource("/rt/one/events"); + source.onmessage = (event) => seen.push(JSON.parse(event.data) as Record); + + await vi.advanceTimersByTimeAsync(0); + // What a run is blocked on is not in the ordinary history a client resumes + // past, so a page that read only the log would draw the run as stopped with + // nothing to answer. + expect(seen).toEqual([{ kind: "ask_request", ask: { id: "ask-1" } }]); + + await vi.advanceTimersByTimeAsync(200); + expect(String(fetchMock.mock.calls[1][0])).toBe("/rt/one/events/replay?lastEventId=3"); + + source.close(); + fetchMock.mockRestore(); + vi.useRealTimers(); + }); + it("resumes from the cursor its URL carries", async () => { vi.useFakeTimers(); const fetchMock = vi.spyOn(globalThis, "fetch").mockResolvedValue(new Response(JSON.stringify({ @@ -129,7 +158,9 @@ describe("remote Studio binary framing", () => { await vi.advanceTimersByTimeAsync(STREAM_SILENCE_MS * 3); stop(); expect(opened).toBe(1); - for (const [input] of fetchMock.mock.calls) expect(String(input)).toMatch(/^\/rt\/one\/events\/replay\?/); + // Every read goes to the replay route: the attach carries no cursor, and + // every later read resumes from the watermark. + for (const [input] of fetchMock.mock.calls) expect(String(input)).toMatch(/^\/rt\/one\/events\/replay(\?|$)/); fetchMock.mockRestore(); vi.unstubAllGlobals(); vi.useRealTimers(); diff --git a/desktop/frontend-next/src/port/cloud_remote.ts b/desktop/frontend-next/src/port/cloud_remote.ts index 3997a9fee4..18c205f2fc 100644 --- a/desktop/frontend-next/src/port/cloud_remote.ts +++ b/desktop/frontend-next/src/port/cloud_remote.ts @@ -173,21 +173,27 @@ interface ReplayAnswer { } // Stands in for EventSource over the relay by polling /events/replay. A cursor -// the URL carries is where a replacement stream resumes; without one it attaches -// at the watermark, and asking past it returns no frames, not the whole log. -// Frames the SSE transport writes on its own (the watermark, a gap) are said -// here too, or the stream reads as silent to whoever watches it. +// the URL carries is where a replacement stream resumes. Frames the SSE +// transport writes on its own (the watermark, a gap) are said here too, or the +// stream reads as silent to whoever watches it. export class RemoteEventSource { onmessage: ((event: MessageEvent) => void) | null = null; private closed = false; private after: number | null; private readonly replay: string; + // What an attach asks for. Only the main event stream carries connection-local + // state to hand back — the pending prompt — so only its attach asks for + // nothing and is answered with that prompt. The tree stream reads the tree, + // not history, so its attach asks past the watermark as it always has: + // replaying the tree's history would only re-read the tree. + private readonly attachQuery: string; constructor(url: string) { const target = new URL(url, location.origin); const cursor = Number(target.searchParams.get("lastEventId") ?? ""); this.after = target.searchParams.has("lastEventId") && Number.isSafeInteger(cursor) && cursor >= 0 ? cursor : null; this.replay = `${target.pathname}/replay`; + this.attachQuery = target.pathname.endsWith("/tree/events") ? `?lastEventId=${Number.MAX_SAFE_INTEGER}` : ""; queueMicrotask(() => void this.poll()); } @@ -203,12 +209,15 @@ export class RemoteEventSource { while (!this.closed) { let active = false; try { - const response = await fetch(`${this.replay}?lastEventId=${this.after ?? Number.MAX_SAFE_INTEGER}`, { credentials: "same-origin" }); + // No cursor yet is an attach; the main stream's attach asks for nothing + // so the kernel answers with the waiting prompt, the tree stream's asks + // past the watermark. Every later read resumes from where it stands. + const query = this.after !== null ? `?lastEventId=${this.after}` : this.attachQuery; + const response = await fetch(`${this.replay}${query}`, { credentials: "same-origin" }); if (response.ok && !this.closed) { const body = await response.json() as ReplayAnswer; const watermark = typeof body.watermark === "number" ? body.watermark : null; - const attaching = this.after === null; - const frames = attaching ? [] : body.frames ?? []; + const frames = body.frames ?? []; active = frames.length > 0; if (this.after === null) this.after = watermark ?? 0; else if (body.complete === false) { @@ -303,6 +312,12 @@ async function connect(deviceId: string, nativeFetch: typeof fetch, useBootstrap } } +// ownFetch is the page's own fetch, bound before remoteHub replaces +// globalThis.fetch with one that routes same-origin requests over the tunnel. +// A hosted page has to read its own origin — the build it is running — not the +// kernel's, so anything that reads the page itself uses this and not the global. +export const ownFetch: typeof fetch = globalThis.fetch.bind(globalThis); + export async function remoteHub(deviceId: string): Promise { const nativeFetch = globalThis.fetch.bind(globalThis); let first: RemoteConnection; diff --git a/desktop/frontend-next/src/port/livestream.test.ts b/desktop/frontend-next/src/port/livestream.test.ts index 740bcc0e01..28e9a819bb 100644 --- a/desktop/frontend-next/src/port/livestream.test.ts +++ b/desktop/frontend-next/src/port/livestream.test.ts @@ -82,3 +82,61 @@ describe("a stream that goes silent", () => { stop(); }); }); + +describe("a stream that gives its connection up", () => { + it("closes the connection, keeps its place, and resumes from the last frame it folded", () => { + const opened = stubStreams(); + vi.stubGlobal("fetch", async () => ({ ok: true, json: async () => ({ frames: [], complete: true }) })); + const seen: number[] = []; + const sub = new SsePort("", "r1").subscribe((ev) => ev.seq && seen.push(ev.seq)); + opened[0].say('{"kind":"stream_watermark","seq":0}'); + opened[0].say('{"kind":"turn_started","seq":1}'); + sub.pause!(); + expect(opened).toHaveLength(1); + expect(opened[0].closed).toBe(true); + // Silence while paused is not a dead stream: nothing replaces it. + vi.advanceTimersByTime(STREAM_SILENCE_MS * 3); + expect(opened).toHaveLength(1); + sub.resume!(); + expect(opened).toHaveLength(2); + expect(opened[1].url).toBe("/events?lastEventId=1"); + opened[1].say('{"kind":"turn_started","seq":1}'); + opened[1].say('{"kind":"turn_done","seq":2}'); + expect(seen).toEqual([1, 2]); + sub(); + }); + + it("recovers the frames that went by while it was paused, in order", async () => { + const opened = stubStreams(); + vi.stubGlobal("fetch", async (url: string) => { + expect(url).toContain("/events/replay?lastEventId=1"); + return { ok: true, json: async () => ({ frames: [{ kind: "text", seq: 2 }, { kind: "turn_done", seq: 3 }], complete: true }) }; + }); + vi.useRealTimers(); + const seen: number[] = []; + const sub = new SsePort("", "r1").subscribe((ev) => ev.seq && seen.push(ev.seq)); + opened[0].say('{"kind":"stream_watermark","seq":0}'); + opened[0].say('{"kind":"turn_started","seq":1}'); + sub.pause!(); + sub.resume!(); + // The server's first word on the new connection states where it is: past the + // cursor, so the missing frames are fetched back before anything newer folds. + opened[1].say('{"kind":"stream_watermark","seq":3}'); + opened[1].say('{"kind":"turn_started","seq":4}'); + await new Promise((r) => setTimeout(r, 20)); + expect(seen).toEqual([1, 2, 3, 4]); + sub(); + }); + + it("says it has a hole when it paused before the stream ever stated a position", () => { + const opened = stubStreams(); + const gaps: number[] = []; + const sub = new SsePort("", "r1").subscribe(() => {}, () => gaps.push(1)); + sub.pause!(); + expect(gaps).toEqual([]); + sub.resume!(); + expect(gaps).toEqual([1]); + expect(opened[1].url).toBe("/events"); + sub(); + }); +}); diff --git a/desktop/frontend-next/src/port/livestream.ts b/desktop/frontend-next/src/port/livestream.ts index 1ad51b8650..6c143bdac0 100644 --- a/desktop/frontend-next/src/port/livestream.ts +++ b/desktop/frontend-next/src/port/livestream.ts @@ -5,27 +5,34 @@ export const STREAM_SILENCE_MS = 45_000; const CHECK_MS = 5_000; +/** The disposer of a live stream, which can also give its connection back and + * take one again. A paused stream keeps whatever cursor `url` reads, so the + * connection that replaces it asks for exactly what went by in between. */ +export type LiveStream = (() => void) & { pause: () => void; resume: () => void }; + /** Holds one EventSource open and replaces it once it has gone silent. `url` * names where to resume, so a replacement asks for what the dead one missed. * Timers stop while a phone sleeps, so waking and coming back online check at * once rather than on the next tick. */ -export function openLiveStream(url: () => string, onData: (raw: string) => void, silenceMs = STREAM_SILENCE_MS): () => void { - let es: EventSource; +export function openLiveStream(url: () => string, onData: (raw: string) => void, silenceMs = STREAM_SILENCE_MS): LiveStream { + let es: EventSource | null = null; + let paused = false; let heard = 0; const open = () => { heard = Date.now(); - es = new EventSource(url(), { withCredentials: true }); - es.onopen = () => { + const next = new EventSource(url(), { withCredentials: true }); + es = next; + next.onopen = () => { heard = Date.now(); }; - es.onmessage = (m) => { + next.onmessage = (m) => { heard = Date.now(); onData(m.data); }; }; const check = () => { - if (Date.now() - heard < silenceMs) return; - es.close(); + if (paused || Date.now() - heard < silenceMs) return; + es?.close(); open(); }; const wake = () => { @@ -38,12 +45,28 @@ export function openLiveStream(url: () => string, onData: (raw: string) => void, document.addEventListener("visibilitychange", wake); addEventListener("online", check); } - return () => { - clearInterval(timer); - if (dom) { - document.removeEventListener("visibilitychange", wake); - removeEventListener("online", check); - } - es.close(); - }; + const stop: LiveStream = Object.assign( + () => { + clearInterval(timer); + if (dom) { + document.removeEventListener("visibilitychange", wake); + removeEventListener("online", check); + } + es?.close(); + }, + { + pause: () => { + if (paused) return; + paused = true; + es?.close(); + es = null; + }, + resume: () => { + if (!paused) return; + paused = false; + open(); + }, + }, + ); + return stop; } diff --git a/desktop/frontend-next/src/port/port.ts b/desktop/frontend-next/src/port/port.ts index d68b4f708d..7438ed8411 100644 --- a/desktop/frontend-next/src/port/port.ts +++ b/desktop/frontend-next/src/port/port.ts @@ -1,6 +1,10 @@ import type { SurfaceActionRequest, SurfaceActionRecord } from "./surface"; import { DeliveryError, HttpError } from "./http_error"; import type { DeliveryFault } from "./http_error"; + +/** What `subscribe` hands back: the disposer, and where the transport has one, a + * way to give its connection up and take it again without losing the cursor. */ +export type Subscription = (() => void) & { pause?: () => void; resume?: () => void }; import type { Attachment, DroppedRef, KeptText } from "./attachment"; export { DeliveryError, HttpError }; export type { DeliveryFault }; @@ -567,7 +571,7 @@ export interface AgentPort { // model folds one ordered stream and folds nothing twice. The number belongs // to the stream, not to the model — several snapshots resume from the lowest // of their watermarks, because a duplicate folds and a gap does not. - subscribe(onEvent: (ev: WireEvent) => void, onGap?: () => void, bootstrap?: () => Promise): () => void; + subscribe(onEvent: (ev: WireEvent) => void, onGap?: () => void, bootstrap?: () => Promise): Subscription; // chips carries skill chips beside the line; without it the line is parsed // as typed, so a kernel that predates chips still runs a leading one. diff --git a/desktop/frontend-next/src/port/sse.ts b/desktop/frontend-next/src/port/sse.ts index 13509175df..269665e994 100644 --- a/desktop/frontend-next/src/port/sse.ts +++ b/desktop/frontend-next/src/port/sse.ts @@ -7,7 +7,7 @@ import type { StoragePlan, StorageQuery, StorageState } from "./storage"; import type { ExecutionGraphRead, TrajectoryRead, WireEvent } from "./wire"; import { host } from "./host"; import { current } from "../i18n"; -import { openLiveStream } from "./livestream"; +import { subscribeStream } from "./sse_subscribe"; // The running project is the default, so its requests stay the bare path they // have always been and only a cross-project read carries the folder. @@ -522,162 +522,10 @@ export class SsePort extends SseWorkspace implements AgentPort { return this.post("/rewind/undo", { transactionId }); } - /** subscribe delivers the kernel's events, in order, with no holes it did not - * announce. Frames that matter are numbered, so a break in the numbers is - * the client noticing it missed something rather than a gap it renders as a - * quiet turn — and the two transports differ only in how they fetch the - * missing frames back, not in what a gap means. - * - * onGap fires when the replay log can no longer close one. The hole is the - * transport's fact; which authority each read model returns to — /history, - * the run graph — is the caller's. - * - * bootstrap is how a read model with a snapshot of its own joins without a - * seam: it reads one and answers with the frame that snapshot is at least as - * new as. Frames are held from before the transport is attached until the - * replay after that number has landed, so nothing between the two is - * skipped. The number is the whole stream's, never one model's — a frame of - * any kind carries the cursor past it, or the next number reads as a hole. */ + /** The kernel's events in order, with no holes it did not announce; see + * subscribeStream for how a gap, a bootstrap and a paused connection work. */ subscribe(onEvent: (ev: WireEvent) => void, onGap?: () => void, bootstrap?: () => Promise) { - // Frames arriving while a recovery request is in flight wait for it: the - // whole point is that the reducer sees one ordered stream, and delivering - // the new frame first would put a result ahead of the dispatch it answers. - // null until the stream states a position: 0 cannot tell a subscriber that - // just attached from one that has seen the stream start. - let seen: number | null = null; - // A bootstrapping subscription holds from before the transport is attached: - // the shell's bus has no connection to open, so it is already delivering, - // and a frame taken live here would be folded ahead of the replay's. - let booting = !!bootstrap; - let recovering = booting; - let held: WireEvent[] = []; - let live = true; - - const deliver = (ev: WireEvent) => { - // The cursor is the record of what has been folded, and the replay and - // the live stream reach the same frames — a snapshot's own watermark - // included. Whichever arrives second is the same fact twice. - if (ev.seq && seen !== null && ev.seq <= seen) return; - if (ev.seq) seen = Math.max(seen ?? 0, ev.seq); - onEvent(ev); - }; - - const flush = () => { - recovering = false; - const queued = held; - held = []; - for (const ev of queued) deliver(ev); - }; - - const recover = async (after: number) => { - recovering = true; - try { - const res = await fetch(`${this.base}/events/replay?lastEventId=${after}`, { credentials: "same-origin" }); - if (!res.ok) throw new Error(String(res.status)); - const body = (await res.json()) as { frames?: WireEvent[]; complete?: boolean }; - if (!live) return; - for (const ev of body.frames ?? []) deliver(ev); - if (!body.complete) onGap?.(); - } catch { - // The frames are gone and asking again would not bring them back. The - // transcript is the one source that can still answer. - if (live) onGap?.(); - } finally { - flush(); - } - }; - - // What the snapshot and the replay overlap on folds twice onto the same - // state; what falls between them is gone for good. So the read names where - // it stands, and the stream resumes from there — one number, one stream. - const bootstrapFrom = async (read: () => Promise) => { - let at: number | null = null; - try { - at = await read(); - } catch { - // Nothing to resume onto. Holding the stream any longer would stall the - // view rather than repair it, so it runs live and says it is behind. - if (live) onGap?.(); - } - if (!live) return; - if (at === null) { - booting = false; - flush(); - return; - } - seen = at; - booting = false; - await recover(at); - }; - - const accept = (ev: WireEvent) => { - // The stream describing itself: a watermark states the number the client - // should have reached, which is the only way to notice a frame lost at - // the end of a turn. Neither reaches the reducer. - if (ev.kind === "stream_watermark") { - // The first one states where this subscriber attached: a live-only - // subscription replays nothing before it, so nothing before it was lost. - // A bootstrap states that position from its snapshot instead, and taking - // it from here would baseline the cursor over frames already held. - if (seen === null) { - if (!booting) seen = ev.seq ?? 0; - return; - } - if (!recovering && ev.seq && ev.seq > seen) void recover(seen); - return; - } - if (ev.kind === "stream_gap") { - // The hole reaches the caller either way — it is the whole stream's, - // not the bootstrapping model's — but the number it carries is not a - // position this subscriber reached. - if (!booting) seen = Math.max(seen ?? 0, ev.seq ?? 0); - onGap?.(); - return; - } - if (recovering) { - held.push(ev); - return; - } - // Numbering that goes backwards is a different stream, not a gap: the - // server restarted and is counting from one again. Inferred rather than - // carried as a stream id — a resumed client is never sent a lower number - // than it holds, so nothing else produces this. Without it the client - // compares against a watermark that will never be reached again and - // silently stops noticing losses. - if (ev.seq && seen && ev.seq < seen) { - seen = ev.seq; - onGap?.(); - onEvent(ev); - return; - } - if (ev.seq && seen && ev.seq > seen + 1) { - held.push(ev); - void recover(seen); - return; - } - deliver(ev); - }; - - const feed = (raw: string) => { - try { - accept(JSON.parse(raw) as WireEvent); - } catch { - // A malformed frame must not tear down the stream. - } - }; - // EventSource resumes on its own: it reconnects carrying the last id it - // saw, and the server replays from there. Recovery here is for the frames - // shed without the connection ever dropping, and a replacement stream for - // one that went silent without dropping resumes from the same cursor. - const detach = openLiveStream( - () => this.base + "/events" + (seen === null ? "" : `?lastEventId=${seen}`), - feed, - ); - if (bootstrap) void bootstrapFrom(bootstrap); - return () => { - live = false; - detach(); - }; + return subscribeStream(this.base, onEvent, onGap, bootstrap); } submit(text: string, chips?: ChipCall) { diff --git a/desktop/frontend-next/src/port/sse_subscribe.ts b/desktop/frontend-next/src/port/sse_subscribe.ts new file mode 100644 index 0000000000..c1445f1e77 --- /dev/null +++ b/desktop/frontend-next/src/port/sse_subscribe.ts @@ -0,0 +1,180 @@ +import type { WireEvent } from "./wire"; +import type { Subscription } from "./port"; +import { openLiveStream } from "./livestream"; + +/** subscribe delivers the kernel's events, in order, with no holes it did not + * announce. Frames that matter are numbered, so a break in the numbers is + * the client noticing it missed something rather than a gap it renders as a + * quiet turn — and the two transports differ only in how they fetch the + * missing frames back, not in what a gap means. + * + * onGap fires when the replay log can no longer close one. The hole is the + * transport's fact; which authority each read model returns to — /history, + * the run graph — is the caller's. + * + * bootstrap is how a read model with a snapshot of its own joins without a + * seam: it reads one and answers with the frame that snapshot is at least as + * new as. Frames are held from before the transport is attached until the + * replay after that number has landed, so nothing between the two is + * skipped. The number is the whole stream's, never one model's — a frame of + * any kind carries the cursor past it, or the next number reads as a hole. */ +export function subscribeStream(base: string, onEvent: (ev: WireEvent) => void, onGap?: () => void, bootstrap?: () => Promise): Subscription { + // Frames arriving while a recovery request is in flight wait for it: the + // whole point is that the reducer sees one ordered stream, and delivering + // the new frame first would put a result ahead of the dispatch it answers. + // null until the stream states a position: 0 cannot tell a subscriber that + // just attached from one that has seen the stream start. + let seen: number | null = null; + // A bootstrapping subscription holds from before the transport is attached: + // the shell's bus has no connection to open, so it is already delivering, + // and a frame taken live here would be folded ahead of the replay's. + let booting = !!bootstrap; + let recovering = booting; + let held: WireEvent[] = []; + let live = true; + + const deliver = (ev: WireEvent) => { + // The cursor is the record of what has been folded, and the replay and + // the live stream reach the same frames — a snapshot's own watermark + // included. Whichever arrives second is the same fact twice. + if (ev.seq && seen !== null && ev.seq <= seen) return; + if (ev.seq) seen = Math.max(seen ?? 0, ev.seq); + onEvent(ev); + }; + + const flush = () => { + recovering = false; + const queued = held; + held = []; + for (const ev of queued) deliver(ev); + }; + + const recover = async (after: number) => { + recovering = true; + try { + const res = await fetch(`${base}/events/replay?lastEventId=${after}`, { credentials: "same-origin" }); + if (!res.ok) throw new Error(String(res.status)); + const body = (await res.json()) as { frames?: WireEvent[]; complete?: boolean }; + if (!live) return; + for (const ev of body.frames ?? []) deliver(ev); + if (!body.complete) onGap?.(); + } catch { + // The frames are gone and asking again would not bring them back. The + // transcript is the one source that can still answer. + if (live) onGap?.(); + } finally { + flush(); + } + }; + + // What the snapshot and the replay overlap on folds twice onto the same + // state; what falls between them is gone for good. So the read names where + // it stands, and the stream resumes from there — one number, one stream. + const bootstrapFrom = async (read: () => Promise) => { + let at: number | null = null; + try { + at = await read(); + } catch { + // Nothing to resume onto. Holding the stream any longer would stall the + // view rather than repair it, so it runs live and says it is behind. + if (live) onGap?.(); + } + if (!live) return; + if (at === null) { + booting = false; + flush(); + return; + } + seen = at; + booting = false; + await recover(at); + }; + + const accept = (ev: WireEvent) => { + // The stream describing itself: a watermark states the number the client + // should have reached, which is the only way to notice a frame lost at + // the end of a turn. Neither reaches the reducer. + if (ev.kind === "stream_watermark") { + // The first one states where this subscriber attached: a live-only + // subscription replays nothing before it, so nothing before it was lost. + // A bootstrap states that position from its snapshot instead, and taking + // it from here would baseline the cursor over frames already held. + if (seen === null) { + if (!booting) seen = ev.seq ?? 0; + return; + } + if (!recovering && ev.seq && ev.seq > seen) void recover(seen); + return; + } + if (ev.kind === "stream_gap") { + // The hole reaches the caller either way — it is the whole stream's, + // not the bootstrapping model's — but the number it carries is not a + // position this subscriber reached. + if (!booting) seen = Math.max(seen ?? 0, ev.seq ?? 0); + onGap?.(); + return; + } + if (recovering) { + held.push(ev); + return; + } + // Numbering that goes backwards is a different stream, not a gap: the + // server restarted and is counting from one again. Inferred rather than + // carried as a stream id — a resumed client is never sent a lower number + // than it holds, so nothing else produces this. Without it the client + // compares against a watermark that will never be reached again and + // silently stops noticing losses. + if (ev.seq && seen && ev.seq < seen) { + seen = ev.seq; + onGap?.(); + onEvent(ev); + return; + } + if (ev.seq && seen && ev.seq > seen + 1) { + held.push(ev); + void recover(seen); + return; + } + deliver(ev); + }; + + const feed = (raw: string) => { + try { + accept(JSON.parse(raw) as WireEvent); + } catch { + // A malformed frame must not tear down the stream. + } + }; + // EventSource resumes on its own: it reconnects carrying the last id it + // saw, and the server replays from there. Recovery here is for the frames + // shed without the connection ever dropping, and a replacement stream for + // one that went silent without dropping resumes from the same cursor. + const detach = openLiveStream( + () => base + "/events" + (seen === null ? "" : `?lastEventId=${seen}`), + feed, + ); + if (bootstrap) void bootstrapFrom(bootstrap); + // A subscription that gave its connection up before the stream ever stated a + // position has nothing to resume from, so what went by is a hole it cannot + // name: it says so, and the caller re-reads. + let pausedBlind = false; + return Object.assign( + () => { + live = false; + detach(); + }, + { + pause: () => { + pausedBlind = seen === null; + detach.pause(); + }, + resume: () => { + detach.resume(); + if (pausedBlind) { + pausedBlind = false; + onGap?.(); + } + }, + }, + ); +} diff --git a/desktop/frontend-next/src/port/streambudget.test.ts b/desktop/frontend-next/src/port/streambudget.test.ts new file mode 100644 index 0000000000..ab2c362388 --- /dev/null +++ b/desktop/frontend-next/src/port/streambudget.test.ts @@ -0,0 +1,99 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { LIVE_PANE_STREAMS, StreamBudget } from "./streambudget"; +import type { Subscription } from "./port"; + +beforeEach(() => vi.useFakeTimers()); +afterEach(() => vi.useRealTimers()); + +function pane() { + const calls: string[] = []; + const sub = Object.assign(() => calls.push("close"), { pause: () => calls.push("pause"), resume: () => calls.push("resume") }) as Subscription; + return { sub, calls, refresh: vi.fn() }; +} + +const idle = { visible: false, busy: false }; + +// The browser gives one origin six connections; the tree holds one and a request +// needs one, so at most this many panes may hold a stream at once. +describe("the panes that hold a live stream", () => { + it("never exceeds the budget, however many panes are open", () => { + const budget = new StreamBudget(); + const panes = Array.from({ length: 8 }, () => pane()); + panes.forEach((p, i) => budget.attach(`r${i}`, p.sub, p.refresh, i === 0 ? { visible: true, busy: false } : idle)); + expect(budget.liveIds()).toHaveLength(LIVE_PANE_STREAMS); + expect(budget.liveIds()).toContain("r0"); + }); + + it("keeps the visible pane first, then the busy ones, with 8 panes and 4 running", () => { + const budget = new StreamBudget(); + const panes = Array.from({ length: 8 }, () => pane()); + panes.forEach((p, i) => budget.attach(`r${i}`, p.sub, p.refresh, { visible: i === 7, busy: i >= 3 && i <= 6 })); + const live = budget.liveIds(); + expect(live).toHaveLength(3); + expect(live).toContain("r7"); + expect(live.filter((id) => id !== "r7").every((id) => ["r3", "r4", "r5", "r6"].includes(id))).toBe(true); + }); + + it("gives an idle pane's connection back and takes it again when the pane is shown", () => { + const budget = new StreamBudget(); + const panes = Array.from({ length: 5 }, () => pane()); + panes.forEach((p, i) => budget.attach(`r${i}`, p.sub, p.refresh, { visible: i === 0, busy: false })); + expect(panes.filter((p) => p.calls.includes("pause"))).toHaveLength(2); + const paused = panes.findIndex((p) => p.calls.includes("pause")); + budget.update("r0", idle); + budget.update(`r${paused}`, { visible: true, busy: false }); + expect(panes[paused].calls).toContain("resume"); + expect(budget.liveIds()).toContain(`r${paused}`); + expect(budget.liveIds()).toHaveLength(3); + }); + + it("reads a busy pane that lost its stream by polling, and stops when it is no longer owed", () => { + const budget = new StreamBudget(() => 1); + const shown = pane(); + const running = pane(); + budget.attach("a", shown.sub, shown.refresh, { visible: true, busy: false }); + budget.attach("b", running.sub, running.refresh, { visible: false, busy: true }); + expect(running.calls).toEqual(["pause"]); + vi.advanceTimersByTime(5_000); + expect(running.refresh).toHaveBeenCalledTimes(1); + expect(shown.refresh).not.toHaveBeenCalled(); + budget.update("b", idle); + vi.advanceTimersByTime(20_000); + expect(running.refresh).toHaveBeenCalledTimes(1); + }); + + it("frees a place when a pane goes away", () => { + const budget = new StreamBudget(() => 1); + const a = pane(); + const b = pane(); + const releaseA = budget.attach("a", a.sub, a.refresh, { visible: true, busy: false }); + budget.attach("b", b.sub, b.refresh, idle); + expect(b.calls).toEqual(["pause"]); + releaseA(); + expect(b.calls).toEqual(["pause", "resume"]); + }); + + it("leaves a transport that cannot pause alone", () => { + const budget = new StreamBudget(() => 1); + const bare = Object.assign(() => {}, {}) as Subscription; + budget.attach("a", bare, () => {}, idle); + budget.attach("b", bare, () => {}, idle); + expect(budget.liveIds()).toEqual(["a", "b"]); + }); +}); + +describe("the number of live streams a person chose", () => { + it("follows the limit as it is read, giving places out again when it changes", () => { + let limit = 1; + const budget = new StreamBudget(() => limit); + const panes = Array.from({ length: 4 }, () => pane()); + panes.forEach((p, i) => budget.attach(`r${i}`, p.sub, p.refresh, { visible: i === 0, busy: false })); + expect(budget.liveIds()).toEqual(["r0"]); + limit = 3; + budget.resettle(); + expect(budget.liveIds()).toHaveLength(3); + limit = 1; + budget.resettle(); + expect(budget.liveIds()).toHaveLength(1); + }); +}); diff --git a/desktop/frontend-next/src/port/streambudget.ts b/desktop/frontend-next/src/port/streambudget.ts new file mode 100644 index 0000000000..0de3ed30b4 --- /dev/null +++ b/desktop/frontend-next/src/port/streambudget.ts @@ -0,0 +1,96 @@ +import type { Subscription } from "./port"; +import { liveStreamBudget, onLiveStreamBudgetChange } from "../state/prefs"; + +// A browser talking HTTP/1.1 opens at most six connections to one origin, and an +// event stream keeps its connection for as long as it lives. Every open pane held +// one, the session tree held another, and from the fifth pane on nothing else — +// Stop, submit, a status read — could get a connection: the window read as stuck. +// So only a few panes keep a live stream. The rest give theirs up and take it +// again, from the cursor they stopped at, when they have something to say. +/** The automatic budget: with the session list's stream that is four of six. */ +export const LIVE_PANE_STREAMS = 3; +const POLL_MS = 5_000; + +export interface Want { + /** On screen: needs frames as they come. */ + visible: boolean; + /** Running or waiting on the person: worth a live stream while there is room. */ + busy: boolean; +} + +interface Entry { + sub: Subscription; + refresh: () => void; + want: Want; + at: number; + live: boolean; +} + +/** Decides which panes hold a live event stream. Visible panes come first, then + * busy ones, then the rest, the most recently changed first within each. A pane + * that is busy but did not make the cut is read by polling instead. */ +export class StreamBudget { + private readonly entries = new Map(); + private timer: ReturnType | null = null; + private tick = 0; + + constructor(private readonly limit: () => number = () => LIVE_PANE_STREAMS) {} + + /** Registers a pane's subscription; the returned function removes it. */ + attach(id: string, sub: Subscription, refresh: () => void, want: Want): () => void { + const entry: Entry = { sub, refresh, want, at: ++this.tick, live: true }; + this.entries.set(id, entry); + this.settle(); + return () => { + if (this.entries.get(id) === entry) this.entries.delete(id); + this.settle(); + }; + } + + /** Says what a pane needs now. */ + update(id: string, want: Want) { + const entry = this.entries.get(id); + if (!entry || (entry.want.visible === want.visible && entry.want.busy === want.busy)) return; + entry.want = want; + entry.at = ++this.tick; + this.settle(); + } + + /** Gives the places out again; call when the limit has changed. */ + resettle() { + this.settle(); + } + + /** Panes with a live stream, for tests and diagnostics. */ + liveIds(): string[] { + return [...this.entries].filter(([, e]) => e.live).map(([id]) => id); + } + + private rank(e: Entry) { + return e.want.visible ? 0 : e.want.busy ? 1 : 2; + } + + private settle() { + const order = [...this.entries].sort(([, a], [, b]) => this.rank(a) - this.rank(b) || b.at - a.at); + order.forEach(([, entry], i) => { + const keep = !entry.sub.pause || i < this.limit(); + if (keep && !entry.live) entry.sub.resume?.(); + if (!keep && entry.live) entry.sub.pause?.(); + entry.live = keep; + }); + this.poll(); + } + + private poll() { + const owed = [...this.entries.values()].some((e) => !e.live && e.want.busy); + if (owed && !this.timer) { + this.timer = setInterval(() => this.entries.forEach((e) => { if (!e.live && e.want.busy) e.refresh(); }), POLL_MS); + } else if (!owed && this.timer) { + clearInterval(this.timer); + this.timer = null; + } + } +} + +export const streamBudget = new StreamBudget(() => liveStreamBudget() ?? LIVE_PANE_STREAMS); +if (typeof window !== "undefined") onLiveStreamBudgetChange(() => streamBudget.resettle()); diff --git a/desktop/frontend-next/src/state/prefs.ts b/desktop/frontend-next/src/state/prefs.ts index 6573f44fe7..8dbe478f3a 100644 --- a/desktop/frontend-next/src/state/prefs.ts +++ b/desktop/frontend-next/src/state/prefs.ts @@ -504,3 +504,45 @@ export function onShowsSessionTabsChange(fn: () => void): () => void { sessionTabListeners.delete(fn); }; } + +// How many panes may hold a live event stream at once. Unset is automatic: the +// most the browser's connection limit leaves room for. A browser opens six +// connections to one origin and the session list holds one, so the number is held +// between one and four; anything else reads as automatic. +const LIVE_STREAM_BUDGET_KEY = "rx-live-stream-budget"; +export const MAX_LIVE_STREAM_BUDGET = 4; +const liveStreamBudgetListeners = new Set<() => void>(); + +/** The chosen number, or null for automatic. */ +export function liveStreamBudget(): number | null { + try { + const raw = localStorage.getItem(LIVE_STREAM_BUDGET_KEY); + if (raw === null || !/^[1-9][0-9]*$/.test(raw)) return null; + const n = Number(raw); + return n >= 1 && n <= MAX_LIVE_STREAM_BUDGET ? n : null; + } catch { + return null; + } +} + +export function setLiveStreamBudget(n: number | null): void { + try { + if (n !== null && Number.isSafeInteger(n) && n >= 1 && n <= MAX_LIVE_STREAM_BUDGET) localStorage.setItem(LIVE_STREAM_BUDGET_KEY, String(n)); + else localStorage.removeItem(LIVE_STREAM_BUDGET_KEY); + } catch { + /* the choice holds for this window and is forgotten on the next */ + } + liveStreamBudgetListeners.forEach((fn) => fn()); +} + +export function onLiveStreamBudgetChange(fn: () => void): () => void { + liveStreamBudgetListeners.add(fn); + const storage = (event: StorageEvent) => { + if (event.key === LIVE_STREAM_BUDGET_KEY || event.key === null) fn(); + }; + window.addEventListener("storage", storage); + return () => { + liveStreamBudgetListeners.delete(fn); + window.removeEventListener("storage", storage); + }; +} diff --git a/desktop/frontend-next/src/styles/LAYOUT.md b/desktop/frontend-next/src/styles/LAYOUT.md index 396fce0f13..d2c61a9d8f 100644 --- a/desktop/frontend-next/src/styles/LAYOUT.md +++ b/desktop/frontend-next/src/styles/LAYOUT.md @@ -1550,6 +1550,13 @@ what exists. - The cards a reader reads or acts on (approval, ask, remember, guardian review, extension card, receipt, compaction) share one shell: `--r-md` radius, a 1px `--border`, the `--surface` ground, and a header strip with no fill of its own. - State is never a border colour: a warning or error card keeps the shell's border and says so in the glyph ink or a chip. +## Notice bar + +- A host notice is one sentence with a glyph and, under it, the kernel's diagnostic in the muted meta size. +- There is no header strip, box, wash or side rule. Info has a dot; warn and error share the warning glyph, and only its ink (`--warn-ink`, `--err-ink`) says which. +- A screen reader gets the level as a visually hidden prefix, and an error is `role="alert"`. +- The card shell (`data-k="host"`) is for receipts, guardian reviews and compaction; notices no longer use it. + ## Transcript Line - A tool call is a Line: a bare status glyph, the title, the mono subject and the meta on one row over a `--hair` bottom border, with no box, icon square or vertical track. Expanding it opens the detail in place. diff --git a/desktop/frontend-next/src/styles/notice.css b/desktop/frontend-next/src/styles/notice.css new file mode 100644 index 0000000000..2bd8696985 --- /dev/null +++ b/desktop/frontend-next/src/styles/notice.css @@ -0,0 +1,11 @@ +.flow .notice-bar { display: flex; align-items: flex-start; gap: 8px; margin: 8px 0; padding: 4px 2px; border-radius: var(--r-sm); background: transparent; color: var(--muted); font: var(--w-reg) var(--fs-ui)/1.55 var(--ui); } +.notice-bar .nb-ic { flex: none; width: 12px; height: 12px; margin-top: 4px; } +.notice-bar .nb-dot { flex: none; width: 6px; height: 6px; margin: 8px 3px 0; border-radius: 50%; background: var(--faint); } +.notice-bar[data-lvl="warn"] .nb-ic { color: var(--warn-ink); } +.notice-bar[data-lvl="err"] .nb-ic { color: var(--err-ink); } +.notice-bar .nb-body { min-width: 0; flex: 1; } +.notice-bar .t { display: block; color: var(--text); overflow-wrap: anywhere; } +.notice-bar .ntimes { margin-left: 6px; color: var(--faint); font-weight: var(--w-reg); } +.notice-bar .nwhy { display: block; margin-top: 2px; font-size: var(--fs-meta); line-height: 1.6; color: var(--faint); white-space: pre-wrap; overflow-wrap: anywhere; } +.notice-bar .nrule { display: block; margin-top: 2px; font: var(--w-reg) var(--fs-meta)/1.6 var(--mono); color: var(--muted); white-space: pre-wrap; overflow-wrap: anywhere; } +.notice-bar .nmd { margin-top: 2px; color: var(--text); overflow-wrap: anywhere; } diff --git a/desktop/frontend-next/src/ui/Appearance.tsx b/desktop/frontend-next/src/ui/Appearance.tsx index d0a60c5054..44a7ecf502 100644 --- a/desktop/frontend-next/src/ui/Appearance.tsx +++ b/desktop/frontend-next/src/ui/Appearance.tsx @@ -15,6 +15,7 @@ import { WindowSection } from "./WindowSection"; import { LiveGraceSection } from "./LiveGraceSection"; import { LegacySessionsSection } from "./LegacySessionsSection"; import { OpenSessionLimitSection } from "./OpenSessionLimitSection"; +import { LiveStreamBudgetSection } from "./LiveStreamBudgetSection"; import { SessionQuickSection } from "./SessionQuickSection"; import { NavRailSection } from "./NavRailSection"; import { GraphicsSection } from "./GraphicsSection"; @@ -554,6 +555,7 @@ export function Appearance({ port, theme, onTheme, contrast, onContrast, weight, + diff --git a/desktop/frontend-next/src/ui/LiveStreamBudgetSection.interaction.test.tsx b/desktop/frontend-next/src/ui/LiveStreamBudgetSection.interaction.test.tsx new file mode 100644 index 0000000000..a4cc67ab4e --- /dev/null +++ b/desktop/frontend-next/src/ui/LiveStreamBudgetSection.interaction.test.tsx @@ -0,0 +1,47 @@ +// @vitest-environment jsdom +import { afterEach, beforeEach, expect, it } from "vitest"; +import { cleanup, render, screen } from "@testing-library/react"; +import userEvent from "@testing-library/user-event"; +import "./testkit"; +import { LiveStreamBudgetSection } from "./LiveStreamBudgetSection"; +import { liveStreamBudget, setLiveStreamBudget } from "../state/prefs"; + +beforeEach(() => localStorage.clear()); +afterEach(cleanup); + +const box = () => screen.getByRole("textbox", { name: "实时连接的标签数" }) as HTMLInputElement; + +it("starts empty, which is automatic, and nothing is stored", () => { + render(); + expect(box().value).toBe(""); + expect(liveStreamBudget()).toBeNull(); + expect(localStorage.getItem("rx-live-stream-budget")).toBeNull(); +}); + +it("saves a number from one to four once it is committed", async () => { + render(); + await userEvent.type(box(), "2{Enter}"); + expect(liveStreamBudget()).toBe(2); + expect(localStorage.getItem("rx-live-stream-budget")).toBe("2"); +}); + +it("reads a number outside the range as automatic and shows that", async () => { + render(); + await userEvent.type(box(), "9{Enter}"); + expect(liveStreamBudget()).toBeNull(); + expect(box().value).toBe(""); + expect(localStorage.getItem("rx-live-stream-budget")).toBeNull(); +}); + +it("returns to automatic when the box is emptied, and shows a choice made in another window", async () => { + setLiveStreamBudget(3); + render(); + expect(box().value).toBe("3"); + await userEvent.clear(box()); + await userEvent.tab(); + expect(liveStreamBudget()).toBeNull(); + cleanup(); + localStorage.setItem("rx-live-stream-budget", "4"); + render(); + expect(box().value).toBe("4"); +}); diff --git a/desktop/frontend-next/src/ui/LiveStreamBudgetSection.tsx b/desktop/frontend-next/src/ui/LiveStreamBudgetSection.tsx new file mode 100644 index 0000000000..52ed9e3eee --- /dev/null +++ b/desktop/frontend-next/src/ui/LiveStreamBudgetSection.tsx @@ -0,0 +1,51 @@ +import { useEffect, useState, useSyncExternalStore } from "react"; +import { t } from "../i18n"; +import { liveStreamBudget, MAX_LIVE_STREAM_BUDGET, onLiveStreamBudgetChange, setLiveStreamBudget } from "../state/prefs"; +import { Group } from "./Group"; + +// How many open session tabs may keep a live connection at once. Empty is +// automatic. A browser holds six connections to one address and the session list +// takes one, so the window keeps a few live and reads the rest when they are shown. +export function LiveStreamBudgetSection() { + const chosen = useSyncExternalStore(onLiveStreamBudgetChange, liveStreamBudget, liveStreamBudget); + const [draft, setDraft] = useState(chosen === null ? "" : String(chosen)); + useEffect(() => setDraft(chosen === null ? "" : String(chosen)), [chosen]); + const commit = () => { + setLiveStreamBudget(draft === "" ? null : Number(draft)); + const now = liveStreamBudget(); + setDraft(now === null ? "" : String(now)); + }; + return ( + +
+ + + {t("留空表示自动；可填 1 到 {max}", { max: MAX_LIVE_STREAM_BUDGET })} + + + setDraft(e.target.value.replace(/\D/g, ""))} + onBlur={commit} + data-action-change="appearance.live-stream-budget" + data-action-keydown="appearance.live-stream-budget" + onKeyDown={(e) => { + if (e.key !== "Enter") return; + commit(); + e.currentTarget.blur(); + }} + /> + {t("个")} + +
+ + ); +} diff --git a/desktop/frontend-next/src/ui/Pane.tsx b/desktop/frontend-next/src/ui/Pane.tsx index 6027a6c136..b69f3b3e9b 100644 --- a/desktop/frontend-next/src/ui/Pane.tsx +++ b/desktop/frontend-next/src/ui/Pane.tsx @@ -1,4 +1,5 @@ import { memo, useCallback, useEffect, useMemo, useReducer, useRef, useState } from "react"; +import { streamBudget } from "../port/streambudget"; import { loadReceiptPref, receiptPrefOf } from "../state/receiptpref"; import { ReceiptPrefContext } from "../state/foldpref"; import { money } from "../i18n/format"; @@ -192,9 +193,8 @@ function PaneView({ port, hub, rt, title, active, visible, sideHost, side, onFoc // A hole in the stream is the transport's fact; which authority answers it is // each model's own. The transcript is rebuilt from /history, the run graph // from the read the kernel rebuilds — one gap, two different re-reads. - useEffect( - () => - port.subscribe( + useEffect(() => { + const sub = port.subscribe( (ev) => { surfaceReplay.observe(ev); compact.onEvent(ev); @@ -221,9 +221,18 @@ function PaneView({ port, hub, rt, title, active, visible, sideHost, side, onFoc rebuild(); refreshStatus(); }, - ), - [port, pacer, rt.id, onTurnDone, reloadMcp, rebuild, refreshStatus, revalue, compact.onEvent, compact.release, surfaceReplay.observe], - ); + ); + // Only a few panes keep a live stream (see streambudget.ts); this one says + // whether it is on screen or busy and the budget gives or takes its connection. + const release = streamBudget.attach(rt.id, sub, refreshStatus, wantRef.current); + return () => { + release(); + sub(); + }; + }, [port, pacer, rt.id, onTurnDone, reloadMcp, rebuild, refreshStatus, revalue, compact.onEvent, compact.release, surfaceReplay.observe]); + const wantRef = useRef({ visible, busy: running }); + wantRef.current = { visible, busy: running }; + useEffect(() => streamBudget.update(rt.id, { visible, busy: running }), [rt.id, visible, running]); // What the rows cover is dispatched before the rows themselves, so the table // never spends a frame showing a prefix as if it were the whole record. diff --git a/desktop/frontend-next/src/ui/cards/NoticeCard.test.tsx b/desktop/frontend-next/src/ui/cards/NoticeCard.test.tsx index d1f6e2a3c7..2394ff9540 100644 --- a/desktop/frontend-next/src/ui/cards/NoticeCard.test.tsx +++ b/desktop/frontend-next/src/ui/cards/NoticeCard.test.tsx @@ -41,34 +41,37 @@ describe("a notice card", () => { } as Partial); expect(box.textContent).toContain("src/b.go"); }); - it("names its speaker rather than opening as a bare paragraph", () => { + it("says its severity to a screen reader and draws it as a glyph, not a header strip", () => { const box = draw(); - expect(box.querySelector(".hl .nm")?.textContent).toBe(t("警告")); - expect(box.querySelector(".hl .tag")?.textContent).toBe(t("主机")); + expect(box.querySelector(".sr-only")?.textContent).toBe(`${t("警告")}：`); + expect(box.querySelector(".nb-ic")).not.toBeNull(); + expect(box.querySelector(".hl, .tag, .find")).toBeNull(); + expect(draw({ level: "info" }).querySelector(".nb-dot")).not.toBeNull(); }); - it("wears its severity where the gutter can follow it", () => { - expect(draw().querySelector(".call")?.getAttribute("data-lvl")).toBe("warn"); - expect(draw({ level: "error" }).querySelector(".call")?.getAttribute("data-lvl")).toBe("err"); - expect(draw({ level: "info" }).querySelector(".call")?.getAttribute("data-lvl")).toBeNull(); + it("wears its severity on the bar so the glyph ink can follow it", () => { + expect(draw().querySelector(".notice-bar")?.getAttribute("data-lvl")).toBe("warn"); + expect(draw({ level: "error" }).querySelector(".notice-bar")?.getAttribute("data-lvl")).toBe("err"); + expect(draw({ level: "error" }).querySelector(".notice-bar")?.getAttribute("role")).toBe("alert"); + expect(draw({ level: "info" }).querySelector(".notice-bar")?.getAttribute("data-lvl")).toBe("info"); }); // The kernel writes its own English for the log. A code this build has a // sentence for is the one the reader gets; only an unknown code falls back. it("shows this build's sentence for a code it knows", () => { - const said = draw().querySelector(".find .t")?.textContent ?? ""; + const said = draw().querySelector(".notice-bar .t")?.textContent ?? ""; expect(said).toContain(t("同一个检查连着几轮都报同样的结果，是否继续由你定")); expect(said).not.toContain("The same check has failed"); }); it("keeps the kernel's diagnostic as the second half", () => { - const why = draw().querySelector(".find .why")?.textContent ?? ""; + const why = draw().querySelector(".notice-bar .nwhy")?.textContent ?? ""; expect(why).toContain("3 rounds"); }); it("shows an unknown code's own English rather than nothing", () => { const box = draw({ code: "some_future_code", text: "something the kernel said" }); - expect(box.querySelector(".find .t")?.textContent).toContain("something the kernel said"); + expect(box.querySelector(".notice-bar .t")?.textContent).toContain("something the kernel said"); }); }); @@ -89,8 +92,8 @@ describe("a hand-back notice", () => { const box = draw(handBack); await waitFor(() => expect(box.querySelector(".nmd strong")).not.toBeNull(), LOADED); expect(box.querySelectorAll(".nmd ol li")).toHaveLength(2); - expect(box.querySelector(".find .why")).toBeNull(); - expect(box.querySelector(".find .t")?.textContent).toContain(t("等待你的输入")); + expect(box.querySelector(".notice-bar .nwhy")).toBeNull(); + expect(box.querySelector(".notice-bar .t")?.textContent).toContain(t("等待你的输入")); }); it("keeps it to the sanitized subset a reply gets", async () => { @@ -106,31 +109,31 @@ describe("a hand-back notice", () => { describe("a coded notice with a payload", () => { it("words the currency change from the stored value and draws no second line for it", () => { const box = draw({ code: "display_currency", text: "fee display currency set to USD (resolved: USD)", detail: "USD" }); - expect(box.querySelector(".find .t")?.textContent).toBe(t("费用显示币种已设为 {mode}", { mode: "USD" })); - expect(box.querySelector(".find .why")).toBeNull(); + expect(box.querySelector(".notice-bar .t")?.textContent).toBe(t("费用显示币种已设为 {mode}", { mode: "USD" })); + expect(box.querySelector(".notice-bar .nwhy")).toBeNull(); const auto = draw({ code: "display_currency", text: "kernel", detail: "" }); - expect(auto.querySelector(".find .t")?.textContent).toBe(t("费用显示币种已设为 {mode}", { mode: "auto" })); + expect(auto.querySelector(".notice-bar .t")?.textContent).toBe(t("费用显示币种已设为 {mode}", { mode: "auto" })); }); it("words the recovered inbox from its count and draws no raw payload line", () => { const box = draw({ level: "warn", code: "inbox_recovered", text: "Recovered 1 pending instruction(s). Inbox is paused — review with /queue before resuming.", detail: '{"count":1}' }); - expect(box.querySelector(".find .t")?.textContent).toBe(t("已恢复 {n} 条未完成的指令。待发送已暂停，请先在输入框上方的队列里查看，再点“继续派发”", { n: 1 })); - expect(box.querySelector(".find .t")?.textContent).not.toContain("Recovered"); - expect(box.querySelector(".find .why")).toBeNull(); + expect(box.querySelector(".notice-bar .t")?.textContent).toBe(t("已恢复 {n} 条未完成的指令。待发送已暂停，请先在输入框上方的队列里查看，再点“继续派发”", { n: 1 })); + expect(box.querySelector(".notice-bar .t")?.textContent).not.toContain("Recovered"); + expect(box.querySelector(".notice-bar .nwhy")).toBeNull(); }); it("keeps the kernel's English when the recovered-inbox payload is not readable", () => { const box = draw({ level: "warn", code: "inbox_recovered", text: "Recovered 2 pending instruction(s).", detail: "garbled" }); - expect(box.querySelector(".find .t")?.textContent).toContain("Recovered 2 pending"); + expect(box.querySelector(".notice-bar .t")?.textContent).toContain("Recovered 2 pending"); }); it("words the dormant permission rules from their payload and draws no raw payload line", () => { const detail = '{"rules":[{"list":"ask","rule":"rm","tool":"rm"},{"list":"deny","rule":"git reset","tool":"git reset"}]}'; const box = draw({ level: "warn", code: "permission_rules_dormant", text: "Permission rules that name no tool match nothing", detail }); - expect(box.querySelector(".find .t")?.textContent).toBe(t("有 {n} 条权限规则没有对应的工具，匹配不到任何调用，因此起不到限制作用（如「{list}」里的 {rule}）；到「设置 → 权限」里删除或改写，命令要写成 Bash(命令:*)", { n: 2, list: t("询问"), rule: "rm" })); - expect(box.querySelector(".find .why")).toBeNull(); + expect(box.querySelector(".notice-bar .t")?.textContent).toBe(t("有 {n} 条权限规则没有对应的工具，匹配不到任何调用，因此起不到限制作用（如「{list}」里的 {rule}）；到「设置 → 权限」里删除或改写，命令要写成 Bash(命令:*)", { n: 2, list: t("询问"), rule: "rm" })); + expect(box.querySelector(".notice-bar .nwhy")).toBeNull(); const garbled = draw({ level: "warn", code: "permission_rules_dormant", text: "Permission rules that name no tool match nothing", detail: "garbled" }); - expect(garbled.querySelector(".find .t")?.textContent).toContain("match nothing"); + expect(garbled.querySelector(".notice-bar .t")?.textContent).toContain("match nothing"); }); it("wraps the user's unapplied guidance in this build's sentence and keeps their words verbatim", () => { @@ -139,20 +142,20 @@ describe("a coded notice with a payload", () => { text: "Guidance was not applied because the turn ended before it could be processed. Send it again if it is still needed:\n同步最新的个人开发管理", detail: "同步最新的个人开发管理", }); - const said = box.querySelector(".find .t")?.textContent ?? ""; + const said = box.querySelector(".notice-bar .t")?.textContent ?? ""; expect(said).toContain(t("引导没有生效：这一轮在处理它之前就结束了。如果仍然需要，请再发送一次：")); expect(said).not.toContain("Guidance was not applied"); - expect(box.querySelector(".find .why")?.textContent).toBe("同步最新的个人开发管理"); + expect(box.querySelector(".notice-bar .nwhy")?.textContent).toBe("同步最新的个人开发管理"); }); }); describe("a refused-input notice", () => { it("says the message is queued rather than asking to resend it", () => { const box = draw({ level: "warn", code: "input_refused", detail: undefined, text: "input was not accepted: the session is being switched — please resend" }); - const said = box.querySelector(".find .t")?.textContent ?? ""; + const said = box.querySelector(".notice-bar .t")?.textContent ?? ""; expect(said).toBe(t("会话、工作区或运行时正在切换，这条消息没能立刻发出；它已排入队列，切换结束后会自动发送，不需要重发")); expect(said).not.toContain("resend"); - expect(box.querySelector(".find .why")).toBeNull(); + expect(box.querySelector(".notice-bar .nwhy")).toBeNull(); }); }); @@ -164,16 +167,16 @@ describe("a /compact notice", () => { text: "compaction failed: summarizer request failed: upstream said no", detail: "summary_failed", }); - const said = box.querySelector(".find .t")?.textContent ?? ""; + const said = box.querySelector(".notice-bar .t")?.textContent ?? ""; expect(said).toBe("压缩失败：生成摘要的请求失败了"); expect(said).not.toContain("upstream"); - expect(box.querySelector(".find .why")).toBeNull(); + expect(box.querySelector(".notice-bar .nwhy")).toBeNull(); }); it("words a decline from its code, and an empty code as the no-class decline", () => { - expect(draw({ code: "compact_declined", text: "x", detail: "input_unchanged" }).querySelector(".find .t")?.textContent) + expect(draw({ code: "compact_declined", text: "x", detail: "input_unchanged" }).querySelector(".notice-bar .t")?.textContent) .toBe("无需压缩：上下文自上次整理后没有变化"); - expect(draw({ code: "compact_declined", text: "x", detail: "" }).querySelector(".find .t")?.textContent) + expect(draw({ code: "compact_declined", text: "x", detail: "" }).querySelector(".notice-bar .t")?.textContent) .toBe("无需压缩：没有值得折叠的内容"); }); @@ -184,15 +187,15 @@ describe("a /compact notice", () => { text: "Automatic compaction is paused: the last attempt did not finish (summary_failed).", detail: "summary_failed", }); - const said = box.querySelector(".find .t")?.textContent ?? ""; + const said = box.querySelector(".notice-bar .t")?.textContent ?? ""; expect(said).toBe("自动压缩暂缓，上次尝试没有完成：生成摘要的请求失败了"); expect(said).not.toContain("summary_failed"); - expect(box.querySelector(".find .why")).toBeNull(); + expect(box.querySelector(".notice-bar .nwhy")).toBeNull(); }); it("keeps the kernel's text for a code this build cannot word", () => { const box = draw({ code: "compact_failed", text: "compaction failed: kernel english", detail: "future_code" }); - expect(box.querySelector(".find .t")?.textContent).toBe("compaction failed: kernel english"); + expect(box.querySelector(".notice-bar .t")?.textContent).toBe("compaction failed: kernel english"); }); it("words a skipped extension from its payload and hides the raw payload", () => { @@ -209,7 +212,7 @@ describe("a /compact notice", () => { }); describe("a background job notice", () => { - const text = (over: Partial) => draw({ level: "info", ...over }).querySelector(".find .t")?.textContent ?? ""; + const text = (over: Partial) => draw({ level: "info", ...over }).querySelector(".notice-bar .t")?.textContent ?? ""; const payload = JSON.stringify({ kind: "bash", id: "bash-126", label: "make build" }); it("is worded from its typed payload, not the kernel's English", () => { @@ -226,8 +229,8 @@ describe("a background job notice", () => { it("names the failed job and keeps its own error underneath", () => { const box = draw({ code: "job_failed", text: "background bash failed: bash-1 — boom", detail: JSON.stringify({ kind: "bash", id: "bash-1", label: "make build", error: "exit status 2" }) }); - expect(box.querySelector(".find .t")?.textContent).toBe(t("后台任务 {name} 失败，需要处理", { name: "make build" })); - expect(box.querySelector(".why")?.textContent).toBe("exit status 2"); + expect(box.querySelector(".notice-bar .t")?.textContent).toBe(t("后台任务 {name} 失败，需要处理", { name: "make build" })); + expect(box.querySelector(".nwhy")?.textContent).toBe("exit status 2"); }); it("keeps the text of a failure stored without a payload", () => { diff --git a/desktop/frontend-next/src/ui/cards/NoticeCard.tsx b/desktop/frontend-next/src/ui/cards/NoticeCard.tsx index d6e3023d79..126e631b84 100644 --- a/desktop/frontend-next/src/ui/cards/NoticeCard.tsx +++ b/desktop/frontend-next/src/ui/cards/NoticeCard.tsx @@ -2,14 +2,12 @@ import type { Item } from "../../state/session"; import { t } from "../../i18n"; import { leftoverDetail, noticeWording } from "../../i18n/notice_wording"; import { workspaceLeaseDetail } from "../../i18n/workspace_lease"; -import { Sym } from "../Sym"; +import { StudioIcon } from "../StudioIcon"; import { LazyMarkdown } from "../LazyMarkdown"; -// One of the host's cards, so it is built like the others: a gutter, a headline -// naming the speaker, and the body under it. The body keeps the kernel's two -// halves apart — the headline it wrote for a person, and the diagnostic under -// it, which is what .why is for everywhere in this UI. Severity is a colour bar -// here as in .guard and .find, and the gutter follows it too. +// A host notice is a bar, not a card: one sentence with a glyph, and under it +// the diagnostic the kernel wrote beside the headline it wrote for a person. +// Severity is the glyph's ink and nothing else — no box, no wash, no side rule. // These name a rule as their detail: a command or target, set as code. const PERMISSION = new Set(["permission_saved", "permission_covered"]); // These carry what the model wrote as their detail, so it reads the way the @@ -26,31 +24,22 @@ export function NoticeCard({ item }: { item: Extract }) { ? workspaceLeaseDetail(claim, item.code !== "workspace_lease_resumed") || item.detail : leftoverDetail(item.code, item.detail, vars); return ( -
-
- -
-
-
- {t(lvl === "err" ? "出错" : lvl === "warn" ? "警告" : "提示")} - {t("主机")} -
-
-
- - {wording ? t(wording.text, vars) : item.text} - {/* Repeats are folded rather than stacked: three identical lines - say the same thing as one and a count, and bury whatever came - before them. */} - {item.count && item.count > 1 ? ×{item.count} : null} - - {detail && (PERMISSION.has(item.code ?? "") - ? {item.detail} - : AUTHORED.has(item.code ?? "") - ?
- : {detail})} -
-
+
+ {t(lvl === "err" ? "出错" : lvl === "warn" ? "警告" : "提示")}： + {lvl ?