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.scoped(
93 Effect.gen(function* () {
94 // Polling active branches without a connected client wastes GitHub requests.
95 const thread = { ...threadShell("idle-branch-thread"), branch: "feature/idle" };
96 const project: OrchestrationProjectShell = {
97 id: thread.projectId,
98 title: "Project",
99 workspaceRoot: "/workspace/project",
100 defaultModelSelection: null,
101 scripts: [],
102 createdAt: "2026-08-01T00:00:00.000Z",
103 updatedAt: "2026-08-01T00:00:00.000Z",
104 };
105 const activation = yield* Deferred.make<void>();
106 const domainEvents = yield* Queue.unbounded<OrchestrationV2DomainEvent>();
107 const turnLookup = yield* Deferred.make<void>();
108 const lookups: Array<{ readonly refresh: boolean }> = [];
109 const dependencies = Layer.mergeAll(
110 Layer.mock(Orchestrator.OrchestratorV2)({
111 streamDomainEvents: Stream.fromQueue(domainEvents),
112 getPullRequestSnapshot: () =>
113 Effect.succeed({
114 schemaVersion: 2,
115 snapshotSequence: 1,
116 threads: [thread],
117 archivedThreads: [],
118 }),
119 }),
120 Layer.mock(ProjectStore.ProjectStoreV2)({
121 listShells: () => Effect.succeed([project]),
122 getShell: () => Effect.succeedSome(project),
123 }),
124 Layer.mock(GitManager.GitManager)({
125 branchPullRequest: (_input, options) =>
126 Effect.gen(function* () {
127 const refresh = options?.refresh ?? false;
128 lookups.push({ refresh });
129 if (refresh) yield* Deferred.succeed(turnLookup, undefined);
130 return null;
131 }),
132 }),
133 Layer.mock(PullRequestService.PullRequestService)({}),
134 Layer.mock(RepositoryIdentityResolver.RepositoryIdentityResolver)({
135 resolve: () =>
136 Effect.succeed({
137 canonicalKey: "github.com/acme/project",
138 locator: {
139 source: "git-remote" as const,
140 remoteName: "origin",
141 remoteUrl: "git@github.com:acme/project.git",
142 },
143 provider: SourceControlProviderKind.make("github"),
144 owner: "acme",
145 name: "project",
146 }),
147 }),
148 // No client holds an activity lease.
149 Layer.mock(BackgroundPolicy.BackgroundPolicy)({
150 shouldRunScopeWork: () => Effect.succeed(false),
151 shouldRunOpportunisticWork: Effect.succeed(false),
152 }),
153 Layer.succeed(ServerActivation.ServerActivation, Deferred.await(activation)),
154 Layer.succeed(
155 SecureRandom.SecureRandom,
156 SecureRandom.make((size) => new Uint8Array(size).fill(1)),
157 ),
158 FileSystem.layerNoop({}),
159 Path.layer,
160 );
161 yield* Effect.gen(function* () {
162 const service = yield* ThreadPullRequestService.make;
163 yield* service.start();
164 yield* Deferred.succeed(activation, undefined);
165
166 // A finished turn still discovers the pull request for its branch.
167 yield* Queue.offer(domainEvents, {
168 type: "checkpoint.captured",
169 threadId: thread.id,
170 } as OrchestrationV2DomainEvent);
171 yield* Deferred.await(turnLookup);
172 yield* service.drain;
173 const lookupsBeforeIdle = lookups.length;
174
175 yield* TestClock.adjust("10 minutes");
176 yield* service.drain;
177
178 expect(lookups.slice(lookupsBeforeIdle)).toEqual([]);
179 }).pipe(Effect.provide(dependencies));
180 }),
181 ),
182 );
183
184 it.effect("looks up a finished turn's branch once for all its turn-end events", () =>
185 Effect.scoped(
186 Effect.gen(function* () {
187 const thread = { ...threadShell("turn-end-thread"), branch: "t3/turn-end" };
188 const otherThread = {
189 ...threadShell("other-turn-end-thread"),
190 branch: "t3/other-turn-end",
191 };
192 const project: OrchestrationProjectShell = {
193 id: thread.projectId,
194 title: "Project",
195 workspaceRoot: "/workspace/project",
196 defaultModelSelection: null,
197 scripts: [],
198 createdAt: "2026-08-01T00:00:00.000Z",
199 updatedAt: "2026-08-01T00:00:00.000Z",
200 };
201 // A turn end commits its checkpoints and the run's end together. The queued
202 // non-refresh request ahead of them must still become a refresh when they join it.
203 const turnEnd = [
204 { type: "thread.metadata-updated", threadId: thread.id },
205 { type: "checkpoint.captured", threadId: thread.id },
206 { type: "checkpoint.captured", threadId: thread.id },
207 { type: "run.updated", threadId: thread.id, payload: { status: "completed" } },
208 { type: "checkpoint.captured", threadId: otherThread.id },
209 { type: "run.updated", threadId: otherThread.id, payload: { status: "completed" } },
210 ] as Array<OrchestrationV2DomainEvent>;
211 // Arrives during the thread's own lookup, which may already have read the old host state.
212 const duringLookup = [
213 { type: "checkpoint.captured", threadId: thread.id },
214 { type: "thread.metadata-updated", threadId: thread.id },
215 ] as Array<OrchestrationV2DomainEvent>;
216 const activation = yield* Deferred.make<void>();
217 const sweepStarted = yield* Deferred.make<void>();
218 const turnEndQueued = yield* Deferred.make<void>();
219 const lookupStarted = yield* Deferred.make<void>();
220 const followUpQueued = yield* Deferred.make<void>();
221 const lookups: Array<{ readonly branch: string; readonly refresh: boolean }> = [];
222 const dependencies = Layer.mergeAll(
223 Layer.mock(Orchestrator.OrchestratorV2)({
224 // The batch arrives while the worker is busy with the startup sweep.
225 streamDomainEvents: Stream.fromEffect(Deferred.await(sweepStarted)).pipe(
226 Stream.drain,
227 Stream.concat(Stream.fromIterable(turnEnd)),
228 Stream.concat(
229 Stream.fromEffect(Deferred.succeed(turnEndQueued, undefined)).pipe(Stream.drain),
230 ),
231 Stream.concat(Stream.fromEffect(Deferred.await(lookupStarted)).pipe(Stream.drain)),
232 Stream.concat(Stream.fromIterable(duringLookup)),
233 Stream.concat(
234 Stream.fromEffect(Deferred.succeed(followUpQueued, undefined)).pipe(Stream.drain),
235 ),
236 Stream.concat(Stream.never),
237 ),
238 getPullRequestSnapshot: ({ threadId }) =>
239 Effect.succeed({
240 schemaVersion: 2,
241 snapshotSequence: 1,
242 threads: [thread, otherThread].filter(
243 (candidate) => threadId === undefined || candidate.id === threadId,
244 ),
245 archivedThreads: [],
246 }),
247 }),
248 Layer.mock(ProjectStore.ProjectStoreV2)({
249 listShells: () => Effect.succeed([project]),
250 getShell: () => Effect.succeedSome(project),
251 }),
252 Layer.mock(GitManager.GitManager)({
253 branchPullRequest: (input, options) =>
254 Effect.gen(function* () {
255 lookups.push({ branch: input.branch, refresh: options?.refresh ?? false });
256 if (!(yield* Deferred.isDone(turnEndQueued))) {
257 yield* Deferred.succeed(sweepStarted, undefined);
258 yield* Deferred.await(turnEndQueued);
259 } else if (
260 input.branch === thread.branch &&
261 !(yield* Deferred.isDone(followUpQueued))
262 ) {
263 yield* Deferred.succeed(lookupStarted, undefined);
264 yield* Deferred.await(followUpQueued);
265 }
266 return null;
267 }),
268 }),
269 Layer.mock(PullRequestService.PullRequestService)({}),
270 Layer.mock(RepositoryIdentityResolver.RepositoryIdentityResolver)({
271 resolve: () =>
272 Effect.succeed({
273 canonicalKey: "github.com/acme/project",
274 locator: {
275 source: "git-remote" as const,
276 remoteName: "origin",
277 remoteUrl: "git@github.com:acme/project.git",
278 },
279 provider: SourceControlProviderKind.make("github"),
280 owner: "acme",
281 name: "project",
282 }),
283 }),
284 Layer.mock(BackgroundPolicy.BackgroundPolicy)({
285 shouldRunScopeWork: () => Effect.succeed(false),
286 shouldRunOpportunisticWork: Effect.succeed(false),
287 }),
288 Layer.succeed(ServerActivation.ServerActivation, Deferred.await(activation)),
289 Layer.succeed(
290 SecureRandom.SecureRandom,
291 SecureRandom.make((size) => new Uint8Array(size).fill(1)),
292 ),
293 FileSystem.layerNoop({}),
294 Path.layer,
295 );
296 yield* Effect.gen(function* () {
297 const service = yield* ThreadPullRequestService.make;
298 yield* service.start();
299 yield* Deferred.succeed(activation, undefined);
300 yield* Deferred.await(followUpQueued);
301 yield* service.drain;
302 }).pipe(Effect.provide(dependencies));
303
304 // The startup sweep, one refresh lookup per thread for its turn-end batch, then one
305 // follow-up refresh for the events that arrived during the thread's lookup.
306 expect(lookups).toEqual([
307 { branch: thread.branch, refresh: false },
308 { branch: otherThread.branch, refresh: false },
309 { branch: thread.branch, refresh: true },
310 { branch: otherThread.branch, refresh: true },
311 { branch: thread.branch, refresh: true },
312 ]);
313 }),
314 ),
315 );
316
317 it.effect.each([
318 {
319 worktree: "a linked worktree",
320 existingPaths: ["/workspace/worktree", "/workspace/worktree/.git"],
321 unreadablePaths: [],
322 cwd: "/workspace/worktree",
323 },
324 // A folder that outlived its .git makes git fail with not_a_repository.
325 {
326 worktree: "a folder without .git",
327 existingPaths: ["/workspace/worktree"],
328 unreadablePaths: [],
329 cwd: "/workspace/project",
330 },
331 {
332 worktree: "a folder whose .git cannot be checked",
333 existingPaths: ["/workspace/worktree"],
334 unreadablePaths: ["/workspace/worktree/.git"],
335 cwd: "/workspace/worktree",
336 },
337 {
338 worktree: "a missing folder",
339 existingPaths: [],
340 unreadablePaths: [],
341 cwd: "/workspace/project",
342 },
343 ] as const)(
344 "looks up a branch whose worktree is $worktree in $cwd",
345 ({ existingPaths, unreadablePaths, cwd }) =>
346 Effect.scoped(
347 Effect.gen(function* () {
348 const thread = {
349 ...threadShell("worktree-thread"),
350 settledOverride: "settled" as const,
351 settledAt: NOW,
352 branch: "feature/worktree",
353 worktreePath: "/workspace/worktree",
354 };
355 const project: OrchestrationProjectShell = {
356 id: thread.projectId,
357 title: "Project",
358 workspaceRoot: "/workspace/project",
359 defaultModelSelection: null,
360 scripts: [],
361 createdAt: "2026-08-01T00:00:00.000Z",
362 updatedAt: "2026-08-01T00:00:00.000Z",
363 };
364 const activation = yield* Deferred.make<void>();
365 const sweepStarted = yield* Deferred.make<void>();
366 const lookupCwds: Array<string> = [];
367 const dependencies = Layer.mergeAll(
368 Layer.mock(Orchestrator.OrchestratorV2)({
369 streamDomainEvents: Stream.never,
370 getPullRequestSnapshot: () =>
371 Deferred.succeed(sweepStarted, undefined).pipe(
372 Effect.as({
373 schemaVersion: 2,
374 snapshotSequence: 1,
375 threads: [thread],
376 archivedThreads: [],
377 }),
378 ),
379 }),
380 Layer.mock(ProjectStore.ProjectStoreV2)({
381 listShells: () => Effect.succeed([project]),
382 getShell: () => Effect.succeedSome(project),
383 }),
384 Layer.mock(GitManager.GitManager)({
385 branchPullRequest: (input) =>
386 Effect.sync(() => {
387 lookupCwds.push(input.cwd);
388 return null;
389 }),
390 }),
391 Layer.mock(PullRequestService.PullRequestService)({}),
392 Layer.mock(RepositoryIdentityResolver.RepositoryIdentityResolver)({
393 resolve: () =>
394 Effect.succeed({
395 canonicalKey: "github.com/acme/project",
396 locator: {
397 source: "git-remote" as const,
398 remoteName: "origin",
399 remoteUrl: "git@github.com:acme/project.git",
400 },
401 provider: SourceControlProviderKind.make("github"),
402 owner: "acme",
403 name: "project",
404 }),
405 }),
406 Layer.mock(BackgroundPolicy.BackgroundPolicy)({}),
407 Layer.succeed(ServerActivation.ServerActivation, Deferred.await(activation)),
408 Layer.succeed(
409 SecureRandom.SecureRandom,
410 SecureRandom.make((size) => new Uint8Array(size).fill(1)),
411 ),
412 FileSystem.layerNoop({
413 exists: (path) =>
414 unreadablePaths.some((entry) => entry === path)
415 ? Effect.fail(
416 PlatformError.systemError({
417 _tag: "PermissionDenied",
418 module: "FileSystem",
419 method: "exists",
420 pathOrDescriptor: path,
421 }),
422 )
423 : Effect.succeed(existingPaths.some((entry) => entry === path)),
424 }),
425 Path.layer,
426 );
427 yield* Effect.gen(function* () {
428 const service = yield* ThreadPullRequestService.make;
429 yield* service.start();
430 yield* Deferred.succeed(activation, undefined);
431 yield* Deferred.await(sweepStarted);
432 yield* service.drain;
433 }).pipe(Effect.provide(dependencies));
434
435 expect(lookupCwds).toEqual([cwd]);
436 }),
437 ),
438 );
439
440 it.effect("leaves delegated child threads to their own refresh instead of the sweep", () =>
441 Effect.scoped(
442 Effect.gen(function* () {
443 const parent = {
444 ...threadShell("parent-thread"),
445 branch: "feature/parent",
446 worktreePath: "/workspace/parent",
447 };
448 const child: OrchestrationV2ThreadShell = {
449 ...threadShell("child-thread"),
450 branch: "t3/reply-ok",
451 worktreePath: "/workspace/child",
452 lineage: {
453 rootThreadId: parent.id,
454 parentThreadId: parent.id,
455 relationshipToParent: "subagent",
456 },
457 };
458 const project: OrchestrationProjectShell = {
459 id: parent.projectId,
460 title: "Project",
461 workspaceRoot: "/workspace/project",
462 defaultModelSelection: null,
463 scripts: [],
464 createdAt: "2026-08-01T00:00:00.000Z",
465 updatedAt: "2026-08-01T00:00:00.000Z",
466 };
467 const activation = yield* Deferred.make<void>();
468 const domainEvents = yield* Queue.unbounded<OrchestrationV2DomainEvent>();
469 const sweeps = yield* Queue.unbounded<void>();
470 const childLookup = yield* Deferred.make<void>();
471 const lookups: Array<{ readonly branch: string; readonly refresh: boolean }> = [];
472 const dependencies = Layer.mergeAll(
473 Layer.mock(Orchestrator.OrchestratorV2)({
474 streamDomainEvents: Stream.fromQueue(domainEvents),
475 getPullRequestSnapshot: ({ threadId }) =>
476 threadId !== undefined
477 ? Effect.succeed({ snapshotSequence: 1, threads: [child] })
478 : Queue.offer(sweeps, undefined).pipe(
479 Effect.as({
480 schemaVersion: 2,
481 snapshotSequence: 1,
482 threads: [parent, child],
483 archivedThreads: [],
484 }),
485 ),
486 }),
487 Layer.mock(ProjectStore.ProjectStoreV2)({
488 listShells: () => Effect.succeed([project]),
489 getShell: () => Effect.succeedSome(project),
490 }),
491 Layer.mock(GitManager.GitManager)({
492 branchPullRequest: (input, options) =>
493 Effect.gen(function* () {
494 const refresh = options?.refresh ?? false;
495 lookups.push({ branch: input.branch, refresh });
496 if (input.branch === child.branch && refresh) {
497 yield* Deferred.succeed(childLookup, undefined);
498 }
499 return null;
500 }),
501 }),
502 Layer.mock(PullRequestService.PullRequestService)({}),
503 Layer.mock(RepositoryIdentityResolver.RepositoryIdentityResolver)({
504 resolve: () =>
505 Effect.succeed({
506 canonicalKey: "github.com/acme/project",
507 locator: {
508 source: "git-remote" as const,
509 remoteName: "origin",
510 remoteUrl: "git@github.com:acme/project.git",
511 },
512 provider: SourceControlProviderKind.make("github"),
513 owner: "acme",
514 name: "project",
515 }),
516 }),
517 Layer.mock(BackgroundPolicy.BackgroundPolicy)({
518 shouldRunScopeWork: () => Effect.succeed(true),
519 shouldRunOpportunisticWork: Effect.succeed(true),
520 }),
521 Layer.succeed(ServerActivation.ServerActivation, Deferred.await(activation)),
522 Layer.succeed(
523 SecureRandom.SecureRandom,
524 SecureRandom.make((size) => new Uint8Array(size).fill(1)),
525 ),
526 FileSystem.layerNoop({}),
527 Path.layer,
528 );
529 yield* Effect.gen(function* () {
530 const service = yield* ThreadPullRequestService.make;
531 yield* service.start();
532 yield* Deferred.succeed(activation, undefined);
533
534 // Startup backfill sweep, then one periodic sweep.
535 yield* Queue.take(sweeps);
536 yield* service.drain;
537 yield* TestClock.adjust("1 minute");
538 yield* Queue.take(sweeps);
539 yield* service.drain;
540 const sweepLookups = lookups.map((lookup) => lookup.branch).toSorted();
541
542 // The child's own turn still discovers its pull request.
543 yield* Queue.offer(domainEvents, {
544 type: "checkpoint.captured",
545 threadId: child.id,
546 } as OrchestrationV2DomainEvent);
547 yield* Deferred.await(childLookup);
548 yield* service.drain;
549
550 expect(sweepLookups).toEqual([parent.branch, parent.branch]);
551 expect(lookups.slice(sweepLookups.length)).toEqual([
552 { branch: child.branch, refresh: true },
553 ]);
554 }).pipe(Effect.provide(dependencies));
555 }),
556 ),
557 );
558
559 it.effect.each([
560 { state: "open", checkedOut: false },
561 { state: "merged", checkedOut: false },
562 { state: "closed", checkedOut: false },
563 { state: "merged", checkedOut: true },
564 ] as const)(
565 "associates a $state branch PR (checkedOut=$checkedOut) only with threads that predate its completion or saved it",
566 ({ state, checkedOut }) =>
567 Effect.scoped(
568 Effect.gen(function* () {
569 const reference = {
570 projectId: ProjectId.make("project-1"),
571 repository: "acme/project",
572 number: 42,
573 url: "https://github.com/acme/project/pull/42",
574 };
575 const terminalAt = "2026-09-19T00:00:00.000Z";
576 const detected = {
577 number: 42,
578 url: reference.url,
579 title: "Previous staging work",
580 baseRef: "main",
581 headRef: "staging",
582 state,
583 updatedAt: terminalAt,
584 closedAt: state === "closed" ? terminalAt : null,
585 mergedAt: state === "merged" ? terminalAt : null,
586 repositoryKey: "github.com/acme/project",
587 };
588 const threads = [
589 { ...threadShell("older"), createdAt: DateTime.makeUnsafe("2026-09-18T00:00:00Z") },
590 { ...threadShell("same-time"), createdAt: DateTime.makeUnsafe(terminalAt) },
591 threadShell("new"),
592 { ...threadShell("saved"), branchPullRequest: reference },
593 { ...threadShell("explicit"), linkedPullRequest: reference },
594 ].map((thread) => ({ ...thread, branch: "staging" }));
595 const project: OrchestrationProjectShell = {
596 id: reference.projectId,
597 title: "Project",
598 workspaceRoot: "/workspace/project",
599 defaultModelSelection: null,
600 scripts: [],
601 createdAt: "2026-08-01T00:00:00.000Z",
602 updatedAt: "2026-08-01T00:00:00.000Z",
603 };
604 const activation = yield* Deferred.make<void>();
605 const sweepStarted = yield* Deferred.make<void>();
606 const commands: Array<OrchestrationV2ServerCommand> = [];
607 const dependencies = Layer.mergeAll(
608 Layer.mock(Orchestrator.OrchestratorV2)({
609 streamDomainEvents: Stream.never,
610 getPullRequestSnapshot: () =>
611 Deferred.succeed(sweepStarted, undefined).pipe(
612 Effect.as({ snapshotSequence: 1, threads }),
613 ),
614 dispatch: (command) =>
615 Effect.sync(() => {
616 commands.push(command);
617 return { sequence: 2, storedEvents: [] };
618 }),
619 }),
620 Layer.mock(ProjectStore.ProjectStoreV2)({
621 listShells: () => Effect.succeed([project]),
622 getShell: () => Effect.succeedSome(project),
623 }),
624 Layer.mock(GitManager.GitManager)({
625 branchPullRequest: () => Effect.succeed(detected),
626 checkedOutPullRequest: () => Effect.succeed(checkedOut),
627 }),
628 Layer.mock(PullRequestService.PullRequestService)({
629 summary: () =>
630 Effect.succeed({
631 ...detected,
632 ...reference,
633 provider: SourceControlProviderKind.make("github"),
634 baseBranch: detected.baseRef,
635 headBranch: detected.headRef,
636 }),
637 }),
638 Layer.mock(RepositoryIdentityResolver.RepositoryIdentityResolver)({
639 resolve: () =>
640 Effect.succeed({
641 canonicalKey: "github.com/acme/project",
642 locator: {
643 source: "git-remote" as const,
644 remoteName: "origin",
645 remoteUrl: "git@github.com:acme/project.git",
646 },
647 provider: SourceControlProviderKind.make("github"),
648 owner: "acme",
649 name: "project",
650 }),
651 }),
652 Layer.mock(BackgroundPolicy.BackgroundPolicy)({}),
653 Layer.succeed(ServerActivation.ServerActivation, Deferred.await(activation)),
654 Layer.succeed(
655 SecureRandom.SecureRandom,
656 SecureRandom.make((size) => new Uint8Array(size).fill(1)),
657 ),
658 FileSystem.layerNoop({}),
659 Path.layer,
660 );
661 yield* Effect.gen(function* () {
662 const service = yield* ThreadPullRequestService.make;
663 yield* service.start();
664 yield* Deferred.succeed(activation, undefined);
665 yield* Deferred.await(sweepStarted);
666 yield* service.drain;
667 }).pipe(Effect.provide(dependencies));
668
669 expect(
670 commands.map((command) => {
671 if (command.type !== "thread.pull-request.sync") throw new Error(command.type);
672 expect(command).not.toHaveProperty("linkedPullRequest");
673 return [command.threadId, command.branchPullRequest];
674 }),
675 ).toEqual(
676 (state === "open" || checkedOut
677 ? ["older", "same-time", "new", "explicit"]
678 : ["older", "same-time"]
679 ).map((id) => [id, reference]),
680 );
681 }),
682 ),
683 );
684
685 it.effect.each([
686 { state: "closed", sameBranch: false },
687 { state: "merged", sameBranch: false },
688 { state: "closed", sameBranch: true },
689 { state: "merged", sameBranch: true },
690 ] as const)(
691 "restores a $state automatic PR only for its own branch (sameBranch=$sameBranch)",
692 ({ state, sameBranch }) =>
693 Effect.scoped(
694 Effect.gen(function* () {
695 const savedReference = {
696 projectId: ProjectId.make("project-1"),
697 repository: "acme/project",
698 number: 42,
699 url: "https://github.com/acme/project/pull/42",
700 };
701 const thread = {
702 ...threadShell("branch-pr-thread"),
703 branch: "main",
704 branchPullRequest: savedReference,
705 linkedPullRequest: savedReference,
706 };
707 const project: OrchestrationProjectShell = {
708 id: thread.projectId,
709 title: "Project",
710 workspaceRoot: "/workspace/project",
711 defaultModelSelection: null,
712 scripts: [],
713 createdAt: "2026-08-01T00:00:00.000Z",
714 updatedAt: "2026-08-01T00:00:00.000Z",
715 };
716 const activation = yield* Deferred.make<void>();
717 const sweepStarted = yield* Deferred.make<void>();
718 const commands: Array<OrchestrationV2ServerCommand> = [];
719 let snapshotThread: unknown;
720 const dependencies = Layer.mergeAll(
721 Layer.mock(Orchestrator.OrchestratorV2)({
722 streamDomainEvents: Stream.never,
723 getPullRequestSnapshot: (options) =>
724 Effect.gen(function* () {
725 const store = yield* ProjectionStore.ProjectionStoreV2;
726 yield* store.apply({
727 id: EventId.make(`event:${thread.id}:created`),
728 type: "thread.created",
729 threadId: thread.id,
730 occurredAt: NOW,
731 payload: { ...thread, lastVisitedAt: thread.lastVisitedAt ?? null },
732 });
733 const sql = yield* SqlClient.SqlClient;
734 yield* sql`INSERT INTO orchestration_events (
735 sequence, event_id, aggregate_kind, stream_id, stream_version, event_type,
736 occurred_at, actor_kind, payload_json, metadata_json, application_event_version
737 ) VALUES (7, 'pr-discovery', 'thread', ${thread.id}, 1, 'thread.created',
738 '2026-09-20T00:00:00.000Z', 'system', '{}', '{}', 2)`;
739 for (const table of [
740 "orchestration_v2_projection_turn_items",
741 "orchestration_v2_projection_messages",
742 "orchestration_v2_projection_provider_threads",
743 "orchestration_v2_projection_runs",
744 ])
745 yield* sql.unsafe(`DROP TABLE ${table}`);
746 const snapshot = yield* store.getPullRequestSnapshot(options);
747 snapshotThread = snapshot.threads[0];
748 yield* Deferred.succeed(sweepStarted, undefined);
749 return snapshot;
750 }).pipe(
751 Effect.provide(
752 ProjectionStore.layer.pipe(Layer.provideMerge(SqlitePersistence.layerMemory)),
753 ),
754 Effect.orDie,
755 ),
756 dispatch: (command) =>
757 Effect.sync(() => {
758 commands.push(command);
759 return { sequence: 2, storedEvents: [] };
760 }),
761 }),
762 Layer.mock(ProjectStore.ProjectStoreV2)({
763 listShells: () => Effect.succeed([project]),
764 getShell: () => Effect.succeedSome(project),
765 }),
766 Layer.mock(GitManager.GitManager)({
767 branchPullRequest: () => Effect.succeed(null),
768 }),
769 Layer.mock(PullRequestService.PullRequestService)({
770 summary: () =>
771 Effect.succeed({
772 ...savedReference,
773 provider: SourceControlProviderKind.make("github"),
774 title: "Previous branch work",
775 state,
776 headBranch: sameBranch ? thread.branch : "feature/previous-work",
777 baseBranch: "release",
778 updatedAt: "2026-09-19T00:00:00.000Z",
779 }),
780 }),
781 Layer.mock(RepositoryIdentityResolver.RepositoryIdentityResolver)({
782 resolve: () =>
783 Effect.succeed({
784 canonicalKey: "github.com/acme/project",
785 locator: {
786 source: "git-remote" as const,
787 remoteName: "origin",
788 remoteUrl: "git@github.com:acme/project.git",
789 },
790 provider: SourceControlProviderKind.make("github"),
791 owner: "acme",
792 name: "project",
793 }),
794 }),
795 Layer.mock(BackgroundPolicy.BackgroundPolicy)({}),
796 Layer.succeed(ServerActivation.ServerActivation, Deferred.await(activation)),
797 Layer.succeed(
798 SecureRandom.SecureRandom,
799 SecureRandom.make((size) => new Uint8Array(size).fill(1)),
800 ),
801 FileSystem.layerNoop({}),
802 Path.layer,
803 );
804 yield* Effect.gen(function* () {
805 const service = yield* ThreadPullRequestService.make;
806 yield* service.start();
807 yield* Deferred.succeed(activation, undefined);
808 yield* Deferred.await(sweepStarted);
809 yield* service.drain;
810 }).pipe(Effect.provide(dependencies));
811
812 expect(snapshotThread).toHaveProperty("createdAt", NOW);
813 if (sameBranch) {
814 expect(commands).toEqual([]);
815 } else {
816 expect(commands).toHaveLength(1);
817 expect(commands[0]).toMatchObject({
818 type: "thread.pull-request.sync",
819 threadId: thread.id,
820 snapshotSequence: 7,
821 expected: {
822 workspaceRoot: project.workspaceRoot,
823 branch: thread.branch,
824 worktreePath: thread.worktreePath,
825 linkedPullRequest: savedReference,
826 branchPullRequest: savedReference,
827 },
828 branchPullRequest: null,
829 });
830 expect(commands[0]).not.toHaveProperty("linkedPullRequest");
831 }
832 }),
833 ),
834 );
835});