clover's git

the-knife

1	import { createCommandPermissions } from "./commandPermissions.ts";
2	import { RpcPermissionGuard } from "../rpc/client.ts";
3	import { vi } from "vite-plus/test";
4	import {
5	EnvironmentId,
6	AuthSourceControlWriteScope,
7	AuthOrchestrationOperateScope,
8	ProjectId,
9	PullRequestOperationError,
10	WS_METHODS,
11	type PullRequestStack,
12	type MergebotRepoQueue,
13	type AuthSessionState,
14	} from "@t3tools/contracts";
15	import { expect, it } from "@effect/vitest";
16	import * as Data from "effect/Data";
17	import * as Deferred from "effect/Deferred";
18	import * as Fiber from "effect/Fiber";
19	import * as TestClock from "effect/testing/TestClock";
20	import * as Effect from "effect/Effect";
21	import * as Latch from "effect/Latch";
22	import * as Layer from "effect/Layer";
23	import * as Option from "effect/Option";
24	import * as PubSub from "effect/PubSub";
25	import * as Stream from "effect/Stream";
26	import * as SubscriptionRef from "effect/SubscriptionRef";
27	import { AsyncResult, Atom, AtomRegistry } from "effect/reactivity";
28
29	import {
30	AVAILABLE_CONNECTION_STATE,
31	ConnectionTransientError,
32	PrimaryConnectionTarget,
33	SshConnectionTarget,
34	type PreparedConnection,
35	type SupervisorConnectionState,
36	} from "../connection/model.ts";
37	import * as EnvironmentRegistry from "../connection/registry.ts";
38	import { SshConnectionProfile, type ConnectionCatalogEntry } from "../connection/catalog.ts";
39	import * as ConnectionProfileStore from "../connection/profileStore.ts";
40	import * as EnvironmentSupervisor from "../connection/supervisor.ts";
41	import type { WsRpcProtocolClient } from "../rpc/protocol.ts";
42	import type { RpcSession } from "../rpc/session.ts";
43	import {
44	createLinkedPullRequestSummaryAtomFamily,
45	createPullRequestEnvironmentAtoms,
46	createPullRequestStackAtomFamily,
47	createMergebotQueueAtomFamily,
48	} from "./pullRequests.ts";
49	import * as PullRequestDiffLoader from "./pullRequestDiffHttp.ts";
50	import { executeAtomQuery } from "./runtime.ts";
51	import { createPullRequestRouter } from "./pullRequestRouting.ts";
52	import { GitHubRoutingPermissions } from "../connection/githubRoutingPermissions.ts";
53
54	const trustedRouting = {
55	get: () => Effect.succeed("read-write" as const),
56	changes: Stream.empty,
57	set: () => Effect.void,
58	forget: () => Effect.void,
59	};
60
61	class MutationRefused extends Data.TaggedError("MutationRefused") {}
62
63	for (const scenario of [
64	"prefers the local environment with the same github account",
65	"falls back before mutation when the local account differs",
66	"never retries an ambiguous mutation failure",
67	"returns a fast local source read without checking alternate identities",
68	"keeps single-environment requests free of identity lookups",
69	"keeps a local origin ahead of another local environment",
70	"keeps mutations on an old origin server without retrying them",
71	"skips an old alternate server before dispatching a mutation",
72	"returns a successful mutation when source invalidation stalls",
73	"does not dispatch after routing permission is revoked during the probe",
74	"keeps mutations on the origin when the alternate is disabled",
75	"does not dispatch after the alternate is disabled during the probe",
76	"preserves a local source account rejection when alternate accounts differ",
77	] as const) {
78	(scenario === "returns a successful mutation when source invalidation stalls"
79	? it.live
80	: it.effect)(scenario, () =>
81	Effect.scoped(
82	Effect.gen(function* () {
83	const calls: string[] = [];
84	const inputs: unknown[] = [];
85	let trusted = true;
86	let disableAlternate = Effect.void;
87	const switchedAccount =
88	scenario === "preserves a local source account rejection when alternate accounts differ";
89	const mismatch =
90	scenario === "falls back before mutation when the local account differs" ||
91	switchedAccount;
92	const ambiguous = scenario === "never retries an ambiguous mutation failure";
93	const reading =
94	scenario === "returns a fast local source read without checking alternate identities";
95	const single = scenario === "keeps single-environment requests free of identity lookups";
96	const localOrigin =
97	scenario === "keeps a local origin ahead of another local environment" ||
98	switchedAccount ||
99	reading;
100	const oldOrigin =
101	scenario === "keeps mutations on an old origin server without retrying them";
102	const oldAlternate =
103	scenario === "skips an old alternate server before dispatching a mutation";
104	const failure = new PullRequestOperationError({
105	operation: switchedAccount ? "routeIdentity" : "runAction",
106	detail: "connection lost after dispatch",
107	});
108	const clientFor = (local: boolean) => {
109	const name = local ? "local" : "origin";
110	return {
111	[local ? WS_METHODS.pullRequestsRoutingIdentity : WS_METHODS.pullRequestsRouting]: (
112	input: unknown,
113	) =>
114	Effect.gen(function* () {
115	if (local) expect(input).toEqual({ host: "github.com" });
116	calls.push(`${name}:identity`);
117	if (
118	local &&
119	scenario === "does not dispatch after the alternate is disabled during the probe"
120	)
121	yield* disableAlternate;
122	if (
123	local &&
124	scenario ===
125	"does not dispatch after routing permission is revoked during the probe"
126	)
127	trusted = false;
128	if ((!local && oldOrigin) || (local && oldAlternate)) {
129	return yield* Effect.die(
130	`Unknown request tag: ${WS_METHODS.pullRequestsRouting}`,
131	);
132	}
133	return {
134	host: "github.com",
135	provider: "github",
136	viewer: "maria-rcks",
137	accountId: local && mismatch ? "456" : "123",
138	};
139	}),
140	[WS_METHODS.pullRequestsRunAction]: (input: unknown) =>
141	Effect.gen(function* () {
142	calls.push(`${name}:mutation`);
143	inputs.push(input);
144	if (
145	!local &&
146	switchedAccount &&
147	(input as { expectedAccountId?: string }).expectedAccountId === "123"
148	)
149	return yield* failure;
150	if (local && ambiguous) return yield* failure;
151	}),
152	[WS_METHODS.pullRequestsSummary]: (input: { allowStale?: boolean }) =>
153	Effect.gen(function* () {
154	calls.push(`${name}:read`);
155	expect(input.allowStale).toBe(false);
156	if (local) return yield* failure;
157	return null;
158	}),
159	[WS_METHODS.pullRequestsInvalidate]: () =>
160	Effect.gen(function* () {
161	calls.push(`${name}:invalidate`);
162	if (
163	!local &&
164	scenario === "returns a successful mutation when source invalidation stalls"
165	) {
166	return yield* Effect.never;
167	}
168	}),
169	} as unknown as WsRpcProtocolClient;
170	};
171	const { environmentRegistry, supervisor } = yield* makeTestRuntime(
172	clientFor(false),
173	single ? undefined : clientFor(true),
174	localOrigin,
175	);
176	disableAlternate = SubscriptionRef.update(
177	environmentRegistry.entries,
178	(entries) =>
179	new Map(
180	[...entries].map(([id, entry]) => [
181	id,
182	id === TARGET.environmentId ? entry : { ...entry, enabled: false },
183	]),
184	),
185	);
186	const disabled =
187	scenario === "keeps mutations on the origin when the alternate is disabled";
188	const disabledDuringProbe =
189	scenario === "does not dispatch after the alternate is disabled during the probe";
190	if (disabled) yield* disableAlternate;
191	const input = {
192	projectId: ProjectId.make("project-1"),
193	host: "github.com",
194	repository: "acme/web",
195	number: 7,
196	action: "merge" as const,
197	};
198	const route = createPullRequestRouter();
199	const result = yield* (
200	reading
201	? route(WS_METHODS.pullRequestsSummary, input)
202	: route(WS_METHODS.pullRequestsRunAction, input)
203	).pipe(
204	Effect.provideService(EnvironmentRegistry.EnvironmentRegistry, environmentRegistry),
205	Effect.provideService(RpcPermissionGuard, { authorize: () => Effect.void }),
206	Effect.provideService(GitHubRoutingPermissions, {
207	...trustedRouting,
208	get: () => Effect.succeed(trusted ? "read-write" : "off"),
209	}),
210	Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
211	Effect.result,
212	);
213
214	if (switchedAccount) {
215	expect(result).toMatchObject({ _tag: "Failure", failure });
216	expect(calls.filter((call) => call.endsWith(":mutation"))).toEqual(["origin:mutation"]);
217	expect(inputs).toEqual([{ ...input, expectedAccountId: "123" }]);
218	} else if (ambiguous) {
219	expect(result).toMatchObject({ _tag: "Failure", failure });
220	expect(calls.filter((call) => call.endsWith(":mutation"))).toEqual(["local:mutation"]);
221	} else {
222	expect(result._tag).toBe("Success");
223	if (single || disabled) {
224	expect(calls.filter((call) => !call.endsWith(":invalidate"))).toEqual([
225	"origin:mutation",
226	]);
227	} else if (reading) expect(calls).toEqual(["origin:read"]);
228	else if (
229	mismatch ||
230	localOrigin ||
231	oldOrigin ||
232	oldAlternate ||
233	!trusted ||
234	disabledDuringProbe
235	)
236	expect(calls.filter((call) => call.endsWith(":mutation"))).toEqual(["origin:mutation"]);
237	else {
238	expect(calls.filter((call) => call.endsWith(":mutation"))).toEqual(["local:mutation"]);
239	expect(calls).toContain("origin:invalidate");
240	expect(inputs).toEqual([{ ...input, expectedAccountId: "123" }]);
241	}
242	}
243	}),
244	),
245	);
246	}
247
248	it.effect("shares active read identity probes while keeping later reads and mutations fresh", () =>
249	Effect.scoped(
250	Effect.gen(function* () {
251	const probes: string[] = [];
252	const reads: number[] = [];
253	const mutations: string[] = [];
254	let accountId = "123";
255	const clientFor = (local: boolean) =>
256	({
257	[local ? WS_METHODS.pullRequestsRoutingIdentity : WS_METHODS.pullRequestsRouting]: () =>
258	Effect.yieldNow.pipe(
259	Effect.andThen(
260	Effect.sync(() => {
261	probes.push(local ? "local" : "origin");
262	return { host: "github.com", provider: "github", viewer: "maria", accountId };
263	}),
264	),
265	),
266	[WS_METHODS.pullRequestsSummary]: (input: {
267	number: number;
268	expectedAccountId: string;
269	}) =>
270	Effect.sync(() => {
271	expect(local).toBe(true);
272	expect(input.expectedAccountId).toBe(accountId);
273	reads.push(input.number);
274	return null;
275	}),
276	[WS_METHODS.pullRequestsRunAction]: (input: { expectedAccountId: string }) =>
277	Effect.sync(() => mutations.push(input.expectedAccountId)),
278	[WS_METHODS.pullRequestsInvalidate]: () => Effect.void,
279	}) as unknown as WsRpcProtocolClient;
280	const { environmentRegistry, supervisor } = yield* makeTestRuntime(
281	clientFor(false),
282	clientFor(true),
283	);
284	const reference = {
285	projectId: ProjectId.make("project-1"),
286	repository: "acme/web",
287	number: 7,
288	};
289	yield* Effect.gen(function* () {
290	yield* Effect.all(
291	Array.from({ length: 4 }, (_, index) =>
292	createPullRequestRouter()(WS_METHODS.pullRequestsSummary, {
293	...reference,
294	number: index + 1,
295	}),
296	),
297	{ concurrency: "unbounded" },
298	);
299	expect(probes).toEqual(["origin", "local"]);
300	expect(reads.toSorted()).toEqual([1, 2, 3, 4]);
301	accountId = "456";
302	yield* createPullRequestRouter()(WS_METHODS.pullRequestsSummary, reference);
303	yield* createPullRequestRouter()(WS_METHODS.pullRequestsRunAction, {
304	...reference,
305	action: "close",
306	});
307	expect(probes).toEqual(["origin", "local", "origin", "local", "origin", "local"]);
308	expect(mutations).toEqual(["456"]);
309	}).pipe(
310	Effect.provideService(EnvironmentRegistry.EnvironmentRegistry, environmentRegistry),
311	Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
312	Effect.provideService(GitHubRoutingPermissions, trustedRouting),
313	Effect.provideService(RpcPermissionGuard, { authorize: () => Effect.void }),
314	);
315	}),
316	),
317	);
318
319	it.effect.each([
320	{ probe: "origin", cancelled: "first" },
321	{ probe: "origin", cancelled: "waiter" },
322	{ probe: "alternate", cancelled: "first" },
323	{ probe: "alternate", cancelled: "waiter" },
324	] as const)(
325	"preserves a shared $probe probe's other reader when the $cancelled cancels",
326	({ probe, cancelled }) =>
327	Effect.scoped(
328	Effect.gen(function* () {
329	const started = yield* Deferred.make<void>();
330	const release = yield* Deferred.make<void>();
331	let probes = 0;
332	const clientFor = (local: boolean) =>
333	({
334	[local ? WS_METHODS.pullRequestsRoutingIdentity : WS_METHODS.pullRequestsRouting]: () =>
335	Effect.gen(function* () {
336	if (local === (probe === "alternate")) {
337	probes++;
338	yield* Deferred.succeed(started, undefined);
339	yield* Deferred.await(release);
340	}
341	return {
342	host: "github.com",
343	provider: "github",
344	viewer: "maria",
345	accountId: "123",
346	};
347	}),
348	[WS_METHODS.pullRequestsSummary]: () => Effect.succeed(null),
349	}) as unknown as WsRpcProtocolClient;
350	const { environmentRegistry, supervisor } = yield* makeTestRuntime(
351	clientFor(false),
352	clientFor(true),
353	);
354	const read = () =>
355	createPullRequestRouter()(WS_METHODS.pullRequestsSummary, {
356	projectId: ProjectId.make("project-1"),
357	repository: "acme/web",
358	number: 7,
359	}).pipe(
360	Effect.provideService(EnvironmentRegistry.EnvironmentRegistry, environmentRegistry),
361	Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
362	Effect.provideService(GitHubRoutingPermissions, trustedRouting),
363	);
364	const first = yield* read().pipe(Effect.forkChild);
365	yield* Deferred.await(started);
366	const waiter = yield* read().pipe(Effect.forkChild);
367	yield* Effect.yieldNow;
368	yield* Fiber.interrupt(cancelled === "first" ? first : waiter);
369	yield* Deferred.succeed(release, undefined);
370	expect(yield* Fiber.join(cancelled === "first" ? waiter : first)).toBeNull();
371	expect(probes).toBe(cancelled === "first" ? 2 : 1);
372	yield* read();
373	expect(probes).toBe(cancelled === "first" ? 3 : 2);
374	}),
375	),
376	);
377
378	it.effect.each(["revoked", "disabled"] as const)(
379	"rechecks routing after an alternate with a shared probe is %s",
380	(change) =>
381	Effect.scoped(
382	Effect.gen(function* () {
383	const started = yield* Deferred.make<void>();
384	const release = yield* Deferred.make<void>();
385	const reads: string[] = [];
386	let trusted = true;
387	let probes = 0;
388	const clientFor = (local: boolean) =>
389	({
390	[local ? WS_METHODS.pullRequestsRoutingIdentity : WS_METHODS.pullRequestsRouting]: () =>
391	Effect.gen(function* () {
392	if (local) {
393	probes++;
394	yield* Deferred.succeed(started, undefined);
395	yield* Deferred.await(release);
396	}
397	return {
398	host: "github.com",
399	provider: "github",
400	viewer: "maria",
401	accountId: "123",
402	};
403	}),
404	[WS_METHODS.pullRequestsSummary]: () =>
405	Effect.sync(() => {
406	reads.push(local ? "local" : "origin");
407	return null;
408	}),
409	}) as unknown as WsRpcProtocolClient;
410	const { environmentRegistry, supervisor } = yield* makeTestRuntime(
411	clientFor(false),
412	clientFor(true),
413	);
414	const readers = yield* Effect.all(
415	Array.from({ length: 2 }, () =>
416	createPullRequestRouter()(WS_METHODS.pullRequestsSummary, {
417	projectId: ProjectId.make("project-1"),
418	repository: "acme/web",
419	number: 7,
420	}),
421	),
422	{ concurrency: "unbounded" },
423	).pipe(
424	Effect.provideService(EnvironmentRegistry.EnvironmentRegistry, environmentRegistry),
425	Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
426	Effect.provideService(GitHubRoutingPermissions, {
427	...trustedRouting,
428	get: () => Effect.sync(() => (trusted ? "read-write" : "off")),
429	}),
430	Effect.forkChild,
431	);
432	yield* Deferred.await(started);
433	yield* Effect.yieldNow;
434	if (change === "revoked") trusted = false;
435	else
436	yield* SubscriptionRef.update(
437	environmentRegistry.entries,
438	(entries) =>
439	new Map(
440	[...entries].map(([id, entry]) => [
441	id,
442	id === TARGET.environmentId ? entry : { ...entry, enabled: false },
443	]),
444	),
445	);
446	yield* Deferred.succeed(release, undefined);
447	expect(yield* Fiber.join(readers)).toEqual([null, null]);
448	expect(probes).toBe(1);
449	expect(reads).toEqual(["origin", "origin"]);
450	}),
451	),
452	);
453
454	it.effect.each(["origin", "alternate"] as const)(
455	"keeps pending metadata from an old %s connection separate from its replacement",
456	(reconnected) =>
457	Effect.scoped(
458	Effect.gen(function* () {
459	const started = yield* Deferred.make<void>();
460	const release = yield* Deferred.make<void>();
461	const probes: string[] = [];
462	const clientFor = (local: boolean, replacement = false) =>
463	({
464	[local ? WS_METHODS.pullRequestsRoutingIdentity : WS_METHODS.pullRequestsRouting]: () =>
465	Effect.gen(function* () {
466	if (local === (reconnected === "alternate")) {
467	probes.push(replacement ? "new" : "old");
468	if (!replacement) {
469	yield* Deferred.succeed(started, undefined);
470	yield* Deferred.await(release);
471	}
472	}
473	return {
474	host: "github.com",
475	provider: "github",
476	viewer: "maria",
477	accountId: "123",
478	};
479	}),
480	[WS_METHODS.pullRequestsSummary]: () => Effect.succeed(null),
481	}) as unknown as WsRpcProtocolClient;
482	const { environmentRegistry, supervisor, localSupervisor } = yield* makeTestRuntime(
483	clientFor(false),
484	clientFor(true),
485	);
486	const read = () =>
487	createPullRequestRouter()(WS_METHODS.pullRequestsSummary, {
488	projectId: ProjectId.make("project-1"),
489	repository: "acme/web",
490	number: 7,
491	}).pipe(
492	Effect.provideService(EnvironmentRegistry.EnvironmentRegistry, environmentRegistry),
493	Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
494	Effect.provideService(GitHubRoutingPermissions, trustedRouting),
495	);
496	const first = yield* read().pipe(Effect.forkChild);
497	yield* Deferred.await(started);
498	yield* SubscriptionRef.set(
499	(reconnected === "origin" ? supervisor : localSupervisor).session,
500	Option.some(session(clientFor(reconnected === "alternate", true))),
501	);
502	expect(yield* read()).toBeNull();
503	expect(probes).toEqual(["old", "new"]);
504	yield* Deferred.succeed(release, undefined);
505	expect(yield* Fiber.join(first)).toBeNull();
506	}),
507	),
508	);
509
510	it.effect.each(["github", "gitlab", "bitbucket", "azure-devops"] as const)(
511	"routes %s viewed marks to their storage environment",
512	(provider) =>
513	Effect.scoped(
514	Effect.gen(function* () {
515	const reference = {
516	projectId: ProjectId.make("project-1"),
517	host: "github.com",
518	repository: "acme/web",
519	number: 7,
520	};
521	const calls: { environment: string; operation: string; input: unknown }[] = [];
522	const clientFor = (environment: string) =>
523	({
524	[WS_METHODS.pullRequestsRouting]: () =>
525	Effect.succeed({
526	host: reference.host,
527	provider,
528	accountId: "123",
529	viewer: "maria-rcks",
530	}),
531	[WS_METHODS.pullRequestsRoutingIdentity]: () =>
532	Effect.succeed({
533	host: reference.host,
534	provider,
535	accountId: "123",
536	viewer: "maria-rcks",
537	}),
538	[WS_METHODS.pullRequestsFilesViewed]: (input: unknown) =>
539	Effect.sync(() => {
540	calls.push({ environment, operation: "read", input });
541	return { files: [{ path: "a.ts", state: "viewed" }], truncated: false };
542	}),
543	[WS_METHODS.pullRequestsSetFilesViewed]: (input: unknown) =>
544	Effect.sync(() => {
545	calls.push({ environment, operation: "write", input });
546	}),
547	[WS_METHODS.pullRequestsInvalidate]: (input: unknown) =>
548	Effect.sync(() => {
549	calls.push({ environment, operation: "invalidate", input });
550	}),
551	}) as unknown as WsRpcProtocolClient;
552	const { environmentRegistry, supervisor } = yield* makeTestRuntime(
553	clientFor("origin"),
554	clientFor("local"),
555	);
556	const files = [{ path: "a.ts", viewed: false }];
557	const route = createPullRequestRouter();
558	yield* Effect.gen(function* () {
559	expect(yield* route(WS_METHODS.pullRequestsFilesViewed, reference)).toEqual({
560	files: [{ path: "a.ts", state: "viewed" }],
561	truncated: false,
562	});
563	yield* route(WS_METHODS.pullRequestsSetFilesViewed, { ...reference, files });
564	}).pipe(
565	Effect.provideService(EnvironmentRegistry.EnvironmentRegistry, environmentRegistry),
566	Effect.provideService(GitHubRoutingPermissions, trustedRouting),
567	Effect.provideService(RpcPermissionGuard, { authorize: () => Effect.void }),
568	Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
569	);
570	const environment = provider === "github" ? "local" : "origin";
571	const guard = provider === "github" ? { expectedAccountId: "123" } : {};
572	expect(calls.filter((call) => call.operation !== "invalidate")).toEqual([
573	{ environment, operation: "read", input: { ...reference, allowStale: false, ...guard } },
574	{ environment, operation: "write", input: { ...reference, files, ...guard } },
575	]);
576	const invalidations = calls.filter((call) => call.operation === "invalidate");
577	if (provider === "github") expect(invalidations.length).toBeGreaterThan(0);
578	else expect(invalidations).toEqual([]);
579	for (const call of invalidations) {
580	expect(call.input).toEqual({
581	reference: expect.objectContaining(reference),
582	filesViewedOnly: true,
583	});
584	}
585	}),
586	),
587	);
588
589	const TARGET = new PrimaryConnectionTarget({
590	environmentId: EnvironmentId.make("environment-1"),
591	label: "Test environment",
592	httpBaseUrl: "https://environment.example.test",
593	wsBaseUrl: "wss://environment.example.test",
594	});
595
596	function session(client: WsRpcProtocolClient): RpcSession {
597	return {
598	client: {
599	...client,
600	[WS_METHODS.pullRequestsInvalidate]:
601	client[WS_METHODS.pullRequestsInvalidate] ?? (() => Effect.void),
602	},
603	initialConfig: client[WS_METHODS.serverGetConfig]?.({}).pipe(Effect.orDie) ?? Effect.never,
604	subscribeServerConfig: (input) => client.subscribeServerConfig(input),
605	ready: Effect.void,
606	probe: Effect.void,
607	closed: Effect.never,
608	};
609	}
610
611	const makeTestRuntime = Effect.fn("makeTestRuntime")(function* (
612	client: WsRpcProtocolClient,
613	localClient?: WsRpcProtocolClient,
614	localOrigin = false,
615	) {
616	const originTarget = localOrigin
617	? new PrimaryConnectionTarget({
618	...TARGET,
619	httpBaseUrl: "http://localhost:3774",
620	wsBaseUrl: "ws://localhost:3774",
621	})
622	: TARGET;
623	const connectionState: SupervisorConnectionState = {
624	...AVAILABLE_CONNECTION_STATE,
625	desired: true,
626	network: "online",
627	phase: "connected",
628	attempt: 1,
629	generation: 1,
630	};
631	const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
632	target: originTarget,
633	state: yield* SubscriptionRef.make(connectionState),
634	session: yield* SubscriptionRef.make(Option.some(session(client))),
635	prepared: yield* SubscriptionRef.make(Option.none<PreparedConnection>()),
636	connect: Effect.void,
637	disconnect: Effect.void,
638	retryNow: Effect.void,
639	} satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
640	const localTarget = new PrimaryConnectionTarget({
641	environmentId: EnvironmentId.make("local-environment"),
642	label: "Local environment",
643	httpBaseUrl: "http://localhost:3773",
644	wsBaseUrl: "ws://localhost:3773",
645	});
646	const localSupervisor = {
647	...supervisor,
648	target: localTarget,
649	session: yield* SubscriptionRef.make(Option.some(session(localClient ?? client))),
650	};
651	const environmentRegistry = EnvironmentRegistry.EnvironmentRegistry.of({
652	entries: yield* SubscriptionRef.make<ReadonlyMap<EnvironmentId, ConnectionCatalogEntry>>(
653	new Map([
654	[
655	originTarget.environmentId,
656	{ target: originTarget, profile: Option.none(), enabled: true },
657	],
658	...(localClient === undefined
659	? []
660	: [
661	[
662	localTarget.environmentId,
663	{ target: localTarget, profile: Option.none(), enabled: true },
664	] as const,
665	]),
666	]),
667	),
668	run: (environmentId, effect) =>
669	Effect.provideService(
670	effect,
671	EnvironmentSupervisor.EnvironmentSupervisor,
672	environmentId === localTarget.environmentId ? localSupervisor : supervisor,
673	),
674	runStream: (_environmentId, stream) =>
675	Stream.provideService(stream, EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
676	followStream: (_environmentId, stream) =>
677	Stream.provideService(stream, EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
678	} as EnvironmentRegistry.EnvironmentRegistry["Service"]);
679	const runtime = Atom.runtime(
680	Layer.merge(
681	Layer.succeed(EnvironmentRegistry.EnvironmentRegistry, environmentRegistry),
682	Layer.succeed(
683	PullRequestDiffLoader.PullRequestDiffLoader,
684	PullRequestDiffLoader.PullRequestDiffLoader.of({ load: () => Effect.die("unused") }),
685	),
686	),
687	);
688	const atoms = createPullRequestEnvironmentAtoms(runtime);
689	const registry = yield* Effect.acquireRelease(Effect.sync(AtomRegistry.make), (registry) =>
690	Effect.sync(() => registry.dispose()),
691	);
692	return { runtime, atoms, registry, environmentRegistry, supervisor, localSupervisor };
693	});
694
695	it.effect("updates Mergebot badges from the stream and clears disconnected snapshots", () =>
696	Effect.scoped(
697	Effect.gen(function* () {
698	const snapshots = yield* SubscriptionRef.make<MergebotRepoQueue | null>({
699	repo: "acme/web",
700	url: "https://queue.example.test",
701	status: "ready",
702	processing: false,
703	pullRequests: [],
704	});
705	const client = {
706	[WS_METHODS.mergebotSubscribe]: () => SubscriptionRef.changes(snapshots),
707	[WS_METHODS.mergebotQueue]: () => Effect.die("live badges must not poll"),
708	} as unknown as WsRpcProtocolClient;
709	const { runtime, registry } = yield* makeTestRuntime(client);
710	const queue = createMergebotQueueAtomFamily(runtime)({
711	environmentId: TARGET.environmentId,
712	input: { repository: "acme/web" },
713	});
714	registry.mount(queue);
715	expect(yield* AtomRegistry.getResult(registry, queue)).toMatchObject({ processing: false });
716
717	const changed = yield* Deferred.make<void>();
718	const disconnected = yield* Deferred.make<void>();
719	const unsubscribe = registry.subscribe(queue, (result) => {
720	if (!AsyncResult.isSuccess(result)) return;
721	if (result.value?.processing) Deferred.doneUnsafe(changed, Effect.void);
722	if (result.value === null) Deferred.doneUnsafe(disconnected, Effect.void);
723	});
724	yield* SubscriptionRef.update(snapshots, (snapshot) =>
725	snapshot === null ? null : { ...snapshot, processing: true },
726	);
727	yield* Deferred.await(changed);
728	expect(yield* AtomRegistry.getResult(registry, queue)).toMatchObject({ processing: true });
729	yield* SubscriptionRef.set(snapshots, null);
730	yield* Deferred.await(disconnected);
731	expect(yield* AtomRegistry.getResult(registry, queue)).toBeNull();
732	unsubscribe();
733	}),
734	),
735	);
736
737	it.effect.each(["default", "origin-off", "destination-off", "read-only"] as const)(
738	"does not probe another environment with %s routing permission",
739	(permission) =>
740	Effect.scoped(
741	Effect.gen(function* () {
742	const calls: string[] = [];
743	const failure = new PullRequestOperationError({
744	operation: "summary",
745	detail: "source failed",
746	});
747	const client = {
748	[WS_METHODS.pullRequestsSummary]: () =>
749	Effect.suspend(() => {
750	calls.push("source-read");
751	return Effect.fail(failure);
752	}),
753	[WS_METHODS.pullRequestsRunAction]: () =>
754	Effect.sync(() => {
755	calls.push("source-write");
756	}),
757	} as unknown as WsRpcProtocolClient;
758	const alternate = {
759	[WS_METHODS.pullRequestsRoutingIdentity]: () =>
760	Effect.die("untrusted environment was contacted"),
761	} as unknown as WsRpcProtocolClient;
762	const { environmentRegistry, supervisor } = yield* makeTestRuntime(client, alternate);
763	const ref = {
764	projectId: ProjectId.make("project-1"),
765	repository: "private/repo",
766	number: 7,
767	};
768	const request = Effect.gen(function* () {
769	const route = createPullRequestRouter();
770	if (permission !== "read-only")
771	yield* route(WS_METHODS.pullRequestsSummary, ref).pipe(Effect.flip);
772	yield* route(WS_METHODS.pullRequestsRunAction, { ...ref, action: "merge" });
773	}).pipe(
774	Effect.provideService(EnvironmentRegistry.EnvironmentRegistry, environmentRegistry),
775	Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
776	Effect.provideService(RpcPermissionGuard, { authorize: () => Effect.void }),
777	);
778	yield* permission === "default"
779	? request
780	: request.pipe(
781	Effect.provideService(RpcPermissionGuard, { authorize: () => Effect.void }),
782	Effect.provideService(GitHubRoutingPermissions, {
783	...trustedRouting,
784	get: (entry) =>
785	Effect.succeed(
786	permission === "read-only"
787	? "read"
788	: (entry.target.environmentId === TARGET.environmentId) ===
789	(permission === "origin-off")
790	? "off"
791	: "read-write",
792	),
793	}),
794	);
795	expect(calls).toEqual(
796	permission === "read-only" ? ["source-write"] : ["source-read", "source-write"],
797	);
798	}),
799	),
800	);
801
802	it.effect.each(
803	(["origin", "destination"] as const).flatMap((side) =>
804	(["matching", "changed", "missing", "unavailable", "failed"] as const).map((stored) => ({
805	side,
806	stored,
807	})),
808	),
809	)("checks the current $side SSH profile before routing with $stored storage", ({ side, stored }) =>
810	Effect.scoped(
811	Effect.gen(function* () {
812	const calls: string[] = [];
813	const identity = {
814	host: "github.com",
815	provider: "github",
816	viewer: "maria",
817	accountId: "123",
818	};
819	const client = {
820	[WS_METHODS.pullRequestsRouting]: () =>
821	Effect.sync(() => {
822	calls.push("source-probe");
823	return identity;
824	}),
825	[WS_METHODS.pullRequestsRunAction]: (input: { expectedAccountId?: string }) =>
826	Effect.gen(function* () {
827	if (input.expectedAccountId !== undefined) {
828	return yield* new PullRequestOperationError({
829	operation: "routeIdentity",
830	detail: "Source guard refused.",
831	});
832	}
833	calls.push("source-write");
834	}),
835	[WS_METHODS.pullRequestsInvalidate]: () => Effect.void,
836	} as unknown as WsRpcProtocolClient;
837	const alternate = {
838	[WS_METHODS.pullRequestsRoutingIdentity]: () =>
839	Effect.sync(() => {
840	calls.push("alternate-probe");
841	return identity;
842	}),
843	[WS_METHODS.pullRequestsRunAction]: () =>
844	Effect.sync(() => {
845	calls.push("alternate-write");
846	}),
847	[WS_METHODS.pullRequestsInvalidate]: () => Effect.void,
848	} as unknown as WsRpcProtocolClient;
849	const { environmentRegistry, supervisor } = yield* makeTestRuntime(client, alternate);
850	const environmentId =
851	side === "origin" ? TARGET.environmentId : EnvironmentId.make("local-environment");
852	const profile = new SshConnectionProfile({
853	connectionId: "ssh-1",
854	environmentId,
855	label: "SSH",
856	target: { alias: "work", hostname: "work.example.test", username: "maria", port: 22 },
857	});
858	yield* SubscriptionRef.update(environmentRegistry.entries, (entries) =>
859	new Map(entries).set(environmentId, {
860	target: new SshConnectionTarget({
861	environmentId,
862	connectionId: profile.connectionId,
863	label: "SSH",
864	}),
865	profile: Option.some(profile),
866	enabled: true,
867	}),
868	);
869	const read = Effect.suspend(() =>
870	stored === "failed"
871	? Effect.fail(
872	new ConnectionTransientError({
873	reason: "remote-unavailable",
874	detail: "Profile storage unavailable.",
875	}),
876	)
877	: Effect.succeed(
878	stored === "missing"
879	? Option.none()
880	: Option.some(
881	stored === "changed"
882	? new SshConnectionProfile({
883	...profile,
884	target: { ...profile.target, hostname: "replacement.example.test" },
885	})
886	: profile,
887	),
888	),
889	);
890	const route = createPullRequestRouter()(WS_METHODS.pullRequestsRunAction, {
891	projectId: ProjectId.make("project-1"),
892	repository: "private/repo",
893	number: 7,
894	action: "merge",
895	}).pipe(
896	Effect.provideService(EnvironmentRegistry.EnvironmentRegistry, environmentRegistry),
897	Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
898	// This also represents a user enabling the stale catalog entry after re-resolution.
899	Effect.provideService(GitHubRoutingPermissions, trustedRouting),
900	Effect.provideService(RpcPermissionGuard, { authorize: () => Effect.void }),
901	);
902	yield* stored === "unavailable"
903	? route
904	: route.pipe(
905	Effect.provideService(ConnectionProfileStore.ConnectionProfileStore, {
906	get: () => read,
907	put: () => Effect.die("unused"),
908	remove: () => Effect.die("unused"),
909	}),
910	);
911	expect(calls).toEqual(
912	stored === "matching"
913	? ["source-probe", "alternate-probe", "alternate-write"]
914	: ["source-write"],
915	);
916	}),
917	),
918	);
919
920	it.live.each(["origin", "alternate"] as const)(
921	"bounds a stalled %s metadata probe without repeating a strict source read",
922	(probe) =>
923	Effect.scoped(
924	Effect.gen(function* () {
925	let sourceReads = 0;
926	const identity = {
927	host: "github.com",
928	provider: "github",
929	viewer: "maria",
930	accountId: "123",
931	};
932	const client = {
933	[WS_METHODS.pullRequestsRouting]: () =>
934	probe === "origin" ? Effect.never : Effect.succeed(identity),
935	[WS_METHODS.pullRequestsSummary]: () =>
936	Effect.suspend(() => {
937	sourceReads += 1;
938	return Effect.fail(
939	new PullRequestOperationError({ operation: "summary", detail: "source failed" }),
940	);
941	}),
942	} as unknown as WsRpcProtocolClient;
943	const alternate = {
944	[WS_METHODS.pullRequestsRoutingIdentity]: () => Effect.never,
945	} as unknown as WsRpcProtocolClient;
946	const { environmentRegistry, supervisor } = yield* makeTestRuntime(client, alternate);
947	const error = yield* createPullRequestRouter()(WS_METHODS.pullRequestsSummary, {
948	projectId: ProjectId.make("project-1"),
949	repository: "acme/web",
950	number: 7,
951	allowStale: false,
952	}).pipe(
953	Effect.provideService(EnvironmentRegistry.EnvironmentRegistry, environmentRegistry),
954	Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
955	Effect.provideService(GitHubRoutingPermissions, trustedRouting),
956	Effect.provideService(RpcPermissionGuard, { authorize: () => Effect.void }),
957	Effect.flip,
958	);
959	expect(error._tag).toBe("PullRequestOperationError");
960	expect(sourceReads).toBe(1);
961	}),
962	),
963	);
964
965	it.effect.each(
966	(["pending", "pending-local", "failed-local", "failed", "offline"] as const).map(
967	(source) =>
968	[
969	source === "offline"
970	? "returns held source data only after both fresh paths fail"
971	: `uses one shared reader with a ${source} source`,
972	source,
973	] as const,
974	),
975	)("%s", ([, source]) =>
976	Effect.scoped(
977	Effect.gen(function* () {
978	const started = yield* Deferred.make<void>();
979	const calls: string[] = [];
980	const clientFor = (local: boolean) =>
981	({
982	[local ? WS_METHODS.pullRequestsRoutingIdentity : WS_METHODS.pullRequestsRouting]: () =>
983	Effect.succeed({
984	host: "github.com",
985	provider: "github",
986	viewer: "maria-rcks",
987	accountId: "123",
988	}),
989	[WS_METHODS.pullRequestsSummary]: (input: { allowStale?: boolean }) =>
990	Effect.gen(function* () {
991	calls.push(local ? "local" : input.allowStale === false ? "origin" : "held");
992	if (source === "offline" && !local && input.allowStale === undefined) {
993	return { state: "open" };
994	}
995	expect(input.allowStale).toBe(false);
996	if (local && source !== "offline") return null;
997	if (source !== "pending" && source !== "pending-local")
998	return yield* new PullRequestOperationError({
999	operation: "summary",
1000	detail: "github unreachable",
1001	});
1002	yield* Deferred.succeed(started, undefined);
1003	return yield* Effect.never;
1004	}),
1005	}) as unknown as WsRpcProtocolClient;
1006	const { environmentRegistry, supervisor } = yield* makeTestRuntime(
1007	clientFor(false),
1008	clientFor(true),
1009	source === "failed-local" || source === "pending-local",
1010	);
1011	const request = createPullRequestRouter()(WS_METHODS.pullRequestsSummary, {
1012	projectId: ProjectId.make("project-1"),
1013	repository: "acme/web",
1014	number: 7,
1015	}).pipe(
1016	Effect.provideService(EnvironmentRegistry.EnvironmentRegistry, environmentRegistry),
1017	Effect.provideService(GitHubRoutingPermissions, trustedRouting),
1018	Effect.provideService(RpcPermissionGuard, { authorize: () => Effect.void }),
1019	Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
1020	);
1021	const fiber = yield* request.pipe(Effect.forkChild);
1022	if (source === "pending-local") {
1023	yield* Deferred.await(started);
1024	yield* TestClock.adjust("30 seconds");
1025	}
1026	const result = yield* Fiber.join(fiber);
1027	if (source === "offline") {
1028	expect(result).toEqual({ state: "open" });
1029	expect(calls).toEqual(["local", "origin", "held"]);
1030	} else {
1031	expect(result).toBeNull();
1032	expect(calls).toEqual(
1033	source === "failed-local" || source === "pending-local" ? ["origin", "local"] : ["local"],
1034	);
1035	}
1036	}),
1037	),
1038	);
1039
1040	it.live("keeps source workspace metadata when an alternate answers a detail read", () =>
1041	Effect.scoped(
1042	Effect.gen(function* () {
1043	const noted = yield* Deferred.make<unknown>();
1044	const clientFor = (local: boolean) =>
1045	({
1046	[local ? WS_METHODS.pullRequestsRoutingIdentity : WS_METHODS.pullRequestsRouting]: () =>
1047	Effect.succeed({
1048	host: "github.com",
1049	provider: "github",
1050	viewer: "maria-rcks",
1051	accountId: "123",
1052	projectTitle: local ? "local project" : "source project",
1053	workspaceRoot: local ? "/Users/local/repo" : "/srv/source/repo",
1054	}),
1055	[WS_METHODS.pullRequestsDetail]: () =>
1056	local
1057	? Effect.succeed({
1058	projectId: "local-project",
1059	projectTitle: "local project",
1060	workspaceRoot: "/Users/local/repo",
1061	title: "github title",
1062	state: "closed",
1063	updatedAt: "2026-10-10T01:52:36Z",
1064	})
1065	: Effect.never,
1066	[WS_METHODS.pullRequestsReportState]: (input: unknown) =>
1067	Deferred.succeed(noted, input).pipe(Effect.asVoid),
1068	}) as unknown as WsRpcProtocolClient;
1069	const { environmentRegistry, supervisor } = yield* makeTestRuntime(
1070	clientFor(false),
1071	clientFor(true),
1072	);
1073	const result = yield* createPullRequestRouter()(WS_METHODS.pullRequestsDetail, {
1074	projectId: ProjectId.make("project-1"),
1075	repository: "acme/web",
1076	number: 7,
1077	}).pipe(
1078	Effect.provideService(EnvironmentRegistry.EnvironmentRegistry, environmentRegistry),
1079	Effect.provideService(GitHubRoutingPermissions, trustedRouting),
1080	Effect.provideService(RpcPermissionGuard, { authorize: () => Effect.void }),
1081	Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
1082	);
1083	expect(result).toEqual({
1084	projectId: "project-1",
1085	projectTitle: "source project",
1086	workspaceRoot: "/srv/source/repo",
1087	title: "github title",
1088	state: "closed",
1089	updatedAt: "2026-10-10T01:52:36Z",
1090	});
1091	expect(yield* Deferred.await(noted)).toEqual({
1092	reference: { projectId: "project-1", repository: "acme/web", number: 7 },
1093	state: "closed",
1094	});
1095	}),
1096	),
1097	);
1098
1099	it.live(
1100	"refreshes identity before writes and invalidates prior readers across router instances",
1101	() =>
1102	Effect.scoped(
1103	Effect.gen(function* () {
1104	let originAccountId = "123";
1105	const mutations: { environment: string; expectedAccountId: string }[] = [];
1106	const invalidations: { environment: string; input: unknown }[] = [];
1107	const identities: string[] = [];
1108	const clientFor = (local: boolean) => {
1109	const environment = local ? "local" : "origin";
1110	return {
1111	[local ? WS_METHODS.pullRequestsRoutingIdentity : WS_METHODS.pullRequestsRouting]: () =>
1112	Effect.sync(() => {
1113	identities.push(environment);
1114	return {
1115	host: "github.com",
1116	provider: "github",
1117	viewer: "maria-rcks",
1118	accountId: local ? "123" : originAccountId,
1119	};
1120	}),
1121	[WS_METHODS.pullRequestsSummary]: () => (local ? Effect.succeed(null) : Effect.never),
1122	[WS_METHODS.pullRequestsRunAction]: (input: { expectedAccountId: string }) =>
1123	Effect.sync(() => {
1124	mutations.push({ environment, expectedAccountId: input.expectedAccountId });
1125	}),
1126	[WS_METHODS.pullRequestsInvalidate]: (input: unknown) =>
1127	Effect.sync(() => {
1128	invalidations.push({ environment, input });
1129	}),
1130	} as unknown as WsRpcProtocolClient;
1131	};
1132	const { environmentRegistry, supervisor } = yield* makeTestRuntime(
1133	clientFor(false),
1134	clientFor(true),
1135	);
1136	const reference = {
1137	projectId: ProjectId.make("project-1"),
1138	repository: "acme/web",
1139	number: 7,
1140	};
1141	const hostedReference = { ...reference, host: "github.com", allowStale: false };
1142	const route = createPullRequestRouter();
1143	yield* Effect.gen(function* () {
1144	yield* route(WS_METHODS.pullRequestsSummary, reference);
1145	originAccountId = "456";
1146	yield* route(WS_METHODS.pullRequestsRunAction, { ...reference, action: "merge" });
1147
1148	expect(identities.filter((environment) => environment === "origin")).toHaveLength(2);
1149	expect(mutations).toEqual([{ environment: "origin", expectedAccountId: "456" }]);
1150	expect(invalidations).toEqual(
1151	expect.arrayContaining([
1152	{ environment: "origin", input: { reference: expect.objectContaining(reference) } },
1153	{ environment: "origin", input: { reference: hostedReference } },
1154	{ environment: "local", input: { reference: hostedReference } },
1155	{ environment: "origin", input: {} },
1156	{ environment: "local", input: {} },
1157	]),
1158	);
1159
1160	invalidations.length = 0;
1161	const refresh = createPullRequestRouter();
1162	yield* refresh(WS_METHODS.pullRequestsInvalidate, { reference });
1163	expect(invalidations).toContainEqual({
1164	environment: "local",
1165	input: { reference: hostedReference },
1166	});
1167
1168	invalidations.length = 0;
1169	yield* refresh(WS_METHODS.pullRequestsInvalidate, {});
1170	expect(invalidations).toContainEqual({ environment: "local", input: {} });
1171	}).pipe(
1172	Effect.provideService(EnvironmentRegistry.EnvironmentRegistry, environmentRegistry),
1173	Effect.provideService(GitHubRoutingPermissions, trustedRouting),
1174	Effect.provideService(RpcPermissionGuard, { authorize: () => Effect.void }),
1175	Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
1176	);
1177	}),
1178	),
1179	);
1180
1181	it.effect("keeps concurrent diff file reads on different hosts separate", () =>
1182	Effect.scoped(
1183	Effect.gen(function* () {
1184	const release = yield* Latch.make();
1185	const started = yield* Latch.make();
1186	const calls: string[] = [];
1187	const client = {
1188	[WS_METHODS.pullRequestsDiffFileContents]: (input: { readonly host: string }) =>
1189	Effect.gen(function* () {
1190	calls.push(input.host);
1191	yield* started.open;
1192	yield* release.await;
1193	return { oldContents: "", newContents: input.host };
1194	}),
1195	} as unknown as WsRpcProtocolClient;
1196	const { atoms, registry } = yield* makeTestRuntime(client);
1197	const input = {
1198	projectId: ProjectId.make("project-1"),
1199	repository: "acme/web",
1200	number: 1,
1201	changeType: "change",
1202	oldPath: "src/app.ts",
1203	newPath: "src/app.ts",
1204	} as const;
1205	const first = atoms.diffFileContents.run(registry, {
1206	environmentId: TARGET.environmentId,
1207	input: { ...input, host: "github.com" },
1208	});
1209	yield* started.await;
1210	const second = atoms.diffFileContents.run(registry, {
1211	environmentId: TARGET.environmentId,
1212	input: { ...input, host: "github.example.com" },
1213	});
1214	yield* release.open;
1215
1216	const results = yield* Effect.promise(() => Promise.all([first, second]));
1217	expect(results).toMatchObject([
1218	{ _tag: "Success", value: { newContents: "github.com" } },
1219	{ _tag: "Success", value: { newContents: "github.example.com" } },
1220	]);
1221	expect(calls).toEqual(["github.com", "github.example.com"]);
1222	}),
1223	),
1224	);
1225
1226	it.effect("queues merge preparation with actions without restarting it on refresh", () =>
1227	Effect.scoped(
1228	Effect.gen(function* () {
1229	const started = yield* Deferred.make<void>();
1230	const release = yield* Deferred.make<void>();
1231	const refreshEvents = yield* PubSub.unbounded<number>();
1232	const calls: string[] = [];
1233	const client = {
1234	[WS_METHODS.pullRequestsSubscribeRefreshes]: () => Stream.fromPubSub(refreshEvents),
1235	[WS_METHODS.pullRequestsDetail]: (input: { number: number; allowStale?: boolean }) =>
1236	Effect.gen(function* () {
1237	expect(input.allowStale).toBe(false);
1238	calls.push(`detail:${input.number}`);
1239	yield* Deferred.succeed(started, undefined);
1240	yield* Deferred.await(release);
1241	return {
1242	state: "open",
1243	isDraft: false,
1244	capabilities: { actions: ["merge"], stackActions: true },
1245	viewerPermissions: { actions: ["merge"] },
1246	};
1247	}),
1248	[WS_METHODS.pullRequestsStack]: (input: { number: number; allowStale?: boolean }) =>
1249	Effect.sync(() => {
1250	expect(input.allowStale).toBe(false);
1251	calls.push(`stack:${input.number}`);
1252	return null;
1253	}),
1254	[WS_METHODS.pullRequestsRunAction]: (input: {
1255	action: string;
1256	number: number;
1257	mergeMethod?: string;
1258	}) =>
1259	Effect.sync(() => {
1260	if (input.action === "merge") expect(input.mergeMethod).toBe("squash");
1261	expect(input).not.toHaveProperty("resolveMergeMethod");
1262	calls.push(`${input.action}:${input.number}`);
1263	}),
1264	} as unknown as WsRpcProtocolClient;
1265	const { atoms, registry } = yield* makeTestRuntime(client);
1266	const unmount = registry.mount(
1267	atoms.refreshes({ environmentId: TARGET.environmentId, input: {} }),
1268	);
1269	yield* Effect.addFinalizer(() => Effect.sync(unmount));
1270	const reference = { projectId: ProjectId.make("project-1"), repository: "acme/web" };
1271	const first = atoms.runAction.run(registry, {
1272	environmentId: TARGET.environmentId,
1273	input: { ...reference, number: 1, action: "merge", resolveMergeMethod: () => "squash" },
1274	});
1275	yield* Deferred.await(started);
1276	const second = atoms.runAction.run(registry, {
1277	environmentId: TARGET.environmentId,
1278	input: { ...reference, number: 2, action: "close" },
1279	});
1280	yield* PubSub.publish(refreshEvents, 1);
1281	expect(calls).toEqual(["detail:1"]);
1282	yield* Deferred.succeed(release, undefined);
1283	const results = yield* Effect.promise(() => Promise.all([first, second]));
1284	expect(results.every(AsyncResult.isSuccess)).toBe(true);
1285	expect(calls).toEqual(["detail:1", "stack:1", "merge:1", "close:2"]);
1286	}),
1287	),
1288	);
1289
1290	it.effect.each(["closed", "draft", "permission", "stack", "method"] as const)(
1291	"rejects an unsafe quick merge with %s and keeps later actions running",
1292	(reason) =>
1293	Effect.scoped(
1294	Effect.gen(function* () {
1295	const calls: string[] = [];
1296	const client = {
1297	[WS_METHODS.pullRequestsDetail]: () =>
1298	Effect.succeed({
1299	state: reason === "closed" ? "closed" : "open",
1300	isDraft: reason === "draft",
1301	capabilities: { actions: ["merge"], stackActions: true },
1302	viewerPermissions: { actions: reason === "permission" ? [] : ["merge"] },
1303	}),
1304	[WS_METHODS.pullRequestsStack]: () => Effect.succeed(reason === "stack" ? {} : null),
1305	[WS_METHODS.pullRequestsRunAction]: (input: { action: string }) =>
1306	Effect.sync(() => calls.push(input.action)),
1307	} as unknown as WsRpcProtocolClient;
1308	const { atoms, registry } = yield* makeTestRuntime(client);
1309	const target = {
1310	environmentId: TARGET.environmentId,
1311	input: { projectId: ProjectId.make("project-1"), repository: "acme/web", number: 1 },
1312	};
1313	const merge = atoms.runAction.run(registry, {
1314	...target,
1315	input: {
1316	...target.input,
1317	action: "merge",
1318	resolveMergeMethod: () => {
1319	if (reason === "method") throw new Error("No merge method is available.");
1320	return "squash";
1321	},
1322	},
1323	});
1324	const close = atoms.runAction.run(registry, {
1325	...target,
1326	input: { ...target.input, action: "close" },
1327	});
1328	const results = yield* Effect.promise(() => Promise.all([merge, close]));
1329	expect(results.map((result) => result._tag)).toEqual(["Failure", "Success"]);
1330	expect(calls).toEqual(["close"]);
1331	}),
1332	),
1333	);
1334
1335	it.effect("keeps a close batch ordered and continues after a refused close", () =>
1336	Effect.scoped(
1337	Effect.gen(function* () {
1338	const calls: number[] = [];
1339	const started = yield* Deferred.make<void>();
1340	const release = yield* Deferred.make<void>();
1341	const client = {
1342	[WS_METHODS.pullRequestsRunAction]: (input: { number: number; action: string }) =>
1343	Effect.gen(function* () {
1344	expect(input.action).toBe("close");
1345	calls.push(input.number);
1346	if (input.number === 6) {
1347	yield* Deferred.succeed(started, undefined);
1348	yield* Deferred.await(release);
1349	}
1350	if (input.number === 5)
1351	return yield* new PullRequestOperationError({
1352	operation: "runAction",
1353	detail: "You cannot close this pull request.",
1354	});
1355	}),
1356	} as unknown as WsRpcProtocolClient;
1357	const { atoms, registry } = yield* makeTestRuntime(client);
1358	const batch = [6, 5, 4].map((number) =>
1359	atoms.runAction.run(registry, {
1360	environmentId: TARGET.environmentId,
1361	input: {
1362	projectId: ProjectId.make("project-1"),
1363	repository: "acme/web",
1364	number,
1365	action: "close",
1366	},
1367	}),
1368	);
1369	yield* Deferred.await(started);
1370	expect(calls).toEqual([6]);
1371	yield* Deferred.succeed(release, undefined);
1372	const results = yield* Effect.promise(() => Promise.all(batch));
1373	expect(results.map((result) => result._tag)).toEqual(["Success", "Failure", "Success"]);
1374	expect(calls).toEqual([6, 5, 4]);
1375	}),
1376	),
1377	);
1378
1379	it.effect("keeps hover previews fresh after edits and turns", () =>
1380	Effect.scoped(
1381	Effect.gen(function* () {
1382	const refreshEvents = yield* PubSub.unbounded<number>();
1383	let title = "Original title";
1384	let reads = 0;
1385	const read = (input: unknown) =>
1386	Effect.sync(() => {
1387	expect(input).toEqual(reference);
1388	reads++;
1389	return { ...reference, title };
1390	});
1391	const reference = {
1392	projectId: ProjectId.make("project-1"),
1393	repository: "acme/web",
1394	number: 1,
1395	host: "github.example.com",
1396	};
1397	const client = {
1398	[WS_METHODS.pullRequestsSubscribeRefreshes]: () => Stream.fromPubSub(refreshEvents),
1399	[WS_METHODS.pullRequestsPreview]: read,
1400	[WS_METHODS.pullRequestsDetail]: read,
1401	[WS_METHODS.pullRequestsUpdate]: (input: { title: string }) =>
1402	Effect.sync(() => {
1403	title = input.title;
1404	}),
1405	[WS_METHODS.pullRequestsRunAction]: () =>
1406	Effect.sync(() => {
1407	title = "Closed";
1408	}),
1409	[WS_METHODS.pullRequestsInvalidate]: () => Effect.void,
1410	} as unknown as WsRpcProtocolClient;
1411	const { atoms, registry } = yield* makeTestRuntime(client);
1412	const target = { environmentId: TARGET.environmentId, input: reference };
1413	const preview = atoms.preview(target);
1414	const unmount = registry.mount(preview);
1415	yield* Effect.addFinalizer(() => Effect.sync(unmount));
1416	expect(
1417	(yield* AtomRegistry.getResult(registry, preview, { suspendOnWaiting: true })).title,
1418	).toBe(title);
1419	yield* Effect.promise(() => executeAtomQuery(registry, preview));
1420	expect(reads).toBe(1);
1421	const edited = yield* Effect.promise(() =>
1422	atoms.update.run(registry, { ...target, input: { ...reference, title: "Edited" } }),
1423	);
1424	expect(AsyncResult.isSuccess(edited)).toBe(true);
1425	expect(
1426	(yield* AtomRegistry.getResult(registry, preview, { suspendOnWaiting: true })).title,
1427	).toBe("Edited");
1428	yield* Effect.promise(() =>
1429	atoms.runAction.run(registry, { ...target, input: { ...reference, action: "close" } }),
1430	);
1431	expect(
1432	(yield* AtomRegistry.getResult(registry, preview, { suspendOnWaiting: true })).title,
1433	).toBe("Closed");
1434	title = "Refreshed";
1435	yield* Effect.promise(() =>
1436	atoms.invalidate.run(registry, {
1437	environmentId: target.environmentId,
1438	input: { reference },
1439	}),
1440	);
1441	expect(
1442	(yield* AtomRegistry.getResult(registry, preview, { suspendOnWaiting: true })).title,
1443	).toBe("Refreshed");
1444	const refreshed = Latch.makeUnsafe();
1445	const stop = registry.subscribe(preview, (result) => {
1446	if (AsyncResult.isSuccess(result) && result.value.title === "After turn")
1447	refreshed.openUnsafe();
1448	});
1449	yield* Effect.addFinalizer(() => Effect.sync(stop));
1450	title = "After turn";
1451	yield* PubSub.publish(refreshEvents, 1);
1452	yield* refreshed.await;
1453	}),
1454	),
1455	);
1456
1457	it.effect("shares close, reopen, and merge with an untouched client's mounted PR readers", () =>
1458	Effect.scoped(
1459	Effect.gen(function* () {
1460	const revision = yield* SubscriptionRef.make(0);
1461	let state: "open" | "closed" | "merged" = "open";
1462	const client = {
1463	[WS_METHODS.pullRequestsSubscribeRefreshes]: () => SubscriptionRef.changes(revision),
1464	[WS_METHODS.pullRequestsSummary]: () => Effect.sync(() => ({ state })),
1465	[WS_METHODS.pullRequestsDetail]: () => Effect.sync(() => ({ state })),
1466	[WS_METHODS.pullRequestsList]: () =>
1467	Effect.sync(() => ({ entries: [{ number: 1, state }] })),
1468	[WS_METHODS.pullRequestsRunAction]: (input: {
1469	readonly action: "close" | "reopen" | "merge";
1470	}) =>
1471	Effect.gen(function* () {
1472	state =
1473	input.action === "close" ? "closed" : input.action === "merge" ? "merged" : "open";
1474	yield* SubscriptionRef.update(revision, (value) => value + 1);
1475	}),
1476	} as unknown as WsRpcProtocolClient;
1477	const writer = yield* makeTestRuntime(client);
1478	const reader = yield* makeTestRuntime(client);
1479	const target = {
1480	environmentId: TARGET.environmentId,
1481	input: {
1482	projectId: ProjectId.make("project-1"),
1483	host: "github.example.com",
1484	repository: "acme/web",
1485	number: 1,
1486	},
1487	};
1488	const detail = reader.atoms.detail(target);
1489	const summary = createLinkedPullRequestSummaryAtomFamily(
1490	reader.runtime,
1491	reader.atoms.refreshes,
1492	)(target);
1493	const list = reader.atoms.list({
1494	environmentId: TARGET.environmentId,
1495	input: { state: "all" },
1496	});
1497	const unmountDetail = reader.registry.mount(detail);
1498	const unmountSummary = reader.registry.mount(summary);
1499	const unmountList = reader.registry.mount(list);
1500	yield* Effect.addFinalizer(() =>
1501	Effect.sync(() => {
1502	unmountDetail();
1503	unmountSummary();
1504	unmountList();
1505	}),
1506	);
1507	expect((yield* AtomRegistry.getResult(reader.registry, detail)).state).toBe("open");
1508	expect((yield* AtomRegistry.getResult(reader.registry, summary)).state).toBe("open");
1509	expect((yield* AtomRegistry.getResult(reader.registry, list)).entries[0]?.state).toBe("open");
1510
1511	for (const [action, expected] of [
1512	["close", "closed"],
1513	["reopen", "open"],
1514	["merge", "merged"],
1515	] as const) {
1516	const detailChanged = Latch.makeUnsafe();
1517	const summaryChanged = Latch.makeUnsafe();
1518	const listChanged = Latch.makeUnsafe();
1519	const stops = [
1520	reader.registry.subscribe(detail, (result) => {
1521	if (AsyncResult.isSuccess(result) && result.value.state === expected) {
1522	detailChanged.openUnsafe();
1523	}
1524	}),
1525	reader.registry.subscribe(summary, (result) => {
1526	if (AsyncResult.isSuccess(result) && result.value.state === expected) {
1527	summaryChanged.openUnsafe();
1528	}
1529	}),
1530	reader.registry.subscribe(list, (result) => {
1531	if (AsyncResult.isSuccess(result) && result.value.entries[0]?.state === expected) {
1532	listChanged.openUnsafe();
1533	}
1534	}),
1535	];
1536	yield* Effect.addFinalizer(() => Effect.sync(() => stops.forEach((stop) => stop())));
1537	const result = yield* Effect.promise(() =>
1538	writer.atoms.runAction.run(writer.registry, {
1539	...target,
1540	input: { ...target.input, action },
1541	}),
1542	);
1543	expect(AsyncResult.isSuccess(result)).toBe(true);
1544	// The second client receives only the server push: no local refresh or timer tick.
1545	yield* detailChanged.await;
1546	yield* summaryChanged.await;
1547	yield* listChanged.await;
1548	expect((yield* AtomRegistry.getResult(reader.registry, detail)).state).toBe(expected);
1549	expect((yield* AtomRegistry.getResult(reader.registry, summary)).state).toBe(expected);
1550	expect((yield* AtomRegistry.getResult(reader.registry, list)).entries[0]?.state).toBe(
1551	expected,
1552	);
1553	stops.forEach((stop) => stop());
1554	}
1555	}),
1556	),
1557	);
1558
1559	it.effect("refreshes pull request activity after a comment is updated", () =>
1560	Effect.scoped(
1561	Effect.gen(function* () {
1562	const refreshEvents = yield* PubSub.unbounded<number>();
1563	let commentBody = "old comment";
1564	const client = {
1565	[WS_METHODS.pullRequestsSubscribeRefreshes]: () => Stream.fromPubSub(refreshEvents),
1566	[WS_METHODS.pullRequestsActivity]: () =>
1567	Effect.succeed({
1568	author: null,
1569	reviewers: [],
1570	comments: [
1571	{
1572	id: "comment-1",
1573	kind: "issue-comment",
1574	author: null,
1575	body: commentBody,
1576	createdAt: "2026-08-24T00:00:00Z",
1577	url: null,
1578	path: null,
1579	reviewState: null,
1580	reactions: [],
1581	},
1582	],
1583	commentCount: 1,
1584	commentsTruncated: false,
1585	reviewThreads: [],
1586	commits: [],
1587	reactions: [],
1588	}),
1589	[WS_METHODS.pullRequestsUpdateComment]: (input: { readonly body: string }) =>
1590	Effect.sync(() => {
1591	commentBody = input.body;
1592	}),
1593	} as unknown as WsRpcProtocolClient;
1594	const { atoms, registry } = yield* makeTestRuntime(client);
1595	const reference = {
1596	projectId: ProjectId.make("project-1"),
1597	host: "github.example.com",
1598	repository: "acme/web",
1599	number: 1,
1600	} as const;
1601	const activity = atoms.activity({ environmentId: TARGET.environmentId, input: reference });
1602	const unmount = registry.mount(activity);
1603	yield* Effect.addFinalizer(() => Effect.sync(unmount));
1604
1605	const initial = yield* Effect.promise(() => executeAtomQuery(registry, activity));
1606	expect(AsyncResult.isSuccess(initial)).toBe(true);
1607	if (!AsyncResult.isSuccess(initial)) {
1608	return yield* Effect.die("activity did not load");
1609	}
1610	expect(initial.value.comments[0]?.body).toBe("old comment");
1611
1612	const update = yield* Effect.promise(() =>
1613	atoms.updateComment.run(registry, {
1614	environmentId: TARGET.environmentId,
1615	input: { ...reference, commentId: "comment-1", kind: "issue-comment", body: "updated" },
1616	}),
1617	);
1618
1619	expect(AsyncResult.isSuccess(update)).toBe(true);
1620	expect(
1621	(yield* AtomRegistry.getResult(registry, activity, { suspendOnWaiting: true })).comments[0]
1622	?.body,
1623	).toBe("updated");
1624	const refreshed = Latch.makeUnsafe();
1625	const stop = registry.subscribe(activity, (result) => {
1626	if (AsyncResult.isSuccess(result) && result.value.comments[0]?.body === "after turn") {
1627	refreshed.openUnsafe();
1628	}
1629	});
1630	yield* Effect.addFinalizer(() => Effect.sync(stop));
1631
1632	commentBody = "after turn";
1633	yield* PubSub.publish(refreshEvents, 1);
1634	yield* refreshed.await;
1635
1636	expect(
1637	(yield* AtomRegistry.getResult(registry, activity, { suspendOnWaiting: true })).comments[0]
1638	?.body,
1639	).toBe("after turn");
1640	}),
1641	),
1642	);
1643
1644	it.effect("refreshes checks without refreshing full detail", () =>
1645	Effect.scoped(
1646	Effect.gen(function* () {
1647	let detailReads = 0;
1648	let checksReads = 0;
1649	const client = {
1650	[WS_METHODS.pullRequestsSubscribeRefreshes]: () => Stream.never,
1651	[WS_METHODS.pullRequestsDetail]: () =>
1652	Effect.sync(() => {
1653	detailReads++;
1654	return { title: "PR" };
1655	}),
1656	[WS_METHODS.pullRequestsChecks]: () =>
1657	Effect.sync(() => {
1658	checksReads++;
1659	return { state: checksReads === 1 ? "open" : "merged", checks: [] };
1660	}),
1661	} as unknown as WsRpcProtocolClient;
1662	const { atoms, registry } = yield* makeTestRuntime(client);
1663	const target = {
1664	environmentId: TARGET.environmentId,
1665	input: { projectId: ProjectId.make("project-1"), repository: "acme/web", number: 1 },
1666	};
1667	const detail = atoms.detail(target);
1668	const checks = atoms.checks(target);
1669	yield* AtomRegistry.mount(registry, detail);
1670	yield* AtomRegistry.mount(registry, checks);
1671	yield* AtomRegistry.getResult(registry, detail);
1672	expect((yield* AtomRegistry.getResult(registry, checks))?.state).toBe("open");
1673	registry.refresh(checks);
1674	expect(
1675	(yield* AtomRegistry.getResult(registry, checks, { suspendOnWaiting: true }))?.state,
1676	).toBe("merged");
1677	expect(detailReads).toBe(1);
1678	expect(checksReads).toBe(2);
1679	}),
1680	),
1681	);
1682
1683	it.effect.each([false, true])(
1684	"routes checks through one reader with old alternate: %s",
1685	(oldAlternate) =>
1686	Effect.scoped(
1687	Effect.gen(function* () {
1688	const calls: string[] = [];
1689	const clientFor = (local: boolean) =>
1690	({
1691	[WS_METHODS.serverGetConfig]: () =>
1692	Effect.succeed({
1693	environment: {
1694	capabilities: local && oldAlternate ? {} : { pullRequestChecks: true },
1695	},
1696	}),
1697	[local ? WS_METHODS.pullRequestsRoutingIdentity : WS_METHODS.pullRequestsRouting]: () =>
1698	Effect.succeed({
1699	host: "github.com",
1700	provider: "github",
1701	viewer: "viewer",
1702	accountId: "123",
1703	}),
1704	[WS_METHODS.pullRequestsChecks]: () =>
1705	Effect.gen(function* () {
1706	calls.push(local ? "local" : "origin");
1707	if (local && oldAlternate)
1708	return yield* Effect.die("Unknown request tag: pullRequests.checks");
1709	return { state: "open", checks: [] };
1710	}),
1711	}) as unknown as WsRpcProtocolClient;
1712	const { environmentRegistry, supervisor } = yield* makeTestRuntime(
1713	clientFor(false),
1714	clientFor(true),
1715	);
1716	const result = yield* createPullRequestRouter()(WS_METHODS.pullRequestsChecks, {
1717	projectId: ProjectId.make("project-1"),
1718	repository: "acme/web",
1719	number: 1,
1720	}).pipe(
1721	Effect.provideService(EnvironmentRegistry.EnvironmentRegistry, environmentRegistry),
1722	Effect.provideService(GitHubRoutingPermissions, trustedRouting),
1723	Effect.provideService(RpcPermissionGuard, { authorize: () => Effect.void }),
1724	Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
1725	);
1726	expect(result).toEqual({ state: "open", checks: [] });
1727	expect(calls).toEqual(oldAlternate ? ["origin"] : ["local"]);
1728	}),
1729	),
1730	);
1731
1732	it.effect("keeps live detail reads separate from reads that allow stale data", () =>
1733	Effect.scoped(
1734	Effect.gen(function* () {
1735	const client = {
1736	[WS_METHODS.pullRequestsSubscribeRefreshes]: () => Stream.never,
1737	[WS_METHODS.pullRequestsDetail]: (input: { allowStale?: boolean }) =>
1738	Effect.succeed({ title: input.allowStale === false ? "latest checks" : "cached checks" }),
1739	} as unknown as WsRpcProtocolClient;
1740	const { atoms, registry } = yield* makeTestRuntime(client);
1741	const target = {
1742	environmentId: TARGET.environmentId,
1743	input: { projectId: ProjectId.make("project-1"), repository: "acme/web", number: 1 },
1744	};
1745	const cached = atoms.detail(target);
1746	const live = atoms.detail({ ...target, input: { ...target.input, allowStale: false } });
1747	expect((yield* AtomRegistry.getResult(registry, cached)).title).toBe("cached checks");
1748	expect((yield* AtomRegistry.getResult(registry, live)).title).toBe("latest checks");
1749	}),
1750	),
1751	);
1752
1753	it.effect("updates cached labels after successful edits without rereading the host", () =>
1754	Effect.scoped(
1755	Effect.gen(function* () {
1756	let detailReads = 0;
1757	let candidateReads = 0;
1758	let refuse = false;
1759	let failDetail = false;
1760	const existing = { name: "existing", color: "111111" };
1761	const addedLabel = { name: "new", color: "abcdef" };
1762	const detailRefreshStarted = yield* Latch.make();
1763	const releaseDetailRefresh = yield* Latch.make();
1764	const client = {
1765	[WS_METHODS.pullRequestsSubscribeRefreshes]: () => Stream.never,
1766	[WS_METHODS.pullRequestsDetail]: () =>
1767	Effect.gen(function* () {
1768	detailReads++;
1769	if (failDetail) {
1770	yield* detailRefreshStarted.open;
1771	yield* releaseDetailRefresh.await;
1772	return yield* new MutationRefused();
1773	}
1774	return { title: "keep this title", labels: [existing] };
1775	}),
1776	[WS_METHODS.pullRequestsLabelCandidates]: () =>
1777	Effect.sync(() => {
1778	candidateReads++;
1779	return {
1780	candidates: [
1781	{ ...existing, description: null, isApplied: true },
1782	{ ...addedLabel, description: "description", isApplied: false },
1783	],
1784	truncated: false,
1785	};
1786	}),
1787	[WS_METHODS.pullRequestsSetLabels]: () =>
1788	refuse ? Effect.fail(new MutationRefused()) : Effect.void,
1789	} as unknown as WsRpcProtocolClient;
1790	const { atoms, registry } = yield* makeTestRuntime(client);
1791	const target = {
1792	environmentId: TARGET.environmentId,
1793	input: {
1794	projectId: ProjectId.make("project-1"),
1795	repository: "acme/web",
1796	number: 1,
1797	host: "github.example.com",
1798	},
1799	};
1800	const detail = atoms.detail(target);
1801	const candidates = atoms.labelCandidates(target);
1802	registry.mount(detail);
1803	const unmountCandidates = registry.mount(candidates);
1804	yield* AtomRegistry.getResult(registry, detail, { suspendOnWaiting: true });
1805	yield* AtomRegistry.getResult(registry, candidates, { suspendOnWaiting: true });
1806
1807	const added = yield* Effect.promise(() =>
1808	atoms.setLabels.run(registry, {
1809	...target,
1810	input: {
1811	host: target.input.host,
1812	projectId: target.input.projectId,
1813	repository: target.input.repository,
1814	number: target.input.number,
1815	labels: ["new"],
1816	applied: true,
1817	},
1818	}),
1819	);
1820	expect(AsyncResult.isSuccess(added)).toBe(true);
1821	expect(yield* AtomRegistry.getResult(registry, detail)).toEqual({
1822	title: "keep this title",
1823	labels: [existing, addedLabel],
1824	});
1825	unmountCandidates();
1826	registry.mount(atoms.labelCandidates(target));
1827	expect((yield* AtomRegistry.getResult(registry, candidates)).candidates[1]).toEqual({
1828	...addedLabel,
1829	description: "description",
1830	isApplied: true,
1831	});
1832
1833	for (const name of ["existing", "new"]) {
1834	refuse = name === "new";
1835	const result = yield* Effect.promise(() =>
1836	atoms.setLabels.run(registry, {
1837	...target,
1838	input: { ...target.input, labels: [name], applied: false },
1839	}),
1840	);
1841	expect(result._tag).toBe(refuse ? "Failure" : "Success");
1842	expect((yield* AtomRegistry.getResult(registry, detail)).labels).toEqual([addedLabel]);
1843	expect((yield* AtomRegistry.getResult(registry, candidates)).candidates).toMatchObject([
1844	{ name: "existing", isApplied: false },
1845	{ name: "new", isApplied: true },
1846	]);
1847	}
1848	expect(detailReads).toBe(1);
1849	expect(candidateReads).toBe(1);
1850
1851	failDetail = true;
1852	registry.refresh(detail);
1853	yield* detailRefreshStarted.await;
1854	expect(registry.get(detail).waiting).toBe(true);
1855	expect(Option.getOrThrow(AsyncResult.value(registry.get(detail))).labels).toEqual([
1856	addedLabel,
1857	]);
1858	yield* releaseDetailRefresh.open;
1859	yield* Effect.exit(AtomRegistry.getResult(registry, detail, { suspendOnWaiting: true }));
1860	expect(AsyncResult.isFailure(registry.get(detail))).toBe(true);
1861	expect(Option.getOrThrow(AsyncResult.value(registry.get(detail))).labels).toEqual([
1862	addedLabel,
1863	]);
1864	failDetail = false;
1865	registry.refresh(detail);
1866	expect(
1867	(yield* AtomRegistry.getResult(registry, detail, { suspendOnWaiting: true })).labels,
1868	).toEqual([existing]);
1869	expect(detailReads).toBe(3);
1870	}),
1871	),
1872	);
1873
1874	it.effect("updates reviewer requests and enriched reviewers without rereading the host", () =>
1875	Effect.scoped(
1876	Effect.gen(function* () {
1877	let reads = 0;
1878	let refuse = false;
1879	const actor = { login: "reviewer", name: "Reviewer", avatarUrl: null };
1880	const hostActor = { ...actor, login: "Reviewer" };
1881	let hostRequested = false;
1882	let reviewed = false;
1883	let pauseActivity = false;
1884	const activityStarted = yield* Latch.make();
1885	const client = {
1886	[WS_METHODS.pullRequestsSubscribeRefreshes]: () => Stream.never,
1887	[WS_METHODS.pullRequestsDetail]: () =>
1888	Effect.sync(() => {
1889	reads++;
1890	return { reviewers: hostRequested ? [hostActor] : [] };
1891	}),
1892	[WS_METHODS.pullRequestsActivity]: () =>
1893	Effect.gen(function* () {
1894	reads++;
1895	if (pauseActivity) {
1896	pauseActivity = false;
1897	yield* activityStarted.open;
1898	return yield* Effect.never;
1899	}
1900	return {
1901	reviewers: hostRequested ? [hostActor] : [],
1902	comments: reviewed ? [{ kind: "review-comment", author: hostActor }] : [],
1903	};
1904	}),
1905	[WS_METHODS.pullRequestsReviewerCandidates]: (input: { number: number }) =>
1906	input.number === 2
1907	? Effect.never
1908	: Effect.sync(() => {
1909	reads++;
1910	return {
1911	candidates: [{ ...actor, id: "12", kind: "user", isRequested: false }],
1912	truncated: false,
1913	};
1914	}),
1915	[WS_METHODS.pullRequestsRequestReviewers]: (input: { requested: boolean }) =>
1916	refuse
1917	? Effect.fail(new MutationRefused())
1918	: Effect.sync(() => {
1919	hostRequested = input.requested;
1920	}),
1921	} as unknown as WsRpcProtocolClient;
1922	const { atoms, registry } = yield* makeTestRuntime(client);
1923	const target = {
1924	environmentId: TARGET.environmentId,
1925	input: {
1926	projectId: ProjectId.make("project-1"),
1927	repository: "acme/web",
1928	number: 1,
1929	host: "github.example.com",
1930	},
1931	};
1932	const detail = atoms.detail(target);
1933	const activity = atoms.activity(target);
1934	const candidates = atoms.reviewerCandidates(target);
1935	registry.mount(detail);
1936	registry.mount(activity);
1937	registry.mount(candidates);
1938	yield* AtomRegistry.getResult(registry, detail, { suspendOnWaiting: true });
1939	yield* AtomRegistry.getResult(registry, activity, { suspendOnWaiting: true });
1940	yield* AtomRegistry.getResult(registry, candidates, { suspendOnWaiting: true });
1941	const request = (requested: boolean, reference = target) =>
1942	Effect.promise(() =>
1943	atoms.requestReviewers.run(registry, {
1944	...reference,
1945	input: { ...reference.input, reviewers: [{ id: "12", kind: "user" }], requested },
1946	}),
1947	);
1948	for (const operation of ["request", "refuse", "remove"]) {
1949	refuse = operation === "refuse";
1950	expect((yield* request(operation === "request"))._tag).toBe(refuse ? "Failure" : "Success");
1951	const expected = operation === "remove" ? [] : [actor];
1952	expect((yield* AtomRegistry.getResult(registry, detail)).reviewers).toEqual(expected);
1953	expect((yield* AtomRegistry.getResult(registry, activity)).reviewers).toEqual(expected);
1954	expect((yield* AtomRegistry.getResult(registry, candidates)).candidates).toMatchObject([
1955	{ isRequested: operation !== "remove" },
1956	]);
1957	}
1958	expect(reads).toBe(3);
1959	// A slow activity read started before the write must not hide the new request.
1960	pauseActivity = true;
1961	registry.refresh(activity);
1962	yield* activityStarted.await;
1963	expect(AsyncResult.isSuccess(yield* request(true))).toBe(true);
1964	expect(
1965	(yield* AtomRegistry.getResult(registry, activity, { suspendOnWaiting: true })).reviewers,
1966	).toEqual([hostActor]);
1967	expect(reads).toBe(5);
1968	expect(AsyncResult.isSuccess(yield* request(false))).toBe(true);
1969	expect((yield* AtomRegistry.getResult(registry, activity)).reviewers).toEqual([]);
1970
1971	reviewed = true;
1972	yield* request(true);
1973	registry.refresh(activity);
1974	yield* AtomRegistry.getResult(registry, activity, { suspendOnWaiting: true });
1975	yield* request(false);
1976	expect((yield* AtomRegistry.getResult(registry, activity)).reviewers).toEqual([hostActor]);
1977
1978	// A caller without an open picker still needs authoritative reviewer identities.
1979	const otherTarget = { ...target, input: { ...target.input, number: 2 } };
1980	const otherDetail = atoms.detail(otherTarget);
1981	registry.mount(otherDetail);
1982	yield* AtomRegistry.getResult(registry, otherDetail, { suspendOnWaiting: true });
1983	yield* request(true, otherTarget);
1984	expect(
1985	(yield* AtomRegistry.getResult(registry, otherDetail, { suspendOnWaiting: true }))
1986	.reviewers,
1987	).toEqual([hostActor]);
1988	}),
1989	),
1990	);
1991
1992	it.effect("refreshes stack state after reopening and head SHAs after a turn", () =>
1993	Effect.scoped(
1994	Effect.gen(function* () {
1995	const refreshEvents = yield* PubSub.unbounded<number>();
1996	let state: "closed" | "open" = "closed";
1997	let headSha = "old-head";
1998	const client = {
1999	[WS_METHODS.pullRequestsSubscribeRefreshes]: () => Stream.fromPubSub(refreshEvents),
2000	[WS_METHODS.pullRequestsStack]: () =>
2001	Effect.sync(
2002	() =>
2003	({
2004	id: "stack-1",
2005	number: 1,
2006	url: "https://github.com/acme/web/pull/1",
2007	base: "main",
2008	layers: [
2009	{
2010	number: 1,
2011	headBranch: "feature",
2012	headSha,
2013	state,
2014	isDraft: false,
2015	},
2016	],
2017	}) satisfies PullRequestStack,
2018	),
2019	} as unknown as WsRpcProtocolClient;
2020	const { runtime, registry } = yield* makeTestRuntime(client);
2021	const stacks = createPullRequestStackAtomFamily(runtime);
2022	const stack = stacks({
2023	environmentId: TARGET.environmentId,
2024	input: {
2025	projectId: ProjectId.make("project-1"),
2026	repository: "acme/web",
2027	number: 1,
2028	},
2029	});
2030	const unmount = registry.mount(stack);
2031	yield* Effect.addFinalizer(() => Effect.sync(unmount));
2032	yield* Effect.promise(() => executeAtomQuery(registry, stack));
2033	expect((yield* AtomRegistry.getResult(registry, stack))?.layers[0]?.state).toBe("closed");
2034	state = "open";
2035	registry.refresh(stack);
2036	expect(
2037	(yield* AtomRegistry.getResult(registry, stack, { suspendOnWaiting: true }))?.layers[0]
2038	?.state,
2039	).toBe("open");
2040
2041	const refreshed = Latch.makeUnsafe();
2042	const stop = registry.subscribe(stack, (result) => {
2043	if (AsyncResult.isSuccess(result) && result.value?.layers[0]?.headSha === "new-head") {
2044	refreshed.openUnsafe();
2045	}
2046	});
2047	yield* Effect.addFinalizer(() => Effect.sync(stop));
2048	headSha = "new-head";
2049	yield* PubSub.publish(refreshEvents, 1);
2050	yield* refreshed.await;
2051	expect((yield* AtomRegistry.getResult(registry, stack))?.layers[0]?.headSha).toBe("new-head");
2052	}),
2053	),
2054	);
2055
2056	// Transport fixtures have a source-control-only session; authorization edge cases
2057	// are exercised by commandPermissions.test.ts.
2058	vi.mock("./session.ts", () => ({
2059	createEnvironmentSessionAtoms: () => ({ sessionStateAtom: grantedSessions }),
2060	}));
2061	const grantedSessions = Atom.family((_id: EnvironmentId) =>
2062	Atom.make<AsyncResult.AsyncResult<AuthSessionState>>(
2063	AsyncResult.success({
2064	authenticated: true,
2065	auth: {
2066	policy: "remote-reachable" as const,
2067	bootstrapMethods: [],
2068	sessionMethods: [],
2069	sessionCookieName: "test",
2070	},
2071	scopes: [AuthSourceControlWriteScope],
2072	}),
2073	),
2074	);
2075
2076	it.effect("denies a routed write when only the origin has source-control permission", () =>
2077	Effect.scoped(
2078	Effect.gen(function* () {
2079	let writes = 0;
2080	const identity = { host: "github.com", provider: "github", viewer: "test", accountId: "123" };
2081	const client = {
2082	[WS_METHODS.pullRequestsRouting]: () => Effect.succeed(identity),
2083	[WS_METHODS.pullRequestsRoutingIdentity]: () => Effect.succeed(identity),
2084	[WS_METHODS.pullRequestsRunAction]: () =>
2085	Effect.sync(() => {
2086	writes++;
2087	}),
2088	} as unknown as WsRpcProtocolClient;
2089	const fixture = yield* makeTestRuntime(client, client);
2090	const local = EnvironmentId.make("local-environment");
2091	const stop = fixture.registry.mount(grantedSessions(local));
2092	yield* Effect.addFinalizer(() => Effect.sync(stop));
2093	fixture.registry.set(
2094	grantedSessions(local),
2095	AsyncResult.success({
2096	authenticated: true,
2097	scopes: [],
2098	auth: {
2099	policy: "remote-reachable" as const,
2100	bootstrapMethods: [],
2101	sessionMethods: [],
2102	sessionCookieName: "test",
2103	},
2104	}),
2105	);
2106	const error = yield* createPullRequestRouter()(WS_METHODS.pullRequestsRunAction, {
2107	projectId: ProjectId.make("project"),
2108	repository: "acme/repo",
2109	number: 1,
2110	action: "merge",
2111	}).pipe(
2112	Effect.provideService(EnvironmentRegistry.EnvironmentRegistry, fixture.environmentRegistry),
2113	Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, fixture.supervisor),
2114	Effect.provideService(GitHubRoutingPermissions, trustedRouting),
2115	Effect.provideService(RpcPermissionGuard, {
2116	authorize: (id, method, input) =>
2117	createCommandPermissions(fixture.runtime, method).authorize(
2118	fixture.registry,
2119	id,
2120	input,
2121	),
2122	}),
2123	Effect.flip,
2124	);
2125	expect(error._tag).toBe("EnvironmentAuthorizationError");
2126	expect(writes).toBe(0);
2127	}),
2128	),
2129	);
2130
2131	it.effect("publishes the accepted guide run before the follow-up read finishes", () =>
2132	Effect.scoped(
2133	Effect.gen(function* () {
2134	const accepted = {
2135	guide: null,
2136	currentHeadSha: "head",
2137	run: { _tag: "Running" as const, headSha: "head", startedAt: "2026-10-11T00:00:00.000Z" },
2138	};
2139	let reads = 0;
2140	const refreshStarted = yield* Latch.make();
2141	const client = {
2142	[WS_METHODS.pullRequestsSubscribeRefreshes]: () => Stream.never,
2143	[WS_METHODS.pullRequestsReviewGuide]: () =>
2144	Effect.gen(function* () {
2145	if (++reads > 1) {
2146	yield* refreshStarted.open;
2147	return yield* Effect.never;
2148	}
2149	return { guide: null, run: null, currentHeadSha: "head" };
2150	}),
2151	[WS_METHODS.pullRequestsGenerateReviewGuide]: () => Effect.succeed(accepted),
2152	} as unknown as WsRpcProtocolClient;
2153	const { atoms, registry } = yield* makeTestRuntime(client);
2154	const target = {
2155	environmentId: TARGET.environmentId,
2156	input: { projectId: ProjectId.make("project-1"), repository: "acme/web", number: 1 },
2157	};
2158	registry.update(
2159	grantedSessions(target.environmentId),
2160	AsyncResult.map((session) => ({
2161	...session,
2162	scopes: [...(session.scopes ?? []), AuthOrchestrationOperateScope],
2163	})),
2164	);
2165	const query = atoms.reviewGuide(target);
2166	registry.mount(query);
2167	yield* AtomRegistry.getResult(registry, query, { suspendOnWaiting: true });
2168	const result = yield* Effect.promise(() => atoms.generateReviewGuide.run(registry, target));
2169	expect(result._tag).toBe("Success");
2170	yield* refreshStarted.await;
2171	expect(Option.getOrNull(AsyncResult.value(registry.get(query)))).toEqual(accepted);
2172	}),
2173	),
2174	);