From 8c21ecb0f53afc12fc5af4376715b11a2be83f5e Mon Sep 17 00:00:00 2001 From: Logan Johnson Date: Fri, 4 Sep 2026 22:44:35 -0400 Subject: [PATCH 1/4] fix(desktop): diagnose and explicitly recover lifecycle receivers Signed-off-by: Logan Johnson --- desktop/src/features/agents/AGENTS.md | 7 +- .../features/agents/desktopLifecycle.test.mjs | 136 +++++++++++++ .../src/features/agents/desktopLifecycle.ts | 191 +++++++++++------- .../agents/desktopLifecycleDiagnostics.ts | 41 ++++ .../ui/DesktopLifecycleControl.test.mjs | 23 ++- .../agents/ui/DesktopLifecycleControl.tsx | 28 ++- .../e2e/top-chrome-zoom-clearance.spec.ts | 4 +- 7 files changed, 348 insertions(+), 82 deletions(-) create mode 100644 desktop/src/features/agents/desktopLifecycleDiagnostics.ts diff --git a/desktop/src/features/agents/AGENTS.md b/desktop/src/features/agents/AGENTS.md index 020be239764..e78fb43b26f 100644 --- a/desktop/src/features/agents/AGENTS.md +++ b/desktop/src/features/agents/AGENTS.md @@ -308,7 +308,12 @@ with a TypeScript lookup table or an id comparison in a component. Known Desktops exposes an owner-private, explicitly selected agent+Desktop Stop, not inferred agent location. The app-scoped receiver subscribes live only; -reopening never replays commands. An explicit retry republishes the exact request; +reopening never replays commands. Receiver initialization reports a safe failure +stage without exposing raw transport/IPC exceptions. Its scope-owned notification +can explicitly retry the **receiver** with a fresh live-only subscription; that +must discard queued callbacks from the retired receiver, not retry an operation. +A readiness timeout is unconfirmed delivery, not a failed initialization; late +EOSE clears that warning after successful projection. An explicit operation retry republishes the exact request; the relay redelivers stored Stop duplicates without repeating relay side effects. The receiver returns saved results or Unknown, never repeats a consumed Stop. Native owner-delegation and community checks diff --git a/desktop/src/features/agents/desktopLifecycle.test.mjs b/desktop/src/features/agents/desktopLifecycle.test.mjs index b15daccfcff..33de28d85ba 100644 --- a/desktop/src/features/agents/desktopLifecycle.test.mjs +++ b/desktop/src/features/agents/desktopLifecycle.test.mjs @@ -238,3 +238,139 @@ test("receiver projects history without executing it and invalidates live work o assert.equal(f.errors.length, 1); close(); }); + +for (const stage of [ + "subscription", + "history", + "projection", + "reconciliation", +]) { + test(`receiver reports safe ${stage} failure and never admits queued work`, async () => { + const f = fixture(); + const secret = "private key /home/private bearer secret"; + if (stage === "subscription") + f.relay.subscribeLive = async () => { + throw Error(secret); + }; + if (stage === "history") + f.relay.fetchEvents = async () => { + throw Error(secret); + }; + const ipc = async (command, args) => { + if ( + command === "observe_desktop_placement" && + ((stage === "projection" && !args.reconcile) || + (stage === "reconciliation" && args.reconcile)) + ) + throw Error(secret); + return f.ipc(command, args); + }; + const started = receiveLifecycle( + scope, + () => true, + (e) => f.errors.push(e), + ipc, + f.relay, + ); + f.deliver({ id: "queued", kind: 50182 }); + await assert.rejects(started, (error) => { + assert.match(error.message, new RegExp(`${stage}: request failed`)); + assert.doesNotMatch(error.message, /private|bearer|secret/); + return true; + }); + await tick(); + assert.equal( + f.calls.filter(([c]) => c === "receive_desktop_lifecycle").length, + 0, + ); + assert.deepEqual( + f.errors, + [], + "discarded callbacks cannot replace the startup diagnosis", + ); + }); +} + +test("explicit receiver recovery discards old queued work and projects history without replay", async () => { + const f = fixture(); + let rejectHistory; + f.relay.fetchEvents = () => + new Promise((_, reject) => { + rejectHistory = reject; + }); + const failed = receiveLifecycle( + scope, + () => true, + () => {}, + f.ipc, + f.relay, + ); + await tick(); + f.deliver({ id: "old-live", kind: 50182 }); + rejectHistory(Error("Timed out while loading channel history.")); + await assert.rejects(failed, /history: history timed out/); + f.relay.fetchEvents = async () => [{ id: "old-live", kind: 50182 }]; + const close = await receiveLifecycle( + scope, + () => true, + () => {}, + f.ipc, + f.relay, + ); + await tick(); + assert.equal( + f.calls.filter(([c]) => c === "receive_desktop_lifecycle").length, + 0, + ); + f.deliver({ id: "new-live", kind: 50182 }); + await tick(); + assert.equal( + f.calls.filter(([c]) => c === "receive_desktop_lifecycle").length, + 1, + ); + f.deliver({ id: "closed-queue", kind: 50182 }); + close(); + await tick(); + assert.equal( + f.calls.filter(([c]) => c === "receive_desktop_lifecycle").length, + 1, + ); +}); + +test("readiness timeout is distinct, late EOSE recovers, CLOSED retires old callbacks", async () => { + const f = fixture(); + let notify, deliver; + let closed = 0, + ready = 0; + f.relay.subscribeLive = async (_filter, event, onReady, timeout) => { + assert.equal(timeout, 5000); + deliver = event; + notify = onReady; + notify("timeout"); + return () => { + closed++; + }; + }; + const close = await receiveLifecycle( + scope, + () => true, + (e) => f.errors.push(e), + f.ipc, + f.relay, + () => ready++, + ); + assert.match(f.errors[0], /readiness timed out/); + assert.equal(ready, 0); + notify("eose"); + assert.equal(ready, 1); + notify("closed"); + deliver({ id: "late", kind: 50182 }); + await tick(); + assert.equal(closed, 1); + assert.equal( + f.calls.filter(([c]) => c === "receive_desktop_lifecycle").length, + 0, + ); + assert.match(f.errors.at(-1), /subscription closed/); + close(); +}); diff --git a/desktop/src/features/agents/desktopLifecycle.ts b/desktop/src/features/agents/desktopLifecycle.ts index 7af772aa181..1fb321c67ac 100644 --- a/desktop/src/features/agents/desktopLifecycle.ts +++ b/desktop/src/features/agents/desktopLifecycle.ts @@ -9,6 +9,11 @@ import { sendStop, } from "./desktopStop"; +import { + LifecycleReceiverError, + receiverStep, +} from "./desktopLifecycleDiagnostics"; + export const DESKTOP_LIFECYCLE = 50182; export const DESKTOP_LIFECYCLE_RESULT = 50183; export type LifecycleAction = "start" | "restart" | "status"; @@ -99,27 +104,33 @@ export function lifecycleClient( let before_id: string | undefined; for (let page = 0; page < 64; page++) { check(); - const events = await relay.fetchEvents({ - kinds: [DESKTOP_STOP, DESKTOP_LIFECYCLE], - authors: [scope.owner], - limit: 256, - until, - before_id, - }); + const events = await receiverStep("history", () => + relay.fetchEvents({ + kinds: [DESKTOP_STOP, DESKTOP_LIFECYCLE], + authors: [scope.owner], + limit: 256, + until, + before_id, + }), + ); check(); // No effects while a partial page could still hide a dominating Start. - await ipc("observe_desktop_placement", { - ...scope, - events, - reconcile: false, - }); + await receiverStep("projection", () => + ipc("observe_desktop_placement", { + ...scope, + events, + reconcile: false, + }), + ); check(); if (events.length < 256) { - await ipc("observe_desktop_placement", { - ...scope, - events: [], - reconcile: true, - }); + await receiverStep("reconciliation", () => + ipc("observe_desktop_placement", { + ...scope, + events: [], + reconcile: true, + }), + ); check(); return; } @@ -250,14 +261,21 @@ export function lifecycleClient( } /** Subscribe first, then project history; live commands wait for complete - * initialization. Reconnect gets a new client epoch and never replays history. */ + * initialization. Explicit receiver retry starts a fresh live-only subscription, + * never retries an operation or executes historical commands. */ export async function receiveLifecycle( scope: DesktopScope, active: () => boolean, onError: (message: string) => void, ipc = invoke, relay = relayClient, + onReady: () => void = () => {}, ) { + let stopped = false; + let stopSubscription = () => {}; + let synced = false; + let subscriptionReady = false; + const valid = () => active() && !stopped; let client: ReturnType; let initialized: () => void = () => {}; const ready = new Promise((resolve) => { @@ -265,70 +283,99 @@ export async function receiveLifecycle( }); let chain = Promise.resolve(); let pending = 0; - const close = await relay.subscribeLive( - { - kinds: [DESKTOP_LIFECYCLE, DESKTOP_STOP], - authors: [scope.owner], - limit: 0, - }, - (event) => { - if (!active()) return; - if (pending >= 16) { - onError("Desktop lifecycle receiver is busy; outcome is unconfirmed."); - return; - } - pending++; - chain = chain - .then(async () => { - await ready; - client.check(); - await ipc("observe_desktop_placement", { - ...scope, - events: [event], - reconcile: true, - }); - client.check(); - const result = await ipc( - event.kind === DESKTOP_STOP - ? "receive_desktop_stop" - : "receive_desktop_lifecycle", - { ...scope, event }, + const unsubscribe = await receiverStep("subscription", () => + relay.subscribeLive( + { + kinds: [DESKTOP_LIFECYCLE, DESKTOP_STOP], + authors: [scope.owner], + limit: 0, + }, + (event) => { + if (!valid()) return; + if (pending >= 16) { + onError( + "Desktop lifecycle receiver is busy; outcome is unconfirmed.", ); - client.check(); - if (result) - await relay.publishEvent( - result, - "Result delivery unconfirmed", - "Result delivery failed", - client.check, + return; + } + pending++; + chain = chain + .then(async () => { + await ready; + if (!valid()) return; + client.check(); + await ipc("observe_desktop_placement", { + ...scope, + events: [event], + reconcile: true, + }); + client.check(); + const result = await ipc( + event.kind === DESKTOP_STOP + ? "receive_desktop_stop" + : "receive_desktop_lifecycle", + { ...scope, event }, ); - }) - .catch(() => { - if (active()) - onError( - "Desktop lifecycle result is unconfirmed. No automatic operation retry.", - ); - }) - .finally(() => { - pending--; - }); - }, - (readiness) => { - if (active() && readiness !== "eose") - onError("Desktop lifecycle receiver is unavailable."); - }, + client.check(); + if (result) + await relay.publishEvent( + result, + "Result delivery unconfirmed", + "Result delivery failed", + client.check, + ); + }) + .catch(() => { + if (valid()) + onError( + "Desktop lifecycle result is unconfirmed. No automatic operation retry.", + ); + }) + .finally(() => { + pending--; + }); + }, + (readiness) => { + if (!valid()) return; + subscriptionReady = readiness === "eose"; + if (readiness === "closed") { + stopped = true; + stopSubscription(); + onError( + "Desktop lifecycle receiver subscription closed. Retry the receiver to accept new requests.", + ); + } else if (readiness === "timeout") { + onError( + "Desktop lifecycle subscription readiness timed out. Delivery is unconfirmed.", + ); + } else if (synced) onReady(); + }, + 5000, + ), ); - client = lifecycleClient(scope, active, ipc, relay); + const close = () => { + stopped = true; + // Unsubscribe may fail on a dead socket; it must not revive this receiver + // or leave an unhandled promise. A retry always owns a new subscription. + void Promise.resolve() + .then(unsubscribe) + .catch(() => {}); + }; + stopSubscription = close; + client = lifecycleClient(scope, valid, ipc, relay); try { + if (stopped) throw new LifecycleReceiverError("subscription", "closed"); await client.sync(); client.check(); + synced = true; initialized(); + if (subscriptionReady) onReady(); } catch (error) { close(); - // Release queued callbacks into a permanently invalidated client. - client = lifecycleClient(scope, () => false, ipc, relay); initialized(); - throw error; + throw error instanceof LifecycleReceiverError + ? error + : new LifecycleReceiverError("initialization", error); } return close; } diff --git a/desktop/src/features/agents/desktopLifecycleDiagnostics.ts b/desktop/src/features/agents/desktopLifecycleDiagnostics.ts new file mode 100644 index 00000000000..aebe01c41a9 --- /dev/null +++ b/desktop/src/features/agents/desktopLifecycleDiagnostics.ts @@ -0,0 +1,41 @@ +/** Safe receiver diagnostics: raw IPC/transport errors can contain private data. */ +type Stage = + | "initialization" + | "subscription" + | "history" + | "projection" + | "reconciliation"; + +export class LifecycleReceiverError extends Error { + constructor(stage: Stage, error: unknown) { + const value = error instanceof Error ? error.message : error; + const reason = + value === "closed" + ? "subscription closed" + : value === "Desktop lifecycle scope changed" + ? "scope changed" + : value === "Relay session is terminal; cannot reconnect." + ? "relay session requires reconnection" + : value === "Timed out while loading channel history." + ? "history timed out" + : "request failed"; + super(`Desktop lifecycle receiver is unavailable (${stage}: ${reason}).`); + } +} + +export function receiverErrorMessage(error: unknown): string { + return error instanceof LifecycleReceiverError + ? error.message + : "Desktop lifecycle receiver is unavailable (initialization failed)."; +} + +export async function receiverStep( + stage: Stage, + action: () => Promise, +): Promise { + try { + return await action(); + } catch (error) { + throw new LifecycleReceiverError(stage, error); + } +} diff --git a/desktop/src/features/agents/ui/DesktopLifecycleControl.test.mjs b/desktop/src/features/agents/ui/DesktopLifecycleControl.test.mjs index f41068c6136..3ac174ac536 100644 --- a/desktop/src/features/agents/ui/DesktopLifecycleControl.test.mjs +++ b/desktop/src/features/agents/ui/DesktopLifecycleControl.test.mjs @@ -157,6 +157,8 @@ test("receiver failure is a scope-owned notification, not pre-shell layout", asy let closed = 0; let rejectLate; let delayed = false; + let storageAvailable = false; + let subscribed = 0; relayClient.fetchEvents = async () => { if (delayed) return new Promise((_, reject) => { @@ -165,13 +167,16 @@ test("receiver failure is a scope-owned notification, not pre-shell layout", asy return []; }; relayClient.subscribeLive = async (_filter, _event, onReadiness) => { + subscribed++; readiness = onReadiness; + onReadiness("eose"); return () => { closed++; }; }; window.__TAURI_INTERNALS__ = { invoke: async () => { + if (storageAvailable) return; throw new Error("fixture: storage unavailable"); }, }; @@ -193,7 +198,7 @@ test("receiver failure is a scope-owned notification, not pre-shell layout", asy assert.equal(warnings().length, 1); assert.equal( warnings()[0].title, - "Desktop lifecycle receiver is unavailable.", + "Desktop lifecycle receiver is unavailable (projection: request failed).", ); assert.equal(warnings()[0].duration, Infinity); assert.equal(warnings()[0].closeButton, true); @@ -203,6 +208,16 @@ test("receiver failure is a scope-owned notification, not pre-shell layout", asy 1, "repeated failures update one notification", ); + const retryAction = warnings()[0].action; + assert.equal(retryAction.label, "Retry receiver"); + storageAvailable = true; + await React.act(async () => retryAction.onClick()); + assert.equal(subscribed, 2, "explicit recovery starts a new live receiver"); + assert.equal( + warnings().length, + 0, + "successful recovery removes its warning", + ); await React.act(async () => root.render( React.createElement(DesktopLifecycleReceiver, { scope: null }), @@ -227,7 +242,11 @@ test("receiver failure is a scope-owned notification, not pre-shell layout", asy 0, "late startup rejection must not recreate the warning", ); - assert.equal(closed, 2, "both failed subscriptions are released"); + assert.equal( + closed, + 3, + "failed, recovered and retired subscriptions are released", + ); } finally { await React.act(async () => root.unmount()); relayClient.fetchEvents = originals.fetch; diff --git a/desktop/src/features/agents/ui/DesktopLifecycleControl.tsx b/desktop/src/features/agents/ui/DesktopLifecycleControl.tsx index d05efee2511..0b6b63d67be 100644 --- a/desktop/src/features/agents/ui/DesktopLifecycleControl.tsx +++ b/desktop/src/features/agents/ui/DesktopLifecycleControl.tsx @@ -9,6 +9,7 @@ import { type LifecycleOutcome, } from "../desktopLifecycle"; import { useRelayAgentsQuery } from "../hooks"; +import { receiverErrorMessage } from "../desktopLifecycleDiagnostics"; export function DesktopLifecycleReceiver({ scope, @@ -16,6 +17,7 @@ export function DesktopLifecycleReceiver({ scope: DesktopScope | null; }) { const { owner, community } = scope ?? {}; + const [attempt, retry] = useState(0); useEffect(() => { if (!owner || !community) return; let active = true; @@ -30,22 +32,40 @@ export function DesktopLifecycleReceiver({ id: notification, duration: Infinity, closeButton: true, + action: { + label: "Retry receiver", + onClick: () => { + if (!active) return; + active = false; + close?.(); + retry(attempt + 1); + }, + }, }); }; - void receiveLifecycle({ owner, community }, () => active, reportError) + void receiveLifecycle( + { owner, community }, + () => active, + reportError, + undefined, + undefined, + () => { + if (active && notification !== undefined) toast.dismiss(notification); + }, + ) .then((fn) => { if (active) close = fn; else fn(); }) - .catch(() => { - reportError("Desktop lifecycle receiver is unavailable."); + .catch((error) => { + reportError(receiverErrorMessage(error)); }); return () => { active = false; close?.(); if (notification !== undefined) toast.dismiss(notification); }; - }, [owner, community]); + }, [owner, community, attempt]); return null; } function message(outcome: LifecycleOutcome) { diff --git a/desktop/tests/e2e/top-chrome-zoom-clearance.spec.ts b/desktop/tests/e2e/top-chrome-zoom-clearance.spec.ts index 1d4d6795a76..97d6b69cb1f 100644 --- a/desktop/tests/e2e/top-chrome-zoom-clearance.spec.ts +++ b/desktop/tests/e2e/top-chrome-zoom-clearance.spec.ts @@ -118,9 +118,7 @@ test.describe("top chrome macOS traffic-light clearance under text zoom", () => // A failed global receiver must remain visible without entering the shell's // layout flow. This also forces the error to settle before measuring chrome. await expect( - page.getByText("Desktop lifecycle receiver is unavailable.", { - exact: true, - }), + page.getByText(/Desktop lifecycle receiver is unavailable/), ).toBeVisible(); // Lock the native and webview placements together: removing this explicit From 7fdaa65b9710d806f94a4b91e334f7a0d4e9bac4 Mon Sep 17 00:00:00 2001 From: Logan Johnson Date: Fri, 4 Sep 2026 23:03:24 -0400 Subject: [PATCH 2/4] fix(desktop): observe receiver closure after subscription readiness Signed-off-by: Logan Johnson --- .../features/agents/desktopLifecycle.test.mjs | 11 +- .../src/features/agents/desktopLifecycle.ts | 34 ++- .../ui/DesktopLifecycleControl.test.mjs | 12 +- desktop/src/shared/api/relayClientSession.ts | 25 +- desktop/src/shared/api/relayClientShared.ts | 15 + desktop/src/shared/api/relayClosedRecovery.ts | 17 +- .../api/relayLiveSubscriptionState.test.mjs | 268 ++++++++++++++++++ 7 files changed, 355 insertions(+), 27 deletions(-) create mode 100644 desktop/src/shared/api/relayLiveSubscriptionState.test.mjs diff --git a/desktop/src/features/agents/desktopLifecycle.test.mjs b/desktop/src/features/agents/desktopLifecycle.test.mjs index 33de28d85ba..2a0ff0f314d 100644 --- a/desktop/src/features/agents/desktopLifecycle.test.mjs +++ b/desktop/src/features/agents/desktopLifecycle.test.mjs @@ -342,10 +342,17 @@ test("readiness timeout is distinct, late EOSE recovers, CLOSED retires old call let notify, deliver; let closed = 0, ready = 0; - f.relay.subscribeLive = async (_filter, event, onReady, timeout) => { + f.relay.subscribeLive = async ( + _filter, + event, + _onReady, + timeout, + options, + ) => { assert.equal(timeout, 5000); deliver = event; - notify = onReady; + assert.equal(options.closedRecovery, "explicit"); + notify = options.onState; notify("timeout"); return () => { closed++; diff --git a/desktop/src/features/agents/desktopLifecycle.ts b/desktop/src/features/agents/desktopLifecycle.ts index 1fb321c67ac..b81e4826052 100644 --- a/desktop/src/features/agents/desktopLifecycle.ts +++ b/desktop/src/features/agents/desktopLifecycle.ts @@ -335,22 +335,26 @@ export async function receiveLifecycle( pending--; }); }, - (readiness) => { - if (!valid()) return; - subscriptionReady = readiness === "eose"; - if (readiness === "closed") { - stopped = true; - stopSubscription(); - onError( - "Desktop lifecycle receiver subscription closed. Retry the receiver to accept new requests.", - ); - } else if (readiness === "timeout") { - onError( - "Desktop lifecycle subscription readiness timed out. Delivery is unconfirmed.", - ); - } else if (synced) onReady(); - }, + undefined, 5000, + { + closedRecovery: "explicit", + onState: (readiness) => { + if (!valid()) return; + subscriptionReady = readiness === "eose"; + if (readiness === "closed") { + stopped = true; + stopSubscription(); + onError( + "Desktop lifecycle receiver subscription closed. Retry the receiver to accept new requests.", + ); + } else if (readiness === "timeout") { + onError( + "Desktop lifecycle subscription readiness timed out. Delivery is unconfirmed.", + ); + } else if (synced) onReady(); + }, + }, ), ); const close = () => { diff --git a/desktop/src/features/agents/ui/DesktopLifecycleControl.test.mjs b/desktop/src/features/agents/ui/DesktopLifecycleControl.test.mjs index 3ac174ac536..35e52f7aab0 100644 --- a/desktop/src/features/agents/ui/DesktopLifecycleControl.test.mjs +++ b/desktop/src/features/agents/ui/DesktopLifecycleControl.test.mjs @@ -166,10 +166,16 @@ test("receiver failure is a scope-owned notification, not pre-shell layout", asy }); return []; }; - relayClient.subscribeLive = async (_filter, _event, onReadiness) => { + relayClient.subscribeLive = async ( + _filter, + _event, + _onReadiness, + _timeout, + options, + ) => { subscribed++; - readiness = onReadiness; - onReadiness("eose"); + readiness = options.onState; + readiness("eose"); return () => { closed++; }; diff --git a/desktop/src/shared/api/relayClientSession.ts b/desktop/src/shared/api/relayClientSession.ts index 2f94a65bf9b..a7c0bded02d 100644 --- a/desktop/src/shared/api/relayClientSession.ts +++ b/desktop/src/shared/api/relayClientSession.ts @@ -16,6 +16,7 @@ import { getTextPayload, toRelayFrames, type ConnectionState, + type LiveSubscriptionOptions, type LiveSubscriptionReadiness, type PendingEvent, type RelaySubscription, @@ -419,8 +420,15 @@ export class RelayClient { onEvent: (event: RelayEvent) => void, onReady?: (readiness: LiveSubscriptionReadiness) => void, readinessTimeoutMs?: number, + options?: LiveSubscriptionOptions, ) { - return this.subscribe(filter, onEvent, onReady, readinessTimeoutMs); + return this.subscribe( + filter, + onEvent, + onReady, + readinessTimeoutMs, + options, + ); } async subscribeToChannelMentionEvents( channelId: string, @@ -609,6 +617,7 @@ export class RelayClient { onEvent: (event: RelayEvent) => void, onReady?: (readiness: LiveSubscriptionReadiness) => void, readinessTimeoutMs = 250, + options: LiveSubscriptionOptions = {}, ) { await this.ensureConnected(); @@ -621,16 +630,18 @@ export class RelayClient { resolve(); }; }); - const fallbackTimeout = window.setTimeout( - () => resolveReady("timeout"), - readinessTimeoutMs, - ); + const fallbackTimeout = window.setTimeout(() => { + options.onState?.("timeout"); + resolveReady("timeout"); + }, readinessTimeoutMs); this.subscriptions.set(subId, { mode: "live", filter, onEvent, resolveReady, + onState: options.onState, + closedRecovery: options.closedRecovery ?? "shared", }); try { @@ -1066,8 +1077,12 @@ export class RelayClient { continue; } subscription.resolveReady?.("closed"); + subscription.onState?.("closed"); subscription.resolveReady = undefined; clearClosedRetry(subscription); + if (subscription.closedRecovery === "explicit") { + this.subscriptions.delete(subId); + } } for (const [eventId, pendingEvent] of this.pendingEvents) { window.clearTimeout(pendingEvent.timeout); diff --git a/desktop/src/shared/api/relayClientShared.ts b/desktop/src/shared/api/relayClientShared.ts index 8bd6379d7a9..925c74b573d 100644 --- a/desktop/src/shared/api/relayClientShared.ts +++ b/desktop/src/shared/api/relayClientShared.ts @@ -70,11 +70,26 @@ type FirstEventSubscription = { export type LiveSubscriptionReadiness = "eose" | "closed" | "timeout"; +/** + * Optional lifecycle policy for a live subscription. + * + * Most subscriptions keep the shared reconnect/CLOSED recovery behavior. A + * command receiver can instead request explicit recovery: every CLOSED retires + * that subscription and `onState` remains observable after initial EOSE so the + * owning UI can offer a deliberate fresh subscription. + */ +export type LiveSubscriptionOptions = { + onState?: (state: LiveSubscriptionReadiness) => void; + closedRecovery?: "shared" | "explicit"; +}; + type LiveSubscription = { mode: "live"; filter: RelaySubscriptionFilter; onEvent: (event: RelayEvent) => void; resolveReady?: (readiness: LiveSubscriptionReadiness) => void; + onState?: (state: LiveSubscriptionReadiness) => void; + closedRecovery: "shared" | "explicit"; lastSeenCreatedAt?: number; /** * Lower bound of a reconnect backfill window that has not yet completed. diff --git a/desktop/src/shared/api/relayClosedRecovery.ts b/desktop/src/shared/api/relayClosedRecovery.ts index 9d30a233e7b..8f1a68287d2 100644 --- a/desktop/src/shared/api/relayClosedRecovery.ts +++ b/desktop/src/shared/api/relayClosedRecovery.ts @@ -124,10 +124,24 @@ function recoverLiveSubscriptionFromClosed({ sendReq: (subId: string, filter: RelaySubscriptionFilter) => Promise; }) { subscription.resolveReady?.("closed"); + subscription.onState?.("closed"); subscription.resolveReady = undefined; const closedClass = classifyRelayClosed(message); + if (closedClass === "rate-limited") { + const hintSeconds = parseRateLimitHint(message); + activateRateLimit(hintSeconds); + } + + if (subscription.closedRecovery === "explicit") { + // Command receivers must not survive CLOSED into shared re-subscription. + // Their owner presents an explicit fresh-receiver action instead. + clearClosedRetry(subscription); + subscriptions.delete(subId); + return; + } + if (closedClass === "terminal") { // Auth/access/filter failure — permanently remove the subscription so it // doesn't silently loop. @@ -146,9 +160,7 @@ function recoverLiveSubscriptionFromClosed({ let delayMs = backoffMs; if (closedClass === "rate-limited") { - // Activate the gate so concurrent operations back off too. const hintSeconds = parseRateLimitHint(message); - activateRateLimit(hintSeconds); // Use the gate's actual remaining time so a shorter hint arriving under a // longer active gate does not schedule a premature retry that just gets // another CLOSED. The fallback covers the gate-inactive edge case @@ -259,6 +271,7 @@ export function handleSubscriptionEose({ if (generation !== undefined) markReconnectLiveEose(subscription, generation); subscription.resolveReady?.("eose"); + subscription.onState?.("eose"); subscription.resolveReady = undefined; subscription.closedRetryAttempt = 0; clearClosedRetry(subscription); diff --git a/desktop/src/shared/api/relayLiveSubscriptionState.test.mjs b/desktop/src/shared/api/relayLiveSubscriptionState.test.mjs new file mode 100644 index 00000000000..587bfae1f28 --- /dev/null +++ b/desktop/src/shared/api/relayLiveSubscriptionState.test.mjs @@ -0,0 +1,268 @@ +import assert from "node:assert/strict"; +import test from "node:test"; + +let fakeNow = 0; +let nextTimerId = 1; +const pendingTimers = new Map(); +const sentFrames = []; + +globalThis.window = { + setTimeout: (fn, ms) => { + const id = nextTimerId++; + pendingTimers.set(id, { fn, fireAt: fakeNow + ms }); + return id; + }, + clearTimeout: (id) => pendingTimers.delete(id), + __TAURI_INTERNALS__: { + invoke: async (command, args) => { + if (command === "plugin:websocket|send") sentFrames.push(args); + }, + }, +}; +Date.now = () => fakeNow; + +const { RelayClient } = await import("./relayClientSession.ts"); +const { receiveLifecycle } = await import("@/features/agents/desktopLifecycle"); +const { resetRateLimitGate } = await import("./relayRateLimitGate.ts"); + +function resetHarness() { + fakeNow = 0; + nextTimerId = 1; + pendingTimers.clear(); + sentFrames.length = 0; + resetRateLimitGate(); +} + +function connectedClient() { + const client = new RelayClient(); + client.wsId = 7; + return client; +} + +function sentProtocolFrames(type) { + return sentFrames + .map(({ message }) => JSON.parse(message.data)) + .filter((frame) => frame[0] === type); +} + +async function flushUntil(predicate, attempts = 40) { + for (let attempt = 0; attempt < attempts; attempt++) { + if (predicate()) return; + await Promise.resolve(); + } + assert.fail("condition did not become true before the microtask limit"); +} + +async function flushMicrotasks(attempts = 10) { + for (let attempt = 0; attempt < attempts; attempt++) await Promise.resolve(); +} + +function tickTo(time) { + fakeNow = time; + for (;;) { + const due = [...pendingTimers.entries()].filter( + ([, timer]) => timer.fireAt <= fakeNow, + ); + if (!due.length) return; + for (const [id, timer] of due) { + if (!pendingTimers.delete(id)) continue; + timer.fn(); + } + } +} + +function deliver(client, frame) { + return client.handleWsMessage( + { type: "Text", data: JSON.stringify(frame) }, + client.connectionGeneration, + ); +} + +async function openLive(client, options, onEvent = () => {}, onReady) { + const opened = client.subscribeLive( + { kinds: [50182, 50180], authors: ["owner"], limit: 0 }, + onEvent, + onReady, + 5000, + options, + ); + await flushUntil(() => sentProtocolFrames("REQ").length > 0); + const subId = sentProtocolFrames("REQ").at(-1)[1]; + return { opened, subId }; +} + +test("persistent state reports timeout, late EOSE, then CLOSED through RelayClient", async () => { + resetHarness(); + const client = connectedClient(); + const states = []; + const readiness = []; + const { opened, subId } = await openLive( + client, + { closedRecovery: "explicit", onState: (state) => states.push(state) }, + () => {}, + (state) => readiness.push(state), + ); + + tickTo(5000); + const close = await opened; + assert.deepEqual(states, ["timeout"]); + assert.deepEqual(readiness, ["timeout"]); + + await deliver(client, ["EOSE", subId]); + await deliver(client, ["CLOSED", subId, "restricted: access revoked"]); + + assert.deepEqual(states, ["timeout", "eose", "closed"]); + assert.deepEqual(readiness, ["timeout", "eose"]); + assert.equal(client.subscriptions.has(subId), false); + await close(); +}); + +for (const [label, message] of [ + ["terminal", "restricted: access revoked"], + ["retryable", "error: storage temporarily unavailable"], +]) { + test(`explicit recovery retires an EOSE-ready ${label} CLOSED without re-REQ`, async () => { + resetHarness(); + const client = connectedClient(); + const states = []; + const { opened, subId } = await openLive(client, { + closedRecovery: "explicit", + onState: (state) => states.push(state), + }); + await deliver(client, ["EOSE", subId]); + const close = await opened; + + await deliver(client, ["CLOSED", subId, message]); + tickTo(60_000); + await Promise.resolve(); + + assert.deepEqual(states, ["eose", "closed"]); + assert.equal(client.subscriptions.has(subId), false); + assert.equal(sentProtocolFrames("REQ").length, 1); + await close(); + }); +} + +test("connection reset reports CLOSED and retires only explicit-recovery subscriptions", async () => { + resetHarness(); + const client = connectedClient(); + const explicitStates = []; + const explicit = await openLive(client, { + closedRecovery: "explicit", + onState: (state) => explicitStates.push(state), + }); + await deliver(client, ["EOSE", explicit.subId]); + const closeExplicit = await explicit.opened; + + const ordinary = await openLive(client); + await deliver(client, ["EOSE", ordinary.subId]); + const closeOrdinary = await ordinary.opened; + + client.resetConnection(new Error("fixture connection reset")); + + assert.deepEqual(explicitStates, ["eose", "closed"]); + assert.equal(client.subscriptions.has(explicit.subId), false); + assert.equal( + client.subscriptions.has(ordinary.subId), + true, + "default live subscribers retain shared reconnect recovery", + ); + + await closeExplicit(); + await closeOrdinary(); + client.disconnect(); +}); + +test("default live subscribers retain shared retry after retryable CLOSED", async () => { + resetHarness(); + const client = connectedClient(); + const ordinary = await openLive(client); + await deliver(client, ["EOSE", ordinary.subId]); + const close = await ordinary.opened; + + await deliver(client, [ + "CLOSED", + ordinary.subId, + "error: storage temporarily unavailable", + ]); + assert.equal(client.subscriptions.has(ordinary.subId), true); + + tickTo(1000); + await flushUntil(() => sentProtocolFrames("REQ").length === 2); + assert.equal(client.subscriptions.has(ordinary.subId), true); + + await close(); +}); + +test("real explicit CLOSED fences queued lifecycle work and a deliberate retry is fresh", async () => { + resetHarness(); + const client = connectedClient(); + let resolveHistory; + client.fetchEvents = () => + new Promise((resolve) => { + resolveHistory = resolve; + }); + const ipcCalls = []; + const ipc = async (command) => { + ipcCalls.push(command); + return null; + }; + const errors = []; + const receiving = receiveLifecycle( + { owner: "owner", community: "wss://one.example" }, + () => true, + (error) => errors.push(error), + ipc, + client, + ); + + await flushUntil(() => sentProtocolFrames("REQ").length === 1); + const retiredSubId = sentProtocolFrames("REQ")[0][1]; + await deliver(client, ["EOSE", retiredSubId]); + await flushUntil(() => resolveHistory !== undefined); + await deliver(client, [ + "EVENT", + retiredSubId, + { id: "queued-old", kind: 50182, created_at: 1 }, + ]); + tickTo(20); + await deliver(client, ["CLOSED", retiredSubId, "restricted: access revoked"]); + resolveHistory([]); + + await assert.rejects(receiving, /receiver is unavailable/); + await flushMicrotasks(); + assert.equal( + ipcCalls.includes("receive_desktop_lifecycle"), + false, + "queued work from the retired receiver must not execute", + ); + assert.match(errors.at(-1), /subscription closed/); + + client.fetchEvents = async () => []; + const retried = receiveLifecycle( + { owner: "owner", community: "wss://one.example" }, + () => true, + (error) => errors.push(error), + ipc, + client, + ); + await flushUntil(() => sentProtocolFrames("REQ").length === 2); + const freshSubId = sentProtocolFrames("REQ")[1][1]; + assert.notEqual(freshSubId, retiredSubId); + await deliver(client, ["EOSE", freshSubId]); + const close = await retried; + + await deliver(client, [ + "EVENT", + freshSubId, + { id: "new-live", kind: 50182, created_at: 2 }, + ]); + tickTo(40); + await flushUntil(() => ipcCalls.includes("receive_desktop_lifecycle")); + assert.equal( + ipcCalls.filter((command) => command === "receive_desktop_lifecycle") + .length, + 1, + ); + await close(); +}); From a2380832e656db01b5cc17137cc94eca5ee51be5 Mon Sep 17 00:00:00 2001 From: Logan Johnson Date: Fri, 4 Sep 2026 23:40:53 -0400 Subject: [PATCH 3/4] fix(desktop): bound lifecycle receiver recovery without command replay Signed-off-by: Logan Johnson --- desktop/src/features/agents/AGENTS.md | 10 +- .../src/features/agents/desktopLifecycle.ts | 18 +- .../agents/desktopLifecycleReceiver.test.mjs | 443 ++++++++++++++++++ .../agents/desktopLifecycleReceiver.ts | 142 ++++++ .../ui/DesktopLifecycleControl.test.mjs | 19 +- .../agents/ui/DesktopLifecycleControl.tsx | 34 +- desktop/src/shared/api/relayClientSession.ts | 25 +- desktop/src/shared/api/relayClientShared.ts | 22 +- desktop/src/shared/api/relayClosedRecovery.ts | 11 +- .../api/relayLiveSubscriptionState.test.mjs | 132 +++++- .../e2e/top-chrome-zoom-clearance.spec.ts | 5 +- 11 files changed, 800 insertions(+), 61 deletions(-) create mode 100644 desktop/src/features/agents/desktopLifecycleReceiver.test.mjs create mode 100644 desktop/src/features/agents/desktopLifecycleReceiver.ts diff --git a/desktop/src/features/agents/AGENTS.md b/desktop/src/features/agents/AGENTS.md index e78fb43b26f..5f67eedbaac 100644 --- a/desktop/src/features/agents/AGENTS.md +++ b/desktop/src/features/agents/AGENTS.md @@ -309,9 +309,13 @@ with a TypeScript lookup table or an id comparison in a component. Known Desktops exposes an owner-private, explicitly selected agent+Desktop Stop, not inferred agent location. The app-scoped receiver subscribes live only; reopening never replays commands. Receiver initialization reports a safe failure -stage without exposing raw transport/IPC exceptions. Its scope-owned notification -can explicitly retry the **receiver** with a fresh live-only subscription; that -must discard queued callbacks from the retired receiver, not retry an operation. +stage without exposing raw transport/IPC exceptions. Transient initialization +failures and transient CLOSED states recover through a bounded receiver-owner +budget; each attempt uses a fresh live-only subscription and repeats +projection-only sync before admission. Terminal closure or exhausted recovery +stays in the scope-owned notification, whose deliberate retry resets the receiver +budget. Recovery must discard queued callbacks from the retired receiver, not +retry an operation, and must respect the relay rate-limit gate. A readiness timeout is unconfirmed delivery, not a failed initialization; late EOSE clears that warning after successful projection. An explicit operation retry republishes the exact request; the relay redelivers stored Stop duplicates without repeating relay side effects. diff --git a/desktop/src/features/agents/desktopLifecycle.ts b/desktop/src/features/agents/desktopLifecycle.ts index b81e4826052..c5f00cafe2b 100644 --- a/desktop/src/features/agents/desktopLifecycle.ts +++ b/desktop/src/features/agents/desktopLifecycle.ts @@ -1,5 +1,6 @@ import { invoke } from "@tauri-apps/api/core"; import { relayClient } from "@/shared/api/relayClient"; +import type { LiveSubscriptionClosedRecovery } from "@/shared/api/relayClientShared"; import type { RelayEvent } from "@/shared/api/types"; import type { DesktopScope } from "./desktopList"; import { @@ -270,8 +271,10 @@ export async function receiveLifecycle( ipc = invoke, relay = relayClient, onReady: () => void = () => {}, + onClosed?: (recovery: LiveSubscriptionClosedRecovery) => void, ) { let stopped = false; + let released = false; let stopSubscription = () => {}; let synced = false; let subscriptionReady = false; @@ -339,15 +342,20 @@ export async function receiveLifecycle( 5000, { closedRecovery: "explicit", - onState: (readiness) => { + onState: (readiness, closed) => { if (!valid()) return; subscriptionReady = readiness === "eose"; if (readiness === "closed") { stopped = true; stopSubscription(); - onError( - "Desktop lifecycle receiver subscription closed. Retry the receiver to accept new requests.", - ); + if (onClosed) + onClosed( + closed ?? { classification: "terminal", retryAfterMs: 0 }, + ); + else + onError( + "Desktop lifecycle receiver subscription closed. Retry the receiver to accept new requests.", + ); } else if (readiness === "timeout") { onError( "Desktop lifecycle subscription readiness timed out. Delivery is unconfirmed.", @@ -358,6 +366,8 @@ export async function receiveLifecycle( ), ); const close = () => { + if (released) return; + released = true; stopped = true; // Unsubscribe may fail on a dead socket; it must not revive this receiver // or leave an unhandled promise. A retry always owns a new subscription. diff --git a/desktop/src/features/agents/desktopLifecycleReceiver.test.mjs b/desktop/src/features/agents/desktopLifecycleReceiver.test.mjs new file mode 100644 index 00000000000..70b96890052 --- /dev/null +++ b/desktop/src/features/agents/desktopLifecycleReceiver.test.mjs @@ -0,0 +1,443 @@ +import assert from "node:assert/strict"; +import test from "node:test"; +import { receiveLifecycle } from "./desktopLifecycle.ts"; +import { + ownLifecycleReceiver, + RECEIVER_RECOVERY_DELAYS_MS, +} from "./desktopLifecycleReceiver.ts"; + +const scope = { owner: "owner", community: "wss://one.example" }; + +async function flushUntil(predicate, attempts = 40) { + for (let attempt = 0; attempt < attempts; attempt++) { + if (predicate()) return; + await Promise.resolve(); + } + assert.fail("condition did not become true before the microtask limit"); +} + +function timers() { + let nextId = 1; + const pending = new Map(); + return { + setTimer(callback, delayMs) { + const id = nextId++; + pending.set(id, { callback, delayMs }); + return id; + }, + clearTimer(id) { + pending.delete(id); + }, + fireNext() { + const entry = pending.entries().next().value; + assert.ok(entry, "expected a pending recovery timer"); + pending.delete(entry[0]); + entry[1].callback(); + }, + pending, + }; +} + +test("first subscribe failure recovers without a reconnect callback and syncs before admission", async () => { + const clock = timers(); + let subscribeCalls = 0; + let liveEvent; + let activeSubscriptions = 0; + let closed = 0; + let ready = 0; + const calls = []; + const relay = { + getSessionEpoch: () => 1, + getConnectionGeneration: () => 1, + subscribeLive: async (filter, onEvent, _onReady, _timeout, options) => { + assert.deepEqual(filter, { + kinds: [50182, 50180], + authors: [scope.owner], + limit: 0, + }); + subscribeCalls++; + if (subscribeCalls === 1) + throw new Error("fixture first connection rejected"); + activeSubscriptions++; + liveEvent = onEvent; + onEvent({ id: "during-sync", kind: 50182, created_at: 1 }); + options.onState("eose"); + return () => { + activeSubscriptions--; + closed++; + }; + }, + fetchEvents: async () => { + calls.push("history"); + return []; + }, + publishEvent: async () => {}, + }; + const ipc = async (command, args) => { + if (command === "observe_desktop_placement") { + calls.push(args.reconcile ? `projection:${args.events.length}` : "page"); + return; + } + if (command === "receive_desktop_lifecycle") { + calls.push("admission"); + return null; + } + throw new Error(command); + }; + const errors = []; + const stop = ownLifecycleReceiver( + scope, + (error) => errors.push(error), + () => ready++, + { + ...clock, + waitForRateLimit: async () => {}, + startReceiver: (receiverScope, active, onError, onReady, onClosed) => + receiveLifecycle( + receiverScope, + active, + onError, + ipc, + relay, + onReady, + onClosed, + ), + }, + ); + + await flushUntil(() => clock.pending.size === 1); + assert.equal(subscribeCalls, 1); + assert.deepEqual(errors, []); + assert.equal(clock.pending.values().next().value.delayMs, 1_000); + + clock.fireNext(); + await flushUntil(() => calls.includes("admission")); + assert.equal(subscribeCalls, 2); + assert.equal(activeSubscriptions, 1); + assert.equal(ready, 1); + assert.deepEqual(calls, [ + "history", + "page", + "projection:0", + "projection:1", + "admission", + ]); + + liveEvent({ id: "ordinary-live", kind: 50182, created_at: 2 }); + await flushUntil( + () => calls.filter((call) => call === "admission").length === 2, + ); + stop(); + await flushUntil(() => activeSubscriptions === 0); + assert.equal(activeSubscriptions, 0); + assert.equal(closed, 1); +}); + +test("transient CLOSED before readiness retires and replaces the whole receiver", async () => { + const clock = timers(); + let subscribeCalls = 0; + let closeCount = 0; + let ready = 0; + const relay = { + getSessionEpoch: () => 1, + getConnectionGeneration: () => 1, + subscribeLive: async (_filter, _onEvent, _onReady, _timeout, options) => { + subscribeCalls++; + if (subscribeCalls === 1) + options.onState("closed", { + classification: "retryable", + retryAfterMs: 0, + }); + else options.onState("eose"); + return () => closeCount++; + }, + fetchEvents: async () => [], + publishEvent: async () => {}, + }; + const stop = ownLifecycleReceiver( + scope, + () => {}, + () => ready++, + { + ...clock, + waitForRateLimit: async () => {}, + startReceiver: (receiverScope, active, onError, onReady, onClosed) => + receiveLifecycle( + receiverScope, + active, + onError, + async () => {}, + relay, + onReady, + onClosed, + ), + }, + ); + + await flushUntil(() => clock.pending.size === 1); + await flushUntil(() => closeCount === 1); + assert.equal(subscribeCalls, 1); + assert.equal(closeCount, 1); + clock.fireNext(); + await flushUntil(() => ready === 1); + assert.equal(subscribeCalls, 2); + stop(); + await flushUntil(() => closeCount === 2); +}); + +test("scope cancellation clears recovery timers and closes a late subscription", async () => { + const timerClock = timers(); + let starts = 0; + const stopTimerOwner = ownLifecycleReceiver( + scope, + () => {}, + () => {}, + { + ...timerClock, + waitForRateLimit: async () => {}, + startReceiver: async () => { + starts++; + throw new Error("transient"); + }, + }, + ); + await flushUntil(() => timerClock.pending.size === 1); + stopTimerOwner(); + assert.equal(timerClock.pending.size, 0); + assert.equal(starts, 1); + + let finishSubscribe; + let lateCloseCount = 0; + let historyCalls = 0; + const relay = { + getSessionEpoch: () => 1, + getConnectionGeneration: () => 1, + subscribeLive: () => + new Promise((resolve) => { + finishSubscribe = () => resolve(() => lateCloseCount++); + }), + fetchEvents: async () => { + historyCalls++; + return []; + }, + publishEvent: async () => {}, + }; + const stopLateOwner = ownLifecycleReceiver( + scope, + () => {}, + () => {}, + { + ...timers(), + waitForRateLimit: async () => {}, + startReceiver: (receiverScope, active, onError, onReady, onClosed) => + receiveLifecycle( + receiverScope, + active, + onError, + async () => {}, + relay, + onReady, + onClosed, + ), + }, + ); + await flushUntil(() => typeof finishSubscribe === "function"); + stopLateOwner(); + finishSubscribe(); + await flushUntil(() => lateCloseCount === 1); + assert.equal(historyCalls, 0, "cancelled receiver must not begin sync"); +}); + +test("scope cancellation during sync fences reconciliation, admission, readiness, and errors", async () => { + let resolveHistory; + let deliver; + let closeCount = 0; + let ready = 0; + const errors = []; + const calls = []; + const relay = { + getSessionEpoch: () => 1, + getConnectionGeneration: () => 1, + subscribeLive: async (_filter, onEvent, _onReady, _timeout, options) => { + deliver = onEvent; + options.onState("eose"); + return () => closeCount++; + }, + fetchEvents: () => + new Promise((resolve) => { + resolveHistory = resolve; + }), + publishEvent: async () => {}, + }; + const stop = ownLifecycleReceiver( + scope, + (error) => errors.push(error), + () => ready++, + { + ...timers(), + waitForRateLimit: async () => {}, + startReceiver: (receiverScope, active, onError, onReady, onClosed) => + receiveLifecycle( + receiverScope, + active, + onError, + async (command) => { + calls.push(command); + return null; + }, + relay, + onReady, + onClosed, + ), + }, + ); + await flushUntil(() => resolveHistory !== undefined); + deliver({ id: "queued", kind: 50182, created_at: 1 }); + stop(); + resolveHistory([]); + await flushUntil(() => closeCount === 1); + + assert.deepEqual(calls, []); + assert.deepEqual(errors, []); + assert.equal(ready, 0); +}); + +test("initializer recovery exhausts the bounded budget and preserves its safe terminal outcome", async () => { + const clock = timers(); + let starts = 0; + const errors = []; + ownLifecycleReceiver( + scope, + (error) => errors.push(error), + () => {}, + { + ...clock, + waitForRateLimit: async () => {}, + startReceiver: async () => { + starts++; + throw new Error("raw private initializer detail"); + }, + }, + ); + + for ( + let attempt = 0; + attempt < RECEIVER_RECOVERY_DELAYS_MS.length; + attempt++ + ) { + await flushUntil(() => clock.pending.size === 1); + assert.equal( + clock.pending.values().next().value.delayMs, + RECEIVER_RECOVERY_DELAYS_MS[attempt], + ); + clock.fireNext(); + } + await flushUntil(() => errors.length === 1); + assert.equal(starts, 4); + assert.equal(clock.pending.size, 0); + assert.equal( + errors[0], + "Desktop lifecycle receiver is unavailable (initialization failed).", + ); + assert.doesNotMatch(errors[0], /private|detail/); +}); + +test("transient CLOSED recovery is bounded across successful receivers; terminal stays manual", async () => { + const clock = timers(); + const closures = []; + let starts = 0; + let closes = 0; + const errors = []; + const stop = ownLifecycleReceiver( + scope, + (error) => errors.push(error), + () => {}, + { + ...clock, + waitForRateLimit: async () => {}, + startReceiver: async (_scope, _active, _onError, onReady, onClosed) => { + starts++; + onReady(); + closures.push(onClosed); + return () => closes++; + }, + }, + ); + await flushUntil(() => closures.length === 1); + + for ( + let attempt = 0; + attempt < RECEIVER_RECOVERY_DELAYS_MS.length; + attempt++ + ) { + closures.at(-1)({ classification: "retryable", retryAfterMs: 0 }); + assert.equal( + clock.pending.values().next().value.delayMs, + RECEIVER_RECOVERY_DELAYS_MS[attempt], + ); + clock.fireNext(); + await flushUntil(() => closures.length === attempt + 2); + } + closures.at(-1)({ classification: "retryable", retryAfterMs: 0 }); + assert.equal(clock.pending.size, 0); + assert.equal(starts, 4); + assert.equal(closes, 4); + assert.equal(errors.length, 1); + assert.match(errors[0], /Retry the receiver/); + stop(); + + const terminalClock = timers(); + let terminalClosed; + const terminalErrors = []; + ownLifecycleReceiver( + scope, + (error) => terminalErrors.push(error), + () => {}, + { + ...terminalClock, + startReceiver: async (_scope, _active, _onError, _onReady, onClosed) => { + terminalClosed = onClosed; + return () => {}; + }, + }, + ); + await flushUntil(() => terminalClosed !== undefined); + terminalClosed({ classification: "terminal", retryAfterMs: 0 }); + assert.equal(terminalClock.pending.size, 0); + assert.equal(terminalErrors.length, 1); +}); + +test("rate-limited recovery honors both the delay and shared gate, and cancellation fences it", async () => { + const clock = timers(); + let releaseGate; + const gate = new Promise((resolve) => { + releaseGate = resolve; + }); + const closures = []; + let starts = 0; + const stop = ownLifecycleReceiver( + scope, + () => {}, + () => {}, + { + ...clock, + waitForRateLimit: () => gate, + startReceiver: async (_scope, _active, _onError, _onReady, onClosed) => { + starts++; + closures.push(onClosed); + return () => {}; + }, + }, + ); + await flushUntil(() => closures.length === 1); + closures[0]({ classification: "rate-limited", retryAfterMs: 8_000 }); + assert.equal(clock.pending.values().next().value.delayMs, 8_000); + clock.fireNext(); + await Promise.resolve(); + assert.equal(starts, 1, "fresh subscription must wait for the active gate"); + stop(); + releaseGate(); + await Promise.resolve(); + await Promise.resolve(); + assert.equal(starts, 1, "scope cancellation must fence the late gate result"); +}); diff --git a/desktop/src/features/agents/desktopLifecycleReceiver.ts b/desktop/src/features/agents/desktopLifecycleReceiver.ts new file mode 100644 index 00000000000..12a2a81ff29 --- /dev/null +++ b/desktop/src/features/agents/desktopLifecycleReceiver.ts @@ -0,0 +1,142 @@ +import type { LiveSubscriptionClosedRecovery } from "@/shared/api/relayClientShared"; +import { waitForRateLimit } from "@/shared/api/relayRateLimitGate"; +import type { DesktopScope } from "./desktopList"; +import { receiveLifecycle } from "./desktopLifecycle"; +import { receiverErrorMessage } from "./desktopLifecycleDiagnostics"; + +export const RECEIVER_RECOVERY_DELAYS_MS = [1_000, 2_000, 4_000] as const; + +const CLOSED_MESSAGE = + "Desktop lifecycle receiver subscription closed. Retry the receiver to accept new requests."; + +type StartReceiver = ( + scope: DesktopScope, + active: () => boolean, + onError: (message: string) => void, + onReady: () => void, + onClosed: (recovery: LiveSubscriptionClosedRecovery) => void, +) => Promise<() => void>; + +type ReceiverOwnerDependencies = { + startReceiver?: StartReceiver; + waitForRateLimit?: () => Promise; + setTimer?: (callback: () => void, delayMs: number) => number; + clearTimer?: (timer: number) => void; +}; + +/** + * Owns one lifecycle receiver scope. Recovery always creates a fresh live-only + * subscription and lets receiveLifecycle repeat projection sync before it + * admits events. The attempt budget belongs to this owner and is intentionally + * not reset by a successful EOSE/sync followed by another CLOSED. + */ +export function ownLifecycleReceiver( + scope: DesktopScope, + onError: (message: string) => void, + onReady: () => void, + dependencies: ReceiverOwnerDependencies = {}, +) { + const startReceiver: StartReceiver = + dependencies.startReceiver ?? + ((receiverScope, active, reportError, ready, closed) => + receiveLifecycle( + receiverScope, + active, + reportError, + undefined, + undefined, + ready, + closed, + )); + const waitForGate = dependencies.waitForRateLimit ?? waitForRateLimit; + const setTimer = + dependencies.setTimer ?? + ((callback, delayMs) => window.setTimeout(callback, delayMs)); + const clearTimer = + dependencies.clearTimer ?? ((timer) => window.clearTimeout(timer)); + + let stopped = false; + let generation = 0; + let recoveryAttempt = 0; + let timer: number | undefined; + let closeCurrent: (() => void) | undefined; + + const current = (token: number) => !stopped && generation === token; + + const retireCurrent = () => { + const close = closeCurrent; + closeCurrent = undefined; + close?.(); + }; + + const recover = ( + token: number, + terminalMessage: string, + recovery: LiveSubscriptionClosedRecovery, + ) => { + if (!current(token)) return; + generation++; + retireCurrent(); + + if ( + recovery.classification === "terminal" || + recoveryAttempt >= RECEIVER_RECOVERY_DELAYS_MS.length + ) { + onError(terminalMessage); + return; + } + + const delayMs = Math.max( + RECEIVER_RECOVERY_DELAYS_MS[recoveryAttempt], + recovery.retryAfterMs, + ); + recoveryAttempt++; + const waitingGeneration = generation; + timer = setTimer(() => { + timer = undefined; + if (stopped || generation !== waitingGeneration) return; + void waitForGate().then(() => { + if (!stopped && generation === waitingGeneration) start(); + }); + }, delayMs); + }; + + const start = () => { + const token = ++generation; + void startReceiver( + scope, + () => current(token), + (message) => { + if (current(token)) onError(message); + }, + () => { + if (current(token)) onReady(); + }, + (recovery) => recover(token, CLOSED_MESSAGE, recovery), + ) + .then((close) => { + if (current(token)) closeCurrent = close; + else close(); + }) + .catch((error) => { + if (current(token)) + recover(token, receiverErrorMessage(error), { + classification: "retryable", + retryAfterMs: 0, + }); + }); + }; + + start(); + + return () => { + if (stopped) return; + stopped = true; + generation++; + if (timer !== undefined) { + clearTimer(timer); + timer = undefined; + } + retireCurrent(); + }; +} diff --git a/desktop/src/features/agents/ui/DesktopLifecycleControl.test.mjs b/desktop/src/features/agents/ui/DesktopLifecycleControl.test.mjs index 35e52f7aab0..17aff55e75a 100644 --- a/desktop/src/features/agents/ui/DesktopLifecycleControl.test.mjs +++ b/desktop/src/features/agents/ui/DesktopLifecycleControl.test.mjs @@ -136,7 +136,7 @@ test("mounted Start exposes unavailable provisioning and exact retry; Restart re } }); -test("receiver failure is a scope-owned notification, not pre-shell layout", async () => { +test("terminal receiver failure is a scope-owned notification, not pre-shell layout", async () => { const originalRaf = globalThis.requestAnimationFrame; globalThis.requestAnimationFrame = (fn) => setTimeout(fn, 0); const dom = new JSDOM("
", { @@ -157,7 +157,6 @@ test("receiver failure is a scope-owned notification, not pre-shell layout", asy let closed = 0; let rejectLate; let delayed = false; - let storageAvailable = false; let subscribed = 0; relayClient.fetchEvents = async () => { if (delayed) @@ -181,10 +180,7 @@ test("receiver failure is a scope-owned notification, not pre-shell layout", asy }; }; window.__TAURI_INTERNALS__ = { - invoke: async () => { - if (storageAvailable) return; - throw new Error("fixture: storage unavailable"); - }, + invoke: async () => {}, }; const root = createRoot(document.getElementById("root")); const scope = { owner: "owner", community: "wss://one.example" }; @@ -201,14 +197,18 @@ test("receiver failure is a scope-owned notification, not pre-shell layout", asy 0, "startup must not render in-flow failure UI", ); + assert.equal(warnings().length, 0); + await React.act(async () => + readiness("closed", { classification: "terminal", retryAfterMs: 0 }), + ); assert.equal(warnings().length, 1); assert.equal( warnings()[0].title, - "Desktop lifecycle receiver is unavailable (projection: request failed).", + "Desktop lifecycle receiver subscription closed. Retry the receiver to accept new requests.", ); assert.equal(warnings()[0].duration, Infinity); assert.equal(warnings()[0].closeButton, true); - readiness("closed"); + readiness("closed", { classification: "terminal", retryAfterMs: 0 }); assert.equal( warnings().length, 1, @@ -216,7 +216,6 @@ test("receiver failure is a scope-owned notification, not pre-shell layout", asy ); const retryAction = warnings()[0].action; assert.equal(retryAction.label, "Retry receiver"); - storageAvailable = true; await React.act(async () => retryAction.onClick()); assert.equal(subscribed, 2, "explicit recovery starts a new live receiver"); assert.equal( @@ -230,7 +229,7 @@ test("receiver failure is a scope-owned notification, not pre-shell layout", asy ), ); assert.equal(warnings().length, 0, "leaving the scope removes its warning"); - readiness("closed"); + readiness("closed", { classification: "terminal", retryAfterMs: 0 }); assert.equal( warnings().length, 0, diff --git a/desktop/src/features/agents/ui/DesktopLifecycleControl.tsx b/desktop/src/features/agents/ui/DesktopLifecycleControl.tsx index 0b6b63d67be..8212eb7258b 100644 --- a/desktop/src/features/agents/ui/DesktopLifecycleControl.tsx +++ b/desktop/src/features/agents/ui/DesktopLifecycleControl.tsx @@ -3,13 +3,9 @@ import { toast } from "sonner"; import { Button } from "@/shared/ui/button"; import type { RelayEvent } from "@/shared/api/types"; import type { DesktopRow, DesktopScope } from "../desktopList"; -import { - lifecycleClient, - receiveLifecycle, - type LifecycleOutcome, -} from "../desktopLifecycle"; +import { lifecycleClient, type LifecycleOutcome } from "../desktopLifecycle"; +import { ownLifecycleReceiver } from "../desktopLifecycleReceiver"; import { useRelayAgentsQuery } from "../hooks"; -import { receiverErrorMessage } from "../desktopLifecycleDiagnostics"; export function DesktopLifecycleReceiver({ scope, @@ -21,7 +17,7 @@ export function DesktopLifecycleReceiver({ useEffect(() => { if (!owner || !community) return; let active = true; - let close: (() => void) | undefined; + let stop = () => {}; let notification: string | number | undefined; const reportError = (message: string) => { if (!active) return; @@ -37,32 +33,18 @@ export function DesktopLifecycleReceiver({ onClick: () => { if (!active) return; active = false; - close?.(); + stop(); retry(attempt + 1); }, }, }); }; - void receiveLifecycle( - { owner, community }, - () => active, - reportError, - undefined, - undefined, - () => { - if (active && notification !== undefined) toast.dismiss(notification); - }, - ) - .then((fn) => { - if (active) close = fn; - else fn(); - }) - .catch((error) => { - reportError(receiverErrorMessage(error)); - }); + stop = ownLifecycleReceiver({ owner, community }, reportError, () => { + if (active && notification !== undefined) toast.dismiss(notification); + }); return () => { active = false; - close?.(); + stop(); if (notification !== undefined) toast.dismiss(notification); }; }, [owner, community, attempt]); diff --git a/desktop/src/shared/api/relayClientSession.ts b/desktop/src/shared/api/relayClientSession.ts index a7c0bded02d..5ba009e6476 100644 --- a/desktop/src/shared/api/relayClientSession.ts +++ b/desktop/src/shared/api/relayClientSession.ts @@ -164,6 +164,12 @@ export class RelayClient { sub.reject(error); } else { clearClosedRetry(sub); + sub.resolveReady?.("closed"); + sub.onState?.("closed", { + classification: "terminal", + retryAfterMs: 0, + }); + sub.resolveReady = undefined; } this.subscriptions.delete(subId); } @@ -635,19 +641,23 @@ export class RelayClient { resolveReady("timeout"); }, readinessTimeoutMs); - this.subscriptions.set(subId, { + const subscription: Extract = { mode: "live", filter, onEvent, resolveReady, onState: options.onState, closedRecovery: options.closedRecovery ?? "shared", - }); + }; + this.subscriptions.set(subId, subscription); try { await this.sendRawWithReconnectRetry( ["REQ", subId, filter], "Failed to restore relay subscription.", + () => + subscription.closedRecovery !== "explicit" || + this.subscriptions.get(subId) === subscription, ); } catch (error) { window.clearTimeout(fallbackTimeout); @@ -709,6 +719,7 @@ export class RelayClient { private async sendRawWithReconnectRetry( payload: unknown[], fallbackMessage: string, + retryStillOwned: () => boolean = () => true, ) { try { await this.sendRaw(payload); @@ -717,8 +728,13 @@ export class RelayClient { error, fallbackMessage, ); + // resetConnection may retire an explicit subscription synchronously. + // Never put its now-ownerless REQ onto the replacement connection; + // shared subscriptions retain their existing reconnect retry behavior. + if (!retryStillOwned()) throw normalizedError; try { await this.ensureConnected(); + if (!retryStillOwned()) throw normalizedError; await this.sendRaw(payload); } catch (retryError) { throw this.recoverFromSocketFailure( @@ -1077,7 +1093,10 @@ export class RelayClient { continue; } subscription.resolveReady?.("closed"); - subscription.onState?.("closed"); + subscription.onState?.("closed", { + classification: options?.reconnect === false ? "terminal" : "retryable", + retryAfterMs: 0, + }); subscription.resolveReady = undefined; clearClosedRetry(subscription); if (subscription.closedRecovery === "explicit") { diff --git a/desktop/src/shared/api/relayClientShared.ts b/desktop/src/shared/api/relayClientShared.ts index 925c74b573d..09064dc6276 100644 --- a/desktop/src/shared/api/relayClientShared.ts +++ b/desktop/src/shared/api/relayClientShared.ts @@ -1,4 +1,5 @@ import type { RelayEvent } from "@/shared/api/types"; +import type { RelayClosedClass } from "@/shared/api/relayClosedPolicy"; /** * Observable connection state for the relay singleton. @@ -70,16 +71,26 @@ type FirstEventSubscription = { export type LiveSubscriptionReadiness = "eose" | "closed" | "timeout"; +export type LiveSubscriptionClosedRecovery = { + classification: RelayClosedClass; + /** Minimum delay before the owner creates a fresh subscription. */ + retryAfterMs: number; +}; + /** * Optional lifecycle policy for a live subscription. * * Most subscriptions keep the shared reconnect/CLOSED recovery behavior. A * command receiver can instead request explicit recovery: every CLOSED retires - * that subscription and `onState` remains observable after initial EOSE so the - * owning UI can offer a deliberate fresh subscription. + * that subscription and `onState` remains observable after initial EOSE. The + * owner receives only a safe recovery class/delay and decides whether to create + * a fresh subscription or require deliberate retry. */ export type LiveSubscriptionOptions = { - onState?: (state: LiveSubscriptionReadiness) => void; + onState?: ( + state: LiveSubscriptionReadiness, + closed?: LiveSubscriptionClosedRecovery, + ) => void; closedRecovery?: "shared" | "explicit"; }; @@ -88,7 +99,10 @@ type LiveSubscription = { filter: RelaySubscriptionFilter; onEvent: (event: RelayEvent) => void; resolveReady?: (readiness: LiveSubscriptionReadiness) => void; - onState?: (state: LiveSubscriptionReadiness) => void; + onState?: ( + state: LiveSubscriptionReadiness, + closed?: LiveSubscriptionClosedRecovery, + ) => void; closedRecovery: "shared" | "explicit"; lastSeenCreatedAt?: number; /** diff --git a/desktop/src/shared/api/relayClosedRecovery.ts b/desktop/src/shared/api/relayClosedRecovery.ts index 8f1a68287d2..d249bf2406d 100644 --- a/desktop/src/shared/api/relayClosedRecovery.ts +++ b/desktop/src/shared/api/relayClosedRecovery.ts @@ -123,10 +123,6 @@ function recoverLiveSubscriptionFromClosed({ message: string; sendReq: (subId: string, filter: RelaySubscriptionFilter) => Promise; }) { - subscription.resolveReady?.("closed"); - subscription.onState?.("closed"); - subscription.resolveReady = undefined; - const closedClass = classifyRelayClosed(message); if (closedClass === "rate-limited") { @@ -134,6 +130,13 @@ function recoverLiveSubscriptionFromClosed({ activateRateLimit(hintSeconds); } + subscription.resolveReady?.("closed"); + subscription.onState?.("closed", { + classification: closedClass, + retryAfterMs: closedClass === "rate-limited" ? rateLimitRemainingMs() : 0, + }); + subscription.resolveReady = undefined; + if (subscription.closedRecovery === "explicit") { // Command receivers must not survive CLOSED into shared re-subscription. // Their owner presents an explicit fresh-receiver action instead. diff --git a/desktop/src/shared/api/relayLiveSubscriptionState.test.mjs b/desktop/src/shared/api/relayLiveSubscriptionState.test.mjs index 587bfae1f28..a52d408bec3 100644 --- a/desktop/src/shared/api/relayLiveSubscriptionState.test.mjs +++ b/desktop/src/shared/api/relayLiveSubscriptionState.test.mjs @@ -5,6 +5,8 @@ let fakeNow = 0; let nextTimerId = 1; const pendingTimers = new Map(); const sentFrames = []; +const sendAttempts = []; +let failNextSend = false; globalThis.window = { setTimeout: (fn, ms) => { @@ -15,7 +17,14 @@ globalThis.window = { clearTimeout: (id) => pendingTimers.delete(id), __TAURI_INTERNALS__: { invoke: async (command, args) => { - if (command === "plugin:websocket|send") sentFrames.push(args); + if (command === "plugin:websocket|send") { + sendAttempts.push(args); + if (failNextSend) { + failNextSend = false; + throw new Error("fixture first send failed"); + } + sentFrames.push(args); + } }, }, }; @@ -30,6 +39,8 @@ function resetHarness() { nextTimerId = 1; pendingTimers.clear(); sentFrames.length = 0; + sendAttempts.length = 0; + failNextSend = false; resetRateLimitGate(); } @@ -117,17 +128,21 @@ test("persistent state reports timeout, late EOSE, then CLOSED through RelayClie await close(); }); -for (const [label, message] of [ - ["terminal", "restricted: access revoked"], - ["retryable", "error: storage temporarily unavailable"], +for (const [label, message, classification] of [ + ["terminal", "restricted: access revoked", "terminal"], + ["retryable", "error: storage temporarily unavailable", "retryable"], ]) { test(`explicit recovery retires an EOSE-ready ${label} CLOSED without re-REQ`, async () => { resetHarness(); const client = connectedClient(); const states = []; + const recoveries = []; const { opened, subId } = await openLive(client, { closedRecovery: "explicit", - onState: (state) => states.push(state), + onState: (state, recovery) => { + states.push(state); + if (recovery) recoveries.push(recovery); + }, }); await deliver(client, ["EOSE", subId]); const close = await opened; @@ -137,6 +152,7 @@ for (const [label, message] of [ await Promise.resolve(); assert.deepEqual(states, ["eose", "closed"]); + assert.deepEqual(recoveries, [{ classification, retryAfterMs: 0 }]); assert.equal(client.subscriptions.has(subId), false); assert.equal(sentProtocolFrames("REQ").length, 1); await close(); @@ -173,6 +189,112 @@ test("connection reset reports CLOSED and retires only explicit-recovery subscri client.disconnect(); }); +test("disconnect reports terminal CLOSED before retiring an explicit subscription", async () => { + resetHarness(); + const client = connectedClient(); + const states = []; + const recoveries = []; + const explicit = await openLive(client, { + closedRecovery: "explicit", + onState: (state, recovery) => { + states.push(state); + if (recovery) recoveries.push(recovery); + }, + }); + await deliver(client, ["EOSE", explicit.subId]); + await explicit.opened; + + client.disconnect(); + + assert.deepEqual(states, ["eose", "closed"]); + assert.deepEqual(recoveries, [ + { classification: "terminal", retryAfterMs: 0 }, + ]); + assert.equal(client.subscriptions.has(explicit.subId), false); +}); + +test("explicit rate-limited CLOSED exposes only classified recovery and gate delay", async () => { + resetHarness(); + const client = connectedClient(); + let recovery; + const explicit = await openLive(client, { + closedRecovery: "explicit", + onState: (state, detail) => { + if (state === "closed") recovery = detail; + }, + }); + await deliver(client, ["EOSE", explicit.subId]); + await explicit.opened; + + await deliver(client, [ + "CLOSED", + explicit.subId, + "rate-limited: private relay detail; retry in 4s", + ]); + + assert.deepEqual(recovery, { + classification: "rate-limited", + retryAfterMs: 4_000, + }); + assert.equal( + JSON.stringify(recovery).includes("private relay detail"), + false, + "raw relay payload must not cross the recovery contract", + ); +}); + +test("explicit retirement during first send failure prevents a fresh ownerless REQ", async () => { + resetHarness(); + const client = connectedClient(); + let ensureCalls = 0; + client.ensureConnected = async () => { + ensureCalls++; + if (ensureCalls > 1) client.wsId = 8; + return client.connectionGeneration; + }; + failNextSend = true; + + await assert.rejects( + client.subscribeLive( + { kinds: [50182], authors: ["owner"], limit: 0 }, + () => {}, + undefined, + 5_000, + { closedRecovery: "explicit" }, + ), + /fixture first send failed/, + ); + + assert.equal( + ensureCalls, + 1, + "retired subscription must not reconnect to retry", + ); + assert.equal(sendAttempts.length, 1); + assert.equal(sentProtocolFrames("REQ").length, 0); + assert.equal(client.subscriptions.size, 0); +}); + +test("ordinary subscription still retries its first failed send", async () => { + resetHarness(); + const client = connectedClient(); + let ensureCalls = 0; + client.ensureConnected = async () => { + ensureCalls++; + if (ensureCalls > 1) client.wsId = 8; + return client.connectionGeneration; + }; + failNextSend = true; + + const close = await client.subscribeLive({ kinds: [9], limit: 0 }, () => {}); + + assert.equal(ensureCalls, 2); + assert.equal(sendAttempts.length, 2); + assert.equal(sentProtocolFrames("REQ").length, 1); + await close(); + client.disconnect(); +}); + test("default live subscribers retain shared retry after retryable CLOSED", async () => { resetHarness(); const client = connectedClient(); diff --git a/desktop/tests/e2e/top-chrome-zoom-clearance.spec.ts b/desktop/tests/e2e/top-chrome-zoom-clearance.spec.ts index 97d6b69cb1f..5a49175d085 100644 --- a/desktop/tests/e2e/top-chrome-zoom-clearance.spec.ts +++ b/desktop/tests/e2e/top-chrome-zoom-clearance.spec.ts @@ -116,10 +116,11 @@ test.describe("top chrome macOS traffic-light clearance under text zoom", () => }); await page.goto("/"); // A failed global receiver must remain visible without entering the shell's - // layout flow. This also forces the error to settle before measuring chrome. + // layout flow. Wait through its bounded 1s/2s/4s recovery budget before + // measuring chrome; an intermediate transient failure need not notify. await expect( page.getByText(/Desktop lifecycle receiver is unavailable/), - ).toBeVisible(); + ).toBeVisible({ timeout: 15_000 }); // Lock the native and webview placements together: removing this explicit // Tauri inset or shifting the nav row regresses the macOS chrome alignment. From 8f2dd70f9633181c8603a37a837addecec19704e Mon Sep 17 00:00:00 2001 From: Logan Johnson Date: Sat, 5 Sep 2026 00:01:45 -0400 Subject: [PATCH 4/4] fix(desktop): do not retry terminal receiver initialization Signed-off-by: Logan Johnson --- desktop/src/features/agents/AGENTS.md | 6 ++- .../agents/desktopLifecycleDiagnostics.ts | 9 ++++ .../agents/desktopLifecycleReceiver.test.mjs | 44 +++++++++++++++++++ .../agents/desktopLifecycleReceiver.ts | 10 ++++- 4 files changed, 65 insertions(+), 4 deletions(-) diff --git a/desktop/src/features/agents/AGENTS.md b/desktop/src/features/agents/AGENTS.md index 5f67eedbaac..d4e046b81a6 100644 --- a/desktop/src/features/agents/AGENTS.md +++ b/desktop/src/features/agents/AGENTS.md @@ -314,8 +314,10 @@ failures and transient CLOSED states recover through a bounded receiver-owner budget; each attempt uses a fresh live-only subscription and repeats projection-only sync before admission. Terminal closure or exhausted recovery stays in the scope-owned notification, whose deliberate retry resets the receiver -budget. Recovery must discard queued callbacks from the retired receiver, not -retry an operation, and must respect the relay rate-limit gate. +budget. A known latched-terminal relay session also reports immediately during +initialization without consuming that budget; unknown and transient failures +remain bounded retries. Recovery must discard queued callbacks from the retired +receiver, not retry an operation, and must respect the relay rate-limit gate. A readiness timeout is unconfirmed delivery, not a failed initialization; late EOSE clears that warning after successful projection. An explicit operation retry republishes the exact request; the relay redelivers stored Stop duplicates without repeating relay side effects. diff --git a/desktop/src/features/agents/desktopLifecycleDiagnostics.ts b/desktop/src/features/agents/desktopLifecycleDiagnostics.ts index aebe01c41a9..cee8e60769e 100644 --- a/desktop/src/features/agents/desktopLifecycleDiagnostics.ts +++ b/desktop/src/features/agents/desktopLifecycleDiagnostics.ts @@ -6,9 +6,17 @@ type Stage = | "projection" | "reconciliation"; +export type LifecycleReceiverFailureClassification = "retryable" | "terminal"; + export class LifecycleReceiverError extends Error { + readonly recoveryClassification: LifecycleReceiverFailureClassification; + constructor(stage: Stage, error: unknown) { const value = error instanceof Error ? error.message : error; + const recoveryClassification = + value === "Relay session is terminal; cannot reconnect." + ? "terminal" + : "retryable"; const reason = value === "closed" ? "subscription closed" @@ -20,6 +28,7 @@ export class LifecycleReceiverError extends Error { ? "history timed out" : "request failed"; super(`Desktop lifecycle receiver is unavailable (${stage}: ${reason}).`); + this.recoveryClassification = recoveryClassification; } } diff --git a/desktop/src/features/agents/desktopLifecycleReceiver.test.mjs b/desktop/src/features/agents/desktopLifecycleReceiver.test.mjs index 70b96890052..6948a55ca65 100644 --- a/desktop/src/features/agents/desktopLifecycleReceiver.test.mjs +++ b/desktop/src/features/agents/desktopLifecycleReceiver.test.mjs @@ -133,6 +133,50 @@ test("first subscribe failure recovers without a reconnect callback and syncs be assert.equal(closed, 1); }); +test("terminal relay-session initialization failure does not consume the recovery budget", async () => { + const clock = timers(); + let subscribeCalls = 0; + const relay = { + getSessionEpoch: () => 1, + getConnectionGeneration: () => 1, + subscribeLive: async () => { + subscribeCalls++; + throw new Error("Relay session is terminal; cannot reconnect."); + }, + fetchEvents: async () => assert.fail("terminal session must not sync"), + publishEvent: async () => {}, + }; + const errors = []; + ownLifecycleReceiver( + scope, + (error) => errors.push(error), + () => {}, + { + ...clock, + waitForRateLimit: async () => {}, + startReceiver: (receiverScope, active, onError, onReady, onClosed) => + receiveLifecycle( + receiverScope, + active, + onError, + async () => + assert.fail("terminal session must not invoke native IPC"), + relay, + onReady, + onClosed, + ), + }, + ); + + await flushUntil(() => errors.length === 1); + assert.equal(subscribeCalls, 1); + assert.equal(clock.pending.size, 0); + assert.equal( + errors[0], + "Desktop lifecycle receiver is unavailable (subscription: relay session requires reconnection).", + ); +}); + test("transient CLOSED before readiness retires and replaces the whole receiver", async () => { const clock = timers(); let subscribeCalls = 0; diff --git a/desktop/src/features/agents/desktopLifecycleReceiver.ts b/desktop/src/features/agents/desktopLifecycleReceiver.ts index 12a2a81ff29..62417287e91 100644 --- a/desktop/src/features/agents/desktopLifecycleReceiver.ts +++ b/desktop/src/features/agents/desktopLifecycleReceiver.ts @@ -2,7 +2,10 @@ import type { LiveSubscriptionClosedRecovery } from "@/shared/api/relayClientSha import { waitForRateLimit } from "@/shared/api/relayRateLimitGate"; import type { DesktopScope } from "./desktopList"; import { receiveLifecycle } from "./desktopLifecycle"; -import { receiverErrorMessage } from "./desktopLifecycleDiagnostics"; +import { + LifecycleReceiverError, + receiverErrorMessage, +} from "./desktopLifecycleDiagnostics"; export const RECEIVER_RECOVERY_DELAYS_MS = [1_000, 2_000, 4_000] as const; @@ -121,7 +124,10 @@ export function ownLifecycleReceiver( .catch((error) => { if (current(token)) recover(token, receiverErrorMessage(error), { - classification: "retryable", + classification: + error instanceof LifecycleReceiverError + ? error.recoveryClassification + : "retryable", retryAfterMs: 0, }); });