From 82029bb10d5d69db46bfdb878634f9d9fe9a4e29 Mon Sep 17 00:00:00 2001 From: h Date: Sat, 5 Sep 2026 00:20:03 +0200 Subject: [PATCH] fix(ui): a stream that went quiet reconnects, the rail says so, the panel opens on the master A phone that slept brings back an SSE socket that never errored and never delivers: the panel read as live for hours while showing nothing new. Every chunk now marks the connection alive and a watchdog drops the socket after 40s of silence - twice the server's keepalive - so the loop reconnects. Each attempt gets its own controller, which is what lets the watchdog (and a manual refresh) end one connection without ending the loop. The connection is admitted in one dot beside the rail's search box, tapped to reconnect at once rather than waiting out the backoff - the pill that used to say this was dropped in the redesign and the panel had no way to own up to being stale. With nothing picked, the shell lands on the freshest open master instead of a prompt to pick something: what a phone wants when the panel opens. An explicit pick, and following the active note, still win. --- ui/src/lib/api/client.ts | 5 +- ui/src/lib/api/live.svelte.ts | 108 ++++++++++++++++++++++++++-- ui/src/lib/api/sse.ts | 6 +- ui/src/lib/panel/index.svelte.ts | 5 ++ ui/src/lib/panel/panel-shell.svelte | 34 ++++++++- ui/src/lib/rail/rail.svelte | 29 ++++++++ 6 files changed, 176 insertions(+), 11 deletions(-) diff --git a/ui/src/lib/api/client.ts b/ui/src/lib/api/client.ts index 961c84a..e24cf67 100644 --- a/ui/src/lib/api/client.ts +++ b/ui/src/lib/api/client.ts @@ -148,7 +148,8 @@ export class ApiClient { async stream( path: string, onEvent: (event: BusEvent) => void, - signal: AbortSignal + signal: AbortSignal, + onActivity?: () => void ): Promise { const response = await fetch(this.url(path), { headers: this.headers({ Accept: "text/event-stream" }), @@ -162,7 +163,7 @@ export class ApiClient { payload ); } - await readSse(response, onEvent, signal); + await readSse(response, onEvent, signal, onActivity); } agents(): Promise { diff --git a/ui/src/lib/api/live.svelte.ts b/ui/src/lib/api/live.svelte.ts index 7fb8a97..de93f89 100644 --- a/ui/src/lib/api/live.svelte.ts +++ b/ui/src/lib/api/live.svelte.ts @@ -6,6 +6,11 @@ export type LiveState = "off" | "connecting" | "open" | "retrying" | "failed"; const BACKOFF_MIN_MS = 1000; const BACKOFF_MAX_MS = 30_000; const JITTER_MS = 250; +// The server sends a keepalive comment every 15s, so twice that of silence +// means the socket is dead even though nothing errored - which is exactly +// what a phone brings back from sleep: a stream that reads as open forever. +const STALE_AFTER_MS = 40_000; +const WATCH_EVERY_MS = 5000; const FATAL = new Set([401, 403, 404]); function sleep(ms: number, signal: AbortSignal): Promise { @@ -41,9 +46,13 @@ export class LiveStream { state = $state("off"); detail = $state(null); private controller: AbortController | null = null; + private inner: AbortController | null = null; private readonly client: () => ApiClient | null; private readonly path: string; private readonly options: LiveOptions; + private seen = 0; + private watcher: ReturnType | null = null; + private wake: (() => void) | null = null; constructor( client: () => ApiClient | null, @@ -65,6 +74,7 @@ export class LiveStream { } const controller = new AbortController(); this.controller = controller; + this.watch(); this.loop(controller.signal).catch((error: unknown) => { this.state = "failed"; this.detail = describe(error); @@ -72,31 +82,108 @@ export class LiveStream { } stop(): void { + this.unwatch(); this.controller?.abort(); this.controller = null; + this.inner = null; this.state = "off"; } + /** + * Reconnect now: the operator tapped the dot, or the app came back to the + * foreground. A failed stream has no loop left to nudge, so it restarts. + */ + refresh(): void { + if (!this.controller || this.state === "failed") { + this.stop(); + this.start(); + return; + } + this.state = "connecting"; + this.detail = null; + this.inner?.abort(); + this.wake?.(); + } + + /** + * A stream that stopped delivering without erroring - the phone slept, the + * network changed under it - reads as open forever. Silence past the + * server's keepalive is the only tell, so the socket is dropped and the + * loop reconnects. + */ + private watch(): void { + if (this.watcher || typeof setInterval === "undefined") { + return; + } + this.watcher = setInterval(() => { + if (this.state !== "open" || Date.now() - this.seen < STALE_AFTER_MS) { + return; + } + this.state = "retrying"; + this.detail = "stream went quiet"; + this.inner?.abort(); + }, WATCH_EVERY_MS); + } + + private unwatch(): void { + if (this.watcher) { + clearInterval(this.watcher); + this.watcher = null; + } + } + private async loop(signal: AbortSignal): Promise { let delay = BACKOFF_MIN_MS; while (!signal.aborted) { this.state = delay === BACKOFF_MIN_MS ? "connecting" : "retrying"; + // Each attempt gets its own controller, so the watchdog (and a manual + // refresh) can drop one socket without ending the loop. + const inner = new AbortController(); + this.inner = inner; + const relay = () => inner.abort(); + signal.addEventListener("abort", relay, { once: true }); // biome-ignore lint/performance/noAwaitInLoops: one connection at a time, by design - const outcome = await this.connectOnce(signal, () => { + const outcome = await this.connectOnce(inner.signal, () => { delay = BACKOFF_MIN_MS; }); + signal.removeEventListener("abort", relay); + this.inner = null; if (signal.aborted || outcome === "fatal") { return; } - await sleep(delay + Math.random() * JITTER_MS, signal); - delay = Math.min(delay * 2, BACKOFF_MAX_MS); + // A socket dropped on purpose reconnects at once; a broken one waits. + const wait = outcome === "now" ? 0 : delay + Math.random() * JITTER_MS; + await this.pause(wait, signal); + delay = + outcome === "now" + ? BACKOFF_MIN_MS + : Math.min(delay * 2, BACKOFF_MAX_MS); } } + /** Backoff that a manual refresh can cut short. */ + private pause(ms: number, signal: AbortSignal): Promise { + if (ms <= 0) { + return Promise.resolve(); + } + return new Promise((resolve) => { + const done = () => { + this.wake = null; + resolve(); + }; + this.wake = done; + sleep(ms, signal).then(() => { + if (this.wake === done) { + done(); + } + }, done); + }); + } + private async connectOnce( signal: AbortSignal, onOpen: () => void - ): Promise<"retry" | "fatal"> { + ): Promise<"retry" | "fatal" | "now"> { try { const client = this.client(); if (!client) { @@ -104,8 +191,9 @@ export class LiveStream { } await this.options.prepare?.(); if (signal.aborted) { - return "fatal"; + return this.controller ? "now" : "fatal"; } + this.seen = Date.now(); await client.stream( this.path, (event) => { @@ -117,19 +205,25 @@ export class LiveStream { } this.options.onEvent(event); }, - signal + signal, + () => { + this.seen = Date.now(); + } ); this.state = "retrying"; this.detail = "stream closed"; return "retry"; } catch (error) { if (signal.aborted) { - return "fatal"; + // Aborted while the session lives on: this was the watchdog or a + // manual refresh, so reconnect without waiting out the backoff. + return this.controller ? "now" : "fatal"; } if (error instanceof ApiError && FATAL.has(error.status)) { this.state = "failed"; this.detail = `${error.status}: ${error.message}`; this.controller = null; + this.unwatch(); return "fatal"; } this.state = "retrying"; diff --git a/ui/src/lib/api/sse.ts b/ui/src/lib/api/sse.ts index ceb1b07..b61891a 100644 --- a/ui/src/lib/api/sse.ts +++ b/ui/src/lib/api/sse.ts @@ -38,7 +38,10 @@ function drain(buffer: string, onEvent: (event: BusEvent) => void): string { export async function readSse( response: Response, onEvent: (event: BusEvent) => void, - signal?: AbortSignal + signal?: AbortSignal, + // Every chunk the socket delivers, keepalive comments included: proof the + // connection is still alive on a phone that just came back from sleep. + onActivity?: () => void ): Promise { const { body } = response; if (!body) { @@ -58,6 +61,7 @@ export async function readSse( if (done) { break; } + onActivity?.(); buffer = drain(buffer + decoder.decode(value, { stream: true }), onEvent); } } finally { diff --git a/ui/src/lib/panel/index.svelte.ts b/ui/src/lib/panel/index.svelte.ts index 30eb488..e794c6b 100644 --- a/ui/src/lib/panel/index.svelte.ts +++ b/ui/src/lib/panel/index.svelte.ts @@ -160,6 +160,11 @@ export class ConversationIndex { this.live.stop(); } + /** The dot was tapped: drop the socket, reload the rows, reconnect. */ + reconnect(): void { + this.live.refresh(); + } + private upsert(row: ConversationSummary): void { const index = this.conversations.findIndex((c) => c.id === row.id); if (index >= 0) { diff --git a/ui/src/lib/panel/panel-shell.svelte b/ui/src/lib/panel/panel-shell.svelte index eb2f190..df28c07 100644 --- a/ui/src/lib/panel/panel-shell.svelte +++ b/ui/src/lib/panel/panel-shell.svelte @@ -1,5 +1,5 @@