1import {
2 EventId,
3 ProjectId,
4 ProviderInstanceId,
5 ThreadId,
6 type OrchestrationProjectShell,
7 type OrchestrationV2DomainEvent,
8 type OrchestrationV2ThreadShell,
9} from "@t3tools/contracts";
10import { describe, expect, it } from "@effect/vitest";
11import * as Crypto from "effect/Crypto";
12import * as DateTime from "effect/DateTime";
13import * as Deferred from "effect/Deferred";
14import * as Effect from "effect/Effect";
15import * as FileSystem from "effect/FileSystem";
16import * as Layer from "effect/Layer";
17import * as Option from "effect/Option";
18import * as PubSub from "effect/PubSub";
19import * as Queue from "effect/Queue";
20import * as Stream from "effect/Stream";
21import * as TestClock from "effect/testing/TestClock";
22
23import * as GitManager from "../git/GitManager.ts";
24import * as ProjectStore from "./ProjectStore.ts";
25import * as RepositoryIdentityResolver from "../project/RepositoryIdentityResolver.ts";
26import * as PullRequestService from "../pullRequest/PullRequestService.ts";
27import * as ServerActivation from "../serverActivation.ts";
28import * as Orchestrator from "./Orchestrator.ts";
29import * as ThreadPullRequestService from "./ThreadPullRequestService.ts";
30
31describe("ThreadPullRequestServiceV2 project guard", () => {
32 it.effect("discovers a repository from a project shell without enrichment", () =>
33 Effect.gen(function* () {
34 const project: OrchestrationProjectShell = {
35 id: ProjectId.make("project-1"),
36 title: "Project",
37 workspaceRoot: "/workspace/project",
38 defaultModelSelection: null,
39 scripts: [],
40 repositoryIdentity: null,
41 createdAt: "2026-08-01T00:00:00.000Z",
42 updatedAt: "2026-08-01T00:00:00.000Z",
43 };
44 let resolvedRoot: string | null = null;
45 const result = yield* ThreadPullRequestService.resolveProjectForPullRequestDiscovery(
46 project,
47 {
48 resolve: (root) => {
49 resolvedRoot = root;
50 return Effect.succeed({
51 canonicalKey: "github.com/pingdotgg/t3code",
52 locator: {
53 source: "git-remote" as const,
54 remoteName: "origin",
55 remoteUrl: "git@github.com:pingdotgg/t3code.git",
56 },
57 provider: "github" as const,
58 displayName: "pingdotgg/t3code",
59 owner: "pingdotgg",
60 name: "t3code",
61 });
62 },
63 },
64 );
65 expect(resolvedRoot).toBe("/workspace/project");
66 expect(result.repository).toBe("pingdotgg/t3code");
67 expect(result.project.repositoryIdentity?.canonicalKey).toBe("github.com/pingdotgg/t3code");
68 }),
69 );
70
71 it.effect("refreshes the cached repository identity only when asked", () =>
72 Effect.gen(function* () {
73 const project: OrchestrationProjectShell = {
74 id: ProjectId.make("project-1"),
75 title: "Project",
76 workspaceRoot: "/workspace/project",
77 defaultModelSelection: null,
78 scripts: [],
79 repositoryIdentity: null,
80 createdAt: "2026-08-01T00:00:00.000Z",
81 updatedAt: "2026-08-01T00:00:00.000Z",
82 };
83 const refreshes: Array<boolean | undefined> = [];
84 const resolver = {
85 resolve: (_root: string, options?: { readonly refresh?: boolean }) => {
86 refreshes.push(options?.refresh);
87 return Effect.succeed(null);
88 },
89 };
90 yield* ThreadPullRequestService.resolveProjectForPullRequestDiscovery(project, resolver);
91 yield* ThreadPullRequestService.resolveProjectForPullRequestDiscovery(project, resolver, {
92 refresh: true,
93 });
94 expect(refreshes).toEqual([false, true]);
95 }),
96 );
97
98 it("rejects a pull-request result when the project root changes before dispatch", () => {
99 const currentProject = Option.some({
100 workspaceRoot: "/workspace/replaced",
101 } satisfies Pick<OrchestrationProjectShell, "workspaceRoot">);
102
103 expect(
104 ThreadPullRequestService.projectWorkspaceMatchesSnapshot(
105 currentProject,
106 "/workspace/original",
107 ),
108 ).toBe(false);
109 });
110
111 it("rejects a pull-request result when the project was deleted before dispatch", () => {
112 expect(
113 ThreadPullRequestService.projectWorkspaceMatchesSnapshot(
114 Option.none<Pick<OrchestrationProjectShell, "workspaceRoot">>(),
115 "/workspace/original",
116 ),
117 ).toBe(false);
118 });
119
120 it("accepts a pull-request result while the project root is unchanged", () => {
121 const currentProject = Option.some({
122 workspaceRoot: "/workspace/original",
123 } satisfies Pick<OrchestrationProjectShell, "workspaceRoot">);
124
125 expect(
126 ThreadPullRequestService.projectWorkspaceMatchesSnapshot(
127 currentProject,
128 "/workspace/original",
129 ),
130 ).toBe(true);
131 });
132});
133
134describe("ThreadPullRequestServiceV2 reads", () => {
135 const NOW = DateTime.makeUnsafe("2026-09-20T00:00:00.000Z");
136 const threadShell = (id: string): OrchestrationV2ThreadShell => {
137 const threadId = ThreadId.make(id);
138 return {
139 id: threadId,
140 projectId: ProjectId.make("project-1"),
141 title: id,
142 providerInstanceId: ProviderInstanceId.make("codex"),
143 modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" },
144 runtimeMode: "full-access",
145 interactionMode: "default",
146 branch: null,
147 worktreePath: null,
148 activeProviderThreadId: null,
149 lineage: { rootThreadId: threadId, parentThreadId: null, relationshipToParent: null },
150 forkedFrom: null,
151 createdBy: "user",
152 creationSource: "web",
153 activeRunId: null,
154 latestRunId: null,
155 status: "idle",
156 pendingRuntimeRequest: null,
157 latestVisibleMessage: null,
158 latestUserMessageAt: null,
159 hasActionableProposedPlan: false,
160 itemCount: 0,
161 visibleItemCount: 0,
162 createdAt: NOW,
163 updatedAt: NOW,
164 archivedAt: null,
165 settledOverride: null,
166 settledAt: null,
167 lastVisitedAt: null,
168 deletedAt: null,
169 };
170 };
171
172 it.effect("an event for one thread reads that thread's shell, not every thread's", () =>
173 Effect.scoped(
174 Effect.gen(function* () {
175 const thread = threadShell("updated-thread");
176 const other = threadShell("other-thread");
177 const activation = yield* Deferred.make<void>();
178 const events = yield* PubSub.unbounded<OrchestrationV2DomainEvent>();
179 // Each read: the thread id for a one-thread read, or the full read's options.
180 const reads = yield* Queue.unbounded<
181 ThreadId | { readonly location?: string; readonly unsettledOnly?: boolean }
182 >();
183 const layerDependencies = Layer.mergeAll(
184 Layer.mock(Orchestrator.OrchestratorV2)({
185 streamDomainEvents: Stream.fromPubSub(events),
186 getShellSnapshot: (options) =>
187 Queue.offer(reads, options ?? {}).pipe(
188 Effect.as({
189 schemaVersion: 2,
190 snapshotSequence: 1,
191 threads: [thread, other],
192 archivedThreads: [],
193 }),
194 ),
195 getThreadEventSequence: () => Effect.succeed(1),
196 getThreadShell: (threadId) =>
197 Queue.offer(reads, threadId).pipe(
198 Effect.as([thread, other].find((candidate) => candidate.id === threadId) ?? null),
199 ),
200 }),
201 Layer.mock(ProjectStore.ProjectStoreV2)({
202 listShells: () => Effect.succeed([]),
203 }),
204 Layer.mock(GitManager.GitManager)({}),
205 Layer.mock(PullRequestService.PullRequestService)({}),
206 Layer.mock(RepositoryIdentityResolver.RepositoryIdentityResolver)({}),
207 Layer.succeed(ServerActivation.ServerActivation, Deferred.await(activation)),
208 Layer.succeed(
209 Crypto.Crypto,
210 Crypto.make({
211 randomBytes: (size) => new Uint8Array(size).fill(1),
212 digest: (_algorithm, data) => Effect.succeed(data),
213 }),
214 ),
215 FileSystem.layerNoop({}),
216 );
217
218 yield* Effect.gen(function* () {
219 const service = yield* ThreadPullRequestService.make;
220 yield* service.start();
221 yield* Deferred.succeed(activation, undefined);
222 // Startup backfill reads every active thread.
223 expect(yield* Queue.take(reads)).toEqual({ location: "active", unsettledOnly: false });
224 yield* service.drain;
225 yield* PubSub.publish(events, {
226 type: "thread.metadata-updated",
227 id: EventId.make("event:metadata"),
228 threadId: thread.id,
229 occurredAt: NOW,
230 payload: {
231 createdBy: thread.createdBy,
232 creationSource: thread.creationSource,
233 id: thread.id,
234 projectId: thread.projectId,
235 title: thread.title,
236 providerInstanceId: thread.providerInstanceId,
237 modelSelection: thread.modelSelection,
238 runtimeMode: thread.runtimeMode,
239 interactionMode: thread.interactionMode,
240 branch: thread.branch,
241 worktreePath: thread.worktreePath,
242 activeProviderThreadId: thread.activeProviderThreadId,
243 lineage: thread.lineage,
244 forkedFrom: null,
245 createdAt: thread.createdAt,
246 updatedAt: thread.updatedAt,
247 archivedAt: null,
248 settledOverride: null,
249 settledAt: null,
250 lastVisitedAt: null,
251 deletedAt: null,
252 },
253 });
254 expect(yield* Queue.take(reads)).toBe(thread.id);
255 yield* service.drain;
256 expect(yield* Queue.size(reads)).toBe(0);
257 }).pipe(Effect.provide(layerDependencies));
258 }),
259 ),
260 );
261
262 it.effect("periodic sweeps read only unsettled threads once backfill is done", () =>
263 Effect.scoped(
264 Effect.gen(function* () {
265 const settled = {
266 ...threadShell("settled-thread"),
267 branch: "feature/settled",
268 settledOverride: "settled" as const,
269 settledAt: NOW,
270 };
271 const activation = yield* Deferred.make<void>();
272 const reads = yield* Queue.unbounded<{
273 readonly location?: string;
274 readonly unsettledOnly?: boolean;
275 }>();
276 const layerDependencies = Layer.mergeAll(
277 Layer.mock(Orchestrator.OrchestratorV2)({
278 streamDomainEvents: Stream.never,
279 getShellSnapshot: (options) =>
280 Queue.offer(reads, options ?? {}).pipe(
281 Effect.as({
282 schemaVersion: 2,
283 snapshotSequence: 1,
284 // The fake honors unsettledOnly like the store does.
285 threads: options?.unsettledOnly ? [] : [settled],
286 archivedThreads: [],
287 }),
288 ),
289 }),
290 Layer.mock(ProjectStore.ProjectStoreV2)({
291 listShells: () => Effect.succeed([]),
292 }),
293 Layer.mock(GitManager.GitManager)({}),
294 Layer.mock(PullRequestService.PullRequestService)({}),
295 Layer.mock(RepositoryIdentityResolver.RepositoryIdentityResolver)({}),
296 Layer.succeed(ServerActivation.ServerActivation, Deferred.await(activation)),
297 Layer.succeed(
298 Crypto.Crypto,
299 Crypto.make({
300 randomBytes: (size) => new Uint8Array(size).fill(1),
301 digest: (_algorithm, data) => Effect.succeed(data),
302 }),
303 ),
304 FileSystem.layerNoop({}),
305 );
306
307 yield* Effect.gen(function* () {
308 const service = yield* ThreadPullRequestService.make;
309 yield* service.start();
310 yield* Deferred.succeed(activation, undefined);
311 // Backfill finds the settled branch thread and must read it again.
312 expect(yield* Queue.take(reads)).toEqual({ location: "active", unsettledOnly: false });
313 yield* service.drain;
314 // Its project is gone, so backfill finishes it on the first pass.
315 yield* TestClock.adjust("1 minute");
316 expect(yield* Queue.take(reads)).toEqual({ location: "active", unsettledOnly: true });
317 yield* service.drain;
318 }).pipe(Effect.provide(layerDependencies));
319 }),
320 ),
321 );
322});