authorgravatar for 130890337+eimexdev@users.noreply.github.comeimexdev <130890337+eimexdev@users.noreply.github.com> 2026-10-03 22:44:54-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-10-04 02:44:54-03:00
logf4f3abf71e98361918e360a89d48780cc65cafd1
treeaccb387096d42e4b4a0293aecaba99043ad0ff54
parentdaa1d0ed947c276cb74b08025b5903e2ffe31278
signature Signed by PGP key B5690EEEBB952194

fix(server): read paginated review replies when watching PRs (#15427)

Co-authored-by: maria-rcks <254055478+maria-rcks@users.noreply.github.com>

11 files changed, 223 insertions(+), 35 deletions(-)

apps/server/src/orchestration-v2/PullRequestWatchReactor.ts+68-7
......@@ -2,6 +2,10 @@ import {
22 CommandId,
33 MessageId,
44 type OrchestrationV2Notification,
5 type PullRequestActivity,
6 type PullRequestComment,
7 type PullRequestRef,
8 type PullRequestThreadCommentsResult,
59 type ThreadPullRequestLink,
610 type ThreadPullRequestWatch,
711} from "@t3tools/contracts";
......@@ -81,6 +85,12 @@ export const make = Effect.gen(function* () {
8185
8286 // Passes in a row that failed, per watch. Kept in memory: a restart only delays the stop.
8387 const readFailures = new Map<string, number>();
88 // Replies past each long thread's first page, per watch, so a pass pages a thread again only
89 // when the host's count of it moves. Kept in memory: a restart pages each thread once more.
90 const threadTails = new Map<
91 string,
92 Map<string, { readonly count: number; readonly comments: ReadonlyArray<PullRequestComment> }>
93 >();
8494
8595 // Host-level identity, with the repository as linked, the way pull request sync reads it.
8696 const identityOf = (link: ThreadPullRequestLink) => ({
......@@ -125,6 +135,60 @@ export const make = Effect.gen(function* () {
125135 },
126136 }).pipe(Effect.catch(() => record(target, null)));
127137
138 const readRemarks = Effect.fn("PullRequestWatchReactor.readRemarks")(
139 function* (target: WatchTarget, reference: PullRequestRef, activity: PullRequestActivity) {
140 // Comment cursors cannot account for missing threads. Only finish a truncated read
141 // when the host confirms that every thread was listed.
142 if (activity.commentsTruncated && activity.reviewThreadsTruncated !== false) return null;
143
144 const key = failureKey(target);
145 let tails = threadTails.get(key);
146 if (tails === undefined) {
147 tails = new Map();
148 threadTails.set(key, tails);
149 }
150 const remarks = [...activity.comments];
151 for (const thread of activity.reviewThreads) {
152 let cursor = thread.nextCommentsCursor ?? null;
153 if (cursor === null) continue;
154 const count = thread.commentCount ?? 0;
155 let tail = tails.get(thread.id);
156 if (tail?.count !== count) {
157 const comments = new Map<string, PullRequestComment>();
158 const cursors = new Set<string>();
159 while (cursor !== null) {
160 if (cursors.has(cursor)) return null;
161 cursors.add(cursor);
162 const page: PullRequestThreadCommentsResult = yield* pullRequests.threadComments({
163 ...reference,
164 threadId: thread.id,
165 cursor,
166 });
167 for (const comment of page.comments) {
168 comments.set(comment.id, {
169 ...comment,
170 kind: "review-comment",
171 path: thread.path,
172 reviewState: null,
173 });
174 }
175 cursor = page.nextCursor;
176 }
177 if (thread.comments.length + comments.size < count) return null;
178 tail = { count, comments: [...comments.values()] };
179 tails.set(thread.id, tail);
180 }
181 remarks.push(...tail.comments);
182 }
183 return remarks.sort((left, right) => left.createdAt.localeCompare(right.createdAt));
184 },
185 Effect.catch((error) =>
186 Effect.logWarning("pull request watch comment pagination failed", { error }).pipe(
187 Effect.as(null),
188 ),
189 ),
190 );
191
128192 const check = Effect.fn("PullRequestWatchReactor.check")(function* (target: WatchTarget) {
129193 const { thread, link, watch } = target;
130194 const pullRequest = identityOf(link);
......@@ -160,13 +224,9 @@ export const make = Effect.gen(function* () {
160224 const [detail, activity] = read.value;
161225 if (detail.state !== "open") return yield* record(target, null);
162226
163 // A degraded read (GitHub's review thread query failed) is truncated with no long thread to
164 // explain it, and would skip review comments, so remarks wait for a later pass. Replies past
165 // the first ten of a long review thread are not read.
166 const degraded =
167 activity.commentsTruncated &&
168 !activity.reviewThreads.some((reviewThread) => reviewThread.nextCommentsCursor !== undefined);
169 const report = evaluatePullRequestWatch(watch, detail, degraded ? null : activity.comments);
227 // Never advance the remark watermark past comments an incomplete read could have missed.
228 const remarks = yield* readRemarks(target, reference, activity);
229 const report = evaluatePullRequestWatch(watch, detail, remarks);
170230 if (report.changes.length > 0) {
171231 return yield* record(
172232 target,
......@@ -192,6 +252,7 @@ export const make = Effect.gen(function* () {
192252 );
193253 const keys = new Set(targets.map(failureKey));
194254 for (const key of readFailures.keys()) if (!keys.has(key)) readFailures.delete(key);
255 for (const key of threadTails.keys()) if (!keys.has(key)) threadTails.delete(key);
195256 yield* Effect.forEach(
196257 targets,
197258 (target) =>
apps/server/src/orchestration-v2/runtimeLayer.test.ts+124-23
......@@ -18,6 +18,8 @@ import {
1818 type OrchestrationV2Run,
1919 ProjectId,
2020 type PullRequestDetail,
21 type PullRequestComment,
22 PullRequestOperationError,
2123 ProviderDriverKind,
2224 ProviderInstanceId,
2325 ProviderThreadId,
......@@ -2307,11 +2309,18 @@ it.layer(TestLayer)("OrchestrationV2LayerLive lifecycle", (it) => {
23072309 }),
23082310 );
23092311
2310 it.effect("wakes a watched thread once for failed checks and a review comment", () =>
2312 it.effect.each([
2313 "single page",
2314 "paginated",
2315 "page failure",
2316 "repeated cursor",
2317 "missing comments",
2318 "thread list truncated",
2319 ])("wakes a watched thread once: %s", (mode) =>
23112320 Effect.gen(function* () {
23122321 const orchestrator = yield* Orchestrator.OrchestratorV2;
2313 const threadId = ThreadId.make("runtime-pull-request-watch-wake");
2314 const projectId = ProjectId.make("pr-watch-wake-project");
2322 const threadId = ThreadId.make(`runtime-pull-request-watch-wake-${mode}`);
2323 const projectId = ProjectId.make(`pr-watch-wake-project-${mode}`);
23152324 yield* seedProject({
23162325 projectId,
23172326 title: "Watch wake",
......@@ -2323,7 +2332,7 @@ it.layer(TestLayer)("OrchestrationV2LayerLive lifecycle", (it) => {
23232332 type: "thread.create",
23242333 createdBy: "user",
23252334 creationSource: "web",
2326 commandId: CommandId.make("pr-watch-wake-create"),
2335 commandId: CommandId.make(`pr-watch-wake-create-${mode}`),
23272336 threadId,
23282337 projectId,
23292338 title: "Watch wake",
......@@ -2337,7 +2346,7 @@ it.layer(TestLayer)("OrchestrationV2LayerLive lifecycle", (it) => {
23372346 const url = "https://github.com/pingdotgg/t3code/pull/7";
23382347 yield* orchestrator.dispatch({
23392348 type: "thread.pull-request.link",
2340 commandId: CommandId.make("pr-watch-wake-link"),
2349 commandId: CommandId.make(`pr-watch-wake-link-${mode}`),
23412350 threadId,
23422351 ...key,
23432352 url,
......@@ -2345,7 +2354,7 @@ it.layer(TestLayer)("OrchestrationV2LayerLive lifecycle", (it) => {
23452354 });
23462355 yield* orchestrator.dispatch({
23472356 type: "thread.pull-request.watch",
2348 commandId: CommandId.make("pr-watch-wake-start"),
2357 commandId: CommandId.make(`pr-watch-wake-start-${mode}`),
23492358 threadId,
23502359 ...key,
23512360 watching: true,
......@@ -2398,6 +2407,38 @@ it.layer(TestLayer)("OrchestrationV2LayerLive lifecycle", (it) => {
23982407 mergeCapabilities: { merge: true, squash: true, rebase: true },
23992408 viewer: "agent-user",
24002409 };
2410 const initialWatch = (yield* orchestrator.getThreadShell(threadId))?.pullRequests?.[0]?.watch;
2411 assert.isDefined(initialWatch);
2412 const remark: PullRequestComment = {
2413 id: "review-1",
2414 kind: "review-comment",
2415 author: { login: "reviewer", name: null, avatarUrl: null },
2416 body: "One more thing.",
2417 createdAt: "2999-01-01T00:00:03.000Z",
2418 url: null,
2419 path: "src/index.ts",
2420 reviewState: null,
2421 };
2422 const firstTen = Array.from({ length: 10 }, (_, index) => ({
2423 ...remark,
2424 id: `old-${index}`,
2425 createdAt: "1900-01-01T00:00:00.000Z",
2426 }));
2427 const eleventh = {
2428 ...remark,
2429 id: "reply-11",
2430 body: "Eleventh reply.",
2431 createdAt: "2999-01-01T00:00:01.000Z",
2432 };
2433 const twelfth = {
2434 ...remark,
2435 id: "reply-12",
2436 body: "Twelfth reply.",
2437 createdAt: "2999-01-01T00:00:02.000Z",
2438 };
2439 const incomplete = mode !== "single page" && mode !== "paginated";
2440 let recovering = false;
2441 let pagesRead = 0;
24012442 const reactor = yield* PullRequestWatchReactor.make.pipe(
24022443 Effect.provide(
24032444 Layer.mergeAll(
......@@ -2406,41 +2447,101 @@ it.layer(TestLayer)("OrchestrationV2LayerLive lifecycle", (it) => {
24062447 detail: () => Effect.succeed(detail),
24072448 activity: () =>
24082449 Effect.succeed({
2409 comments: [
2410 {
2411 id: "review-1",
2412 kind: "review-comment",
2413 author: { login: "reviewer", name: null, avatarUrl: null },
2414 body: "One more thing.",
2415 createdAt: "2999-01-01T00:00:00.000Z",
2416 url: null,
2417 path: "src/index.ts",
2418 reviewState: null,
2419 },
2420 ],
2421 commentCount: 1,
2422 commentsTruncated: false,
2423 reviewThreads: [],
2450 comments:
2451 mode === "single page"
2452 ? [remark]
2453 : [...firstTen, { ...remark, kind: "issue-comment" }],
2454 commentCount: mode === "single page" ? 1 : 13,
2455 commentsTruncated: mode !== "single page",
2456 reviewThreadsTruncated: mode === "thread list truncated" && !recovering,
2457 reviewThreads:
2458 mode === "single page"
2459 ? []
2460 : [
2461 {
2462 id: "review-thread",
2463 path: "src/index.ts",
2464 line: 1,
2465 side: "right",
2466 isResolved: false,
2467 isOutdated: false,
2468 comments: firstTen,
2469 commentCount: 12,
2470 nextCommentsCursor: "after-10",
2471 },
2472 ],
24242473 commits: [],
24252474 }),
2475 threadComments: (input) => {
2476 pagesRead += 1;
2477 assert.equal(input.threadId, "review-thread");
2478 if (input.cursor === "after-10") {
2479 return Effect.succeed({ comments: [eleventh], nextCursor: "after-11" });
2480 }
2481 assert.equal(input.cursor, "after-11");
2482 if (!recovering && mode === "page failure") {
2483 return Effect.fail(
2484 new PullRequestOperationError({
2485 operation: "threadComments",
2486 detail: "Page unavailable",
2487 }),
2488 );
2489 }
2490 if (!recovering && mode === "repeated cursor") {
2491 return Effect.succeed({ comments: [eleventh], nextCursor: "after-11" });
2492 }
2493 if (!recovering && mode === "missing comments") {
2494 return Effect.succeed({ comments: [], nextCursor: null });
2495 }
2496 // Overlapping pages must not report the same reply twice.
2497 return Effect.succeed({ comments: [eleventh, twelfth], nextCursor: null });
2498 },
24262499 }),
24272500 ),
24282501 ),
24292502 );
2503 if (incomplete) {
2504 yield* reactor.sweep;
2505 const held = (yield* orchestrator.getThreadShell(threadId))?.pullRequests?.[0]?.watch;
2506 assert.equal(held?.remarksThrough, initialWatch?.remarksThrough);
2507 assert.deepEqual(held?.remarkIds, initialWatch?.remarkIds);
2508 assert.deepEqual(held?.failedChecks, ["lint"]);
2509 const { messages } = yield* orchestrator.getThreadRecords(threadId, ["messages"]);
2510 assert.deepEqual(
2511 messages.map((message) => message.notification?.summary),
2512 ["#7: checks failed"],
2513 );
2514 // Keep the two notifications ordered independently of their random message IDs.
2515 yield* TestClock.adjust("1 millis");
2516 recovering = true;
2517 }
24302518 yield* reactor.sweep;
2519 // A thread whose count has not moved is not paged again.
2520 const pagesBefore = pagesRead;
24312521 yield* reactor.sweep;
2522 assert.equal(pagesRead, pagesBefore);
24322523
24332524 const { messages } = yield* orchestrator.getThreadRecords(threadId, ["messages"]);
24342525 assert.deepEqual(
24352526 messages.flatMap((message) =>
24362527 message.notification === undefined ? [] : [message.notification.summary],
24372528 ),
2438 ["#7: checks failed, new comments"],
2529 incomplete
2530 ? ["#7: checks failed", "#7: new comments"]
2531 : ["#7: checks failed, new comments"],
24392532 );
2533 if (mode !== "single page") {
2534 const wake = messages.at(-1);
2535 assert.include(wake?.text ?? "", "3 new comments");
2536 assert.include(wake?.text ?? "", "Eleventh reply.");
2537 assert.include(wake?.text ?? "", "Twelfth reply.");
2538 }
24402539 const watch = (yield* orchestrator.getThreadShell(threadId))?.pullRequests?.[0]?.watch;
2540 assert.equal(watch?.remarksThrough, remark.createdAt);
2541 assert.deepEqual(watch?.remarkIds, [remark.id]);
24412542 assert.deepEqual(
24422543 { headSha: watch?.headSha, failedChecks: watch?.failedChecks, wakes: watch?.wakes },
2443 { headSha: "abc1234def", failedChecks: ["lint"], wakes: 0 },
2544 { headSha: "abc1234def", failedChecks: ["lint"], wakes: incomplete ? 1 : 0 },
24442545 );
24452546 }),
24462547 );
apps/server/src/pullRequest/GitHubPullRequestCli.test.ts+10-1
......@@ -3612,7 +3612,14 @@ layer("GitHubPullRequestCli.layer", (it) => {
36123612 Effect.gen(function* () {
36133613 // A host that never runs out of pages: the walk has to end itself.
36143614 mockedExecute.mockReturnValue(
3615 Effect.succeed(output(reviewThreadsPage([thread("PRRT_1", "c1")], "Y3Vyc29yOjE"))),
3615 Effect.succeed(
3616 output(
3617 reviewThreadsPage(
3618 [{ ...thread("PRRT_1", "c1"), comments: threadComments(["c1"], "Y3Vyc29yOjI", 3) }],
3619 "Y3Vyc29yOjE",
3620 ),
3621 ),
3622 ),
36163623 );
36173624 const cli = yield* GitHubPullRequestCli.GitHubPullRequestCli;
36183625
......@@ -3625,6 +3632,7 @@ layer("GitHubPullRequestCli.layer", (it) => {
36253632
36263633 assert.strictEqual(mockedExecute.mock.calls.length, 10);
36273634 assert.isTrue(conversation.truncated);
3635 assert.isTrue(conversation.reviewThreadsTruncated);
36283636 }),
36293637 );
36303638
......@@ -3656,6 +3664,7 @@ layer("GitHubPullRequestCli.layer", (it) => {
36563664 nextCommentsCursor: "Y3Vyc29yOjI",
36573665 });
36583666 assert.isTrue(conversation.truncated);
3667 assert.isFalse(conversation.reviewThreadsTruncated);
36593668 }),
36603669 );
36613670
apps/server/src/pullRequest/GitHubPullRequestCli.ts+1
......@@ -2409,6 +2409,7 @@ export const make = Effect.gen(function* () {
24092409 // where a bound kept some of the words on GitHub.
24102410 commentCount: entries.reduce((total, entry) => total + entry.commentCount, 0),
24112411 truncated: cursor !== null || entries.some((entry) => entry.nextCommentCursor !== null),
2412 reviewThreadsTruncated: cursor !== null,
24122413 reactions,
24132414 reactionsById,
24142415 reviewers,
apps/server/src/pullRequest/GitHubPullRequestProvider.test.ts+2
......@@ -882,6 +882,7 @@ describe("getChangeRequest commits", () => {
882882 reviewThreads: [],
883883 commentCount: 0,
884884 truncated: false,
885 reviewThreadsTruncated: false,
885886 reactions: [],
886887 reactionsById: new Map<string, ReadonlyArray<PullRequestReaction>>(),
887888 reviewers: [],
......@@ -967,6 +968,7 @@ describe("getChangeRequestActivity dismissed reviews", () => {
967968 reviewThreads: [],
968969 commentCount: 0,
969970 truncated: false,
971 reviewThreadsTruncated: false,
970972 reactions: [],
971973 reactionsById: new Map(),
972974 reviewers: [],
apps/server/src/pullRequest/GitHubPullRequestProvider.ts+2
......@@ -412,6 +412,7 @@ export const make = Effect.gen(function* () {
412412 reviewThreads: [],
413413 commentCount: 0,
414414 truncated: true,
415 reviewThreadsTruncated: true,
415416 reviewers: [],
416417 avatarsByLogin: new Map<string, string>(),
417418 botLogins: new Set<string>(),
......@@ -480,6 +481,7 @@ export const make = Effect.gen(function* () {
480481 // are always whole and only the thread walk can stop short of the host.
481482 commentCount: pullRequest.comments.length + reviewThreads.commentCount,
482483 commentsTruncated: reviewThreads.truncated,
484 reviewThreadsTruncated: reviewThreads.reviewThreadsTruncated,
483485 reviewThreads: reviewThreads.reviewThreads.map((thread) => ({
484486 ...thread,
485487 comments: thread.comments.map((comment) => ({
apps/server/src/pullRequest/PullRequestProvider.ts+1
......@@ -254,6 +254,7 @@ export interface ProviderChangeRequestActivity {
254254 */
255255 readonly commentCount: number;
256256 readonly commentsTruncated: boolean;
257 readonly reviewThreadsTruncated?: boolean;
257258 readonly reviewThreads: ReadonlyArray<PullRequestReviewThread>;
258259 readonly commits: ReadonlyArray<PullRequestCommit>;
259260 /** The change request's own reactions, from a host that has them. */
apps/server/src/pullRequest/PullRequestService.test.ts+9-4
......@@ -4318,7 +4318,8 @@ it.effect(
43184318 return Effect.succeed({
43194319 comments: [],
43204320 commentCount: 0,
4321 commentsTruncated: false,
4321 commentsTruncated: true,
4322 reviewThreadsTruncated: true,
43224323 reviewThreads: [],
43234324 commits: [],
43244325 });
......@@ -4342,10 +4343,14 @@ it.effect(
43424343 },
43434344 ]);
43444345
4345 yield* Effect.all([service.activity(reference), service.activity(reference)], {
4346 concurrency: 2,
4347 });
4346 const activities = yield* Effect.all(
4347 [service.activity(reference), service.activity(reference)],
4348 {
4349 concurrency: 2,
4350 },
4351 );
43484352 assert.strictEqual(activityCalls, 1);
4353 assert.isTrue(activities.every((activity) => activity.reviewThreadsTruncated === true));
43494354
43504355 yield* service.invalidate({ reference });
43514356 yield* service.activity(reference);
apps/server/src/pullRequest/PullRequestService.ts+3
......@@ -1783,6 +1783,9 @@ export const make = Effect.gen(function* () {
17831783 comments: activity.comments,
17841784 commentCount: activity.commentCount,
17851785 commentsTruncated: activity.commentsTruncated,
1786 ...(activity.reviewThreadsTruncated === undefined
1787 ? {}
1788 : { reviewThreadsTruncated: activity.reviewThreadsTruncated }),
17861789 reviewThreads: activity.reviewThreads,
17871790 commits: activity.commits,
17881791 ...(activity.reactions === undefined ? {} : { reactions: activity.reactions }),
apps/server/src/pullRequest/gitHubPullRequestJson.ts+1
......@@ -2093,6 +2093,7 @@ export interface GitHubReviewThreadComments {
20932093 /** The host's own count of the conversation, which a bounded read can fall short of. */
20942094 readonly commentCount: number;
20952095 readonly truncated: boolean;
2096 readonly reviewThreadsTruncated: boolean;
20962097 /** The pull request's own reactions, which sit on its description. */
20972098 readonly reactions: ReadonlyArray<PullRequestReaction>;
20982099 /** Reactions by node id, for the comments and reviews the `gh` JSON read carries no reaction on. */
packages/contracts/src/pullRequest.ts+2
......@@ -916,6 +916,8 @@ export const PullRequestActivity = Schema.Struct({
916916 * however long it is.
917917 */
918918 commentsTruncated: Schema.Boolean,
919 /** Whether the thread listing itself is incomplete, apart from pages within a thread. */
920 reviewThreadsTruncated: Schema.optional(Schema.Boolean),
919921 reviewThreads: Schema.Array(PullRequestReviewThread),
920922 commits: Schema.Array(PullRequestCommit),
921923 /**