Fork Horse repository, listed by its owner. Publisher snapshot.
packages/forkhorse-gatekeeper/test/sidebar.test.ts
Snapshot at 34e3175
1import { installDom } from "../../forkhorse-dashboard/test/dom.ts";
2import assert from "node:assert/strict";
3import { createRequire } from "node:module";
4import { setImmediate as turn } from "node:timers/promises";
5import { test } from "node:test";
6import { newMessagePortRpcSession, RpcStub, RpcTarget } from "capnweb";
7import { createRateLimitedCapability } from "../../../cloudflare-os/packages/workshop-frontend/src/rateLimitedCapability.ts";
8import { fixtureState } from "../../forkhorse-dashboard/test/fixtures.ts";
9import { stateView } from "../../forkhorse-dashboard/src/model.ts";
10import { SnapshotFeed } from "../src/feed.ts";
11import { connectSidebar, OWNER_UNAVAILABLE } from "../src/client.ts";
12import { mountSidebar } from "../src/mount.ts";
13import type { SidebarUi } from "../src/ui-types.ts";
14import { FixtureUi } from "./fixture-ui.ts";
15
16const { JSDOM } = createRequire(import.meta.url)("jsdom") as {
17 JSDOM: new (html: string) => {
18 window: { document: Document; MouseEvent: typeof MouseEvent; close(): void };
19 };
20};
21const fixture = () => stateView(fixtureState());
22async function until(predicate: () => boolean) {
23 for (let i = 0; i < 300 && !predicate(); i++) await turn();
24 assert.ok(predicate(), "asynchronous condition settled");
25}
26
27function connection(push = true) {
28 const feed = new SnapshotFeed(fixture());
29 const service = new FixtureUi(feed, push);
30 // Actual frozen upstream limiter, not a copy of its budget math.
31 const limited = createRateLimitedCapability(service, {
32 maxConcurrency: 8,
33 maxCallsPerMinute: 600,
34 maxPendingCalls: 128,
35 onRateLimit: "throttle",
36 label: "Fixture sidebar",
37 });
38 const { port1, port2 } = new MessageChannel();
39 const server = newMessagePortRpcSession(port2, limited.capability);
40 const remote = newMessagePortRpcSession<SidebarUi>(port1);
41 return {
42 feed,
43 service,
44 remote,
45 close() {
46 feed.dispose();
47 limited.dispose();
48 server[Symbol.dispose]();
49 remote[Symbol.dispose]();
50 port1.close();
51 port2.close();
52 },
53 };
54}
55
56test("240 quarter-second pushed batches traverse RPC and pinned limiter without inbound refresh calls", async (t) => {
57 t.mock.timers.enable({ apis: ["setTimeout", "Date"] });
58 const link = connection();
59 t.after(() => link.close());
60 const errors: string[] = [];
61 const client = await connectSidebar(link.remote, (message) => errors.push(message));
62 t.after(() => client.dispose());
63 let changes = 0;
64 client.transport.subscribe!(client.project, () => changes++);
65 for (let i = 1; i <= 240; i++) {
66 const state = fixture();
67 state.packets[0]!.title = `Fixture ${i}`;
68 link.feed.publish(state);
69 t.mock.timers.tick(250);
70 await until(() => changes === i);
71 }
72 assert.equal((await client.transport.getState("demo")).packets[0]!.title, "Fixture 240");
73 assert.deepEqual(link.service.calls, { snapshot: 1, subscribe: 1, poll: 0 });
74 assert.deepEqual(errors, []);
75 await assert.rejects(client.transport.approve("demo", "review-1"), {
76 message: OWNER_UNAVAILABLE,
77 });
78 await assert.rejects(client.transport.promote("demo", "review-1"), {
79 message: OWNER_UNAVAILABLE,
80 });
81 await assert.rejects(client.transport.getState("other"), /another project/);
82});
83
84test("fallback uses one long poll and stays below 600 calls for 240 batches", async (t) => {
85 t.mock.timers.enable({ apis: ["setTimeout", "Date"] });
86 const link = connection(false);
87 t.after(() => link.close());
88 const errors: string[] = [];
89 const client = await connectSidebar(link.remote, (message) => errors.push(message));
90 t.after(() => client.dispose());
91 let changes = 0;
92 client.transport.subscribe!(client.project, () => changes++);
93 for (let i = 1; i <= 240; i++) {
94 const state = fixture();
95 state.packets[0]!.title = `Fallback ${i}`;
96 link.feed.publish(state);
97 t.mock.timers.tick(250);
98 await until(() => changes === i);
99 }
100 assert.ok(link.service.calls.poll <= 241);
101 assert.equal(link.service.calls.snapshot, 1);
102 assert.deepEqual(errors, []);
103});
104
105test("feed coalesces bursts and slow receivers, releases polls, refuses cross-project state", async (t) => {
106 t.mock.timers.enable({ apis: ["setTimeout"] });
107 const feed = new SnapshotFeed(fixture());
108 t.after(() => feed.dispose());
109 let release!: () => void;
110 const received: number[] = [];
111 let disposed = 0;
112 const close = feed.subscribe({
113 async receive(batch) {
114 received.push(batch.version);
115 if (batch.version === 0)
116 await new Promise<void>((resolve) => {
117 release = resolve;
118 });
119 },
120 dispose() {
121 disposed++;
122 },
123 });
124 await until(() => received.length === 1);
125 for (let i = 0; i < 20; i++) feed.publish(fixture());
126 const waiting = feed.waitForChanges(0);
127 await assert.rejects(feed.waitForChanges(0), /one pending poll/);
128 t.mock.timers.tick(250);
129 assert.equal((await waiting).version, 1);
130 feed.publish(fixture());
131 t.mock.timers.tick(250);
132 assert.deepEqual(received, [0]);
133 release();
134 await until(() => received.length === 2);
135 assert.deepEqual(received, [0, 2]);
136 const state = fixture();
137 state.project.id = "foreign";
138 assert.throws(() => feed.publish(state), /bound to one project/);
139 await assert.rejects(feed.waitForChanges(99), /Invalid UI version/);
140 const timeout = feed.waitForChanges(2);
141 t.mock.timers.tick(25_000);
142 assert.equal((await timeout).version, 2);
143 close();
144 assert.equal(disposed, 1);
145 const pending = feed.waitForChanges(2);
146 feed.dispose();
147 await assert.rejects(pending, /closed/);
148});
149
150for (const fixtureMode of [true, false])
151 test(`dashboard renders with fixture=${fixtureMode}, owner actions hidden, and host openPrompt`, async (t) => {
152 const dom = new JSDOM('<div id="root"></div>');
153 const link = connection();
154 t.after(() => link.close());
155 const client = await connectSidebar(link.remote, () => {});
156 t.after(() => client.dispose());
157 installDom(dom.window as unknown as Window & typeof globalThis);
158 const root = dom.window.document.getElementById("root")!;
159 const prompts: string[] = [];
160 const view = mountSidebar(root, {
161 brand: { name: "Fixture product", tagline: "Fixture collaboration" },
162 project: client.project,
163 transport: client.transport,
164 fixture: fixtureMode,
165 async openPrompt(prompt) {
166 prompts.push(prompt);
167 },
168 });
169 t.after(() => {
170 view.dispose();
171 dom.window.close();
172 });
173 await until(() => !!root.querySelector("#reviews"));
174 assert.equal(
175 root.querySelector('button[data-action="approve"], button[data-action="promote"]'),
176 null,
177 );
178 assert.match(
179 root.querySelector(".agent-notice")!.textContent!,
180 /Promote and approve stay with the owner/,
181 );
182 assert.equal(root.querySelector('[role="combobox"]'), null);
183 const ask = [...root.querySelectorAll("button")].find(
184 (button) => button.textContent === "Ask an agent",
185 )!;
186 ask.click();
187 await until(() => prompts.length === 1);
188 assert.match(prompts[0]!, /project demo in Fixture product/);
189 assert.match(prompts[0]!, /Call context\(\) first/);
190 assert.match(prompts[0]!, /never repeat join\(\) to resume/);
191 assert.match(prompts[0]!, /Stop for each approval decision/);
192 assert.match(prompts[0]!, /requestReview\(packet, repoName, sha\)/);
193 assert.match(root.textContent!, /your pushes need approval on your fork/);
194 assert.ok(root.querySelector(".tab-gutter + .content .agent-notice"));
195 assert.ok(root.querySelector(".page-header [data-action=ask]"));
196 if (fixtureMode) assert.match(root.textContent!, /FIXTURE · OFFLINE/);
197 assert.doesNotMatch(
198 root.textContent!,
199 /Owner workspace|Owner Access session|Owner actions are authorized/,
200 );
201 assert.equal(root.querySelector(".sidebar, .topbar"), null);
202 assert.equal(root.querySelector("#open-project"), null);
203 const reviewTab = [...root.querySelectorAll<HTMLButtonElement>('[role="tab"]')].find((button) =>
204 button.textContent!.startsWith("Review"),
205 )!;
206 reviewTab.click();
207 await until(() => !root.querySelector<HTMLElement>("#view-review")!.hidden);
208 assert.equal(root.querySelectorAll('[role="tab"]').length, 5); // Overview, Work, Review, Agents, Races (FH-081)
209 });
210
211test("silent push channel falls back after 30 seconds without overlapping polls", async (t) => {
212 t.mock.timers.enable({ apis: ["setTimeout", "Date"] });
213 const link = connection();
214 t.after(() => link.close());
215 const client = await connectSidebar(link.remote, () => {});
216 t.after(() => client.dispose());
217 await turn();
218 t.mock.timers.tick(30_000);
219 await until(() => link.service.calls.poll === 1);
220 assert.equal(link.service.calls.snapshot, 1);
221 let updates = 0;
222 client.transport.subscribe!("demo", () => updates++);
223 link.feed.publish(fixture());
224 t.mock.timers.tick(250);
225 await until(() => updates === 1);
226 assert.ok(link.service.calls.poll <= 2);
227});
228
229test("failed long poll retries at bounded backoff, keeps last snapshot and renews theme", async (t) => {
230 t.mock.timers.enable({ apis: ["setTimeout", "Date"] });
231 const link = connection();
232 t.after(() => link.close());
233 let down = false,
234 themes = 0;
235 const times: number[] = [];
236 const ui = new (class extends RpcTarget {
237 async snapshot() {
238 times.push(Date.now());
239 if (down) throw new Error("Offline fixture restart");
240 return link.feed.snapshot();
241 }
242 async subscribe(receiver: Parameters<SidebarUi["subscribe"]>[0]) {
243 using forwarded = new RpcStub(receiver);
244 return await link.service.subscribe(forwarded);
245 }
246 async waitForChanges(after: number) {
247 if (down) throw new Error("Offline fixture disconnect");
248 return link.service.waitForChanges(after);
249 }
250 })();
251 const status: string[] = [];
252 const client = await connectSidebar(
253 ui,
254 (message) => status.push(message),
255 async () => {
256 themes++;
257 },
258 );
259 t.after(() => client.dispose());
260 let paused = false;
261 client.transport.live!.subscribe((value) => {
262 paused = value;
263 });
264 await turn();
265 down = true;
266 t.mock.timers.tick(30_000);
267 await until(() => paused);
268 assert.equal((await client.transport.getState("demo")).project.id, "demo");
269 for (const delay of [1000, 2000, 4000, 8000, 16000, 30000, 30000]) {
270 const count = times.length;
271 t.mock.timers.tick(delay - 1);
272 await turn();
273 assert.equal(times.length, count);
274 t.mock.timers.tick(1);
275 await until(() => times.length === count + 1);
276 await turn();
277 }
278 down = false;
279 await Promise.all([client.transport.live!.reconnect(), client.transport.live!.reconnect()]);
280 assert.equal(paused, false);
281 assert.equal(themes, 9);
282 assert.equal(link.service.calls.subscribe, 2);
283 assert.equal(status.at(-1), "");
284 client.dispose();
285 const calls = times.length;
286 t.mock.timers.tick(90_000);
287 await turn();
288 assert.equal(times.length, calls);
289});
290
291for (const automatic of [false, true])
292 test(`hung poll recovers with a fresh capability via ${automatic ? "automatic backoff" : "Reconnect"}`, async (t) => {
293 t.mock.timers.enable({ apis: ["setTimeout", "Date"] });
294 let rejectOld!: (error: Error) => void;
295 let oldPolls = 0,
296 acquired = 0,
297 snapshots = 0,
298 subscriptions = 0,
299 disposed = 0;
300 let currentTheme = "light",
301 appliedTheme = "";
302 const next = fixture();
303 next.packets[0]!.title = "Fresh recovered data";
304 const old = new (class extends RpcTarget {
305 async snapshot() {
306 return { version: 0, state: fixture() };
307 }
308 async subscribe(): Promise<never> {
309 throw new Error("Use fallback");
310 }
311 async waitForChanges(): Promise<never> {
312 oldPolls++;
313 return new Promise((_resolve, reject) => {
314 rejectOld = reject;
315 });
316 }
317 [Symbol.dispose]() {
318 disposed++;
319 }
320 })();
321 const fresh = new (class extends RpcTarget {
322 async snapshot() {
323 snapshots++;
324 return { version: 0, state: next };
325 }
326 async subscribe() {
327 subscriptions++;
328 return new (class extends RpcTarget {
329 async close() {}
330 })();
331 }
332 async waitForChanges() {
333 return new Promise<never>(() => {});
334 }
335 })();
336 const errors: string[] = [];
337 const client = await connectSidebar(
338 old,
339 (message) => errors.push(message),
340 async () => {
341 appliedTheme = currentTheme;
342 },
343 async () => {
344 acquired++;
345 return fresh;
346 },
347 );
348 t.after(() => client.dispose());
349 let paused = false;
350 client.transport.live!.subscribe((value) => {
351 paused = value;
352 });
353 t.mock.timers.tick(0);
354 await until(() => oldPolls === 1);
355 t.mock.timers.tick(30_000);
356 await until(() => paused);
357 currentTheme = "dark";
358 if (automatic) {
359 t.mock.timers.tick(59_999);
360 await turn();
361 assert.equal(acquired, 0);
362 t.mock.timers.tick(1);
363 await until(() => subscriptions === 1);
364 } else
365 await Promise.all([client.transport.live!.reconnect(), client.transport.live!.reconnect()]);
366 assert.equal(acquired, 1);
367 assert.equal(snapshots, 1);
368 assert.equal(subscriptions, 1);
369 assert.equal(oldPolls, 1);
370 assert.equal(disposed, 1);
371 assert.equal(paused, false);
372 assert.equal(appliedTheme, "dark");
373 assert.equal(
374 (await client.transport.getState("demo")).packets[0]!.title,
375 "Fresh recovered data",
376 );
377 const count = errors.length;
378 rejectOld(new Error("Late retired relay rejection"));
379 await turn();
380 await turn();
381 assert.equal(errors.length, count);
382 assert.equal(paused, false);
383 assert.equal(acquired, 1);
384 });
385
386test("a second stalled generation cannot abandon another call until the first settles", async (t) => {
387 t.mock.timers.enable({ apis: ["setTimeout", "Date"] });
388 let finishOld!: () => void,
389 oldPolls = 0,
390 freshSnapshots = 0,
391 acquired = 0;
392 const old = new (class extends RpcTarget {
393 async snapshot() {
394 return { version: 0, state: fixture() };
395 }
396 async subscribe(): Promise<never> {
397 throw new Error("Use fallback");
398 }
399 async waitForChanges() {
400 oldPolls++;
401 await new Promise<void>((resolve) => {
402 finishOld = resolve;
403 });
404 return { version: 999, state: fixture() };
405 }
406 })();
407 const fresh = new (class extends RpcTarget {
408 async snapshot(): Promise<never> {
409 freshSnapshots++;
410 return new Promise(() => {});
411 }
412 async subscribe(): Promise<never> {
413 throw new Error("Not reached");
414 }
415 async waitForChanges(): Promise<never> {
416 throw new Error("Not reached");
417 }
418 })();
419 const client = await connectSidebar(
420 old,
421 () => {},
422 async () => {},
423 async () => {
424 acquired++;
425 return fresh;
426 },
427 );
428 t.after(() => client.dispose());
429 t.mock.timers.tick(0);
430 await until(() => oldPolls === 1);
431 t.mock.timers.tick(30_000);
432 await turn();
433 void client.transport.live!.reconnect();
434 await until(() => freshSnapshots === 1);
435 t.mock.timers.tick(30_000);
436 await turn();
437 await client.transport.live!.reconnect();
438 t.mock.timers.tick(60_000);
439 await turn();
440 assert.equal(acquired, 1);
441 assert.equal(oldPolls, 1);
442 assert.equal(freshSnapshots, 1);
443 finishOld();
444 await turn();
445 await turn();
446 assert.equal(
447 (await client.transport.getState("demo")).packets[0]!.title,
448 fixture().packets[0]!.title,
449 );
450 void client.transport.live!.reconnect();
451 await until(() => acquired === 2);
452});
453
454test("disposing releases the owned subscription stub even when remote close hangs", async () => {
455 let released = 0;
456 const subscription = new (class extends RpcTarget {
457 async close(): Promise<void> {
458 await new Promise<void>(() => {});
459 }
460 [Symbol.dispose]() {
461 released++;
462 }
463 })();
464 const ui = new (class extends RpcTarget {
465 async snapshot() {
466 return { version: 0, state: fixture() };
467 }
468 async subscribe() {
469 return subscription;
470 }
471 async waitForChanges() {
472 return { version: 0, state: fixture() };
473 }
474 })();
475 const client = await connectSidebar(ui, () => {});
476 client.dispose();
477 assert.equal(released, 1);
478});
479
480test("late theme reply from an abandoned generation cannot repaint the recovered theme", async (t) => {
481 t.mock.timers.enable({ apis: ["setTimeout", "Date"] });
482 let themes = 0,
483 acquired = 0,
484 finishOldTheme!: (theme: string) => void;
485 let applied = "",
486 hostTheme = "light";
487 const old = new (class extends RpcTarget {
488 async snapshot() {
489 return { version: 0, state: fixture() };
490 }
491 async subscribe(): Promise<never> {
492 throw new Error("Use fallback");
493 }
494 async waitForChanges(): Promise<never> {
495 throw new Error("Old relay failed");
496 }
497 })();
498 const fresh = new (class extends RpcTarget {
499 async snapshot() {
500 return { version: 0, state: fixture() };
501 }
502 async subscribe() {
503 return new (class extends RpcTarget {
504 async close() {}
505 })();
506 }
507 async waitForChanges() {
508 return new Promise<never>(() => {});
509 }
510 })();
511 const client = await connectSidebar(
512 old,
513 () => {},
514 async (isCurrent) => {
515 themes++;
516 const theme =
517 themes === 2
518 ? await new Promise<string>((resolve) => {
519 finishOldTheme = resolve;
520 })
521 : hostTheme;
522 if (isCurrent()) applied = theme;
523 },
524 async () => {
525 acquired++;
526 return fresh;
527 },
528 );
529 t.after(() => client.dispose());
530 t.mock.timers.tick(0);
531 await turn();
532 await turn();
533 t.mock.timers.tick(1000);
534 await until(() => themes === 2);
535 t.mock.timers.tick(30_000);
536 await turn();
537 hostTheme = "dark";
538 await client.transport.live!.reconnect();
539 assert.equal(acquired, 2);
540 assert.equal(themes, 3);
541 assert.equal(applied, "dark");
542 finishOldTheme("light");
543 await turn();
544 await turn();
545 assert.equal(applied, "dark");
546});
547