Open History to see the last commit
1import {
2 CommandId,
3 isOrchestrationV2WorkActive,
4 type OrchestrationV2TurnItem,
5 type PullRequestSummary,
6 type ThreadId,
7 type ThreadPullRequestKey,
8 type ThreadPullRequestLink,
9 type ThreadPullRequestSnapshot,
10 type ThreadPullRequestStack,
11} from "@t3tools/contracts";
12import { makeDrainableWorker } from "@t3tools/shared/DrainableWorker";
13import {
14 threadPullRequestKeyOf,
15 normalizeThreadPullRequestKey,
16 threadPullRequestKeysEqual,
17 visibleThreadPullRequests,
18} from "../sourceControl/threadPullRequests.ts";
19import { sourceControlDefinitions } from "../sourceControl/sourceControlDefinitions.ts";
20import * as Cause from "effect/Cause";
21import * as Clock from "effect/Clock";
22import * as Context from "effect/Context";
23import * as SecureRandom from "effect/crypto/SecureRandom";
24import * as DateTime from "effect/DateTime";
25import * as Effect from "effect/Effect";
26import * as Layer from "effect/Layer";
27import * as Schema from "effect/Schema";
28import * as Schedule from "effect/Schedule";
29import type * as Scope from "effect/Scope";
30import * as Semaphore from "effect/Semaphore";
31import * as Stream from "effect/Stream";
32
33import * as GitManager from "../git/GitManager.ts";
34import { PullRequestProviderError } from "@t3tools/source-control-core/server/PullRequestProvider";
35import * as PullRequestService from "../pullRequest/PullRequestService.ts";
36import { forkParked } from "../serverActivation.ts";
37import * as Orchestrator from "./Orchestrator.ts";
38import * as ProjectionStore from "./ProjectionStore.ts";
39import { isTerminalRunStatus } from "./ThreadManagementService.ts";
40
41const SLOW_SYNC_INTERVAL_MS = 15 * 60 * 1_000;
42// Every server polls every open linked pull request, and most of them sit untouched for days.
43// One nobody has updated for QUIET_AFTER_MS is read on the quiet cadence instead of each sweep;
44// requested refreshes, watches and a host CLI's merge command still read it at once.
45const QUIET_AFTER_MS = 30 * 60 * 1_000;
46const QUIET_SYNC_INTERVAL_MS = 5 * 60 * 1_000;
47// A pull request closed this long ago is rarely reopened, so the sweep stops reading it, as it
48// does merged ones. A reader that sees it open, or a requested refresh, still reads it at once.
49const CLOSED_FINAL_AFTER_MS = 24 * 60 * 60 * 1_000;
50
51const isChangeRequestCloseCommand = (command: string) =>
52 sourceControlDefinitions.definitions.some(
53 (definition) => definition.changeRequestCloseCommand?.test(command) === true,
54 );
55
56// Where a shell starts a command, past `NAME=value` assignments and wrappers like `timeout 60`.
57// A create command anywhere else, as in `echo "gh pr create"`, is text, not a command that runs.
58const COMMAND_START =
59 "(?:^|[;&|({`\\n]|\\b(?:do|then|else)\\s)\\s*(?:(?:\\w+=\\S*|env|time|nohup|timeout\\s+\\S+)\\s+)*";
60const CREATE_COMMANDS = sourceControlDefinitions.definitions.flatMap(
61 ({ kind, changeRequestCreateCommand: command }) =>
62 command === undefined
63 ? []
64 : // The subcommand must end its shell word: `gh pr create#x` runs another one.
65 [
66 {
67 kind,
68 command,
69 run: new RegExp(`${COMMAND_START}${command.source}(?=$|[\\s;&|)])`, "gu"),
70 },
71 ],
72);
73const createCommandCount = (code: string) =>
74 CREATE_COMMANDS.reduce((count, { run }) => count + [...code.matchAll(run)].length, 0);
75// A push, directly or through a stack tool, that can start checks on a linked pull request.
76const PUSH_COMMAND = new RegExp(
77 `${COMMAND_START}(?:git(?:\\s+-[cC]\\s+\\S+)*\\s+push|gh\\s+stack\\s+(?:push|submit)|cafe\\s+stack\\s+submit|jj\\s+git\\s+push)(?=$|[\\s;&|)])`,
78 "u",
79);
80
81// Codex runs each command as `/bin/zsh -lc <script>`, quoting the script in pieces such as
82// `"a "'`b`'" c"`.
83const SHELL_WRAPPER = /^\s*(?:\S*\/)?(?:ba|z|da|k)?sh\s+-l?c\s+(?=["'])([\s\S]*)$/u;
84const SHELL_WORD_PIECE = /"((?:\\[\s\S]|[^"\\])*)"|'([^']*)'|\\([\s\S])|(\S)/gu;
85// Heredoc bodies, quoted strings and comments are text a command carries, not commands it runs.
86const HEREDOC_BODY =
87 /<<-?[ \t]*(['"]?)(\w+)\1[^\n]*\n(?:(?:[\s\S]*?\n)?[ \t]*\2[ \t]*(?=\n|$)|[\s\S]*$)/gu;
88const QUOTED_OR_COMMENT = /'[^']*'|"(?:\\[\s\S]|[^"\\])*"|(?<=^|[\s;&|(])#[^\n]*/gmu;
89const COMMAND_SEPARATOR = /&&|\|\||[;|\n]/u;
90// Commands that print other change requests' URLs, so output beside them proves nothing.
91const LISTS_CHANGE_REQUESTS =
92 /\b(?:gh\s+(?:pr\s+(?:list|status)|search\s+prs|api)|glab\s+(?:mr\s+list|api))\b/u;
93const URL_LINE = /^[ \t]*(https?:\/\/\S+)[ \t]*$/gmu;
94
95/** The script a shell input runs, unwrapped from Codex's `sh -lc`. */
96function shellScript(input: string): string {
97 const [, wrapped] = SHELL_WRAPPER.exec(input) ?? [];
98 return wrapped === undefined
99 ? input
100 : Array.from(
101 wrapped.matchAll(SHELL_WORD_PIECE),
102 ([, double, single, escaped, bare]) =>
103 double?.replace(/\\(["\\$`])/gu, "$1") ?? single ?? escaped ?? bare,
104 ).join("");
105}
106
107/** A script split at command separators that sit outside quotes. */
108function shellSegments(script: string): ReadonlyArray<string> {
109 const segments: Array<string> = [];
110 let current = "";
111 let quote: string | null = null;
112 for (let index = 0; index < script.length; index++) {
113 const char = script[index] ?? "";
114 if (quote !== null) {
115 current += char;
116 if (char === "\\" && quote === '"') current += script[++index] ?? "";
117 else if (char === quote) quote = null;
118 continue;
119 }
120 if (char === "'" || char === '"') quote = char;
121 if (char === "\\") {
122 current += char + (script[++index] ?? "");
123 continue;
124 }
125 const pair = script.slice(index, index + 2);
126 if (pair === "&&" || pair === "||" || char === ";" || char === "|" || char === "\n") {
127 segments.push(current);
128 current = "";
129 if (pair === "&&" || pair === "||") index++;
130 continue;
131 }
132 current += char;
133 }
134 segments.push(current);
135 return segments;
136}
137
138/** The commands a script runs, without the text it carries. */
139const shellCode = (script: string) =>
140 script.replace(HEREDOC_BODY, "<<").replace(QUOTED_OR_COMMENT, "''");
141
142const isPushCommand = (input: string) => PUSH_COMMAND.test(shellCode(shellScript(input)));
143
144type CommandExecution = Extract<OrchestrationV2TurnItem, { readonly type: "command_execution" }>;
145
146/**
147 * The change requests a shell command opened: URLs on lines of their own in the output of a
148 * command that runs a host's create command. One the command also names, as `gh pr view 12`
149 * does, existed before it ran. More URLs than create commands means something else printed
150 * some, such as `gh pr list`, so none are taken.
151 */
152export function changeRequestsCreatedBy(input: string, output: string) {
153 if (!CREATE_COMMANDS.some(({ command }) => command.test(input))) return [];
154 const script = shellScript(input);
155 const code = shellCode(script);
156 const creates = createCommandCount(code);
157 if (creates === 0) return [];
158 const otherCommands = code
159 .split(COMMAND_SEPARATOR)
160 .filter((part) => createCommandCount(part) === 0)
161 .join("\n");
162 if (LISTS_CHANGE_REQUESTS.test(otherCommands)) return [];
163 const otherScript = shellSegments(script)
164 .filter((segment) => createCommandCount(shellCode(segment)) === 0)
165 .join("\n");
166 const creators = new Set(
167 CREATE_COMMANDS.filter(({ run }) => code.match(run) !== null).map(({ kind }) => kind),
168 );
169 const created = new Map<string, ThreadPullRequestKey & { readonly url: string }>();
170 for (const [, url = ""] of output.matchAll(URL_LINE)) {
171 const parsed = sourceControlDefinitions.parseChangeRequestUrl(url);
172 // Only the host whose create command ran can have printed a new change request.
173 if (parsed === null || !creators.has(parsed.definition.kind)) continue;
174 const { link } = parsed;
175 const named = new RegExp(`(?:^|(?<=[\\s#/=("']))${link.number}(?=$|[\\s);&|/"'])`, "u");
176 // Quoted text in the other commands counts, as in `printf '.../pull/12'`.
177 if (named.test(otherScript)) continue;
178 const key = normalizeThreadPullRequestKey(link);
179 created.set(threadPullRequestKeyOf(key), { ...key, url });
180 }
181 return created.size > creates ? [] : [...created.values()];
182}
183
184const isPullRequestProviderError = Schema.is(PullRequestProviderError);
185
186type SnapshotFields = Omit<ThreadPullRequestSnapshot, "syncedAt">;
187
188interface LinkEntry {
189 readonly thread: ProjectionStore.ProjectionThreadPullRequests;
190 readonly link: ThreadPullRequestLink;
191}
192
193const referenceOf = ({ thread, link }: LinkEntry) => ({
194 projectId: thread.projectId,
195 host: normalizeThreadPullRequestKey(link).host,
196 repository: link.repository,
197 number: link.number,
198});
199
200/** Pauses and batching both belong to a project's host. */
201const pauseKeyOf = (entry: LinkEntry) => {
202 const { projectId, host } = referenceOf(entry);
203 return `${projectId}\0${host}`;
204};
205
206/** Another layer of the stack `url` belongs to, in its host's canonical shape. */
207function siblingChangeRequestUrl(url: string, number: number): string | null {
208 if (!Number.isSafeInteger(number) || number < 1) return null;
209 const parsed = sourceControlDefinitions.parseChangeRequestUrl(url);
210 if (parsed === null) return null;
211 const { definition, link } = parsed;
212 return definition.changeRequestUrl({
213 host: link.host,
214 repository: link.repository,
215 number,
216 webUrl: link.repositoryUrl,
217 });
218}
219
220function snapshotFieldsOf(summary: PullRequestSummary): SnapshotFields {
221 return {
222 state: summary.state,
223 title: summary.title,
224 headBranch: summary.headBranch,
225 baseBranch: summary.baseBranch,
226 isDraft: summary.isDraft ?? false,
227 updatedAt: summary.updatedAt,
228 closedAt: summary.closedAt ?? null,
229 mergedAt: summary.mergedAt ?? null,
230 ...(summary.author === undefined ? {} : { author: summary.author }),
231 ...(summary.additions === undefined ? {} : { additions: summary.additions }),
232 ...(summary.deletions === undefined ? {} : { deletions: summary.deletions }),
233 ...(summary.changedFiles === undefined ? {} : { changedFiles: summary.changedFiles }),
234 ...(summary.reviewDecision === undefined ? {} : { reviewDecision: summary.reviewDecision }),
235 ...(summary.checksState === undefined ? {} : { checksState: summary.checksState }),
236 ...(summary.mergeability === undefined ? {} : { mergeability: summary.mergeability }),
237 };
238}
239
240// GitHub briefly returns unknown after base pushes; other hosts may mean a lasting condition.
241function withKnownMergeability(
242 previous: ThreadPullRequestSnapshot | null,
243 fields: SnapshotFields,
244 provider: PullRequestSummary["provider"],
245): SnapshotFields {
246 if (
247 provider !== "github" ||
248 fields.state !== "open" ||
249 fields.mergeability !== "unknown" ||
250 previous?.updatedAt !== fields.updatedAt
251 )
252 return fields;
253 const known = previous?.state === "open" ? previous.mergeability : undefined;
254 return known === undefined || known === "unknown" ? fields : { ...fields, mergeability: known };
255}
256
257// Rotating ghe.com avatar tokens should not make otherwise identical snapshots differ.
258function avatarIdentity(avatarUrl: string | null | undefined): string | null {
259 if (avatarUrl === null || avatarUrl === undefined) return null;
260 const url = URL.parse(avatarUrl);
261 if (url === null || (url.hostname !== "ghe.com" && !url.hostname.endsWith(".ghe.com"))) {
262 return avatarUrl;
263 }
264 url.searchParams.delete("token");
265 return url.href;
266}
267
268function snapshotFieldsEqual(left: SnapshotFields, right: SnapshotFields): boolean {
269 return (
270 left.state === right.state &&
271 left.title === right.title &&
272 left.headBranch === right.headBranch &&
273 left.baseBranch === right.baseBranch &&
274 left.isDraft === right.isDraft &&
275 left.updatedAt === right.updatedAt &&
276 (left.closedAt ?? null) === (right.closedAt ?? null) &&
277 (left.mergedAt ?? null) === (right.mergedAt ?? null) &&
278 (left.author?.login ?? null) === (right.author?.login ?? null) &&
279 avatarIdentity(left.author?.avatarUrl) === avatarIdentity(right.author?.avatarUrl) &&
280 left.additions === right.additions &&
281 left.deletions === right.deletions &&
282 left.changedFiles === right.changedFiles &&
283 (left.reviewDecision ?? null) === (right.reviewDecision ?? null) &&
284 (left.checksState ?? null) === (right.checksState ?? null) &&
285 left.mergeability === right.mergeability
286 );
287}
288
289function stacksEqual(
290 left: ThreadPullRequestStack | null,
291 right: ThreadPullRequestStack | null,
292): boolean {
293 if (left === null || right === null) return left === right;
294 return (
295 left.kind === right.kind &&
296 left.id === right.id &&
297 left.number === right.number &&
298 left.url === right.url &&
299 left.base === right.base &&
300 left.layers.length === right.layers.length &&
301 left.layers.every((layer, index) => {
302 const other = right.layers[index]!;
303 return (
304 layer.number === other.number &&
305 layer.headBranch === other.headBranch &&
306 layer.state === other.state
307 );
308 })
309 );
310}
311
312/** Stack ids are unique only within a repository, e.g. GitHub's fallback to the stack number. */
313function nativeStackKeyOf(link: ThreadPullRequestLink, stack: ThreadPullRequestStack): string {
314 const key = normalizeThreadPullRequestKey(link);
315 return `${key.host}/${key.repository}#stack:${stack.id}`;
316}
317
318function sameLayerNumbers(left: ThreadPullRequestStack, right: ThreadPullRequestStack): boolean {
319 return (
320 left.layers.length === right.layers.length &&
321 left.layers.every((layer, index) => layer.number === right.layers[index]!.number)
322 );
323}
324
325function skipReason(cause: Cause.Cause<unknown>): string {
326 const error = Cause.squash(cause);
327 return error instanceof Error ? error.message : String(error);
328}
329
330function rateLimitRetryAt(cause: Cause.Cause<unknown>): number | undefined {
331 let error: unknown = Cause.squash(cause);
332 while (error instanceof Error) {
333 if (isPullRequestProviderError(error) && error.reason === "rate-limited") return error.retryAt;
334 error = error.cause;
335 }
336 return undefined;
337}
338
339function isUnsettled(thread: ProjectionStore.ProjectionThreadPullRequests): boolean {
340 return thread.settledOverride !== "settled" && thread.settledAt === null;
341}
342
343export class PullRequestSyncReactor extends Context.Service<
344 PullRequestSyncReactor,
345 {
346 readonly start: () => Effect.Effect<void, never, Scope.Scope>;
347 readonly drain: Effect.Effect<void>;
348 /** Forces reread on the next sweep, including terminal snapshots; rate-limited hosts wait until the pause ends. */
349 readonly requestSync: (key: ThreadPullRequestKey) => Effect.Effect<void>;
350 }
351>()("t3/orchestration-v2/PullRequestSyncReactor") {}
352
353/** @public */
354export const make = Effect.gen(function* () {
355 const engine = yield* Orchestrator.OrchestratorV2;
356 const projections = yield* ProjectionStore.ProjectionStoreV2;
357 const pullRequests = yield* PullRequestService.PullRequestService;
358 const git = yield* GitManager.GitManager;
359 const random = yield* SecureRandom.SecureRandom;
360
361 const lastSyncedAt = new Map<string, number>();
362 const requested = new Set<string>();
363 // Requests arriving during a host read wait for the next sweep.
364 const pendingRequests = new Set<string>();
365 // PR keys wait in `pendingRequests` for one queued sweep, so a burst of links (an agent
366 // linking dozens of pull requests) is read together and shares the summary batches.
367 let requestedSweepQueued = false;
368 const retryStacks = new Set<string>();
369 // Group keys the host answered not-found for; they are read on the slow cadence until found.
370 const missing = new Set<string>();
371 // Pause per project and host because credentials are project-specific; leave paused PRs due until reset.
372 const pausedUntil = new Map<string, number>();
373
374 const isDue = (key: string, entries: ReadonlyArray<LinkEntry>, nowMs: number): boolean => {
375 if (requested.has(key)) return true;
376 const active = entries.filter((entry) => isUnsettled(entry.thread));
377 if (active.length === 0) return false;
378 const last = lastSyncedAt.get(key);
379 const sinceLastMs = last === undefined ? Number.POSITIVE_INFINITY : nowMs - last;
380 if (missing.has(key)) return sinceLastMs >= SLOW_SYNC_INTERVAL_MS;
381 if (retryStacks.has(key) || active.some((entry) => entry.link.snapshot === null)) return true;
382 // Merged snapshots without mergedAt cannot settle threads; reread slowly until supplied.
383 const isFinal = ({ link: { snapshot } }: LinkEntry) =>
384 snapshot?.state === "merged"
385 ? snapshot.mergedAt != null
386 : snapshot?.state === "closed" &&
387 nowMs - Date.parse(snapshot.closedAt ?? "") >= CLOSED_FINAL_AFTER_MS;
388 if (active.every(isFinal)) return false;
389 const open = active.filter((entry) => entry.link.snapshot?.state === "open");
390 if (open.length > 0) {
391 const quiet = open.every((entry) => {
392 const updatedAt = entry.link.snapshot?.updatedAt;
393 return updatedAt != null && nowMs - Date.parse(updatedAt) >= QUIET_AFTER_MS;
394 });
395 return !quiet || sinceLastMs >= QUIET_SYNC_INTERVAL_MS;
396 }
397 return sinceLastMs >= SLOW_SYNC_INTERVAL_MS;
398 };
399
400 const logSkipped =
401 (message: string, fields: Record<string, unknown>) =>
402 <E>(cause: Cause.Cause<E>): Effect.Effect<void, E> =>
403 Cause.hasInterruptsOnly(cause) ? Effect.failCause(cause) : Effect.logWarning(message, fields);
404
405 const sweep = Effect.fn("PullRequestSyncReactor.sweep")(function* (scope: "all" | "requested") {
406 const threads = yield* projections.getThreadsWithPullRequests();
407 const now = yield* DateTime.now;
408 const nowMs = DateTime.toEpochMillis(now);
409 const nowIso = DateTime.formatIso(now);
410
411 const groups = new Map<string, Array<LinkEntry>>();
412 for (const thread of threads) {
413 for (const link of visibleThreadPullRequests(thread.pullRequests ?? [])) {
414 // Reads depend on the project's checkout and credentials, including their failures.
415 const key = `${thread.projectId}\0${threadPullRequestKeyOf(link)}`;
416 const entries = groups.get(key) ?? [];
417 entries.push({ thread, link });
418 groups.set(key, entries);
419 }
420 }
421
422 // A host-level refresh reaches every project, each retaining its own pending retry.
423 for (const [key, entries] of groups) {
424 if (pendingRequests.has(threadPullRequestKeyOf(entries[0]!.link))) requested.add(key);
425 }
426 pendingRequests.clear();
427
428 for (const key of lastSyncedAt.keys()) if (!groups.has(key)) lastSyncedAt.delete(key);
429 for (const key of retryStacks) if (!groups.has(key)) retryStacks.delete(key);
430 for (const key of missing) if (!groups.has(key)) missing.delete(key);
431 for (const key of requested.keys()) if (!groups.has(key)) requested.delete(key);
432
433 const linkedThisSweep = new Set<string>();
434 // Fresh stack reads by group, so a member flagged by a sibling is compared against what the
435 // host answered this sweep, not the saved layers that sibling is about to replace.
436 const stacksReadThisSweep = new Map<string, ThreadPullRequestStack | null>();
437 const persistence = yield* Semaphore.make(1);
438
439 // Effect.fn's stack frames close over these sweep-scoped bodies and their arguments, and a
440 // cache lookup forked inside keeps those frames, so a plain span leaves the sweep collectable.
441 const syncEntry = Effect.fnUntraced(function* (
442 entry: LinkEntry,
443 hostFields: SnapshotFields,
444 fetchedStack: { readonly stack: ThreadPullRequestStack | null } | null,
445 provider: PullRequestSummary["provider"],
446 ) {
447 const { thread, link } = entry;
448 const fields = withKnownMergeability(link.snapshot, hostFields, provider);
449 const nextStack = fetchedStack === null ? link.stack : fetchedStack.stack;
450 const changed =
451 link.snapshot === null ||
452 !snapshotFieldsEqual(link.snapshot, fields) ||
453 !stacksEqual(link.stack, nextStack);
454 // Persist siblings before terminal snapshots trigger settlement; settled threads get snapshots without new links.
455 for (const layer of isUnsettled(thread) ? (fetchedStack?.stack?.layers ?? []) : []) {
456 const layerKey = {
457 host: normalizeThreadPullRequestKey(link).host,
458 repository: link.repository,
459 number: layer.number,
460 };
461 const dedupeKey = `${thread.id}:${threadPullRequestKeyOf(layerKey)}`;
462 if (linkedThisSweep.has(dedupeKey)) continue;
463 // Tombstones count as present: a dismissed layer is never re-added.
464 if (
465 (thread.pullRequests ?? []).some((existing) =>
466 threadPullRequestKeysEqual(existing, layerKey),
467 )
468 ) {
469 continue;
470 }
471 const url = siblingChangeRequestUrl(link.url, layer.number);
472 if (url === null) continue;
473 const uuid = yield* random.uuidV4();
474 yield* engine.dispatch({
475 type: "thread.pull-request.link",
476 commandId: CommandId.make(`server:pr-stack-link:${thread.id}:${uuid}`),
477 threadId: thread.id,
478 ...layerKey,
479 url,
480 source: "stack",
481 });
482 linkedThisSweep.add(dedupeKey);
483 }
484 if (changed) {
485 const uuid = yield* random.uuidV4();
486 yield* engine.dispatch({
487 type: "thread.pull-request-link.sync",
488 commandId: CommandId.make(`server:pr-sync:${thread.id}:${uuid}`),
489 threadId: thread.id,
490 host: normalizeThreadPullRequestKey(link).host,
491 repository: link.repository,
492 number: link.number,
493 snapshot: { ...fields, syncedAt: nowIso },
494 stack: nextStack,
495 });
496 }
497 }, Effect.withSpan("PullRequestSyncReactor.syncEntry"));
498
499 const syncGroup = Effect.fnUntraced(function* (key: string, entries: ReadonlyArray<LinkEntry>) {
500 const first = entries[0]!;
501 const ref = referenceOf(first);
502 const refreshRequested = requested.has(key);
503 // A held merged summary is served without asking the host, so a snapshot still missing its
504 // merge time has to drop it or the slow-cadence re-read would hand back the same gap.
505 const missingMergedAt = entries.some(
506 (entry) => entry.link.snapshot?.state === "merged" && entry.link.snapshot.mergedAt == null,
507 );
508 // A stack retry reads past the cache, or it could get back the layer list it is retrying.
509 const invalidated = refreshRequested || missingMergedAt || retryStacks.has(key);
510 if (invalidated) yield* pullRequests.invalidate({ reference: ref });
511 const summary = yield* pullRequests.summary(ref, { recoverTransientFailure: false }).pipe(
512 Effect.tapError((error) =>
513 error.reason === "not-found"
514 ? Effect.sync(() => {
515 // Not-found is a host answer: it consumes the refresh and starts the backoff.
516 missing.add(key);
517 lastSyncedAt.set(key, nowMs);
518 requested.delete(key);
519 })
520 : Effect.void,
521 ),
522 );
523 missing.delete(key);
524 const fields = snapshotFieldsOf(summary);
525 const needsStack =
526 refreshRequested ||
527 retryStacks.has(key) ||
528 entries.some(
529 (entry) =>
530 entry.link.snapshot === null ||
531 !snapshotFieldsEqual(
532 entry.link.snapshot,
533 withKnownMergeability(entry.link.snapshot, fields, summary.provider),
534 ) ||
535 (summary.stack !== undefined &&
536 (entry.link.stack?.number ?? null) !== (summary.stack?.number ?? null)),
537 );
538 const knownUnstacked =
539 summary.stack === null && entries.every((entry) => entry.link.stack === null);
540 // A sibling can flag this group while its summary read is in flight.
541 if (!invalidated && retryStacks.has(key) && !knownUnstacked) {
542 yield* pullRequests.invalidate({ reference: ref });
543 }
544 // The retry this read answers is consumed before the read, not after it: a sibling may
545 // flag this group again while the read is in flight, and that flag must survive to the
546 // next sweep because the answer below may predate the sibling's layers.
547 if (needsStack) retryStacks.delete(key);
548 const fetchedStack = !needsStack
549 ? null
550 : knownUnstacked
551 ? { stack: null }
552 : yield* pullRequests.stack(ref, { includeDetails: false }).pipe(
553 Effect.map((stack) => ({
554 stack: stack === null ? null : ({ kind: "native", ...stack } as const),
555 })),
556 Effect.catchCauseIf(
557 (cause) => !Cause.hasInterruptsOnly(cause),
558 (cause) =>
559 rateLimitRetryAt(cause) !== undefined
560 ? Effect.sync(() => retryStacks.add(key)).pipe(
561 Effect.andThen(Effect.failCause(cause)),
562 )
563 : Effect.logWarning("pull request stack lookup failed", {
564 key,
565 }).pipe(Effect.as(null)),
566 ),
567 );
568 if (needsStack) {
569 if (fetchedStack === null) {
570 retryStacks.add(key);
571 return;
572 }
573 const fetched = fetchedStack.stack;
574 stacksReadThisSweep.set(key, fetched);
575 // New upper layers leave older members' fields unchanged; differing layer lists need a stack read.
576 // Only flag siblings in the same project: different credentials can yield lists that never agree.
577 if (fetched !== null) {
578 const fetchedKey = nativeStackKeyOf(first.link, fetched);
579 for (const [siblingKey, siblingEntries] of groups) {
580 if (siblingKey === key) continue;
581 if (siblingEntries[0]!.thread.projectId !== first.thread.projectId) continue;
582 const readThisSweep = stacksReadThisSweep.get(siblingKey);
583 const stale = siblingEntries.some(({ link }) => {
584 const stack = readThisSweep === undefined ? link.stack : readThisSweep;
585 return (
586 stack !== null &&
587 nativeStackKeyOf(link, stack) === fetchedKey &&
588 !sameLayerNumbers(stack, fetched)
589 );
590 });
591 if (stale) retryStacks.add(siblingKey);
592 }
593 }
594 }
595 // The host answered, so the cadence clock ticks even if a dispatch below is rejected.
596 lastSyncedAt.set(key, nowMs);
597 requested.delete(key);
598 yield* Effect.forEach(
599 entries,
600 (entry) =>
601 syncEntry(entry, fields, fetchedStack, summary.provider).pipe(
602 persistence.withPermits(1),
603 Effect.catchCause((cause) => {
604 if (!Cause.hasInterruptsOnly(cause)) retryStacks.add(key);
605 return logSkipped("pull request sync skipped", { threadId: entry.thread.id, key })(
606 cause,
607 );
608 }),
609 ),
610 { discard: true },
611 );
612 }, Effect.withSpan("PullRequestSyncReactor.syncGroup"));
613
614 const skips = new Map<string, { count: number; readonly key: string }>();
615 const readGroup = (key: string, entries: ReadonlyArray<LinkEntry>, pauseKey: string) =>
616 syncGroup(key, entries).pipe(
617 Effect.catchCause((cause) => {
618 if (Cause.hasInterruptsOnly(cause)) return Effect.failCause(cause);
619 const retryAt = rateLimitRetryAt(cause);
620 if (retryAt !== undefined) {
621 pausedUntil.set(pauseKey, Math.max(retryAt, pausedUntil.get(pauseKey) ?? 0));
622 }
623 const reason = skipReason(cause);
624 const skip = skips.get(reason);
625 if (skip === undefined) skips.set(reason, { count: 1, key });
626 else skip.count += 1;
627 return Effect.void;
628 }),
629 );
630 const pools = {
631 // Wider than one summary batch, so reads that finish their cache checks at different
632 // times still share batches.
633 batched: { width: 100, reads: yield* Semaphore.make(100) },
634 // One read per pull request, often a CLI process, so these keep the narrower width.
635 unbatched: { width: 25, reads: yield* Semaphore.make(25) },
636 };
637 const isPaused = (pauseKey: string, atMs: number) => (pausedUntil.get(pauseKey) ?? 0) > atMs;
638 // Judged once a slot is free, so a stack retry a sibling flags meanwhile still counts, and
639 // pauses set during the sweep apply at once.
640 const readIfDue = (
641 key: string,
642 entries: ReadonlyArray<LinkEntry>,
643 pauseKey: string,
644 reads: Semaphore.Semaphore,
645 ) =>
646 reads.withPermits(1)(
647 Effect.suspend(() =>
648 isDue(key, entries, nowMs)
649 ? Clock.currentTimeMillis.pipe(
650 Effect.flatMap((startedAtMs) =>
651 isPaused(pauseKey, startedAtMs) ? Effect.void : readGroup(key, entries, pauseKey),
652 ),
653 )
654 : Effect.void,
655 ),
656 );
657 const groupsByHost = Map.groupBy(
658 Array.from(groups).filter(([key]) => scope === "all" || requested.has(key)),
659 ([, entries]) => pauseKeyOf(entries[0]!),
660 );
661 yield* Effect.forEach(
662 groupsByHost,
663 ([pauseKey, hostGroups]) => {
664 // A host paused or with nothing due skips the lookup. Stack retries only flag groups on
665 // the flagging group's own host, so nothing on it becomes due this sweep.
666 if (
667 isPaused(pauseKey, nowMs) ||
668 !hostGroups.some(([key, entries]) => isDue(key, entries, nowMs))
669 ) {
670 return Effect.void;
671 }
672 return pullRequests.batchesReads(referenceOf(hostGroups[0]![1][0]!)).pipe(
673 Effect.flatMap((batches) => {
674 const { width, reads } = batches ? pools.batched : pools.unbatched;
675 return Effect.forEach(
676 hostGroups,
677 ([key, entries]) => readIfDue(key, entries, pauseKey, reads),
678 {
679 concurrency: width,
680 discard: true,
681 },
682 );
683 }),
684 );
685 },
686 { concurrency: "unbounded", discard: true },
687 );
688 for (const [reason, { count, key }] of skips) {
689 yield* Effect.logWarning("pull request sync skipped", { count, key, reason });
690 }
691 });
692
693 const worker = yield* makeDrainableWorker((scope: "all" | "requested") =>
694 Effect.suspend(() => {
695 if (scope === "requested") requestedSweepQueued = false;
696 return pullRequests.withSharedProjectReads(sweep(scope));
697 }).pipe(Effect.catchCause(logSkipped("pull request sync sweep failed", {}))),
698 );
699
700 // Threads that ran a close command this run, with the commands already read for.
701 const closeCommands = new Map<ThreadId, Set<string>>();
702 const refreshUnmergedLinks = (threadId: ThreadId) =>
703 projections.getThreadsWithPullRequests(threadId).pipe(
704 Effect.flatMap((threads) =>
705 Effect.forEach(
706 threads.flatMap((thread) =>
707 visibleThreadPullRequests(thread.pullRequests ?? []).filter(
708 (link) => link.snapshot?.state === "open" || link.snapshot?.state === "closed",
709 ),
710 ),
711 requestSync,
712 { discard: true },
713 ),
714 ),
715 Effect.catchCause(logSkipped("pull request refresh after run skipped", { threadId })),
716 );
717
718 // A thread's sibling syncs and relinks must not re-read a pull request its project's host
719 // could not find; the slow cadence and explicit refreshes still do.
720 const requestUnsyncedLinks = Effect.fnUntraced(
721 function* (threadId: ThreadId, links: ReadonlyArray<ThreadPullRequestLink>) {
722 const unsynced = links.filter((link) => link.snapshot === null);
723 if (unsynced.length === 0) return;
724 const [thread] =
725 missing.size === 0 ? [] : yield* projections.getThreadsWithPullRequests(threadId);
726 for (const link of unsynced) {
727 if (
728 thread !== undefined &&
729 missing.has(`${thread.projectId}\0${threadPullRequestKeyOf(link)}`)
730 ) {
731 continue;
732 }
733 yield* requestSync(link);
734 }
735 },
736 Effect.catchCause(logSkipped("pull request refresh after link update skipped", {})),
737 );
738
739 // An agent that opens a pull request from a shell gets it linked as link_pull_request would.
740 const linkCreatedChangeRequests = Effect.fn("PullRequestSyncReactor.linkCreatedChangeRequests")(
741 function* (threadId: ThreadId, item: CommandExecution) {
742 if (isOrchestrationV2WorkActive(item.status)) return;
743 yield* Effect.forEach(
744 changeRequestsCreatedBy(item.input, item.output ?? ""),
745 (created) =>
746 engine
747 .dispatch({
748 type: "thread.pull-request.link",
749 // Stable per command, so a repeated settlement cannot relink one the user unlinked.
750 commandId: CommandId.make(
751 `server:pr-created-link:${threadId}:${item.id}:${threadPullRequestKeyOf(created)}`,
752 ),
753 threadId,
754 ...created,
755 source: "created",
756 })
757 .pipe(
758 Effect.catchCause(
759 logSkipped("linking a pull request a command created failed", { threadId }),
760 ),
761 ),
762 { discard: true },
763 );
764 },
765 );
766
767 // A push can start new checks, so the thread's watches forget whether checks were running and
768 // hold the thread until they read the pull request again. The run is still going, so the
769 // thread never shows as done in between.
770 const reopenWatchedChecks = Effect.fn("PullRequestSyncReactor.reopenWatchedChecks")(
771 function* (threadId: ThreadId, item: CommandExecution) {
772 if (isOrchestrationV2WorkActive(item.status) || !isPushCommand(item.input)) return;
773 const threads = yield* projections.getThreadsWithPullRequests(threadId);
774 const answered = threads.some((thread) =>
775 (thread.pullRequests ?? []).some((link) => link.watch?.checksPending !== undefined),
776 );
777 if (!answered) return;
778 yield* engine.dispatch({
779 type: "thread.pull-request-watch.checks-reopen",
780 // Stable per command, so a repeated settlement clears an answer read since only once.
781 commandId: CommandId.make(`server:pr-push-checks:${threadId}:${item.id}`),
782 threadId,
783 });
784 },
785 (effect, threadId) =>
786 effect.pipe(
787 Effect.catchCause(
788 logSkipped("reopening a watch's checks after a push failed", { threadId }),
789 ),
790 ),
791 );
792
793 const start: PullRequestSyncReactor["Service"]["start"] = Effect.fn(
794 "PullRequestSyncReactor.start",
795 )(function* () {
796 const events = engine.streamDomainEvents;
797 const stateChanges = Stream.merge(
798 yield* pullRequests.subscribeStateChanges,
799 yield* git.subscribePullRequestStateChanges,
800 );
801 yield* forkParked(
802 Stream.runForEach(stateChanges, requestSync).pipe(
803 Effect.catchCause(logSkipped("pull request state change stream failed", {})),
804 ),
805 );
806 yield* forkParked(
807 Stream.runForEach(events, (event) => {
808 switch (event.type) {
809 case "thread.pull-request-link-updated":
810 return requestUnsyncedLinks(event.threadId, [event.payload]);
811 case "thread.pull-request-synced":
812 return requestUnsyncedLinks(
813 event.threadId,
814 visibleThreadPullRequests(event.payload.pullRequests ?? []),
815 );
816 // An agent can merge or close its pull request from a shell with its host's CLI, which
817 // sends no merge notification. Read the thread's unmerged links fresh once that command
818 // ends, while the agent usually still replies, so settlement can land before the run
819 // ends. Read again when the run ends, since `merge --auto` returns before it merges.
820 case "turn-item.updated": {
821 const item = event.payload;
822 if (item.type !== "command_execution") return Effect.void;
823 const linked = Effect.andThen(
824 linkCreatedChangeRequests(event.threadId, item),
825 reopenWatchedChecks(event.threadId, item),
826 );
827 if (!isChangeRequestCloseCommand(item.input)) return linked;
828 const ended = closeCommands.get(event.threadId) ?? new Set<string>();
829 closeCommands.set(event.threadId, ended);
830 if (isOrchestrationV2WorkActive(item.status) || ended.has(item.id)) return linked;
831 ended.add(item.id);
832 return Effect.andThen(linked, refreshUnmergedLinks(event.threadId));
833 }
834 // Settled threads are not swept, so read their links once to settle with fresh status.
835 case "thread.settled":
836 return refreshUnmergedLinks(event.threadId);
837 case "run.updated":
838 return isTerminalRunStatus(event.payload.status) && closeCommands.delete(event.threadId)
839 ? refreshUnmergedLinks(event.threadId)
840 : Effect.void;
841 default:
842 return Effect.void;
843 }
844 }).pipe(Effect.catchCause(logSkipped("pull request sync event stream failed", {}))),
845 );
846 yield* forkParked(
847 Effect.gen(function* () {
848 yield* worker.enqueue("all");
849 yield* worker.drain;
850 }).pipe(Effect.repeat(Schedule.spaced("1 minute")), Effect.asVoid),
851 );
852 });
853
854 const requestSync: PullRequestSyncReactor["Service"]["requestSync"] = (key) =>
855 Effect.suspend(() => {
856 pendingRequests.add(threadPullRequestKeyOf(key));
857 if (requestedSweepQueued) return Effect.void;
858 requestedSweepQueued = true;
859 return worker.enqueue("requested");
860 });
861
862 return { start, drain: worker.drain, requestSync } satisfies PullRequestSyncReactor["Service"];
863});
864
865export const layer = Layer.effect(PullRequestSyncReactor, make);