packages/forkhorse-gatekeeper/src/client.ts

1import { RpcTarget } from "capnweb";
2import { assertState } from "../../forkhorse-dashboard/src/model.ts";
3import type { DashboardTransport } from "../../forkhorse-dashboard/src/contracts.ts";
4import type { SidebarBatch, SidebarUi, Subscription, UpdateReceiver } from "./ui-types.ts";
5
6export const OWNER_UNAVAILABLE = "Approve and Promote are unavailable for this agent connection.";
7
8function disposeStub(value: object | undefined) {
9  if (value && Symbol.dispose in value) {
10    const dispose = value[Symbol.dispose];
11    if (typeof dispose === "function") dispose.call(value);
12  }
13}
14
15/** One active session and at most one abandoned remote call awaiting settlement. */
16export async function connectSidebar(
17  initialUi: SidebarUi,
18  onError: (message: string) => void,
19  resubscribeTheme: (isCurrent: () => boolean) => Promise<void> = async () => {},
20  acquireUi?: (isCurrent: () => boolean) => Promise<SidebarUi>,
21) {
22  let batch = await initialUi.snapshot();
23  assertState(batch.state);
24  if (!Number.isSafeInteger(batch.version) || batch.version < 0)
25    throw new Error("Invalid sidebar version.");
26  const project = batch.state.project.id;
27  let closed = false;
28  let paused = false;
29  let timer: ReturnType<typeof setTimeout> | undefined;
30  let failures = 0;
31  const deadlines = new Set<ReturnType<typeof setTimeout>>();
32  const listeners = new Set<() => void>();
33  const statusListeners = new Set<(paused: boolean) => void>();
34  type Session = {
35    generation: number;
36    ui?: SidebarUi;
37    receiver?: Receiver;
38    subscription?: Subscription;
39    pending?: Promise<unknown>;
40    stalled: boolean;
41  };
42  let session: Session = { generation: 0, ui: initialUi, stalled: false };
43  let running: { session: Session; promise: Promise<void> } | undefined;
44  let abandoned: Promise<unknown> | undefined;
45  const current = (value: Session) => !closed && value === session;
46
47  function status(value: boolean) {
48    if (closed || paused === value) return;
49    paused = value;
50    onError(value ? "Live updates paused." : "");
51    for (const listener of statusListeners) listener(value);
52  }
53  function accept(owner: Session, next: SidebarBatch, snapshot = false): void {
54    if (!current(owner)) return;
55    assertState(next.state);
56    if (
57      next.state.project.id !== project ||
58      !Number.isSafeInteger(next.version) ||
59      next.version < 0
60    )
61      throw new Error("Invalid sidebar update.");
62    if (!snapshot && next.version < batch.version) throw new Error("Sidebar version moved back.");
63    if (snapshot || next.version > batch.version) {
64      batch = next;
65      for (const listener of listeners) listener();
66    }
67    status(false);
68    failures = 0;
69  }
70  function schedule(operation: () => Promise<void>, delay: number) {
71    clearTimeout(timer);
72    if (!closed)
73      timer = setTimeout(() => {
74        timer = undefined;
75        void operation();
76      }, delay);
77  }
78  async function observe<T>(owner: Session, operation: PromiseLike<T>): Promise<T> {
79    const pending = Promise.resolve(operation);
80    owner.pending = pending;
81    const deadline = setTimeout(() => {
82      if (!current(owner)) return;
83      owner.stalled = true;
84      status(true);
85      // Give normal relay restoration time, then try a fresh capability without a call storm.
86      schedule(reconnect, 60_000);
87    }, 30_000);
88    deadlines.add(deadline);
89    try {
90      const result = await pending;
91      if (!current(owner)) {
92        if (result && typeof result === "object") disposeStub(result);
93        throw new Error("Sidebar session retired.");
94      }
95      return result;
96    } finally {
97      clearTimeout(deadline);
98      deadlines.delete(deadline);
99      if (owner.pending === pending) {
100        owner.pending = undefined;
101        owner.stalled = false;
102      }
103    }
104  }
105  async function release(owner: Session, remote = true) {
106    if (owner.receiver) owner.receiver.active = false;
107    owner.receiver = undefined;
108    const previous = owner.subscription;
109    owner.subscription = undefined;
110    let closing: Promise<void> | undefined;
111    try {
112      if (remote) closing = previous?.close();
113    } catch {
114      /* Broken remote. */
115    }
116    disposeStub(previous);
117    try {
118      if (closing) await observe(owner, closing);
119    } catch {
120      /* Disconnected or retired. */
121    }
122  }
123  function single(owner: Session, operation: () => Promise<void>): Promise<void> {
124    if (!current(owner)) return Promise.resolve();
125    if (running) return running.promise;
126    const entry = { session: owner, promise: Promise.resolve() };
127    running = entry;
128    entry.promise = Promise.resolve()
129      .then(operation)
130      .catch(() => {
131        if (!current(owner)) return; // Late failures cannot overwrite a recovered generation.
132        status(true);
133        schedule(reconnect, Math.min(30_000, 1000 * 2 ** Math.min(failures++, 5)));
134      })
135      .finally(() => {
136        if (running === entry) running = undefined;
137      });
138    return entry.promise;
139  }
140  function watch(owner: Session) {
141    if (current(owner)) schedule(() => single(owner, () => poll(owner)), 30_000);
142  }
143  async function poll(owner: Session) {
144    await release(owner);
145    while (current(owner))
146      accept(owner, await observe(owner, owner.ui!.waitForChanges(batch.version)));
147  }
148  class Receiver extends RpcTarget implements UpdateReceiver {
149    active = true;
150    readonly owner: Session;
151    constructor(owner: Session) {
152      super();
153      this.owner = owner;
154    }
155    async receive(next: SidebarBatch) {
156      if (!this.active || !current(this.owner)) return;
157      accept(this.owner, next);
158      watch(this.owner);
159    }
160  }
161  async function subscribe(owner: Session) {
162    try {
163      owner.receiver = new Receiver(owner);
164      const subscription = await observe(owner, owner.ui!.subscribe(owner.receiver));
165      if (!current(owner)) {
166        disposeStub(subscription);
167        return;
168      }
169      owner.subscription = subscription;
170      status(false);
171      watch(owner);
172    } catch {
173      if (current(owner)) schedule(() => single(owner, () => poll(owner)), 0);
174    }
175  }
176  async function recover(owner: Session) {
177    await release(owner);
178    if (!current(owner)) return;
179    if (acquireUi)
180      owner.ui = await observe(
181        owner,
182        acquireUi(() => current(owner)),
183      );
184    else owner.ui = initialUi;
185    // Host-local and current: apply theme even if the replacement UI snapshot then fails.
186    await observe(
187      owner,
188      resubscribeTheme(() => current(owner)),
189    );
190    accept(owner, await observe(owner, owner.ui.snapshot()), true);
191    if (current(owner)) await subscribe(owner);
192  }
193  async function reconnect(): Promise<void> {
194    if (closed) return;
195    clearTimeout(timer);
196    if (running) {
197      const old = running.session;
198      if (!old.stalled) return running.promise;
199      // A second hung attempt must wait while the first abandoned call is still outstanding.
200      if (abandoned || !acquireUi) {
201        schedule(reconnect, 60_000);
202        return;
203      }
204      const pending = old.pending!;
205      abandoned = pending;
206      void pending.then(
207        () => {
208          if (abandoned === pending) abandoned = undefined;
209        },
210        () => {
211          if (abandoned === pending) abandoned = undefined;
212        },
213      );
214      session = { generation: old.generation + 1, stalled: false };
215      running = undefined;
216      // Local disposal must not add a remote close call behind the already hung operation.
217      void release(old, false);
218      disposeStub(old.ui);
219    } else {
220      const old = session;
221      session = { generation: old.generation + 1, stalled: false };
222      void release(old, false);
223      if (acquireUi) disposeStub(old.ui);
224    }
225    const owner = session;
226    return single(owner, () => recover(owner));
227  }
228  await single(session, async () => {
229    const owner = session;
230    await observe(
231      owner,
232      resubscribeTheme(() => current(owner)),
233    );
234    if (current(owner)) await subscribe(owner);
235  });
236  function requireProject(id: string) {
237    if (id !== project) throw new Error("This sidebar connection is bound to another project.");
238  }
239  const transport: DashboardTransport = {
240    live: {
241      subscribe(listener) {
242        statusListeners.add(listener);
243        listener(paused);
244        return () => statusListeners.delete(listener);
245      },
246      reconnect,
247    },
248    async getState(id) {
249      requireProject(id);
250      return structuredClone(batch.state);
251    },
252    async getActivity(id, after = 0) {
253      requireProject(id);
254      const activity = (batch.state.recentActivity ?? []).filter((event) => event.sequence > after);
255      return { activity, nextAfter: activity.at(-1)?.sequence ?? after };
256    },
257    async approve() {
258      throw new Error(OWNER_UNAVAILABLE);
259    },
260    async promote() {
261      throw new Error(OWNER_UNAVAILABLE);
262    },
263    subscribe(id, invalidate) {
264      requireProject(id);
265      listeners.add(invalidate);
266      return () => listeners.delete(invalidate);
267    },
268  };
269  return {
270    project,
271    transport,
272    dispose() {
273      closed = true;
274      clearTimeout(timer);
275      for (const deadline of deadlines) clearTimeout(deadline);
276      deadlines.clear();
277      listeners.clear();
278      statusListeners.clear();
279      void release(session, false);
280      if (acquireUi) disposeStub(session.ui);
281    },
282  };
283}
284