diff --git a/desktop/src/features/agents/desktopLifecycle.test.mjs b/desktop/src/features/agents/desktopLifecycle.test.mjs new file mode 100644 index 0000000000..b15daccfcf --- /dev/null +++ b/desktop/src/features/agents/desktopLifecycle.test.mjs @@ -0,0 +1,240 @@ +import assert from "node:assert/strict"; +import test from "node:test"; +import { lifecycleClient, receiveLifecycle } from "./desktopLifecycle.ts"; +const scope = { owner: "owner", community: "wss://one.example" }; +const tick = () => new Promise((resolve) => setImmediate(resolve)); +function fixture() { + let epoch = 0, + connection = 1, + active = true, + live; + let placement = ["source", "selection"], + stopOutcome = "stopped"; + let lifecycleOutcome = "running", + history = [], + ackLost = false; + const prepared = [], + sent = [], + calls = [], + errors = []; + const ipc = async (command, args) => { + assert.equal(args.owner, scope.owner); + assert.equal(args.community, scope.community); + calls.push([command, args]); + if (command === "observe_desktop_placement") return; + if (command === "read_desktop_placement") return placement; + if ( + command === "prepare_desktop_lifecycle" || + command === "prepare_desktop_stop" + ) { + const request = { + id: `request-${prepared.length}`, + kind: command.endsWith("_stop") ? 50180 : 50182, + ...args, + }; + prepared.push(request); + return request; + } + if (command === "read_desktop_lifecycle_results") + return args.request.action === "status" ? "running" : lifecycleOutcome; + if (command === "read_desktop_stop_results") return stopOutcome; + if (command === "receive_desktop_lifecycle") + return { id: "result", kind: 50183 }; + throw Error(command); + }; + const relay = { + getSessionEpoch: () => epoch, + getConnectionGeneration: () => connection, + fetchEvents: async (filter) => + filter.kinds.includes(50182) ? history : [], + publishEvent: async (event, _timeout, _failure, check) => { + check(); + sent.push(event); + if (ackLost) throw Error("ACK lost"); + }, + subscribeLive: async (filter, callback) => { + assert.deepEqual(filter, { + kinds: [50182, 50180], + authors: [scope.owner], + limit: 0, + }); + live = callback; + return () => { + live = undefined; + }; + }, + }; + return { + ipc, + relay, + prepared, + sent, + calls, + errors, + client: () => lifecycleClient(scope, () => active, ipc, relay), + changeScope: () => { + epoch++; + }, + disconnect: () => { + connection++; + }, + unmount: () => { + active = false; + }, + stop: (value) => { + stopOutcome = value; + }, + outcome: (value) => { + lifecycleOutcome = value; + }, + place: (value) => { + placement = value; + }, + history: (value) => { + history = value; + }, + loseAck: () => { + ackLost = true; + }, + deliver: (event) => live?.(event), + }; +} + +test("lost ACK/result permits explicit exact-byte retry, not a fresh Start", async () => { + const f = fixture(), + client = f.client(); + const request = await client.start("destination", "agent"); + f.loseAck(); + f.outcome("unknown"); + assert.equal(await client.send(request, 0), "unknown"); + f.outcome("running"); + assert.equal(await client.send(request, 1), "running"); + assert.equal(f.prepared.length, 1); + assert.deepEqual(f.sent, [request, request]); + assert.equal(f.sent[0], f.sent[1]); + assert.equal( + f.calls.filter(([c]) => c === "receive_desktop_lifecycle").length, + 0, + ); +}); + +for (const interruption of ["changeScope", "disconnect", "unmount"]) { + test(`${interruption} during Stop cancels Move before destination prepare or send`, async () => { + const f = fixture(), + client = f.client(); + const publish = f.relay.publishEvent; + f.relay.publishEvent = async (...args) => { + await publish(...args); + if (args[0].kind === 50180) f[interruption](); + }; + await assert.rejects( + client.move("agent", "destination", ["source"], () => {}), + /scope changed/, + ); + assert.equal(f.prepared.filter((r) => r.action === "start").length, 0); + }); +} + +test("failed Stop is final even if a successful outcome appears later", async () => { + const f = fixture(); + f.stop("failed"); + await assert.rejects( + f.client().move("agent", "destination", ["source"], () => {}), + /will not continue later/, + ); + f.stop("stopped"); + await tick(); + assert.equal(f.prepared.filter((r) => r.action === "start").length, 0); +}); + +test("unconfirmed Stop exhausts polling without storing any future Start", async () => { + const f = fixture(); + f.stop("unknown"); + const timer = globalThis.setTimeout; + globalThis.setTimeout = (fn) => { + queueMicrotask(fn); + return 0; + }; + try { + await assert.rejects( + f.client().move("agent", "destination", ["source"], () => {}), + /Could not confirm source Stop/, + ); + assert.equal(f.prepared.filter((r) => r.action === "start").length, 0); + } finally { + globalThis.setTimeout = timer; + } +}); + +test("Move dispatches Start only after Stop success and unchanged placement", async () => { + const f = fixture(); + assert.equal( + await f.client().move("agent", "destination", ["source"], () => {}), + "running", + ); + assert.deepEqual( + f.sent.map((r) => r.action ?? "stop"), + ["status", "stop", "start"], + ); + assert.equal(f.sent.at(-1).desktop, "destination"); + const stopRead = f.calls.findIndex( + ([c]) => c === "read_desktop_stop_results", + ); + const startPrepare = f.calls.findIndex( + ([c, a]) => c === "prepare_desktop_lifecycle" && a.action === "start", + ); + assert.ok(stopRead < startPrepare); +}); + +test("another Desktop's new placement supersedes an in-flight Move", async () => { + const f = fixture(), + publish = f.relay.publishEvent; + f.relay.publishEvent = async (...args) => { + await publish(...args); + if (args[0].kind === 50180) f.place(["third", "new-selection"]); + }; + await assert.rejects( + f.client().move("agent", "destination", ["source"], () => {}), + /Placement changed/, + ); + assert.equal(f.prepared.filter((r) => r.action === "start").length, 0); +}); + +test("Restart resolves current host and binds its Status observation, not destination", async () => { + const f = fixture(); + const request = await f.client().restart("agent", ["source", "other"]); + assert.equal(request.desktop, "source"); + assert.equal(request.observed, f.prepared[0].id); + assert.equal(request.action, "restart"); +}); + +test("receiver projects history without executing it and invalidates live work on disconnect", async () => { + const f = fixture(); + f.history([{ id: "historical", kind: 50182 }]); + const close = await receiveLifecycle( + scope, + () => true, + (e) => f.errors.push(e), + f.ipc, + f.relay, + ); + assert.equal( + f.calls.filter(([c]) => c === "receive_desktop_lifecycle").length, + 0, + ); + f.deliver({ id: "live", kind: 50182 }); + await tick(); + assert.equal( + f.calls.filter(([c]) => c === "receive_desktop_lifecycle").length, + 1, + ); + f.disconnect(); + f.deliver({ id: "late", kind: 50182 }); + await tick(); + assert.equal( + f.calls.filter(([c]) => c === "receive_desktop_lifecycle").length, + 1, + ); + assert.equal(f.errors.length, 1); + close(); +}); diff --git a/desktop/src/features/agents/desktopLifecycle.ts b/desktop/src/features/agents/desktopLifecycle.ts new file mode 100644 index 0000000000..7af772aa18 --- /dev/null +++ b/desktop/src/features/agents/desktopLifecycle.ts @@ -0,0 +1,334 @@ +import { invoke } from "@tauri-apps/api/core"; +import { relayClient } from "@/shared/api/relayClient"; +import type { RelayEvent } from "@/shared/api/types"; +import type { DesktopScope } from "./desktopList"; +import { + DESKTOP_STOP, + prepareStop, + readStopOutcome, + sendStop, +} from "./desktopStop"; + +export const DESKTOP_LIFECYCLE = 50182; +export const DESKTOP_LIFECYCLE_RESULT = 50183; +export type LifecycleAction = "start" | "restart" | "status"; +export type LifecycleOutcome = + | "running" + | "stopped" + | "provisioning_unavailable" + | "failed" + | "unknown"; +export type CurrentHost = { desktop: string; observation: string }; + +/** Captures identity and connection generation across every asynchronous step. */ +export function lifecycleClient( + scope: DesktopScope, + active: () => boolean, + ipc = invoke, + relay = relayClient, +) { + const epoch = relay.getSessionEpoch(); + const connection = relay.getConnectionGeneration(); + const check = () => { + if ( + !active() || + relay.getSessionEpoch() !== epoch || + relay.getConnectionGeneration() !== connection + ) + throw new Error("Desktop lifecycle scope changed"); + }; + const prepare = async ( + desktop: string, + agent: string, + action: LifecycleAction, + observed: string | null = null, + ) => { + check(); + const request = await ipc("prepare_desktop_lifecycle", { + ...scope, + desktop, + agent, + action, + observed, + }); + check(); + return request; + }; + const read = async (request: RelayEvent) => { + check(); + const events = await relay.fetchEvents({ + kinds: [DESKTOP_LIFECYCLE_RESULT], + authors: [scope.owner], + "#e": [request.id], + limit: 16, + }); + check(); + const outcome = await ipc( + "read_desktop_lifecycle_results", + { ...scope, request, events }, + ); + check(); + return outcome; + }; + const send = async ( + request: RelayEvent, + attempts = 15, + ): Promise => { + check(); + try { + await relay.publishEvent( + request, + "Delivery unconfirmed", + "Delivery failed", + check, + ); + } catch { + check(); /* Lost ACK may still have a signed result. */ + } + for (let i = 0; i < attempts; i++) { + check(); + const outcome = await read(request); + check(); + if (outcome !== "unknown") return outcome; + await new Promise((resolve) => setTimeout(resolve, 1000)); + } + return "unknown"; + }; + const sync = async () => { + let until: number | undefined; + 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, + }); + check(); + // No effects while a partial page could still hide a dominating Start. + await ipc("observe_desktop_placement", { + ...scope, + events, + reconcile: false, + }); + check(); + if (events.length < 256) { + await ipc("observe_desktop_placement", { + ...scope, + events: [], + reconcile: true, + }); + check(); + return; + } + const last = events.at(-1); + if (!last || last.id === before_id) + throw new Error("Placement history cursor did not advance"); + until = last.created_at; + before_id = last.id; + } + throw new Error( + "Placement history is incomplete; no launch was dispatched", + ); + }; + const current = async ( + agent: string, + desktops: string[], + ): Promise => { + await sync(); + check(); + const desired = await ipc<[string, string] | null>( + "read_desktop_placement", + { ...scope, agent }, + ); + check(); + // Probe actual native state, never infer current from last-heard/profile. + const candidates = desired ? [desired[0]] : [...new Set(desktops)]; + if (!candidates.length || candidates.length > 32) + throw new Error("Current Desktop is unknown"); + const observations = await Promise.all( + candidates.map(async (desktop) => { + const request = await prepare(desktop, agent, "status"); + return { + desktop, + observation: request.id, + outcome: await send(request, 3), + }; + }), + ); + check(); + const running = observations.filter((o) => o.outcome === "running"); + if ( + running.length !== 1 || + observations.some( + (o) => o.outcome !== "running" && o.outcome !== "stopped", + ) + ) + throw new Error( + "Current Desktop is unknown or ambiguous; choose explicit Start instead", + ); + return running[0]; + }; + const start = async (desktop: string, agent: string) => { + await sync(); + return prepare(desktop, agent, "start"); + }; + const restart = async (agent: string, desktops: string[]) => { + const host = await current(agent, desktops); + check(); + return prepare(host.desktop, agent, "restart", host.observation); + }; + /** Failed/unconfirmed Move is terminal in this invocation. No saved future + * Start, background callback, reopen replay or retry of a failed Move. */ + const move = async ( + agent: string, + destination: string, + desktops: string[], + onStage: (stage: string) => void, + ): Promise => { + const host = await current(agent, desktops); + check(); + if (host.desktop === destination) + throw new Error("Agent is already on the selected Desktop"); + const before = await ipc<[string, string] | null>( + "read_desktop_placement", + { ...scope, agent }, + ); + check(); + const stop = await prepareStop( + scope, + host.desktop, + agent, + active, + ipc, + relay, + ); + check(); + onStage( + "Waiting for source Desktop Stop; destination has not been started.", + ); + try { + await sendStop(scope, stop, active, relay); + } catch { + check(); + } + let stopped = false; + for (let i = 0; i < 15; i++) { + const result = await readStopOutcome(scope, stop, active, ipc, relay); + check(); + if (result === "failed") break; + if (result === "stopped") { + stopped = true; + break; + } + await new Promise((resolve) => setTimeout(resolve, 1000)); + } + if (!stopped) + throw new Error( + "Could not confirm source Stop; destination was not started. This Move will not continue later.", + ); + await sync(); + check(); + const after = await ipc<[string, string] | null>("read_desktop_placement", { + ...scope, + agent, + }); + check(); + // Stop may clear source, but another device's new placement must win. + if (after && (!before || after[1] !== before[1])) + throw new Error( + "Placement changed during Move; destination was not started", + ); + onStage("Source Stop confirmed. Requesting destination Start."); + const request = await prepare(destination, agent, "start"); + check(); + return send(request); + }; + return { check, prepare, read, send, sync, current, start, restart, move }; +} + +/** Subscribe first, then project history; live commands wait for complete + * initialization. Reconnect gets a new client epoch and never replays history. */ +export async function receiveLifecycle( + scope: DesktopScope, + active: () => boolean, + onError: (message: string) => void, + ipc = invoke, + relay = relayClient, +) { + let client: ReturnType; + let initialized: () => void = () => {}; + const ready = new Promise((resolve) => { + initialized = resolve; + }); + 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 }, + ); + client.check(); + if (result) + await relay.publishEvent( + result, + "Result delivery unconfirmed", + "Result delivery failed", + client.check, + ); + }) + .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 = lifecycleClient(scope, active, ipc, relay); + try { + await client.sync(); + client.check(); + initialized(); + } catch (error) { + close(); + // Release queued callbacks into a permanently invalidated client. + client = lifecycleClient(scope, () => false, ipc, relay); + initialized(); + throw error; + } + return close; +} diff --git a/desktop/src/shared/api/relayClientSession.ts b/desktop/src/shared/api/relayClientSession.ts index 3839d01c7e..2f94a65bf9 100644 --- a/desktop/src/shared/api/relayClientSession.ts +++ b/desktop/src/shared/api/relayClientSession.ts @@ -115,6 +115,11 @@ export class RelayClient { getSessionEpoch() { return this.sessionEpoch; } + + /** Invalidates one-shot lifecycle coordinators on a transport interruption. */ + getConnectionGeneration() { + return this.connectionGeneration; + } disconnect() { const error = new Error("Relay disconnected for community switch.");