1import {
2 CommandId,
3 type OrchestrationProjectShell,
4 type OrchestrationV2DomainEvent,
5 type OrchestrationV2ThreadShell,
6 type PullRequestSummary,
7 type ThreadId,
8 type ThreadLinkedPullRequest,
9} from "@t3tools/contracts";
10import { makeDrainableWorker } from "@t3tools/shared/DrainableWorker";
11import * as Cause from "effect/Cause";
12import * as Context from "effect/Context";
13import * as SecureRandom from "effect/crypto/SecureRandom";
14import * as DateTime from "effect/DateTime";
15import * as Effect from "effect/Effect";
16import * as FileSystem from "effect/FileSystem";
17import * as Option from "effect/Option";
18import * as Path from "effect/Path";
19import * as Schedule from "effect/Schedule";
20import type * as Scope from "effect/Scope";
21import * as Stream from "effect/Stream";
22
23import * as BackgroundPolicy from "../background/BackgroundPolicy.ts";
24import * as GitManager from "../git/GitManager.ts";
25import * as GitVcsDriver from "../vcs/GitVcsDriver.ts";
26import * as ProjectStore from "./ProjectStore.ts";
27import * as PullRequestService from "../pullRequest/PullRequestService.ts";
28import * as RepositoryIdentityResolver from "../project/RepositoryIdentityResolver.ts";
29import { forkParked } from "../serverActivation.ts";
30import { sourceControlDefinitions } from "../sourceControl/sourceControlDefinitions.ts";
31import * as Orchestrator from "./Orchestrator.ts";
32
33class ThreadPullRequestServiceV2 extends Context.Service<
34 ThreadPullRequestServiceV2,
35 {
36 readonly start: () => Effect.Effect<void, never, Scope.Scope>;
37 readonly drain: Effect.Effect<void>;
38 }
39>()("t3/orchestration-v2/ThreadPullRequestService/ThreadPullRequestServiceV2") {}
40
41function samePullRequest(
42 left: ThreadLinkedPullRequest | null | undefined,
43 right: ThreadLinkedPullRequest | null,
44): boolean {
45 if (left == null || right === null) return left == null && right === null;
46 return (
47 left.projectId === right.projectId &&
48 left.repository.toLowerCase() === right.repository.toLowerCase() &&
49 left.number === right.number &&
50 left.url === right.url
51 );
52}
53
54function pullRequestPredatesThread(
55 pullRequest: Pick<PullRequestSummary, "state" | "closedAt" | "mergedAt">,
56 createdAt: DateTime.Utc,
57): boolean {
58 const terminalAt =
59 pullRequest.state === "merged"
60 ? pullRequest.mergedAt
61 : pullRequest.state === "closed"
62 ? pullRequest.closedAt
63 : null;
64 return terminalAt != null && Date.parse(terminalAt) < DateTime.toEpochMillis(createdAt);
65}
66
67function pullRequestMatchesProject(
68 pullRequest: GitManager.GitBranchPullRequest,
69 project: OrchestrationProjectShell,
70): boolean {
71 return (
72 pullRequest.repositoryKey !== null &&
73 project.repositoryIdentity != null &&
74 sourceControlDefinitions.canonicalRepositoryKey(pullRequest.repositoryKey) ===
75 sourceControlDefinitions.canonicalRepositoryKey(project.repositoryIdentity.canonicalKey)
76 );
77}
78
79const resolveProjectForPullRequestDiscovery = Effect.fn(
80 "ThreadPullRequestServiceV2.resolveProject",
81)(function* (
82 project: OrchestrationProjectShell,
83 repositoryIdentities: RepositoryIdentityResolver.RepositoryIdentityResolver["Service"],
84 options?: { readonly refresh?: boolean },
85) {
86 // Post-turn discovery refreshes 15-minute identity caches because agents may add remotes.
87 const repositoryIdentity = yield* repositoryIdentities.resolve(project.workspaceRoot, {
88 refresh: options?.refresh ?? false,
89 });
90 return {
91 project: { ...project, repositoryIdentity },
92 repository: sourceControlDefinitions.repositorySelector(repositoryIdentity),
93 };
94});
95
96export function projectWorkspaceMatchesSnapshot(
97 currentProject: Option.Option<Pick<OrchestrationProjectShell, "workspaceRoot">>,
98 expectedWorkspaceRoot: string,
99): boolean {
100 return (
101 Option.isSome(currentProject) && currentProject.value.workspaceRoot === expectedWorkspaceRoot
102 );
103}
104
105const BACKFILL_ATTEMPTS = 5;
106
107interface RefreshRequest {
108 readonly threadId: ThreadId | null;
109 readonly refresh: boolean;
110 readonly backfill?: boolean;
111}
112
113export const make = Effect.gen(function* () {
114 const orchestrator = yield* Orchestrator.OrchestratorV2;
115 const projectStore = yield* ProjectStore.ProjectStoreV2;
116 const git = yield* GitManager.GitManager;
117 const pullRequests = yield* PullRequestService.PullRequestService;
118 const repositoryIdentities = yield* RepositoryIdentityResolver.RepositoryIdentityResolver;
119 const random = yield* SecureRandom.SecureRandom;
120 const fileSystem = yield* FileSystem.FileSystem;
121 const path = yield* Path.Path;
122 const backgroundPolicy = yield* BackgroundPolicy.BackgroundPolicy;
123 const pendingBackfill = new Map<ThreadId, number>();
124
125 const finishBackfill = (threads: ReadonlyArray<Pick<OrchestrationV2ThreadShell, "id">>) => {
126 for (const thread of threads) pendingBackfill.delete(thread.id);
127 };
128 const failBackfill = (threads: ReadonlyArray<Pick<OrchestrationV2ThreadShell, "id">>) => {
129 for (const thread of threads) {
130 const remaining = pendingBackfill.get(thread.id);
131 if (remaining === undefined) continue;
132 if (remaining <= 1) pendingBackfill.delete(thread.id);
133 else pendingBackfill.set(thread.id, remaining - 1);
134 }
135 };
136
137 const readThreadSnapshot = ({ threadId, backfill }: RefreshRequest) =>
138 orchestrator.getPullRequestSnapshot(
139 threadId === null ? { unsettledOnly: !(backfill || pendingBackfill.size > 0) } : { threadId },
140 );
141
142 const synchronize = Effect.fn("ThreadPullRequestServiceV2.synchronize")(function* (
143 request: RefreshRequest,
144 ) {
145 const [threadSnapshot, projectShells] = yield* Effect.all([
146 readThreadSnapshot(request),
147 projectStore.listShells(),
148 ]);
149 // Subagent children refresh after their own runs; sweeping them costs a lookup each per minute.
150 const snapshotThreads =
151 request.threadId === null
152 ? threadSnapshot.threads.filter(
153 (thread) => thread.lineage.relationshipToParent !== "subagent",
154 )
155 : threadSnapshot.threads;
156 const projects = new Map(projectShells.map((project) => [project.id, project]));
157 if (request.backfill) {
158 for (const thread of snapshotThreads) {
159 if (
160 (thread.settledOverride === "settled" || thread.settledAt !== null) &&
161 thread.branchPullRequest == null
162 ) {
163 pendingBackfill.set(thread.id, BACKFILL_ATTEMPTS);
164 }
165 }
166 }
167 const visibleThreadIds = new Set(
168 snapshotThreads.filter((thread) => thread.branch !== null).map((thread) => thread.id),
169 );
170 const checkedIds = request.threadId === null ? [...pendingBackfill.keys()] : [request.threadId];
171 for (const threadId of checkedIds) {
172 if (!visibleThreadIds.has(threadId)) pendingBackfill.delete(threadId);
173 }
174 const threads = snapshotThreads.filter(
175 (thread) =>
176 thread.archivedAt === null &&
177 ((thread.settledOverride !== "settled" && thread.settledAt === null) ||
178 request.threadId !== null ||
179 pendingBackfill.has(thread.id)) &&
180 (thread.branch !== null || thread.branchPullRequest != null),
181 );
182 const groups = Map.groupBy(threads, (thread) =>
183 JSON.stringify([thread.projectId, thread.worktreePath, thread.branch]),
184 );
185
186 yield* Effect.forEach(
187 groups.values(),
188 (group) =>
189 Effect.gen(function* () {
190 const first = group[0]!;
191 const project = projects.get(first.projectId);
192 if (project === undefined) return finishBackfill(group);
193 const { project: resolvedProject, repository } =
194 yield* resolveProjectForPullRequestDiscovery(project, repositoryIdentities, {
195 refresh: request.refresh,
196 });
197 if (first.branch !== null && repository === null) return finishBackfill(group);
198 // A worktree folder can outlive its .git, and git there fails with not_a_repository.
199 // An unreadable .git keeps the worktree rather than failing the whole group.
200 const cwd =
201 first.worktreePath !== null &&
202 (yield* fileSystem
203 .exists(path.join(first.worktreePath, ".git"))
204 .pipe(Effect.orElseSucceed(() => true)))
205 ? first.worktreePath
206 : project.workspaceRoot;
207 // A detected PR is confirmed against fresh git state before it is saved.
208 const detected =
209 first.branch === null
210 ? null
211 : yield* git.branchPullRequest(
212 { cwd, branch: first.branch },
213 { refresh: request.refresh, confirmTerminal: false },
214 );
215 if (detected !== null && !pullRequestMatchesProject(detected, resolvedProject)) {
216 return finishBackfill(group);
217 }
218 const detectedReference =
219 detected !== null && repository !== null
220 ? {
221 projectId: project.id,
222 repository,
223 number: detected.number,
224 url: detected.url,
225 }
226 : null;
227
228 // Only a new association is refused: a thread keeps a PR it already saved, and a PR the
229 // user checked out stays with the threads made for it, however long ago it closed.
230 const isHistoricalFor = (thread: (typeof group)[number]) =>
231 detected !== null &&
232 pullRequestPredatesThread(detected, thread.createdAt) &&
233 !samePullRequest(thread.branchPullRequest, detectedReference);
234 const dropsHistorical =
235 detected !== null &&
236 first.branch !== null &&
237 group.some(isHistoricalFor) &&
238 !(yield* git.checkedOutPullRequest({ cwd, branch: first.branch, url: detected.url }));
239
240 const plans = yield* Effect.forEach(group, (thread) =>
241 Effect.gen(function* () {
242 let branchPullRequest =
243 dropsHistorical && isHistoricalFor(thread) ? null : detectedReference;
244 if (
245 branchPullRequest === null &&
246 thread.branch !== null &&
247 thread.worktreePath === null &&
248 thread.branchPullRequest != null
249 ) {
250 const previous = yield* pullRequests.summary(thread.branchPullRequest, {
251 recoverTransientFailure: false,
252 });
253 if (
254 previous.headBranch === thread.branch &&
255 (previous.state === "merged" || previous.state === "closed")
256 ) {
257 branchPullRequest = thread.branchPullRequest;
258 }
259 }
260
261 let replacement: ThreadLinkedPullRequest | undefined;
262 if (
263 (thread.pullRequests ?? []).length === 0 &&
264 thread.linkedPullRequest != null &&
265 detected?.state === "open" &&
266 detectedReference !== null &&
267 !samePullRequest(thread.linkedPullRequest, detectedReference)
268 ) {
269 const linked = yield* pullRequests.summary(thread.linkedPullRequest, {
270 recoverTransientFailure: false,
271 });
272 if (linked.state === "merged" || linked.state === "closed") {
273 replacement = detectedReference;
274 }
275 }
276 if (
277 samePullRequest(thread.branchPullRequest, branchPullRequest) &&
278 replacement === undefined
279 ) {
280 pendingBackfill.delete(thread.id);
281 return null;
282 }
283 return { thread, branchPullRequest, replacement };
284 }).pipe(
285 Effect.catchCauseIf(
286 (cause) => !Cause.hasInterruptsOnly(cause),
287 (cause) =>
288 Effect.logWarning("thread pull request discovery failed", {
289 threadId: thread.id,
290 cause: Cause.pretty(cause),
291 }).pipe(
292 Effect.tap(() => Effect.sync(() => failBackfill([thread]))),
293 Effect.as(null),
294 ),
295 ),
296 ),
297 );
298 const updates = plans.filter((plan) => plan !== null);
299 if (updates.length === 0) return;
300
301 if (detected !== null && first.branch !== null) {
302 const current = yield* git.branchPullRequest(
303 { cwd, branch: first.branch },
304 { freshGitState: true },
305 );
306 const currentIdentity = yield* repositoryIdentities.resolve(project.workspaceRoot, {
307 refresh: true,
308 });
309 if (
310 current === null ||
311 current.number !== detected.number ||
312 current.url !== detected.url ||
313 current.state !== detected.state ||
314 current.repositoryKey !== detected.repositoryKey ||
315 !pullRequestMatchesProject(current, {
316 ...project,
317 repositoryIdentity: currentIdentity,
318 })
319 ) {
320 return failBackfill(updates.map((update) => update.thread));
321 }
322 }
323
324 yield* Effect.forEach(
325 updates,
326 ({ thread, branchPullRequest, replacement }) =>
327 Effect.gen(function* () {
328 // Discovery yields to network I/O; recheck the project at commit to reject deleted projects or changed checkouts.
329 const currentProject = yield* projectStore.getShell(project.id);
330 if (!projectWorkspaceMatchesSnapshot(currentProject, project.workspaceRoot)) {
331 return failBackfill([thread]);
332 }
333 const uuid = yield* random.uuidV4();
334 yield* orchestrator.dispatch({
335 type: "thread.pull-request.sync",
336 commandId: CommandId.make(`server:thread-pull-request:${thread.id}:${uuid}`),
337 threadId: thread.id,
338 projectId: project.id,
339 snapshotSequence: threadSnapshot.snapshotSequence,
340 expected: {
341 workspaceRoot: project.workspaceRoot,
342 branch: thread.branch,
343 worktreePath: thread.worktreePath,
344 linkedPullRequest: thread.linkedPullRequest ?? null,
345 branchPullRequest: thread.branchPullRequest ?? null,
346 },
347 branchPullRequest,
348 ...(replacement === undefined ? {} : { linkedPullRequest: replacement }),
349 });
350 pendingBackfill.delete(thread.id);
351 }).pipe(
352 Effect.catchCauseIf(
353 (cause) => !Cause.hasInterruptsOnly(cause),
354 (cause) =>
355 Effect.logWarning("thread pull request update failed", {
356 threadId: thread.id,
357 cause: Cause.pretty(cause),
358 }).pipe(Effect.tap(() => Effect.sync(() => failBackfill([thread])))),
359 ),
360 ),
361 { discard: true },
362 );
363 }).pipe(
364 Effect.catchCauseIf(
365 (cause) => !Cause.hasInterruptsOnly(cause),
366 (cause) =>
367 Effect.logWarning("thread branch pull request lookup failed", {
368 threadIds: group.map((thread) => thread.id),
369 cause: Cause.pretty(cause),
370 }).pipe(Effect.tap(() => Effect.sync(() => failBackfill(group)))),
371 ),
372 ),
373 // Batch concurrent branch lookups into one GitHub GraphQL document.
374 { concurrency: 32, discard: true },
375 );
376 });
377
378 // At most one queued request per thread, so a turn end's events share one lookup.
379 // A request after that lookup starts queues another, so later host changes are still read.
380 const queuedThreadRequests = new Map<ThreadId, { threadId: ThreadId; refresh: boolean }>();
381
382 const worker = yield* makeDrainableWorker((request: RefreshRequest) =>
383 Effect.suspend(() => {
384 if (request.threadId !== null) queuedThreadRequests.delete(request.threadId);
385 return synchronize(request);
386 }).pipe(
387 Effect.provideService(GitVcsDriver.BackgroundGitWork, true),
388 Effect.catchCauseIf(
389 (cause) => !Cause.hasInterruptsOnly(cause),
390 (cause) =>
391 Effect.logWarning("thread pull request refresh failed", {
392 cause: Cause.pretty(cause),
393 }),
394 ),
395 ),
396 );
397
398 const enqueueThread = (threadId: ThreadId, refresh: boolean) =>
399 Effect.suspend(() => {
400 const queued = queuedThreadRequests.get(threadId);
401 if (queued !== undefined) {
402 queued.refresh ||= refresh;
403 return Effect.void;
404 }
405 const request = { threadId, refresh };
406 queuedThreadRequests.set(threadId, request);
407 return worker.enqueue(request);
408 });
409
410 const processEvent = (event: OrchestrationV2DomainEvent) => {
411 switch (event.type) {
412 case "thread.created":
413 case "thread.unarchived":
414 case "thread.metadata-updated":
415 return enqueueThread(event.threadId, false);
416 case "thread.unsettled":
417 case "checkpoint.captured":
418 return enqueueThread(event.threadId, true);
419 case "run.updated":
420 if (
421 event.payload.status === "completed" ||
422 event.payload.status === "failed" ||
423 event.payload.status === "cancelled" ||
424 event.payload.status === "interrupted"
425 ) {
426 return enqueueThread(event.threadId, true);
427 }
428 break;
429 }
430 return Effect.void;
431 };
432
433 const start: ThreadPullRequestServiceV2["Service"]["start"] = Effect.fn(
434 "ThreadPullRequestServiceV2.start",
435 )(function* () {
436 yield* forkParked(Stream.runForEach(orchestrator.streamDomainEvents, processEvent));
437 yield* forkParked(
438 Effect.gen(function* () {
439 yield* worker.enqueue({ threadId: null, refresh: false, backfill: true });
440 yield* worker.drain;
441 yield* Effect.gen(function* () {
442 // The sweep only catches pull requests opened outside a turn, which
443 // nobody sees without a client. Thread events still refresh headless work.
444 if (!(yield* backgroundPolicy.shouldRunOpportunisticWork)) return;
445 yield* worker.enqueue({ threadId: null, refresh: false });
446 yield* worker.drain;
447 }).pipe(Effect.repeat(Schedule.spaced("1 minute")), Effect.delay("1 minute"));
448 }).pipe(Effect.asVoid),
449 );
450 });
451
452 return { start, drain: worker.drain } satisfies ThreadPullRequestServiceV2["Service"];
453});