feat(desktop): coordinate failure-final Move and exact lifecycle retry

Signed-off-by: Logan Johnson <loganj@squareup.com>

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();
+});

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<RelayEvent>("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<LifecycleOutcome>(
+      "read_desktop_lifecycle_results",
+      { ...scope, request, events },
+    );
+    check();
+    return outcome;
+  };
+  const send = async (
+    request: RelayEvent,
+    attempts = 15,
+  ): Promise<LifecycleOutcome> => {
+    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<CurrentHost> => {
+    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<LifecycleOutcome> => {
+    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<typeof lifecycleClient>;
+  let initialized: () => void = () => {};
+  const ready = new Promise<void>((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<RelayEvent | null>(
+            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;
+}

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.");