From 2cb1c45fae44896115db07c1996270cc5dfeaaa6 Mon Sep 17 00:00:00 2001 From: Advait Paliwal Date: Tue, 3 Nov 2026 04:43:04 -0500 Subject: [PATCH 1/2] perf(server): PR sync stops reading pull requests closed over a day ago Co-authored-by: Claude Opus 5.5 --- .../PullRequestSyncReactor.test.ts | 43 +++++++++++++++++++ .../PullRequestSyncReactor.ts | 16 ++++--- 2 files changed, 52 insertions(+), 7 deletions(-) diff --git a/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts b/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts index a1bacda0806ed..81055c9990d39 100644 --- a/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts +++ b/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts @@ -979,6 +979,49 @@ describe("PullRequestSyncReactor", () => { ), ); + it.effect("stops reading a pull request a day after it closed, until a refresh asks", () => + Effect.scoped( + Effect.gen(function* () { + yield* TestClock.setTime(Date.parse(NOW)); + const closed = (closedAt: string | null) => ({ state: "closed" as const, closedAt }); + const fixture = yield* makeHarness({ + snapshot: makeSnapshot([ + makeThread("closed", { + pullRequests: [ + makeLink(1, closed("2026-08-20T00:00:00.000Z")), + makeLink(2, closed("2026-08-26T00:00:00.000Z")), + makeLink(3, closed("2026-08-28T11:00:00.000Z")), + makeLink(4, closed(null)), + ], + }), + ]), + summary: (input) => + Effect.succeed(makeSummary(input, { state: "closed", closedAt: NOW })), + }); + const reads = Ref.get(fixture.summaryCalls).pipe( + Effect.map((calls) => calls.map((call) => call.number).toSorted()), + ); + + yield* Effect.gen(function* () { + const reactor = yield* startAndSweep(fixture); + // Closed within the day, or with no close time, keeps the fifteen-minute cadence. + assert.deepStrictEqual(yield* reads, [3, 4]); + for (let index = 0; index < 15; index += 1) yield* sweepAgain(fixture, reactor); + assert.deepStrictEqual(yield* reads, [3, 3, 4, 4]); + + yield* reactor.requestSync({ + host: "github.com", + repository: "owner/repository", + number: 1, + }); + yield* Queue.take(fixture.snapshotReads); + yield* reactor.drain; + assert.deepStrictEqual(yield* reads, [1, 3, 3, 4, 4]); + }).pipe(Effect.provide(fixture.layer)); + }), + ), + ); + it.effect("re-reads a merged pull request until the host supplies its merge time", () => Effect.scoped( Effect.gen(function* () { diff --git a/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts b/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts index 25c062301e35f..cea977b782a42 100644 --- a/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts +++ b/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts @@ -44,6 +44,9 @@ const SLOW_SYNC_INTERVAL_MS = 15 * 60 * 1_000; // requested refreshes, watches and a host CLI's merge command still read it at once. const QUIET_AFTER_MS = 30 * 60 * 1_000; const QUIET_SYNC_INTERVAL_MS = 5 * 60 * 1_000; +// A pull request closed this long ago is rarely reopened, so the sweep stops reading it, as it +// does merged ones. A reader that sees it open, or a requested refresh, still reads it at once. +const CLOSED_FINAL_AFTER_MS = 24 * 60 * 60 * 1_000; const isChangeRequestCloseCommand = (command: string) => sourceControlDefinitions.definitions.some( @@ -351,13 +354,12 @@ export const make = Effect.gen(function* () { if (entries.some((entry) => entry.link.snapshot === null)) return true; const active = entries.filter((entry) => isUnsettled(entry.thread)); // Merged snapshots without mergedAt cannot settle threads; reread slowly until supplied. - if ( - active.every( - (entry) => entry.link.snapshot?.state === "merged" && entry.link.snapshot.mergedAt != null, - ) - ) { - return false; - } + const isFinal = ({ link: { snapshot } }: LinkEntry) => + snapshot?.state === "merged" + ? snapshot.mergedAt != null + : snapshot?.state === "closed" && + nowMs - Date.parse(snapshot.closedAt ?? "") >= CLOSED_FINAL_AFTER_MS; + if (active.every(isFinal)) return false; const last = lastSyncedAt.get(key); const sinceLastMs = last === undefined ? Number.POSITIVE_INFINITY : nowMs - last; const open = active.filter((entry) => entry.link.snapshot?.state === "open"); From cbb81afc7953c57baa618362c55cfd77dfa68d97 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Tue, 3 Nov 2026 12:04:35 +0000 Subject: [PATCH 2/2] fix(server): refresh closed links after pull request commands --- .../PullRequestSyncReactor.test.ts | 81 ++++++++++--------- .../PullRequestSyncReactor.ts | 10 +-- 2 files changed, 50 insertions(+), 41 deletions(-) diff --git a/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts b/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts index 81055c9990d39..0161054781567 100644 --- a/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts +++ b/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts @@ -1622,46 +1622,55 @@ describe("PullRequestSyncReactor", () => { ), ); - it.effect("reads open links fresh as soon as a merge command ends, before its run ends", () => - Effect.scoped( - Effect.gen(function* () { - yield* TestClock.setTime(Date.parse(NOW)); - const state = yield* Ref.make("open"); - const fixture = yield* makeHarness({ - snapshot: makeSnapshot([ - makeThread("agent", { pullRequests: [makeLink(7, { state: "open" })] }), - ]), - summary: (input) => - Ref.get(state).pipe( - Effect.map((current) => - makeSummary(input, current === "merged" ? { state: current, mergedAt: NOW } : {}), + it.effect.each(["open", "closed"] as const)( + "reads %s links fresh as soon as a merge command ends, before its run ends", + (initialState) => + Effect.scoped( + Effect.gen(function* () { + yield* TestClock.setTime(Date.parse(NOW)); + const state = yield* Ref.make(initialState); + const initialSnapshot = { + state: initialState, + closedAt: initialState === "closed" ? "2026-08-20T00:00:00.000Z" : null, + }; + const fixture = yield* makeHarness({ + snapshot: makeSnapshot([ + makeThread("agent", { pullRequests: [makeLink(7, initialSnapshot)] }), + ]), + summary: (input) => + Ref.get(state).pipe( + Effect.map((current) => + makeSummary( + input, + current === "merged" ? { state: current, mergedAt: NOW } : initialSnapshot, + ), + ), ), - ), - }); + }); - yield* Effect.gen(function* () { - const reactor = yield* startAndSweep(fixture); - yield* Ref.set(state, "merged"); - yield* Ref.set(fixture.summaryCalls, []); + yield* Effect.gen(function* () { + const reactor = yield* startAndSweep(fixture); + yield* Ref.set(state, "merged"); + yield* Ref.set(fixture.summaryCalls, []); - yield* Queue.offer( - fixture.domainEvents, - commandRan(ThreadId.make("agent"), "gh pr merge 7"), - ); - yield* Queue.take(fixture.snapshotReads); - yield* Queue.take(fixture.snapshotReads); - yield* reactor.drain; + yield* Queue.offer( + fixture.domainEvents, + commandRan(ThreadId.make("agent"), "gh pr merge 7"), + ); + yield* Queue.take(fixture.snapshotReads); + yield* Queue.take(fixture.snapshotReads); + yield* reactor.drain; - assert.deepStrictEqual( - (yield* Ref.get(fixture.syncCommands)).map((command) => [ - command.number, - command.snapshot.state, - ]), - [[7, "merged"]], - ); - }).pipe(Effect.provide(fixture.layer)); - }), - ), + assert.deepStrictEqual( + (yield* Ref.get(fixture.syncCommands)).map((command) => [ + command.number, + command.snapshot.state, + ]), + [[7, "merged"]], + ); + }).pipe(Effect.provide(fixture.layer)); + }), + ), ); it.effect.each([ diff --git a/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts b/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts index cea977b782a42..9dffe4babb424 100644 --- a/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts +++ b/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts @@ -629,13 +629,13 @@ export const make = Effect.gen(function* () { // Threads that ran a close command this run, with the commands already read for. const closeCommands = new Map>(); - const refreshOpenLinks = (threadId: ThreadId) => + const refreshUnmergedLinks = (threadId: ThreadId) => projections.getThreadsWithPullRequests(threadId).pipe( Effect.flatMap((threads) => Effect.forEach( threads.flatMap((thread) => visibleThreadPullRequests(thread.pullRequests ?? []).filter( - (link) => link.snapshot?.state === "open", + (link) => link.snapshot?.state === "open" || link.snapshot?.state === "closed", ), ), requestSync, @@ -700,7 +700,7 @@ export const make = Effect.gen(function* () { { discard: true }, ); // An agent can merge or close its pull request from a shell with its host's CLI, which - // sends no merge notification. Read the thread's open links fresh once that command + // sends no merge notification. Read the thread's unmerged links fresh once that command // ends, while the agent usually still replies, so settlement can land before the run // ends. Read again when the run ends, since `merge --auto` returns before it merges. case "turn-item.updated": { @@ -712,11 +712,11 @@ export const make = Effect.gen(function* () { closeCommands.set(event.threadId, ended); if (isOrchestrationV2WorkActive(item.status) || ended.has(item.id)) return linked; ended.add(item.id); - return Effect.andThen(linked, refreshOpenLinks(event.threadId)); + return Effect.andThen(linked, refreshUnmergedLinks(event.threadId)); } case "run.updated": return isTerminalRunStatus(event.payload.status) && closeCommands.delete(event.threadId) - ? refreshOpenLinks(event.threadId) + ? refreshUnmergedLinks(event.threadId) : Effect.void; default: return Effect.void;