1import { SourceControlProviderKind } from "@t3tools/contracts";
2import * as Stream from "effect/Stream";
3import {
4 EventId,
5 MessageId,
6 ProjectId,
7 ProviderInstanceId,
8 PullRequestOperationError,
9 RunId,
10 ThreadId,
11 TurnItemId,
12 type OrchestrationV2Command as OrchestrationCommand,
13 type OrchestrationV2DomainEvent,
14 type OrchestrationV2Run,
15 type OrchestrationProjectShell,
16 type PullRequestRef,
17 type PullRequestStack,
18 type PullRequestSummary,
19 type ThreadPullRequestKey,
20 type ThreadPullRequestLink,
21 type ThreadPullRequestSnapshot,
22} from "@t3tools/contracts";
23import { assert, describe, it } from "@effect/vitest";
24import * as Clock from "effect/Clock";
25import * as Crypto from "effect/Crypto";
26import * as DateTime from "effect/DateTime";
27import * as Deferred from "effect/Deferred";
28import * as Effect from "effect/Effect";
29import * as Layer from "effect/Layer";
30import * as Logger from "effect/Logger";
31import * as Queue from "effect/Queue";
32import * as Ref from "effect/Ref";
33import { TestClock } from "effect/testing";
34
35import { PullRequestProviderError } from "@t3tools/source-control-core/server/PullRequestProvider";
36import * as MergebotQueue from "../mergebot/MergebotQueue.ts";
37import * as GitManager from "../git/GitManager.ts";
38import * as PullRequestService from "../pullRequest/PullRequestService.ts";
39import * as ServerActivation from "../serverActivation.ts";
40import * as Orchestrator from "./Orchestrator.ts";
41import * as ProjectionStore from "./ProjectionStore.ts";
42import * as PullRequestSyncReactor from "./PullRequestSyncReactor.ts";
43import type { PullRequestTestThread } from "./testkit/pullRequestFixtures.ts";
44
45const NOW = "2026-08-28T12:00:00.000Z";
46const PROJECT_ID = ProjectId.make("sync-project");
47
48type SyncCommand = Extract<
49 OrchestrationCommand,
50 { readonly type: "thread.pull-request-link.sync" }
51>;
52type LinkCommand = Extract<OrchestrationCommand, { readonly type: "thread.pull-request.link" }>;
53
54const testCrypto = Crypto.make({
55 randomBytes: (size) => new Uint8Array(size).fill(1),
56 digest: (_algorithm, data) => Effect.succeed(data),
57});
58
59function makeProject(id: ProjectId = PROJECT_ID): OrchestrationProjectShell {
60 return {
61 id,
62 title: `Project ${id}`,
63 workspaceRoot: "/workspace/project",
64 defaultModelSelection: null,
65 scripts: [],
66 createdAt: "2026-08-01T00:00:00.000Z",
67 updatedAt: NOW,
68 };
69}
70
71function makeThread(
72 id: string,
73 overrides: Partial<PullRequestTestThread> = {},
74): PullRequestTestThread {
75 return {
76 id: ThreadId.make(id),
77 projectId: PROJECT_ID,
78 title: id,
79 modelSelection: {
80 instanceId: ProviderInstanceId.make("codex"),
81 model: "gpt-5",
82 },
83 runtimeMode: "full-access",
84 interactionMode: "default",
85 branch: null,
86 worktreePath: null,
87 pullRequests: [],
88 createdAt: "2026-08-01T00:00:00.000Z",
89 updatedAt: "2026-08-20T00:00:00.000Z",
90 archivedAt: null,
91 settledOverride: null,
92 settledAt: null,
93 latestUserMessageAt: "2026-08-20T00:00:00.000Z",
94 ...overrides,
95 };
96}
97
98function makeLink(
99 number: number,
100 snapshot: Partial<ThreadPullRequestSnapshot> | null = null,
101 overrides: Partial<ThreadPullRequestLink> = {},
102): ThreadPullRequestLink {
103 return {
104 host: "github.com",
105 repository: "owner/repository",
106 number,
107 url: `https://github.com/owner/repository/pull/${number}`,
108 source: "manual",
109 linkedAt: "2026-08-10T00:00:00.000Z",
110 snapshot:
111 snapshot === null
112 ? null
113 : {
114 state: "open",
115 title: "Pull request",
116 headBranch: "feature",
117 baseBranch: "main",
118 isDraft: false,
119 updatedAt: "2026-08-27T00:00:00.000Z",
120 syncedAt: "2026-08-27T00:00:00.000Z",
121 ...snapshot,
122 },
123 stack: null,
124 ...overrides,
125 };
126}
127
128/** The shell the fake projection store serves: one project and its threads. */
129interface TestShellSnapshot {
130 readonly snapshotSequence: number;
131 readonly projects: ReadonlyArray<OrchestrationProjectShell>;
132 readonly threads: ReadonlyArray<PullRequestTestThread>;
133 readonly updatedAt: string;
134}
135
136function makeSnapshot(
137 threads: ReadonlyArray<PullRequestTestThread>,
138 snapshotSequence = 1,
139): TestShellSnapshot {
140 return {
141 snapshotSequence,
142 projects: [makeProject()],
143 threads,
144 updatedAt: NOW,
145 };
146}
147
148function makeSummary(
149 input: PullRequestRef,
150 overrides: Partial<PullRequestSummary> = {},
151): PullRequestSummary {
152 return {
153 provider: SourceControlProviderKind.make("github"),
154 projectId: input.projectId,
155 repository: input.repository,
156 number: input.number,
157 title: "Pull request",
158 url: `https://github.com/${input.repository}/pull/${input.number}`,
159 state: "open",
160 headBranch: "feature",
161 baseBranch: "main",
162 updatedAt: "2026-08-27T00:00:00.000Z",
163 ...overrides,
164 };
165}
166
167interface HarnessOptions {
168 readonly mergebot?: MergebotQueue.MergebotQueue["Service"]["configuration"];
169 readonly invalidate?: PullRequestService.PullRequestService["Service"]["invalidate"];
170 readonly snapshot: TestShellSnapshot;
171 readonly summary?: (
172 input: PullRequestRef,
173 ) => Effect.Effect<PullRequestSummary, PullRequestOperationError>;
174 readonly stack?: (
175 input: PullRequestRef,
176 ) => Effect.Effect<PullRequestStack | null, PullRequestOperationError>;
177}
178
179const makeHarness = Effect.fn("makePullRequestSyncHarness")(function* (options: HarnessOptions) {
180 const activation = yield* Deferred.make<void>();
181 const snapshots = yield* Ref.make(options.snapshot);
182 const snapshotReads = yield* Queue.unbounded<void>();
183 const shellSnapshotReads = yield* Ref.make(0);
184 const syncCommands = yield* Ref.make<ReadonlyArray<SyncCommand>>([]);
185 const linkCommands = yield* Ref.make<ReadonlyArray<LinkCommand>>([]);
186 const summaryCalls = yield* Ref.make<ReadonlyArray<PullRequestRef>>([]);
187 const stackCalls = yield* Ref.make<ReadonlyArray<PullRequestRef>>([]);
188 const domainEvents = yield* Queue.unbounded<OrchestrationV2DomainEvent>();
189 const stateChanges = yield* Queue.unbounded<ThreadPullRequestKey>();
190
191 const summary: PullRequestService.PullRequestService["Service"]["summary"] = (
192 input,
193 readOptions,
194 ) =>
195 Effect.gen(function* () {
196 assert.strictEqual(readOptions?.recoverTransientFailure, false);
197 yield* Ref.update(summaryCalls, (calls) => [...calls, input]);
198 return yield* options.summary?.(input) ?? Effect.succeed(makeSummary(input));
199 });
200
201 const stack: PullRequestService.PullRequestService["Service"]["stack"] = (input, readOptions) =>
202 Effect.gen(function* () {
203 assert.strictEqual(readOptions?.includeDetails, false);
204 yield* Ref.update(stackCalls, (calls) => [...calls, input]);
205 return yield* options.stack?.(input) ?? Effect.succeed(null);
206 });
207
208 const dispatch: Orchestrator.OrchestratorV2Shape["dispatch"] = (command) => {
209 if (command.type === "thread.pull-request-link.sync") {
210 return Ref.update(syncCommands, (recorded) => [...recorded, command]).pipe(
211 Effect.as({ sequence: 1, storedEvents: [] }),
212 );
213 }
214 if (command.type === "thread.pull-request.link") {
215 return Ref.update(linkCommands, (recorded) => [...recorded, command]).pipe(
216 Effect.as({ sequence: 1, storedEvents: [] }),
217 );
218 }
219 return Effect.die(new Error(`Unexpected command: ${command.type}`));
220 };
221
222 const layerDependencies = Layer.mergeAll(
223 Layer.mock(MergebotQueue.MergebotQueue)({ configuration: options.mergebot ?? null }),
224 Layer.mock(PullRequestService.PullRequestService)({
225 summary,
226 stack,
227 invalidate: options.invalidate ?? (() => Effect.void),
228 subscribeStateChanges: Effect.succeed(Stream.fromQueue(stateChanges)),
229 }),
230 Layer.mock(GitManager.GitManager)({
231 subscribePullRequestStateChanges: Effect.succeed(Stream.empty),
232 }),
233 Layer.mock(ProjectionStore.ProjectionStoreV2)({
234 // Mirrors the store's filter: active threads that have at least one link.
235 getThreadsWithPullRequests: (threadId) =>
236 Queue.offer(snapshotReads, undefined).pipe(
237 Effect.andThen(Ref.get(snapshots)),
238 Effect.map((snapshot) =>
239 snapshot.threads
240 .filter(
241 (thread) =>
242 (threadId === undefined || thread.id === threadId) &&
243 thread.archivedAt === null &&
244 thread.pullRequests.length > 0,
245 )
246 .map((thread) => ({
247 id: thread.id,
248 projectId: thread.projectId,
249 lineage: {
250 parentThreadId: null,
251 relationshipToParent: null,
252 rootThreadId: thread.id,
253 },
254 settledOverride: thread.settledOverride,
255 settledAt: thread.settledAt === null ? null : DateTime.makeUnsafe(thread.settledAt),
256 pullRequests: thread.pullRequests,
257 })),
258 ),
259 ),
260 }),
261 Layer.mock(Orchestrator.OrchestratorV2)({
262 getShellSnapshot: () =>
263 Ref.update(shellSnapshotReads, (count) => count + 1).pipe(
264 Effect.andThen(Effect.die(new Error("pull request sync must not read the shell"))),
265 ),
266 dispatch,
267 streamDomainEvents: Stream.fromQueue(domainEvents),
268 }),
269 Layer.succeed(ServerActivation.ServerActivation, Deferred.await(activation)),
270 Layer.succeed(Crypto.Crypto, testCrypto),
271 );
272
273 return {
274 activation,
275 snapshots,
276 snapshotReads,
277 shellSnapshotReads,
278 syncCommands,
279 linkCommands,
280 summaryCalls,
281 stackCalls,
282 domainEvents,
283 stateChanges,
284 layer: PullRequestSyncReactor.layer.pipe(Layer.provide(layerDependencies)),
285 };
286});
287
288type Harness = Effect.Success<ReturnType<typeof makeHarness>>;
289
290const startAndSweep = Effect.fn("startPullRequestSyncHarness")(function* (fixture: Harness) {
291 const reactor = yield* PullRequestSyncReactor.PullRequestSyncReactor;
292 yield* reactor.start();
293 yield* Deferred.succeed(fixture.activation, undefined);
294 yield* Queue.take(fixture.snapshotReads);
295 yield* reactor.drain;
296 return reactor;
297});
298
299const sweepAgain = Effect.fn("sweepPullRequestSyncHarness")(function* (
300 fixture: Harness,
301 reactor: PullRequestSyncReactor.PullRequestSyncReactor["Service"],
302) {
303 yield* TestClock.adjust("1 minute");
304 yield* Queue.take(fixture.snapshotReads);
305 yield* reactor.drain;
306});
307
308function runUpdated(
309 threadId: ThreadId,
310 status: OrchestrationV2Run["status"],
311): OrchestrationV2DomainEvent {
312 const runId = RunId.make(`run:${threadId}:1`);
313 const at = DateTime.makeUnsafe(NOW);
314 return {
315 type: "run.updated",
316 id: EventId.make(`event:${threadId}:${status}`),
317 threadId,
318 runId,
319 occurredAt: at,
320 payload: {
321 id: runId,
322 threadId,
323 ordinal: 1,
324 providerInstanceId: ProviderInstanceId.make("codex"),
325 modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5" },
326 providerThreadId: null,
327 userMessageId: MessageId.make(`message:${threadId}:1`),
328 rootNodeId: null,
329 activeAttemptId: null,
330 status,
331 requestedAt: at,
332 startedAt: at,
333 completedAt: at,
334 checkpointId: null,
335 contextHandoffId: null,
336 },
337 };
338}
339
340function commandRan(threadId: ThreadId, input: string): OrchestrationV2DomainEvent {
341 const runId = RunId.make(`run:${threadId}:1`);
342 const at = DateTime.makeUnsafe(NOW);
343 return {
344 type: "turn-item.updated",
345 id: EventId.make(`event:${threadId}:command`),
346 threadId,
347 runId,
348 occurredAt: at,
349 payload: {
350 id: TurnItemId.make(`item:${threadId}:command`),
351 threadId,
352 runId,
353 nodeId: null,
354 providerThreadId: null,
355 providerTurnId: null,
356 nativeItemRef: null,
357 parentItemId: null,
358 ordinal: 1,
359 status: "completed",
360 title: null,
361 startedAt: at,
362 completedAt: at,
363 updatedAt: at,
364 type: "command_execution",
365 input,
366 },
367 };
368}
369
370/** What the reactor would have persisted, so the next sweep sees its own writes. */
371function applySync(
372 snapshot: TestShellSnapshot,
373 commands: ReadonlyArray<SyncCommand>,
374): TestShellSnapshot {
375 return {
376 ...snapshot,
377 snapshotSequence: snapshot.snapshotSequence + 1,
378 threads: snapshot.threads.map((thread) => ({
379 ...thread,
380 pullRequests: thread.pullRequests.map((link) => {
381 const command = commands.findLast(
382 (candidate) => candidate.threadId === thread.id && candidate.number === link.number,
383 );
384 return command === undefined
385 ? link
386 : { ...link, snapshot: command.snapshot, stack: command.stack };
387 }),
388 })),
389 };
390}
391
392describe("PullRequestSyncReactor", () => {
393 it.effect("work summaries sync while unsupported stacks retain their existing links", () =>
394 Effect.scoped(
395 Effect.gen(function* () {
396 yield* TestClock.setTime(Date.parse(NOW));
397 const stack = {
398 kind: "native" as const,
399 id: "stack",
400 number: 7,
401 url: "https://github.com/owner/repository/stacks/7",
402 base: "main",
403 layers: [{ number: 7, headBranch: "feature", state: "open" as const }],
404 };
405 const fixture = yield* makeHarness({
406 mergebot: { origin: "https://mergebot.example", repositories: ["owner/repository"] },
407 snapshot: makeSnapshot([
408 makeThread("work", { pullRequests: [makeLink(7, null, { stack })] }),
409 ]),
410 summary: (input) =>
411 Effect.succeed(makeSummary(input, { state: "merged", mergedAt: NOW })),
412 stack: () => Effect.die("Work sync must not read unsupported GitHub stacks"),
413 });
414 yield* Effect.gen(function* () {
415 yield* startAndSweep(fixture);
416 const commands = yield* Ref.get(fixture.syncCommands);
417 assert.strictEqual(commands[0]?.snapshot?.state, "merged");
418 assert.deepStrictEqual(commands[0]?.stack, stack);
419 assert.deepStrictEqual(yield* Ref.get(fixture.stackCalls), []);
420 }).pipe(Effect.provide(fixture.layer));
421 }),
422 ),
423 );
424
425 it.effect("retries a failed stack read after the summary becomes terminal", () =>
426 Effect.scoped(
427 Effect.gen(function* () {
428 yield* TestClock.setTime(Date.parse(NOW));
429 let attempts = 0;
430 const nativeStack: PullRequestStack = {
431 id: "stack",
432 number: 7,
433 url: "https://github.com/owner/repository/stacks/7",
434 base: "main",
435 layers: [{ number: 7, headBranch: "feature", state: "merged" }],
436 };
437 const fixture = yield* makeHarness({
438 snapshot: makeSnapshot([makeThread("one", { pullRequests: [makeLink(7)] })]),
439 summary: (input) =>
440 Effect.succeed(makeSummary(input, { state: "merged", mergedAt: NOW })),
441 stack: () =>
442 ++attempts === 1
443 ? Effect.fail(
444 new PullRequestOperationError({
445 operation: "stack",
446 detail: "temporary failure",
447 }),
448 )
449 : Effect.succeed(nativeStack),
450 });
451 yield* Effect.gen(function* () {
452 const reactor = yield* startAndSweep(fixture);
453 const commands = yield* Ref.get(fixture.syncCommands);
454 yield* Ref.update(fixture.snapshots, (snapshot) => applySync(snapshot, commands));
455 yield* sweepAgain(fixture, reactor);
456 assert.strictEqual(attempts, 2);
457 assert.deepStrictEqual((yield* Ref.get(fixture.syncCommands)).at(-1)?.stack, {
458 kind: "native",
459 ...nativeStack,
460 });
461 }).pipe(Effect.provide(fixture.layer));
462 }),
463 ),
464 );
465
466 it.effect("explicit refresh reads a changed stack even when its PR summary is unchanged", () =>
467 Effect.scoped(
468 Effect.gen(function* () {
469 yield* TestClock.setTime(Date.parse(NOW));
470 const fixture = yield* makeHarness({
471 snapshot: makeSnapshot([makeThread("one", { pullRequests: [makeLink(7, {})] })]),
472 });
473 yield* Effect.gen(function* () {
474 const reactor = yield* startAndSweep(fixture);
475 assert.strictEqual((yield* Ref.get(fixture.stackCalls)).length, 0);
476 yield* reactor.requestSync({
477 host: "github.com",
478 repository: "owner/repository",
479 number: 7,
480 });
481 yield* Queue.take(fixture.snapshotReads);
482 yield* reactor.drain;
483 assert.strictEqual((yield* Ref.get(fixture.stackCalls)).length, 1);
484 }).pipe(Effect.provide(fixture.layer));
485 }),
486 ),
487 );
488 it.effect("snapshots an unsynced link once and writes it to the thread", () =>
489 Effect.scoped(
490 Effect.gen(function* () {
491 yield* TestClock.setTime(Date.parse(NOW));
492 const fixture = yield* makeHarness({
493 snapshot: makeSnapshot([makeThread("one", { pullRequests: [makeLink(42)] })]),
494 summary: (input) =>
495 Effect.succeed(makeSummary(input, { title: "Ship it", isDraft: true })),
496 });
497
498 yield* Effect.gen(function* () {
499 yield* startAndSweep(fixture);
500
501 assert.deepStrictEqual(yield* Ref.get(fixture.summaryCalls), [
502 {
503 projectId: PROJECT_ID,
504 host: "github.com",
505 repository: "owner/repository",
506 number: 42,
507 },
508 ]);
509 assert.deepStrictEqual(
510 (yield* Ref.get(fixture.syncCommands)).map(({ commandId: _, ...rest }) => rest),
511 [
512 {
513 type: "thread.pull-request-link.sync",
514 threadId: ThreadId.make("one"),
515 host: "github.com",
516 repository: "owner/repository",
517 number: 42,
518 snapshot: {
519 state: "open",
520 title: "Ship it",
521 headBranch: "feature",
522 baseBranch: "main",
523 isDraft: true,
524 updatedAt: "2026-08-27T00:00:00.000Z",
525 syncedAt: NOW,
526 closedAt: null,
527 mergedAt: null,
528 },
529 stack: null,
530 },
531 ],
532 );
533 assert.strictEqual((yield* Ref.get(fixture.stackCalls)).length, 1);
534 // Reads only linked threads, never the full shell snapshot of every thread.
535 assert.strictEqual(yield* Ref.get(fixture.shellSnapshotReads), 0);
536 }).pipe(Effect.provide(fixture.layer));
537 }),
538 ),
539 );
540
541 it.effect("asks the host once for a pull request shared by two threads", () =>
542 Effect.scoped(
543 Effect.gen(function* () {
544 yield* TestClock.setTime(Date.parse(NOW));
545 const fixture = yield* makeHarness({
546 snapshot: makeSnapshot([
547 makeThread("one", { pullRequests: [makeLink(42)] }),
548 makeThread("two", {
549 pullRequests: [makeLink(42, null, { repository: "Owner/Repository" })],
550 }),
551 ]),
552 });
553
554 yield* Effect.gen(function* () {
555 yield* startAndSweep(fixture);
556
557 assert.strictEqual((yield* Ref.get(fixture.summaryCalls)).length, 1);
558 const commands = yield* Ref.get(fixture.syncCommands);
559 assert.deepStrictEqual(
560 commands
561 .map((command) => [command.threadId, command.repository] as const)
562 .sort((left, right) => left[0].localeCompare(right[0])),
563 [
564 [ThreadId.make("one"), "owner/repository"],
565 [ThreadId.make("two"), "Owner/Repository"],
566 ],
567 );
568 }).pipe(Effect.provide(fixture.layer));
569 }),
570 ),
571 );
572
573 it.effect("recovers the HTTP port when syncing an older Forgejo link", () =>
574 Effect.scoped(
575 Effect.gen(function* () {
576 const fixture = yield* makeHarness({
577 snapshot: makeSnapshot([
578 makeThread("one", {
579 pullRequests: [
580 makeLink(42, null, {
581 host: "forge.example",
582 url: "http://forge.example:3000/owner/repository/pulls/42",
583 }),
584 ],
585 }),
586 ]),
587 summary: (input) => Effect.succeed(makeSummary(input)),
588 });
589 yield* Effect.gen(function* () {
590 yield* startAndSweep(fixture);
591 assert.strictEqual((yield* Ref.get(fixture.summaryCalls))[0]?.host, "forge.example:3000");
592 assert.strictEqual((yield* Ref.get(fixture.syncCommands))[0]?.host, "forge.example:3000");
593 }).pipe(Effect.provide(fixture.layer));
594 }),
595 ),
596 );
597
598 it.effect("dispatches nothing when the host snapshot is unchanged", () =>
599 Effect.scoped(
600 Effect.gen(function* () {
601 yield* TestClock.setTime(Date.parse(NOW));
602 const fixture = yield* makeHarness({
603 snapshot: makeSnapshot([makeThread("one", { pullRequests: [makeLink(42)] })]),
604 });
605
606 yield* Effect.gen(function* () {
607 const reactor = yield* startAndSweep(fixture);
608 const firstSweep = yield* Ref.get(fixture.syncCommands);
609 assert.strictEqual(firstSweep.length, 1);
610 yield* Ref.update(fixture.snapshots, (snapshot) => applySync(snapshot, firstSweep));
611
612 yield* sweepAgain(fixture, reactor);
613
614 // Still open on an active thread, so the host was asked again, but nothing changed.
615 assert.strictEqual((yield* Ref.get(fixture.summaryCalls)).length, 2);
616 assert.strictEqual((yield* Ref.get(fixture.stackCalls)).length, 1);
617 assert.strictEqual((yield* Ref.get(fixture.syncCommands)).length, 1);
618 }).pipe(Effect.provide(fixture.layer));
619 }),
620 ),
621 );
622
623 it.effect("refreshes closed links through the reactor's project after reopening elsewhere", () =>
624 Effect.scoped(
625 Effect.gen(function* () {
626 yield* TestClock.setTime(Date.parse(NOW));
627 const stale = yield* Ref.make(true);
628 const fixture = yield* makeHarness({
629 snapshot: makeSnapshot([
630 makeThread("first", { pullRequests: [makeLink(42, { state: "closed" })] }),
631 makeThread("second", {
632 projectId: ProjectId.make("second-project"),
633 pullRequests: [makeLink(42, { state: "closed" })],
634 }),
635 ]),
636 invalidate: ({ reference }) =>
637 reference?.projectId === makeProject().id && reference.host === "github.com"
638 ? Ref.set(stale, false)
639 : Effect.void,
640 summary: (input) =>
641 Ref.get(stale).pipe(
642 Effect.map((cached) => makeSummary(input, { state: cached ? "closed" : "open" })),
643 ),
644 });
645 yield* Effect.gen(function* () {
646 const reactor = yield* startAndSweep(fixture);
647 assert.strictEqual((yield* Ref.get(fixture.summaryCalls)).length, 1);
648 yield* reactor.requestSync({
649 host: "github.com",
650 repository: "owner/repository",
651 number: 42,
652 });
653 yield* Queue.take(fixture.snapshotReads);
654 yield* reactor.drain;
655 const commands = yield* Ref.get(fixture.syncCommands);
656 yield* Ref.update(fixture.snapshots, (snapshot) => applySync(snapshot, commands));
657 const snapshot = yield* Ref.get(fixture.snapshots);
658 assert.deepStrictEqual(
659 snapshot.threads.map((thread) => thread.pullRequests[0]?.snapshot?.state),
660 ["open", "open"],
661 );
662 }).pipe(Effect.provide(fixture.layer));
663 }),
664 ),
665 );
666
667 it.effect("stops asking the host once a pull request is merged", () =>
668 Effect.scoped(
669 Effect.gen(function* () {
670 yield* TestClock.setTime(Date.parse(NOW));
671 const fixture = yield* makeHarness({
672 snapshot: makeSnapshot([
673 makeThread("merged", { pullRequests: [makeLink(1, { state: "merged" })] }),
674 ]),
675 });
676
677 yield* Effect.gen(function* () {
678 const reactor = yield* startAndSweep(fixture);
679 yield* sweepAgain(fixture, reactor);
680
681 assert.deepStrictEqual(yield* Ref.get(fixture.summaryCalls), []);
682 assert.deepStrictEqual(yield* Ref.get(fixture.syncCommands), []);
683
684 yield* reactor.requestSync({
685 host: "github.com",
686 repository: "owner/repository",
687 number: 1,
688 });
689 yield* Queue.take(fixture.snapshotReads);
690 yield* reactor.drain;
691
692 assert.deepStrictEqual(
693 (yield* Ref.get(fixture.summaryCalls)).map((call) => call.number),
694 [1],
695 );
696 }).pipe(Effect.provide(fixture.layer));
697 }),
698 ),
699 );
700
701 it.effect("syncs a pull request a reader saw merge without waiting for the sweep", () =>
702 Effect.scoped(
703 Effect.gen(function* () {
704 yield* TestClock.setTime(Date.parse(NOW));
705 const state = yield* Ref.make<PullRequestSummary["state"]>("open");
706 const invalidated = yield* Ref.make<ReadonlyArray<number>>([]);
707 const fixture = yield* makeHarness({
708 snapshot: makeSnapshot([
709 makeThread("agent", { pullRequests: [makeLink(7, { state: "open" })] }),
710 ]),
711 summary: (input) =>
712 Ref.get(state).pipe(
713 Effect.map((current) =>
714 makeSummary(input, current === "merged" ? { state: current, mergedAt: NOW } : {}),
715 ),
716 ),
717 invalidate: ({ reference }) =>
718 Ref.update(invalidated, (numbers) => [...numbers, reference?.number ?? -1]),
719 });
720
721 yield* Effect.gen(function* () {
722 const reactor = yield* startAndSweep(fixture);
723 yield* Ref.set(state, "merged");
724
725 // The clock stays put: the next sweep is still a minute away.
726 yield* Queue.offer(fixture.stateChanges, {
727 host: "github.com",
728 repository: "owner/repository",
729 number: 7,
730 });
731 yield* Queue.take(fixture.snapshotReads);
732 yield* reactor.drain;
733
734 assert.deepStrictEqual(yield* Ref.get(invalidated), [7]);
735 assert.deepStrictEqual(
736 (yield* Ref.get(fixture.syncCommands)).map((command) => command.snapshot.state),
737 ["merged"],
738 );
739 }).pipe(Effect.provide(fixture.layer));
740 }),
741 ),
742 );
743
744 it.effect("discovers externally reopened pull requests after fifteen minutes", () =>
745 Effect.scoped(
746 Effect.gen(function* () {
747 yield* TestClock.setTime(Date.parse(NOW));
748 const state = yield* Ref.make<"closed" | "open">("closed");
749 const fixture = yield* makeHarness({
750 snapshot: makeSnapshot([
751 makeThread("closed", { pullRequests: [makeLink(2, { state: "closed" })] }),
752 ]),
753 summary: (input) =>
754 Ref.get(state).pipe(Effect.map((state) => makeSummary(input, { state }))),
755 });
756 yield* Effect.gen(function* () {
757 const reactor = yield* startAndSweep(fixture);
758 assert.strictEqual((yield* Ref.get(fixture.summaryCalls)).length, 1);
759 yield* Ref.set(state, "open");
760 for (let index = 0; index < 14; index += 1) yield* sweepAgain(fixture, reactor);
761 assert.strictEqual((yield* Ref.get(fixture.summaryCalls)).length, 1);
762 yield* sweepAgain(fixture, reactor);
763 assert.strictEqual((yield* Ref.get(fixture.summaryCalls)).length, 2);
764 assert.strictEqual((yield* Ref.get(fixture.syncCommands)).at(-1)?.snapshot.state, "open");
765 }).pipe(Effect.provide(fixture.layer));
766 }),
767 ),
768 );
769
770 it.effect("preserves a refresh requested while an older host read is in flight", () =>
771 Effect.scoped(
772 Effect.gen(function* () {
773 yield* TestClock.setTime(Date.parse(NOW));
774 const reading = yield* Deferred.make<void>();
775 const release = yield* Deferred.make<void>();
776 let calls = 0;
777 const fixture = yield* makeHarness({
778 snapshot: makeSnapshot([]),
779 summary: (input) =>
780 Effect.gen(function* () {
781 calls += 1;
782 if (calls === 1) {
783 yield* Deferred.succeed(reading, undefined);
784 yield* Deferred.await(release);
785 }
786 return makeSummary(input, { state: calls === 1 ? "closed" : "open" });
787 }),
788 });
789 yield* Effect.gen(function* () {
790 const reactor = yield* startAndSweep(fixture);
791 yield* Ref.set(
792 fixture.snapshots,
793 makeSnapshot([
794 makeThread("closed", { pullRequests: [makeLink(2, { state: "closed" })] }),
795 ]),
796 );
797 const key = { host: "github.com", repository: "owner/repository", number: 2 };
798 yield* reactor.requestSync(key);
799 yield* Deferred.await(reading);
800 yield* reactor.requestSync(key);
801 yield* Deferred.succeed(release, undefined);
802 yield* reactor.drain;
803 assert.strictEqual((yield* Ref.get(fixture.summaryCalls)).length, 2);
804 assert.strictEqual((yield* Ref.get(fixture.syncCommands)).at(-1)?.snapshot.state, "open");
805 }).pipe(Effect.provide(fixture.layer));
806 }),
807 ),
808 );
809
810 it.effect("reads a burst of requested links in one sweep", () =>
811 Effect.scoped(
812 Effect.gen(function* () {
813 yield* TestClock.setTime(Date.parse(NOW));
814 const reading = yield* Deferred.make<void>();
815 const release = yield* Deferred.make<void>();
816 const fixture = yield* makeHarness({
817 snapshot: makeSnapshot([]),
818 summary: (input) =>
819 Effect.gen(function* () {
820 if (input.number === 1) {
821 yield* Deferred.succeed(reading, undefined);
822 yield* Deferred.await(release);
823 }
824 return makeSummary(input);
825 }),
826 });
827 yield* Effect.gen(function* () {
828 const reactor = yield* startAndSweep(fixture);
829 yield* Ref.set(
830 fixture.snapshots,
831 makeSnapshot(
832 [1, 2, 3, 4].map((number) =>
833 makeThread(`thread-${number}`, { pullRequests: [makeLink(number)] }),
834 ),
835 ),
836 );
837 const request = (number: number) =>
838 reactor.requestSync({ host: "github.com", repository: "owner/repository", number });
839 yield* request(1);
840 yield* Deferred.await(reading);
841 yield* Effect.forEach([2, 3, 4], request, { discard: true });
842 yield* Deferred.succeed(release, undefined);
843 yield* reactor.drain;
844 // One sweep for the first link and one for the three that arrived while it read.
845 assert.strictEqual(yield* Queue.size(fixture.snapshotReads), 2);
846 assert.deepStrictEqual(
847 (yield* Ref.get(fixture.summaryCalls)).map((ref) => ref.number).toSorted(),
848 [1, 2, 3, 4],
849 );
850 }).pipe(Effect.provide(fixture.layer));
851 }),
852 ),
853 );
854
855 it.effect("skips the stack read for a pull request the summary places in no stack", () =>
856 Effect.scoped(
857 Effect.gen(function* () {
858 yield* TestClock.setTime(Date.parse(NOW));
859 const fixture = yield* makeHarness({
860 snapshot: makeSnapshot([
861 makeThread("unstacked", { pullRequests: [makeLink(1)] }),
862 makeThread("stacked", { pullRequests: [makeLink(2)] }),
863 makeThread("unknown", { pullRequests: [makeLink(3)] }),
864 ]),
865 summary: (input) =>
866 Effect.succeed(
867 makeSummary(
868 input,
869 input.number === 1
870 ? { stack: null }
871 : input.number === 2
872 ? { stack: { number: 9, position: 1, size: 2, base: "main" } }
873 : {},
874 ),
875 ),
876 });
877 yield* Effect.gen(function* () {
878 yield* startAndSweep(fixture);
879 assert.deepStrictEqual(
880 (yield* Ref.get(fixture.stackCalls)).map((ref) => ref.number).toSorted(),
881 [2, 3],
882 );
883 assert.deepStrictEqual(
884 (yield* Ref.get(fixture.syncCommands)).map((command) => command.number).toSorted(),
885 [1, 2, 3],
886 );
887 }).pipe(Effect.provide(fixture.layer));
888 }),
889 ),
890 );
891
892 it.effect("leaves a settled thread's links unread until the thread is unsettled", () =>
893 Effect.scoped(
894 Effect.gen(function* () {
895 yield* TestClock.setTime(Date.parse(NOW));
896 const fixture = yield* makeHarness({
897 snapshot: makeSnapshot([
898 makeThread("settled", {
899 settledOverride: "settled",
900 settledAt: "2026-08-21T00:00:00.000Z",
901 pullRequests: [makeLink(5, { state: "open" }), makeLink(6, { state: "closed" })],
902 }),
903 ]),
904 });
905
906 yield* Effect.gen(function* () {
907 const reactor = yield* startAndSweep(fixture);
908 for (let index = 0; index < 15; index += 1) yield* sweepAgain(fixture, reactor);
909 assert.deepStrictEqual(yield* Ref.get(fixture.summaryCalls), []);
910
911 yield* Ref.update(fixture.snapshots, (snapshot) => ({
912 ...snapshot,
913 threads: snapshot.threads.map((thread) => ({
914 ...thread,
915 settledOverride: null,
916 settledAt: null,
917 })),
918 }));
919 yield* sweepAgain(fixture, reactor);
920 assert.deepStrictEqual(
921 (yield* Ref.get(fixture.summaryCalls)).map((call) => call.number).toSorted(),
922 [5, 6],
923 );
924 }).pipe(Effect.provide(fixture.layer));
925 }),
926 ),
927 );
928
929 it.effect("auto-links missing native stack layers on active threads only", () =>
930 Effect.scoped(
931 Effect.gen(function* () {
932 yield* TestClock.setTime(Date.parse(NOW));
933 const stack: PullRequestStack = {
934 id: "stack-1",
935 number: 42,
936 url: "https://github.com/owner/repository/stack/1",
937 base: "main",
938 layers: [
939 { number: 41, headBranch: "layer-1", state: "merged" },
940 { number: 42, headBranch: "layer-2", state: "open" },
941 { number: 43, headBranch: "layer-3", state: "open" },
942 ],
943 };
944 const fixture = yield* makeHarness({
945 snapshot: makeSnapshot([
946 makeThread("one", {
947 pullRequests: [
948 makeLink(42),
949 makeLink(41, { state: "merged" }, { source: "stack-dismissed" }),
950 ],
951 }),
952 makeThread("settled", {
953 settledOverride: "settled",
954 settledAt: "2026-08-21T00:00:00.000Z",
955 pullRequests: [makeLink(42)],
956 }),
957 ]),
958 stack: () => Effect.succeed(stack),
959 });
960
961 yield* Effect.gen(function* () {
962 yield* startAndSweep(fixture);
963
964 const syncCommands = yield* Ref.get(fixture.syncCommands);
965 assert.deepStrictEqual(
966 syncCommands.map((command) => [command.threadId, command.stack] as const),
967 [
968 [ThreadId.make("one"), { kind: "native", ...stack }],
969 [ThreadId.make("settled"), { kind: "native", ...stack }],
970 ],
971 );
972 assert.deepStrictEqual(
973 (yield* Ref.get(fixture.linkCommands)).map(({ commandId: _, ...rest }) => rest),
974 [
975 {
976 type: "thread.pull-request.link",
977 threadId: ThreadId.make("one"),
978 host: "github.com",
979 repository: "owner/repository",
980 number: 43,
981 url: "https://github.com/owner/repository/pull/43",
982 source: "stack",
983 },
984 ],
985 );
986 }).pipe(Effect.provide(fixture.layer));
987 }),
988 ),
989 );
990
991 it.effect("keeps existing snapshots and continues when the host fails", () =>
992 Effect.scoped(
993 Effect.gen(function* () {
994 yield* TestClock.setTime(Date.parse(NOW));
995 const fixture = yield* makeHarness({
996 snapshot: makeSnapshot([
997 makeThread("failing", { pullRequests: [makeLink(7, { state: "open" })] }),
998 makeThread("fine", { pullRequests: [makeLink(8)] }),
999 ]),
1000 summary: (input) =>
1001 input.number === 7
1002 ? Effect.fail(
1003 new PullRequestOperationError({ operation: "summary", detail: "host down" }),
1004 )
1005 : Effect.succeed(makeSummary(input)),
1006 });
1007
1008 yield* Effect.gen(function* () {
1009 yield* startAndSweep(fixture);
1010
1011 assert.deepStrictEqual(
1012 (yield* Ref.get(fixture.syncCommands)).map((command) => command.number),
1013 [8],
1014 );
1015 }).pipe(Effect.provide(fixture.layer));
1016 }),
1017 ),
1018 );
1019
1020 it.effect("leaves a rate limited host unread until its pause ends", () => {
1021 const skips: Array<ReadonlyArray<unknown>> = [];
1022 const logger = Logger.make(({ logLevel, message }) => {
1023 const parts = Array.isArray(message) ? message : [message];
1024 if (logLevel === "Warn" && parts[0] === "pull request sync skipped") skips.push(parts);
1025 });
1026 const retryAt = Date.parse(NOW) + 3 * 60_000;
1027 return Effect.scoped(
1028 Effect.gen(function* () {
1029 yield* TestClock.setTime(Date.parse(NOW));
1030 const fixture = yield* makeHarness({
1031 snapshot: makeSnapshot([
1032 makeThread("first", { pullRequests: [makeLink(7, { state: "open" })] }),
1033 makeThread("second", { pullRequests: [makeLink(8, { state: "open" })] }),
1034 makeThread("third", {
1035 pullRequests: [
1036 makeLink(
1037 9,
1038 { state: "open" },
1039 { host: "forge.example", url: "https://forge.example/owner/repository/pulls/9" },
1040 ),
1041 ],
1042 }),
1043 ]),
1044 summary: (input) =>
1045 Effect.gen(function* () {
1046 if (input.host === "github.com" && (yield* Clock.currentTimeMillis) < retryAt) {
1047 const paused = new PullRequestProviderError({
1048 provider: SourceControlProviderKind.make("github"),
1049 operation: "getChangeRequestSummary",
1050 reason: "rate-limited",
1051 detail: "paused",
1052 retryAt,
1053 });
1054 return yield* new PullRequestOperationError({
1055 operation: "summary",
1056 detail: "paused",
1057 cause: paused,
1058 });
1059 }
1060 return makeSummary(input, { state: "open" });
1061 }),
1062 });
1063 const githubReads = Ref.get(fixture.summaryCalls).pipe(
1064 Effect.map((calls) =>
1065 calls.filter((call) => call.host === "github.com").map((call) => call.number),
1066 ),
1067 );
1068
1069 yield* Effect.gen(function* () {
1070 // The first refused read pauses the host, so the sweep does not try the other.
1071 const reactor = yield* startAndSweep(fixture);
1072 assert.deepStrictEqual(yield* githubReads, [7]);
1073 assert.strictEqual(skips.length, 1);
1074 assert.deepInclude(skips[0]![1], { count: 1 });
1075
1076 // A requested refresh waits for the pause like the sweep does.
1077 yield* reactor.requestSync({
1078 host: "github.com",
1079 repository: "owner/repository",
1080 number: 7,
1081 });
1082 yield* Queue.take(fixture.snapshotReads);
1083 yield* reactor.drain;
1084 yield* sweepAgain(fixture, reactor);
1085 yield* sweepAgain(fixture, reactor);
1086 assert.deepStrictEqual(yield* githubReads, [7]);
1087 assert.strictEqual(skips.length, 1);
1088 // Other hosts keep their cadence.
1089 assert.strictEqual((yield* Ref.get(fixture.summaryCalls)).length, 4);
1090
1091 yield* sweepAgain(fixture, reactor);
1092 assert.sameMembers((yield* githubReads).slice(1), [7, 8]);
1093 }).pipe(
1094 // The reactor forks its worker while its layer builds, so the logger must reach it there.
1095 Effect.provide(
1096 fixture.layer.pipe(Layer.provide(Logger.layer([logger], { mergeWithExisting: false }))),
1097 ),
1098 );
1099 }),
1100 );
1101 });
1102
1103 it.effect("pauses the host when only its stack read is rate limited", () =>
1104 Effect.scoped(
1105 Effect.gen(function* () {
1106 yield* TestClock.setTime(Date.parse(NOW));
1107 const retryAt = Date.parse(NOW) + 3 * 60_000;
1108 const fixture = yield* makeHarness({
1109 snapshot: makeSnapshot([makeThread("one", { pullRequests: [makeLink(7, null)] })]),
1110 summary: (input) => Effect.succeed(makeSummary(input, { state: "open" })),
1111 stack: () =>
1112 Effect.gen(function* () {
1113 if ((yield* Clock.currentTimeMillis) >= retryAt) return null;
1114 return yield* new PullRequestOperationError({
1115 operation: "stack",
1116 detail: "paused",
1117 cause: new PullRequestProviderError({
1118 provider: SourceControlProviderKind.make("github"),
1119 operation: "getChangeRequestStack",
1120 reason: "rate-limited",
1121 detail: "paused",
1122 retryAt,
1123 }),
1124 });
1125 }),
1126 });
1127 const reads = Effect.all([
1128 Ref.get(fixture.summaryCalls).pipe(Effect.map((calls) => calls.length)),
1129 Ref.get(fixture.stackCalls).pipe(Effect.map((calls) => calls.length)),
1130 ]);
1131
1132 yield* Effect.gen(function* () {
1133 const reactor = yield* startAndSweep(fixture);
1134 assert.deepStrictEqual(yield* reads, [1, 1]);
1135 yield* sweepAgain(fixture, reactor);
1136 yield* sweepAgain(fixture, reactor);
1137 assert.deepStrictEqual(yield* reads, [1, 1]);
1138
1139 // At retryAt the pull request is read again, stack included.
1140 yield* sweepAgain(fixture, reactor);
1141 assert.deepStrictEqual(yield* reads, [2, 2]);
1142 assert.strictEqual((yield* Ref.get(fixture.syncCommands)).length, 1);
1143 }).pipe(Effect.provide(fixture.layer));
1144 }),
1145 ),
1146 );
1147
1148 it.effect("reads open links fresh only when a run that ran a merge command ends", () =>
1149 Effect.scoped(
1150 Effect.gen(function* () {
1151 yield* TestClock.setTime(Date.parse(NOW));
1152 const state = yield* Ref.make<PullRequestSummary["state"]>("open");
1153 const invalidated = yield* Ref.make<ReadonlyArray<number>>([]);
1154 const fixture = yield* makeHarness({
1155 snapshot: makeSnapshot([
1156 makeThread("agent", { pullRequests: [makeLink(7, { state: "open" })] }),
1157 makeThread("other", { pullRequests: [makeLink(9, { state: "open" })] }),
1158 ]),
1159 summary: (input) =>
1160 Ref.get(state).pipe(
1161 Effect.map((current) =>
1162 makeSummary(input, current === "merged" ? { state: current, mergedAt: NOW } : {}),
1163 ),
1164 ),
1165 invalidate: ({ reference }) =>
1166 Ref.update(invalidated, (numbers) => [...numbers, reference?.number ?? -1]),
1167 });
1168
1169 yield* Effect.gen(function* () {
1170 const reactor = yield* startAndSweep(fixture);
1171 // The agent merges from a shell during its turn, inside the summary cache window.
1172 yield* Ref.set(state, "merged");
1173 yield* Ref.set(fixture.summaryCalls, []);
1174
1175 yield* Queue.offerAll(fixture.domainEvents, [
1176 // A run that only reads its pull request costs no host read when it ends.
1177 commandRan(ThreadId.make("other"), "gh pr view 9"),
1178 runUpdated(ThreadId.make("other"), "completed"),
1179 commandRan(ThreadId.make("agent"), "gh pr merge 7 --squash 2>&1 | tail -3"),
1180 runUpdated(ThreadId.make("agent"), "completed"),
1181 ]);
1182 // The agent thread's lookup, then the requested sweep.
1183 yield* Queue.take(fixture.snapshotReads);
1184 yield* Queue.take(fixture.snapshotReads);
1185 yield* reactor.drain;
1186
1187 assert.deepStrictEqual(yield* Ref.get(invalidated), [7]);
1188 assert.deepStrictEqual(
1189 (yield* Ref.get(fixture.summaryCalls)).map((call) => call.number),
1190 [7],
1191 );
1192 assert.deepStrictEqual(
1193 (yield* Ref.get(fixture.syncCommands)).map((command) => [
1194 command.number,
1195 command.snapshot.state,
1196 ]),
1197 [[7, "merged"]],
1198 );
1199 }).pipe(Effect.provide(fixture.layer));
1200 }),
1201 ),
1202 );
1203});