1import {
2 ProjectId,
3 ProviderInstanceId,
4 SourceControlProviderKind,
5 ThreadId,
6 type OrchestrationProjectShell,
7 type OrchestrationV2DomainEvent,
8 type OrchestrationV2ThreadShell,
9 type OrchestrationV2ServerCommand,
10} from "@t3tools/contracts";
11import { describe, expect, it } from "@effect/vitest";
12import * as SecureRandom from "effect/crypto/SecureRandom";
13import * as DateTime from "effect/DateTime";
14import * as Deferred from "effect/Deferred";
15import * as Effect from "effect/Effect";
16import * as FileSystem from "effect/FileSystem";
17import * as Layer from "effect/Layer";
18import * as Option from "effect/Option";
19import * as Path from "effect/Path";
20import * as PlatformError from "effect/PlatformError";
21import * as Queue from "effect/Queue";
22import * as Stream from "effect/Stream";
23import * as TestClock from "effect/testing/TestClock";
24
25import * as BackgroundPolicy from "../background/BackgroundPolicy.ts";
26import * as GitManager from "../git/GitManager.ts";
27import * as ProjectStore from "./ProjectStore.ts";
28import * as RepositoryIdentityResolver from "../project/RepositoryIdentityResolver.ts";
29import * as PullRequestService from "../pullRequest/PullRequestService.ts";
30import * as ServerActivation from "../serverActivation.ts";
31import * as ProjectionStore from "./ProjectionStore.ts";
32import * as SqlClient from "effect/sql/SqlClient";
33import * as SqlitePersistence from "../persistence/Sqlite.ts";
34import { EventId } from "@t3tools/contracts";
35import * as Orchestrator from "./Orchestrator.ts";
36import * as ThreadPullRequestService from "./ThreadPullRequestService.ts";
37
38describe("ThreadPullRequestServiceV2 project guard", () => {
39 it("rejects a pull-request result when the project root changes before dispatch", () => {
40 const currentProject = Option.some({
41 workspaceRoot: "/workspace/replaced",
42 } satisfies Pick<OrchestrationProjectShell, "workspaceRoot">);
43
44 expect(
45 ThreadPullRequestService.projectWorkspaceMatchesSnapshot(
46 currentProject,
47 "/workspace/original",
48 ),
49 ).toBe(false);
50 });
51});
52
53describe("ThreadPullRequestServiceV2 reads", () => {
54 const NOW = DateTime.makeUnsafe("2026-09-20T00:00:00.000Z");
55 const threadShell = (id: string): OrchestrationV2ThreadShell => {
56 const threadId = ThreadId.make(id);
57 return {
58 id: threadId,
59 projectId: ProjectId.make("project-1"),
60 title: id,
61 providerInstanceId: ProviderInstanceId.make("codex"),
62 modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" },
63 runtimeMode: "full-access",
64 interactionMode: "default",
65 branch: null,
66 worktreePath: null,
67 activeProviderThreadId: null,
68 lineage: { rootThreadId: threadId, parentThreadId: null, relationshipToParent: null },
69 forkedFrom: null,
70 createdBy: "user",
71 creationSource: "web",
72 activeRunId: null,
73 latestRunId: null,
74 status: "idle",
75 pendingRuntimeRequest: null,
76 latestVisibleMessage: null,
77 latestUserMessageAt: null,
78 hasActionableProposedPlan: false,
79 itemCount: 0,
80 visibleItemCount: 0,
81 createdAt: NOW,
82 updatedAt: NOW,
83 archivedAt: null,
84 settledOverride: null,
85 settledAt: null,
86 lastVisitedAt: null,
87 deletedAt: null,
88 };
89 };
90
91 it.effect("does not sweep branch pull requests while no client is connected", () =>
92 Effect.gen(function* () {
93 // Polling active branches without a connected client wastes GitHub requests.
94 const thread = { ...threadShell("idle-branch-thread"), branch: "feature/idle" };
95 const project: OrchestrationProjectShell = {
96 id: thread.projectId,
97 title: "Project",
98 workspaceRoot: "/workspace/project",
99 defaultModelSelection: null,
100 scripts: [],
101 createdAt: "2026-08-01T00:00:00.000Z",
102 updatedAt: "2026-08-01T00:00:00.000Z",
103 };
104 const activation = yield* Deferred.make<void>();
105 const domainEvents = yield* Queue.unbounded<OrchestrationV2DomainEvent>();
106 const turnLookup = yield* Deferred.make<void>();
107 const lookups: Array<{ readonly refresh: boolean }> = [];
108 const dependencies = Layer.mergeAll(
109 Layer.mock(Orchestrator.OrchestratorV2)({
110 streamDomainEvents: Stream.fromQueue(domainEvents),
111 getPullRequestSnapshot: () =>
112 Effect.succeed({
113 schemaVersion: 2,
114 snapshotSequence: 1,
115 threads: [thread],
116 archivedThreads: [],
117 }),
118 }),
119 Layer.mock(ProjectStore.ProjectStoreV2)({
120 listShells: () => Effect.succeed([project]),
121 getShell: () => Effect.succeedSome(project),
122 }),
123 Layer.mock(GitManager.GitManager)({
124 branchPullRequest: (_input, options) =>
125 Effect.gen(function* () {
126 const refresh = options?.refresh ?? false;
127 lookups.push({ refresh });
128 if (refresh) yield* Deferred.succeed(turnLookup, undefined);
129 return null;
130 }),
131 }),
132 Layer.mock(PullRequestService.PullRequestService)({}),
133 Layer.mock(RepositoryIdentityResolver.RepositoryIdentityResolver)({
134 resolve: () =>
135 Effect.succeed({
136 canonicalKey: "github.com/acme/project",
137 locator: {
138 source: "git-remote" as const,
139 remoteName: "origin",
140 remoteUrl: "git@github.com:acme/project.git",
141 },
142 provider: SourceControlProviderKind.make("github"),
143 owner: "acme",
144 name: "project",
145 }),
146 }),
147 // No client holds an activity lease.
148 Layer.mock(BackgroundPolicy.BackgroundPolicy)({
149 shouldRunScopeWork: () => Effect.succeed(false),
150 shouldRunOpportunisticWork: Effect.succeed(false),
151 }),
152 Layer.succeed(ServerActivation.ServerActivation, Deferred.await(activation)),
153 Layer.succeed(
154 SecureRandom.SecureRandom,
155 SecureRandom.make((size) => new Uint8Array(size).fill(1)),
156 ),
157 FileSystem.layerNoop({}),
158 Path.layer,
159 );
160 yield* Effect.gen(function* () {
161 const service = yield* ThreadPullRequestService.make;
162 yield* service.start();
163 yield* Deferred.succeed(activation, undefined);
164
165 // A finished turn still discovers the pull request for its branch.
166 yield* Queue.offer(domainEvents, {
167 type: "checkpoint.captured",
168 threadId: thread.id,
169 } as OrchestrationV2DomainEvent);
170 yield* Deferred.await(turnLookup);
171 yield* service.drain;
172 const lookupsBeforeIdle = lookups.length;
173
174 yield* TestClock.adjust("10 minutes");
175 yield* service.drain;
176
177 expect(lookups.slice(lookupsBeforeIdle)).toEqual([]);
178 }).pipe(Effect.provide(dependencies));
179 }),
180 );
181
182 it.effect("looks up a finished turn's branch once for all its turn-end events", () =>
183 Effect.gen(function* () {
184 const thread = { ...threadShell("turn-end-thread"), branch: "t3/turn-end" };
185 const otherThread = {
186 ...threadShell("other-turn-end-thread"),
187 branch: "t3/other-turn-end",
188 };
189 const project: OrchestrationProjectShell = {
190 id: thread.projectId,
191 title: "Project",
192 workspaceRoot: "/workspace/project",
193 defaultModelSelection: null,
194 scripts: [],
195 createdAt: "2026-08-01T00:00:00.000Z",
196 updatedAt: "2026-08-01T00:00:00.000Z",
197 };
198 // A turn end commits its checkpoints and the run's end together. The queued
199 // non-refresh request ahead of them must still become a refresh when they join it.
200 const turnEnd = [
201 { type: "thread.metadata-updated", threadId: thread.id },
202 { type: "checkpoint.captured", threadId: thread.id },
203 { type: "checkpoint.captured", threadId: thread.id },
204 { type: "run.updated", threadId: thread.id, payload: { status: "completed" } },
205 { type: "checkpoint.captured", threadId: otherThread.id },
206 { type: "run.updated", threadId: otherThread.id, payload: { status: "completed" } },
207 ] as Array<OrchestrationV2DomainEvent>;
208 // Arrives during the thread's own lookup, which may already have read the old host state.
209 const duringLookup = [
210 { type: "checkpoint.captured", threadId: thread.id },
211 { type: "thread.metadata-updated", threadId: thread.id },
212 ] as Array<OrchestrationV2DomainEvent>;
213 const activation = yield* Deferred.make<void>();
214 const sweepStarted = yield* Deferred.make<void>();
215 const turnEndQueued = yield* Deferred.make<void>();
216 const lookupStarted = yield* Deferred.make<void>();
217 const followUpQueued = yield* Deferred.make<void>();
218 const lookups: Array<{ readonly branch: string; readonly refresh: boolean }> = [];
219 const dependencies = Layer.mergeAll(
220 Layer.mock(Orchestrator.OrchestratorV2)({
221 // The batch arrives while the worker is busy with the startup sweep.
222 streamDomainEvents: Stream.fromEffect(Deferred.await(sweepStarted)).pipe(
223 Stream.drain,
224 Stream.concat(Stream.fromIterable(turnEnd)),
225 Stream.concat(
226 Stream.fromEffect(Deferred.succeed(turnEndQueued, undefined)).pipe(Stream.drain),
227 ),
228 Stream.concat(Stream.fromEffect(Deferred.await(lookupStarted)).pipe(Stream.drain)),
229 Stream.concat(Stream.fromIterable(duringLookup)),
230 Stream.concat(
231 Stream.fromEffect(Deferred.succeed(followUpQueued, undefined)).pipe(Stream.drain),
232 ),
233 Stream.concat(Stream.never),
234 ),
235 getPullRequestSnapshot: ({ threadId }) =>
236 Effect.succeed({
237 schemaVersion: 2,
238 snapshotSequence: 1,
239 threads: [thread, otherThread].filter(
240 (candidate) => threadId === undefined || candidate.id === threadId,
241 ),
242 archivedThreads: [],
243 }),
244 }),
245 Layer.mock(ProjectStore.ProjectStoreV2)({
246 listShells: () => Effect.succeed([project]),
247 getShell: () => Effect.succeedSome(project),
248 }),
249 Layer.mock(GitManager.GitManager)({
250 branchPullRequest: (input, options) =>
251 Effect.gen(function* () {
252 lookups.push({ branch: input.branch, refresh: options?.refresh ?? false });
253 if (!(yield* Deferred.isDone(turnEndQueued))) {
254 yield* Deferred.succeed(sweepStarted, undefined);
255 yield* Deferred.await(turnEndQueued);
256 } else if (
257 input.branch === thread.branch &&
258 !(yield* Deferred.isDone(followUpQueued))
259 ) {
260 yield* Deferred.succeed(lookupStarted, undefined);
261 yield* Deferred.await(followUpQueued);
262 }
263 return null;
264 }),
265 }),
266 Layer.mock(PullRequestService.PullRequestService)({}),
267 Layer.mock(RepositoryIdentityResolver.RepositoryIdentityResolver)({
268 resolve: () =>
269 Effect.succeed({
270 canonicalKey: "github.com/acme/project",
271 locator: {
272 source: "git-remote" as const,
273 remoteName: "origin",
274 remoteUrl: "git@github.com:acme/project.git",
275 },
276 provider: SourceControlProviderKind.make("github"),
277 owner: "acme",
278 name: "project",
279 }),
280 }),
281 Layer.mock(BackgroundPolicy.BackgroundPolicy)({
282 shouldRunScopeWork: () => Effect.succeed(false),
283 shouldRunOpportunisticWork: Effect.succeed(false),
284 }),
285 Layer.succeed(ServerActivation.ServerActivation, Deferred.await(activation)),
286 Layer.succeed(
287 SecureRandom.SecureRandom,
288 SecureRandom.make((size) => new Uint8Array(size).fill(1)),
289 ),
290 FileSystem.layerNoop({}),
291 Path.layer,
292 );
293 yield* Effect.gen(function* () {
294 const service = yield* ThreadPullRequestService.make;
295 yield* service.start();
296 yield* Deferred.succeed(activation, undefined);
297 yield* Deferred.await(followUpQueued);
298 yield* service.drain;
299 }).pipe(Effect.provide(dependencies));
300
301 // The startup sweep, one refresh lookup per thread for its turn-end batch, then one
302 // follow-up refresh for the events that arrived during the thread's lookup.
303 expect(lookups).toEqual([
304 { branch: thread.branch, refresh: false },
305 { branch: otherThread.branch, refresh: false },
306 { branch: thread.branch, refresh: true },
307 { branch: otherThread.branch, refresh: true },
308 { branch: thread.branch, refresh: true },
309 ]);
310 }),
311 );
312
313 it.effect.each([
314 {
315 worktree: "a linked worktree",
316 existingPaths: ["/workspace/worktree", "/workspace/worktree/.git"],
317 unreadablePaths: [],
318 cwd: "/workspace/worktree",
319 },
320 // A folder that outlived its .git makes git fail with not_a_repository.
321 {
322 worktree: "a folder without .git",
323 existingPaths: ["/workspace/worktree"],
324 unreadablePaths: [],
325 cwd: "/workspace/project",
326 },
327 {
328 worktree: "a folder whose .git cannot be checked",
329 existingPaths: ["/workspace/worktree"],
330 unreadablePaths: ["/workspace/worktree/.git"],
331 cwd: "/workspace/worktree",
332 },
333 {
334 worktree: "a missing folder",
335 existingPaths: [],
336 unreadablePaths: [],
337 cwd: "/workspace/project",
338 },
339 ] as const)(
340 "looks up a branch whose worktree is $worktree in $cwd",
341 ({ existingPaths, unreadablePaths, cwd }) =>
342 Effect.gen(function* () {
343 const thread = {
344 ...threadShell("worktree-thread"),
345 settledOverride: "settled" as const,
346 settledAt: NOW,
347 branch: "feature/worktree",
348 worktreePath: "/workspace/worktree",
349 };
350 const project: OrchestrationProjectShell = {
351 id: thread.projectId,
352 title: "Project",
353 workspaceRoot: "/workspace/project",
354 defaultModelSelection: null,
355 scripts: [],
356 createdAt: "2026-08-01T00:00:00.000Z",
357 updatedAt: "2026-08-01T00:00:00.000Z",
358 };
359 const activation = yield* Deferred.make<void>();
360 const sweepStarted = yield* Deferred.make<void>();
361 const lookupCwds: Array<string> = [];
362 const dependencies = Layer.mergeAll(
363 Layer.mock(Orchestrator.OrchestratorV2)({
364 streamDomainEvents: Stream.never,
365 getPullRequestSnapshot: () =>
366 Deferred.succeed(sweepStarted, undefined).pipe(
367 Effect.as({
368 schemaVersion: 2,
369 snapshotSequence: 1,
370 threads: [thread],
371 archivedThreads: [],
372 }),
373 ),
374 }),
375 Layer.mock(ProjectStore.ProjectStoreV2)({
376 listShells: () => Effect.succeed([project]),
377 getShell: () => Effect.succeedSome(project),
378 }),
379 Layer.mock(GitManager.GitManager)({
380 branchPullRequest: (input) =>
381 Effect.sync(() => {
382 lookupCwds.push(input.cwd);
383 return null;
384 }),
385 }),
386 Layer.mock(PullRequestService.PullRequestService)({}),
387 Layer.mock(RepositoryIdentityResolver.RepositoryIdentityResolver)({
388 resolve: () =>
389 Effect.succeed({
390 canonicalKey: "github.com/acme/project",
391 locator: {
392 source: "git-remote" as const,
393 remoteName: "origin",
394 remoteUrl: "git@github.com:acme/project.git",
395 },
396 provider: SourceControlProviderKind.make("github"),
397 owner: "acme",
398 name: "project",
399 }),
400 }),
401 Layer.mock(BackgroundPolicy.BackgroundPolicy)({}),
402 Layer.succeed(ServerActivation.ServerActivation, Deferred.await(activation)),
403 Layer.succeed(
404 SecureRandom.SecureRandom,
405 SecureRandom.make((size) => new Uint8Array(size).fill(1)),
406 ),
407 FileSystem.layerNoop({
408 exists: (path) =>
409 unreadablePaths.some((entry) => entry === path)
410 ? Effect.fail(
411 PlatformError.systemError({
412 _tag: "PermissionDenied",
413 module: "FileSystem",
414 method: "exists",
415 pathOrDescriptor: path,
416 }),
417 )
418 : Effect.succeed(existingPaths.some((entry) => entry === path)),
419 }),
420 Path.layer,
421 );
422 yield* Effect.gen(function* () {
423 const service = yield* ThreadPullRequestService.make;
424 yield* service.start();
425 yield* Deferred.succeed(activation, undefined);
426 yield* Deferred.await(sweepStarted);
427 yield* service.drain;
428 }).pipe(Effect.provide(dependencies));
429
430 expect(lookupCwds).toEqual([cwd]);
431 }),
432 );
433
434 it.effect("leaves delegated child threads to their own refresh instead of the sweep", () =>
435 Effect.gen(function* () {
436 const parent = {
437 ...threadShell("parent-thread"),
438 branch: "feature/parent",
439 worktreePath: "/workspace/parent",
440 };
441 const child: OrchestrationV2ThreadShell = {
442 ...threadShell("child-thread"),
443 branch: "t3/reply-ok",
444 worktreePath: "/workspace/child",
445 lineage: {
446 rootThreadId: parent.id,
447 parentThreadId: parent.id,
448 relationshipToParent: "subagent",
449 },
450 };
451 const project: OrchestrationProjectShell = {
452 id: parent.projectId,
453 title: "Project",
454 workspaceRoot: "/workspace/project",
455 defaultModelSelection: null,
456 scripts: [],
457 createdAt: "2026-08-01T00:00:00.000Z",
458 updatedAt: "2026-08-01T00:00:00.000Z",
459 };
460 const activation = yield* Deferred.make<void>();
461 const domainEvents = yield* Queue.unbounded<OrchestrationV2DomainEvent>();
462 const sweeps = yield* Queue.unbounded<void>();
463 const childLookup = yield* Deferred.make<void>();
464 const lookups: Array<{ readonly branch: string; readonly refresh: boolean }> = [];
465 const dependencies = Layer.mergeAll(
466 Layer.mock(Orchestrator.OrchestratorV2)({
467 streamDomainEvents: Stream.fromQueue(domainEvents),
468 getPullRequestSnapshot: ({ threadId }) =>
469 threadId !== undefined
470 ? Effect.succeed({ snapshotSequence: 1, threads: [child] })
471 : Queue.offer(sweeps, undefined).pipe(
472 Effect.as({
473 schemaVersion: 2,
474 snapshotSequence: 1,
475 threads: [parent, child],
476 archivedThreads: [],
477 }),
478 ),
479 }),
480 Layer.mock(ProjectStore.ProjectStoreV2)({
481 listShells: () => Effect.succeed([project]),
482 getShell: () => Effect.succeedSome(project),
483 }),
484 Layer.mock(GitManager.GitManager)({
485 branchPullRequest: (input, options) =>
486 Effect.gen(function* () {
487 const refresh = options?.refresh ?? false;
488 lookups.push({ branch: input.branch, refresh });
489 if (input.branch === child.branch && refresh) {
490 yield* Deferred.succeed(childLookup, undefined);
491 }
492 return null;
493 }),
494 }),
495 Layer.mock(PullRequestService.PullRequestService)({}),
496 Layer.mock(RepositoryIdentityResolver.RepositoryIdentityResolver)({
497 resolve: () =>
498 Effect.succeed({
499 canonicalKey: "github.com/acme/project",
500 locator: {
501 source: "git-remote" as const,
502 remoteName: "origin",
503 remoteUrl: "git@github.com:acme/project.git",
504 },
505 provider: SourceControlProviderKind.make("github"),
506 owner: "acme",
507 name: "project",
508 }),
509 }),
510 Layer.mock(BackgroundPolicy.BackgroundPolicy)({
511 shouldRunScopeWork: () => Effect.succeed(true),
512 shouldRunOpportunisticWork: Effect.succeed(true),
513 }),
514 Layer.succeed(ServerActivation.ServerActivation, Deferred.await(activation)),
515 Layer.succeed(
516 SecureRandom.SecureRandom,
517 SecureRandom.make((size) => new Uint8Array(size).fill(1)),
518 ),
519 FileSystem.layerNoop({}),
520 Path.layer,
521 );
522 yield* Effect.gen(function* () {
523 const service = yield* ThreadPullRequestService.make;
524 yield* service.start();
525 yield* Deferred.succeed(activation, undefined);
526
527 // Startup backfill sweep, then one periodic sweep.
528 yield* Queue.take(sweeps);
529 yield* service.drain;
530 yield* TestClock.adjust("1 minute");
531 yield* Queue.take(sweeps);
532 yield* service.drain;
533 const sweepLookups = lookups.map((lookup) => lookup.branch).toSorted();
534
535 // The child's own turn still discovers its pull request.
536 yield* Queue.offer(domainEvents, {
537 type: "checkpoint.captured",
538 threadId: child.id,
539 } as OrchestrationV2DomainEvent);
540 yield* Deferred.await(childLookup);
541 yield* service.drain;
542
543 expect(sweepLookups).toEqual([parent.branch, parent.branch]);
544 expect(lookups.slice(sweepLookups.length)).toEqual([
545 { branch: child.branch, refresh: true },
546 ]);
547 }).pipe(Effect.provide(dependencies));
548 }),
549 );
550
551 it.effect.each([
552 { state: "open", checkedOut: false },
553 { state: "merged", checkedOut: false },
554 { state: "closed", checkedOut: false },
555 { state: "merged", checkedOut: true },
556 ] as const)(
557 "associates a $state branch PR (checkedOut=$checkedOut) only with threads that predate its completion or saved it",
558 ({ state, checkedOut }) =>
559 Effect.gen(function* () {
560 const reference = {
561 projectId: ProjectId.make("project-1"),
562 repository: "acme/project",
563 number: 42,
564 url: "https://github.com/acme/project/pull/42",
565 };
566 const terminalAt = "2026-09-19T00:00:00.000Z";
567 const detected = {
568 number: 42,
569 url: reference.url,
570 title: "Previous staging work",
571 baseRef: "main",
572 headRef: "staging",
573 state,
574 updatedAt: terminalAt,
575 closedAt: state === "closed" ? terminalAt : null,
576 mergedAt: state === "merged" ? terminalAt : null,
577 repositoryKey: "github.com/acme/project",
578 };
579 const threads = [
580 { ...threadShell("older"), createdAt: DateTime.makeUnsafe("2026-09-18T00:00:00Z") },
581 { ...threadShell("same-time"), createdAt: DateTime.makeUnsafe(terminalAt) },
582 threadShell("new"),
583 { ...threadShell("saved"), branchPullRequest: reference },
584 { ...threadShell("explicit"), linkedPullRequest: reference },
585 ].map((thread) => ({ ...thread, branch: "staging" }));
586 const project: OrchestrationProjectShell = {
587 id: reference.projectId,
588 title: "Project",
589 workspaceRoot: "/workspace/project",
590 defaultModelSelection: null,
591 scripts: [],
592 createdAt: "2026-08-01T00:00:00.000Z",
593 updatedAt: "2026-08-01T00:00:00.000Z",
594 };
595 const activation = yield* Deferred.make<void>();
596 const sweepStarted = yield* Deferred.make<void>();
597 const commands: Array<OrchestrationV2ServerCommand> = [];
598 const dependencies = Layer.mergeAll(
599 Layer.mock(Orchestrator.OrchestratorV2)({
600 streamDomainEvents: Stream.never,
601 getPullRequestSnapshot: () =>
602 Deferred.succeed(sweepStarted, undefined).pipe(
603 Effect.as({ snapshotSequence: 1, threads }),
604 ),
605 dispatch: (command) =>
606 Effect.sync(() => {
607 commands.push(command);
608 return { sequence: 2, storedEvents: [] };
609 }),
610 }),
611 Layer.mock(ProjectStore.ProjectStoreV2)({
612 listShells: () => Effect.succeed([project]),
613 getShell: () => Effect.succeedSome(project),
614 }),
615 Layer.mock(GitManager.GitManager)({
616 branchPullRequest: () => Effect.succeed(detected),
617 checkedOutPullRequest: () => Effect.succeed(checkedOut),
618 }),
619 Layer.mock(PullRequestService.PullRequestService)({
620 summary: () =>
621 Effect.succeed({
622 ...detected,
623 ...reference,
624 provider: SourceControlProviderKind.make("github"),
625 baseBranch: detected.baseRef,
626 headBranch: detected.headRef,
627 }),
628 }),
629 Layer.mock(RepositoryIdentityResolver.RepositoryIdentityResolver)({
630 resolve: () =>
631 Effect.succeed({
632 canonicalKey: "github.com/acme/project",
633 locator: {
634 source: "git-remote" as const,
635 remoteName: "origin",
636 remoteUrl: "git@github.com:acme/project.git",
637 },
638 provider: SourceControlProviderKind.make("github"),
639 owner: "acme",
640 name: "project",
641 }),
642 }),
643 Layer.mock(BackgroundPolicy.BackgroundPolicy)({}),
644 Layer.succeed(ServerActivation.ServerActivation, Deferred.await(activation)),
645 Layer.succeed(
646 SecureRandom.SecureRandom,
647 SecureRandom.make((size) => new Uint8Array(size).fill(1)),
648 ),
649 FileSystem.layerNoop({}),
650 Path.layer,
651 );
652 yield* Effect.gen(function* () {
653 const service = yield* ThreadPullRequestService.make;
654 yield* service.start();
655 yield* Deferred.succeed(activation, undefined);
656 yield* Deferred.await(sweepStarted);
657 yield* service.drain;
658 }).pipe(Effect.provide(dependencies));
659
660 expect(
661 commands.map((command) => {
662 if (command.type !== "thread.pull-request.sync") throw new Error(command.type);
663 expect(command).not.toHaveProperty("linkedPullRequest");
664 return [command.threadId, command.branchPullRequest];
665 }),
666 ).toEqual(
667 (state === "open" || checkedOut
668 ? ["older", "same-time", "new", "explicit"]
669 : ["older", "same-time"]
670 ).map((id) => [id, reference]),
671 );
672 }),
673 );
674
675 it.effect.each([
676 { state: "closed", sameBranch: false },
677 { state: "merged", sameBranch: false },
678 { state: "closed", sameBranch: true },
679 { state: "merged", sameBranch: true },
680 ] as const)(
681 "restores a $state automatic PR only for its own branch (sameBranch=$sameBranch)",
682 ({ state, sameBranch }) =>
683 Effect.gen(function* () {
684 const savedReference = {
685 projectId: ProjectId.make("project-1"),
686 repository: "acme/project",
687 number: 42,
688 url: "https://github.com/acme/project/pull/42",
689 };
690 const thread = {
691 ...threadShell("branch-pr-thread"),
692 branch: "main",
693 branchPullRequest: savedReference,
694 linkedPullRequest: savedReference,
695 };
696 const project: OrchestrationProjectShell = {
697 id: thread.projectId,
698 title: "Project",
699 workspaceRoot: "/workspace/project",
700 defaultModelSelection: null,
701 scripts: [],
702 createdAt: "2026-08-01T00:00:00.000Z",
703 updatedAt: "2026-08-01T00:00:00.000Z",
704 };
705 const activation = yield* Deferred.make<void>();
706 const sweepStarted = yield* Deferred.make<void>();
707 const commands: Array<OrchestrationV2ServerCommand> = [];
708 let snapshotThread: unknown;
709 const dependencies = Layer.mergeAll(
710 Layer.mock(Orchestrator.OrchestratorV2)({
711 streamDomainEvents: Stream.never,
712 getPullRequestSnapshot: (options) =>
713 Effect.gen(function* () {
714 const store = yield* ProjectionStore.ProjectionStoreV2;
715 yield* store.apply({
716 id: EventId.make(`event:${thread.id}:created`),
717 type: "thread.created",
718 threadId: thread.id,
719 occurredAt: NOW,
720 payload: { ...thread, lastVisitedAt: thread.lastVisitedAt ?? null },
721 });
722 const sql = yield* SqlClient.SqlClient;
723 yield* sql`INSERT INTO orchestration_events (
724 sequence, event_id, aggregate_kind, stream_id, stream_version, event_type,
725 occurred_at, actor_kind, payload_json, metadata_json, application_event_version
726 ) VALUES (7, 'pr-discovery', 'thread', ${thread.id}, 1, 'thread.created',
727 '2026-09-20T00:00:00.000Z', 'system', '{}', '{}', 2)`;
728 for (const table of [
729 "orchestration_v2_projection_turn_items",
730 "orchestration_v2_projection_messages",
731 "orchestration_v2_projection_provider_threads",
732 "orchestration_v2_projection_runs",
733 ])
734 yield* sql.unsafe(`DROP TABLE ${table}`);
735 const snapshot = yield* store.getPullRequestSnapshot(options);
736 snapshotThread = snapshot.threads[0];
737 yield* Deferred.succeed(sweepStarted, undefined);
738 return snapshot;
739 }).pipe(
740 Effect.provide(
741 ProjectionStore.layer.pipe(Layer.provideMerge(SqlitePersistence.layerMemory)),
742 ),
743 Effect.orDie,
744 ),
745 dispatch: (command) =>
746 Effect.sync(() => {
747 commands.push(command);
748 return { sequence: 2, storedEvents: [] };
749 }),
750 }),
751 Layer.mock(ProjectStore.ProjectStoreV2)({
752 listShells: () => Effect.succeed([project]),
753 getShell: () => Effect.succeedSome(project),
754 }),
755 Layer.mock(GitManager.GitManager)({
756 branchPullRequest: () => Effect.succeed(null),
757 }),
758 Layer.mock(PullRequestService.PullRequestService)({
759 summary: () =>
760 Effect.succeed({
761 ...savedReference,
762 provider: SourceControlProviderKind.make("github"),
763 title: "Previous branch work",
764 state,
765 headBranch: sameBranch ? thread.branch : "feature/previous-work",
766 baseBranch: "release",
767 updatedAt: "2026-09-19T00:00:00.000Z",
768 }),
769 }),
770 Layer.mock(RepositoryIdentityResolver.RepositoryIdentityResolver)({
771 resolve: () =>
772 Effect.succeed({
773 canonicalKey: "github.com/acme/project",
774 locator: {
775 source: "git-remote" as const,
776 remoteName: "origin",
777 remoteUrl: "git@github.com:acme/project.git",
778 },
779 provider: SourceControlProviderKind.make("github"),
780 owner: "acme",
781 name: "project",
782 }),
783 }),
784 Layer.mock(BackgroundPolicy.BackgroundPolicy)({}),
785 Layer.succeed(ServerActivation.ServerActivation, Deferred.await(activation)),
786 Layer.succeed(
787 SecureRandom.SecureRandom,
788 SecureRandom.make((size) => new Uint8Array(size).fill(1)),
789 ),
790 FileSystem.layerNoop({}),
791 Path.layer,
792 );
793 yield* Effect.gen(function* () {
794 const service = yield* ThreadPullRequestService.make;
795 yield* service.start();
796 yield* Deferred.succeed(activation, undefined);
797 yield* Deferred.await(sweepStarted);
798 yield* service.drain;
799 }).pipe(Effect.provide(dependencies));
800
801 expect(snapshotThread).toHaveProperty("createdAt", NOW);
802 if (sameBranch) {
803 expect(commands).toEqual([]);
804 } else {
805 expect(commands).toHaveLength(1);
806 expect(commands[0]).toMatchObject({
807 type: "thread.pull-request.sync",
808 threadId: thread.id,
809 snapshotSequence: 7,
810 expected: {
811 workspaceRoot: project.workspaceRoot,
812 branch: thread.branch,
813 worktreePath: thread.worktreePath,
814 linkedPullRequest: savedReference,
815 branchPullRequest: savedReference,
816 },
817 branchPullRequest: null,
818 });
819 expect(commands[0]).not.toHaveProperty("linkedPullRequest");
820 }
821 }),
822 );
823
824 it.effect("a saved merged branch PR rereads no git state on each sweep", () =>
825 Effect.gen(function* () {
826 const savedReference = {
827 projectId: ProjectId.make("project-1"),
828 repository: "acme/project",
829 number: 42,
830 url: "https://github.com/acme/project/pull/42",
831 };
832 const thread = {
833 ...threadShell("merged-branch-thread"),
834 branch: "feature/merged",
835 branchPullRequest: savedReference,
836 };
837 const project: OrchestrationProjectShell = {
838 id: thread.projectId,
839 title: "Project",
840 workspaceRoot: "/workspace/project",
841 defaultModelSelection: null,
842 scripts: [],
843 createdAt: "2026-08-01T00:00:00.000Z",
844 updatedAt: "2026-08-01T00:00:00.000Z",
845 };
846 const activation = yield* Deferred.make<void>();
847 const sweeps = yield* Queue.unbounded<void>();
848 const commands: Array<OrchestrationV2ServerCommand> = [];
849 const lookups: Array<Parameters<GitManager.GitManager["Service"]["branchPullRequest"]>[1]> =
850 [];
851 const dependencies = Layer.mergeAll(
852 Layer.mock(Orchestrator.OrchestratorV2)({
853 streamDomainEvents: Stream.never,
854 getPullRequestSnapshot: () =>
855 Queue.offer(sweeps, undefined).pipe(
856 Effect.as({
857 schemaVersion: 2,
858 snapshotSequence: 1,
859 threads: [thread],
860 archivedThreads: [],
861 }),
862 ),
863 dispatch: (command) =>
864 Effect.sync(() => {
865 commands.push(command);
866 return { sequence: 2, storedEvents: [] };
867 }),
868 }),
869 Layer.mock(ProjectStore.ProjectStoreV2)({
870 listShells: () => Effect.succeed([project]),
871 getShell: () => Effect.succeedSome(project),
872 }),
873 Layer.mock(GitManager.GitManager)({
874 branchPullRequest: (_input, options) =>
875 Effect.sync(() => {
876 lookups.push(options);
877 return {
878 number: 42,
879 title: "Merged work",
880 url: savedReference.url,
881 baseRef: "main",
882 headRef: thread.branch,
883 state: "merged" as const,
884 repositoryKey: "github.com/acme/project",
885 updatedAt: "2026-09-19T00:00:00.000Z",
886 closedAt: null,
887 mergedAt: "2026-09-19T00:00:00.000Z",
888 };
889 }),
890 }),
891 Layer.mock(PullRequestService.PullRequestService)({}),
892 Layer.mock(RepositoryIdentityResolver.RepositoryIdentityResolver)({
893 resolve: () =>
894 Effect.succeed({
895 canonicalKey: "github.com/acme/project",
896 locator: {
897 source: "git-remote" as const,
898 remoteName: "origin",
899 remoteUrl: "git@github.com:acme/project.git",
900 },
901 provider: SourceControlProviderKind.make("github"),
902 owner: "acme",
903 name: "project",
904 }),
905 }),
906 // A connected client keeps the periodic sweep running.
907 Layer.mock(BackgroundPolicy.BackgroundPolicy)({
908 shouldRunOpportunisticWork: Effect.succeed(true),
909 }),
910 Layer.succeed(ServerActivation.ServerActivation, Deferred.await(activation)),
911 Layer.succeed(
912 SecureRandom.SecureRandom,
913 SecureRandom.make((size) => new Uint8Array(size).fill(1)),
914 ),
915 FileSystem.layerNoop({}),
916 Path.layer,
917 );
918 yield* Effect.gen(function* () {
919 const service = yield* ThreadPullRequestService.make;
920 yield* service.start();
921 yield* Deferred.succeed(activation, undefined);
922 yield* Queue.take(sweeps);
923 yield* service.drain;
924 yield* TestClock.adjust("1 minute");
925 yield* Queue.take(sweeps);
926 yield* service.drain;
927 }).pipe(Effect.provide(dependencies));
928
929 expect(commands).toEqual([]);
930 expect(lookups).toHaveLength(2);
931 // GitManager rereads git state on request, and by default whenever its
932 // remembered answer is merged or closed.
933 expect(
934 lookups.filter(
935 (options) =>
936 options?.refresh === true ||
937 options?.freshGitState === true ||
938 options?.confirmTerminal !== false,
939 ),
940 ).toEqual([]);
941 }),
942 );
943
944 const mergedPullRequest43 = {
945 number: 43,
946 title: "Replacement work",
947 url: "https://github.com/acme/project/pull/43",
948 baseRef: "main",
949 headRef: "feature/replaced",
950 state: "merged" as const,
951 repositoryKey: "github.com/acme/project",
952 updatedAt: "2026-09-19T00:00:00.000Z",
953 closedAt: null,
954 mergedAt: "2026-09-19T00:00:00.000Z",
955 };
956
957 it.effect.each([
958 { name: "reopened", fresh: { ...mergedPullRequest43, state: "open" as const }, writes: false },
959 { name: "the same merged PR", fresh: mergedPullRequest43, writes: true },
960 ])("saves a remembered merged PR only when fresh git state finds $name", ({ fresh, writes }) =>
961 Effect.gen(function* () {
962 const savedReference = {
963 projectId: ProjectId.make("project-1"),
964 repository: "acme/project",
965 number: 42,
966 url: "https://github.com/acme/project/pull/42",
967 };
968 const thread = {
969 ...threadShell("replaced-branch-thread"),
970 createdAt: DateTime.makeUnsafe("2026-09-01T00:00:00.000Z"),
971 branch: mergedPullRequest43.headRef,
972 branchPullRequest: savedReference,
973 };
974 const project: OrchestrationProjectShell = {
975 id: thread.projectId,
976 title: "Project",
977 workspaceRoot: "/workspace/project",
978 defaultModelSelection: null,
979 scripts: [],
980 createdAt: "2026-08-01T00:00:00.000Z",
981 updatedAt: "2026-08-01T00:00:00.000Z",
982 };
983 const activation = yield* Deferred.make<void>();
984 const sweeps = yield* Queue.unbounded<void>();
985 const commands: Array<OrchestrationV2ServerCommand> = [];
986 const dependencies = Layer.mergeAll(
987 Layer.mock(Orchestrator.OrchestratorV2)({
988 streamDomainEvents: Stream.never,
989 getPullRequestSnapshot: () =>
990 Queue.offer(sweeps, undefined).pipe(
991 Effect.as({
992 schemaVersion: 2,
993 snapshotSequence: 1,
994 threads: [thread],
995 archivedThreads: [],
996 }),
997 ),
998 dispatch: (command) =>
999 Effect.sync(() => {
1000 commands.push(command);
1001 return { sequence: 2, storedEvents: [] };
1002 }),
1003 }),
1004 Layer.mock(ProjectStore.ProjectStoreV2)({
1005 listShells: () => Effect.succeed([project]),
1006 getShell: () => Effect.succeedSome(project),
1007 }),
1008 // Only the fresh read disagrees with the remembered detection.
1009 Layer.mock(GitManager.GitManager)({
1010 branchPullRequest: (_input, options) =>
1011 Effect.succeed(options?.freshGitState === true ? fresh : mergedPullRequest43),
1012 }),
1013 Layer.mock(PullRequestService.PullRequestService)({}),
1014 Layer.mock(RepositoryIdentityResolver.RepositoryIdentityResolver)({
1015 resolve: () =>
1016 Effect.succeed({
1017 canonicalKey: "github.com/acme/project",
1018 locator: {
1019 source: "git-remote" as const,
1020 remoteName: "origin",
1021 remoteUrl: "git@github.com:acme/project.git",
1022 },
1023 provider: SourceControlProviderKind.make("github"),
1024 owner: "acme",
1025 name: "project",
1026 }),
1027 }),
1028 Layer.mock(BackgroundPolicy.BackgroundPolicy)({
1029 shouldRunOpportunisticWork: Effect.succeed(true),
1030 }),
1031 Layer.succeed(ServerActivation.ServerActivation, Deferred.await(activation)),
1032 Layer.succeed(
1033 SecureRandom.SecureRandom,
1034 SecureRandom.make((size) => new Uint8Array(size).fill(1)),
1035 ),
1036 FileSystem.layerNoop({}),
1037 Path.layer,
1038 );
1039 yield* Effect.gen(function* () {
1040 const service = yield* ThreadPullRequestService.make;
1041 yield* service.start();
1042 yield* Deferred.succeed(activation, undefined);
1043 yield* Queue.take(sweeps);
1044 yield* service.drain;
1045 }).pipe(Effect.provide(dependencies));
1046
1047 expect(
1048 commands.map((command) =>
1049 command.type === "thread.pull-request.sync" ? command.branchPullRequest : command.type,
1050 ),
1051 ).toEqual(
1052 writes
1053 ? [
1054 {
1055 projectId: project.id,
1056 repository: "acme/project",
1057 number: 43,
1058 url: mergedPullRequest43.url,
1059 },
1060 ]
1061 : [],
1062 );
1063 }),
1064 );
1065});