diff --git a/desktop/src/features/agents/AGENTS.md b/desktop/src/features/agents/AGENTS.md index e78fb43b26..5f67eedbaa 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 b81e482605..c5f00cafe2 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 0000000000..70b9689005 --- /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 0000000000..12a2a81ff2 --- /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 35e52f7aab..17aff55e75 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 0b6b63d67b..8212eb7258 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 a7c0bded02..5ba009e647 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 925c74b573..09064dc627 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 8f1a68287d..d249bf2406 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 587bfae1f2..a52d408bec 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 97d6b69cb1..5a49175d08 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.