diff --git a/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts b/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts index 0161054781567..8b981c43375fe 100644 --- a/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts +++ b/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts @@ -41,7 +41,7 @@ import * as ServerActivation from "../serverActivation.ts"; import * as Orchestrator from "./Orchestrator.ts"; import * as ProjectionStore from "./ProjectionStore.ts"; import * as PullRequestSyncReactor from "./PullRequestSyncReactor.ts"; -import type { PullRequestTestThread } from "./testkit/pullRequestFixtures.ts"; +import { type PullRequestTestThread, v2PullRequestThread } from "./testkit/pullRequestFixtures.ts"; const NOW = "2026-08-28T12:00:00.000Z"; const PROJECT_ID = ProjectId.make("sync-project"); @@ -161,6 +161,14 @@ function makeSummary( }; } +const missingSummary = Effect.fail( + new PullRequestOperationError({ + operation: "summary", + detail: "Pull request not found", + reason: "not-found", + }), +); + interface HarnessOptions { readonly invalidate?: PullRequestService.PullRequestService["Service"]["invalidate"]; readonly snapshot: TestShellSnapshot; @@ -1262,7 +1270,11 @@ describe("PullRequestSyncReactor", () => { makeThread("settled", { settledOverride: "settled", settledAt: "2026-08-21T00:00:00.000Z", - pullRequests: [makeLink(5, { state: "open" }), makeLink(6, { state: "closed" })], + pullRequests: [ + makeLink(5, { state: "open" }), + makeLink(6, { state: "closed" }), + makeLink(7), + ], }), ]), }); @@ -1283,7 +1295,7 @@ describe("PullRequestSyncReactor", () => { yield* sweepAgain(fixture, reactor); assert.deepStrictEqual( (yield* Ref.get(fixture.summaryCalls)).map((call) => call.number).toSorted(), - [5, 6], + [5, 6, 7], ); }).pipe(Effect.provide(fixture.layer)); }), @@ -1519,6 +1531,152 @@ describe("PullRequestSyncReactor", () => { ), ); + it.effect( + "retries missing links every fifteen minutes and resumes normal reads after recovery", + () => + Effect.scoped( + Effect.gen(function* () { + yield* TestClock.setTime(Date.parse(NOW)); + const found = yield* Ref.make(false); + const fixture = yield* makeHarness({ + snapshot: makeSnapshot([ + makeThread("missing", { pullRequests: [makeLink(9), makeLink(10, {})] }), + ]), + summary: (input) => + Ref.get(found).pipe( + Effect.flatMap((available) => + available ? Effect.succeed(makeSummary(input)) : missingSummary, + ), + ), + }); + yield* Effect.gen(function* () { + const reactor = yield* startAndSweep(fixture); + assert.strictEqual((yield* Ref.get(fixture.summaryCalls)).length, 2); + for (let index = 0; index < 14; index += 1) yield* sweepAgain(fixture, reactor); + assert.strictEqual((yield* Ref.get(fixture.summaryCalls)).length, 2); + yield* sweepAgain(fixture, reactor); + assert.strictEqual((yield* Ref.get(fixture.summaryCalls)).length, 4); + assert.deepStrictEqual(yield* Ref.get(fixture.syncCommands), []); + + yield* Ref.set(found, true); + for (let index = 0; index < 14; index += 1) yield* sweepAgain(fixture, reactor); + assert.strictEqual((yield* Ref.get(fixture.summaryCalls)).length, 4); + yield* sweepAgain(fixture, reactor); + const commands = yield* Ref.get(fixture.syncCommands); + assert.deepStrictEqual( + commands.map((command) => command.number), + [9], + ); + yield* Ref.update(fixture.snapshots, (snapshot) => applySync(snapshot, commands)); + // Recovered quiet open pull requests return to the five-minute cadence. + for (let index = 0; index < 5; index += 1) yield* sweepAgain(fixture, reactor); + assert.deepStrictEqual( + (yield* Ref.get(fixture.summaryCalls)) + .map((call) => call.number) + .toSorted((left, right) => left - right), + [9, 9, 9, 9, 10, 10, 10, 10], + ); + }).pipe(Effect.provide(fixture.layer)); + }), + ), + ); + + it.effect( + "holds missing links through sibling and link updates but honors a reader refresh", + () => + Effect.scoped( + Effect.gen(function* () { + yield* TestClock.setTime(Date.parse(NOW)); + const fixture = yield* makeHarness({ + snapshot: makeSnapshot([makeThread("missing", { pullRequests: [makeLink(9)] })]), + summary: (input) => + input.number === 9 ? missingSummary : Effect.succeed(makeSummary(input)), + }); + const missingReads = Ref.get(fixture.summaryCalls).pipe( + Effect.map((calls) => calls.filter((call) => call.number === 9).length), + ); + yield* Effect.gen(function* () { + const reactor = yield* startAndSweep(fixture); + const linked = makeThread("missing", { pullRequests: [makeLink(9), makeLink(8)] }); + yield* Ref.set(fixture.snapshots, makeSnapshot([linked])); + yield* Queue.offer(fixture.domainEvents, { + type: "thread.pull-request-synced", + id: EventId.make("sibling-synced"), + threadId: linked.id, + occurredAt: DateTime.makeUnsafe(NOW), + payload: { ...v2PullRequestThread(linked), lastVisitedAt: null }, + }); + yield* Queue.take(fixture.snapshotReads); + yield* reactor.drain; + const afterSiblingSync = yield* missingReads; + assert.deepStrictEqual( + (yield* Ref.get(fixture.syncCommands)).map((command) => command.number), + [8], + ); + + yield* Queue.offer(fixture.domainEvents, { + type: "thread.pull-request-link-updated", + id: EventId.make("missing-link-updated"), + threadId: linked.id, + occurredAt: DateTime.makeUnsafe(NOW), + payload: makeLink(9), + }); + // Let the event consumer finish even when it correctly queues no sweep. + yield* TestClock.adjust(0); + yield* reactor.drain; + const afterLinkUpdate = yield* missingReads; + + yield* Queue.offer(fixture.stateChanges, makeLink(9)); + yield* TestClock.adjust(0); + yield* reactor.drain; + const afterReaderRefresh = yield* missingReads; + yield* sweepAgain(fixture, reactor); + assert.deepStrictEqual( + { + afterSiblingSync, + afterLinkUpdate, + afterReaderRefresh, + afterSweep: yield* missingReads, + }, + { afterSiblingSync: 1, afterLinkUpdate: 1, afterReaderRefresh: 2, afterSweep: 2 }, + ); + }).pipe(Effect.provide(fixture.layer)); + }), + ), + ); + + it.effect("a pending stack retry respects the missing summary backoff", () => + Effect.scoped( + Effect.gen(function* () { + yield* TestClock.setTime(Date.parse(NOW)); + const found = yield* Ref.make(true); + const fixture = yield* makeHarness({ + snapshot: makeSnapshot([makeThread("missing", { pullRequests: [makeLink(4)] })]), + summary: (input) => + Ref.get(found).pipe( + Effect.flatMap((available) => + available ? Effect.succeed(makeSummary(input)) : missingSummary, + ), + ), + stack: () => + Effect.fail(new PullRequestOperationError({ operation: "stack", detail: "host down" })), + }); + yield* Effect.gen(function* () { + const reactor = yield* startAndSweep(fixture); + assert.strictEqual((yield* Ref.get(fixture.stackCalls)).length, 1); + yield* Ref.set(found, false); + yield* sweepAgain(fixture, reactor); + assert.strictEqual((yield* Ref.get(fixture.summaryCalls)).length, 2); + for (let index = 0; index < 14; index += 1) yield* sweepAgain(fixture, reactor); + assert.strictEqual((yield* Ref.get(fixture.summaryCalls)).length, 2); + yield* sweepAgain(fixture, reactor); + assert.strictEqual((yield* Ref.get(fixture.summaryCalls)).length, 3); + assert.strictEqual((yield* Ref.get(fixture.stackCalls)).length, 1); + }).pipe(Effect.provide(fixture.layer)); + }), + ), + ); + it.effect.each([ "gh pr merge 7 --squash 2>&1 | tail -3", "cafe pr merge 7 --squash", diff --git a/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts b/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts index 9dffe4babb424..1cebfc2619ad8 100644 --- a/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts +++ b/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts @@ -346,13 +346,19 @@ export const make = Effect.gen(function* () { // linking dozens of pull requests) is read together and shares the summary batches. let requestedSweepQueued = false; const retryStacks = new Set(); + // Group keys the host answered not-found for; they are read on the slow cadence until found. + const missing = new Set(); // Pause per project and host because credentials are project-specific; leave paused PRs due until reset. const pausedUntil = new Map(); const isDue = (key: string, entries: ReadonlyArray, nowMs: number): boolean => { - if (requested.has(key) || retryStacks.has(key)) return true; - if (entries.some((entry) => entry.link.snapshot === null)) return true; + if (requested.has(key)) return true; const active = entries.filter((entry) => isUnsettled(entry.thread)); + if (active.length === 0) return false; + const last = lastSyncedAt.get(key); + const sinceLastMs = last === undefined ? Number.POSITIVE_INFINITY : nowMs - last; + if (missing.has(key)) return sinceLastMs >= SLOW_SYNC_INTERVAL_MS; + if (retryStacks.has(key) || active.some((entry) => entry.link.snapshot === null)) return true; // Merged snapshots without mergedAt cannot settle threads; reread slowly until supplied. const isFinal = ({ link: { snapshot } }: LinkEntry) => snapshot?.state === "merged" @@ -360,8 +366,6 @@ export const make = Effect.gen(function* () { : 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"); if (open.length > 0) { const quiet = open.every((entry) => { @@ -403,6 +407,7 @@ export const make = Effect.gen(function* () { for (const key of lastSyncedAt.keys()) if (!groups.has(key)) lastSyncedAt.delete(key); for (const key of retryStacks) if (!groups.has(key)) retryStacks.delete(key); + for (const key of missing) if (!groups.has(key)) missing.delete(key); for (const key of requested.keys()) if (!groups.has(key)) requested.delete(key); const linkedThisSweep = new Set(); @@ -488,7 +493,19 @@ export const make = Effect.gen(function* () { // A stack retry reads past the cache, or it could get back the layer list it is retrying. const invalidated = refreshRequested || missingMergedAt || retryStacks.has(key); if (invalidated) yield* pullRequests.invalidate({ reference: ref }); - const summary = yield* pullRequests.summary(ref, { recoverTransientFailure: false }); + const summary = yield* pullRequests.summary(ref, { recoverTransientFailure: false }).pipe( + Effect.tapError((error) => + error.reason === "not-found" + ? Effect.sync(() => { + // Not-found is a host answer: it consumes the refresh and starts the backoff. + missing.add(key); + lastSyncedAt.set(key, nowMs); + requested.delete(key); + }) + : Effect.void, + ), + ); + missing.delete(key); const fields = snapshotFieldsOf(summary); const needsStack = refreshRequested || @@ -645,6 +662,27 @@ export const make = Effect.gen(function* () { Effect.catchCause(logSkipped("pull request refresh after run skipped", { threadId })), ); + // A thread's sibling syncs and relinks must not re-read a pull request its project's host + // could not find; the slow cadence and explicit refreshes still do. + const requestUnsyncedLinks = Effect.fnUntraced( + function* (threadId: ThreadId, links: ReadonlyArray) { + const unsynced = links.filter((link) => link.snapshot === null); + if (unsynced.length === 0) return; + const [thread] = + missing.size === 0 ? [] : yield* projections.getThreadsWithPullRequests(threadId); + for (const link of unsynced) { + if ( + thread !== undefined && + missing.has(`${thread.projectId}\0${threadPullRequestKeyOf(link)}`) + ) { + continue; + } + yield* requestSync(link); + } + }, + Effect.catchCause(logSkipped("pull request refresh after link update skipped", {})), + ); + // An agent that opens a pull request from a shell gets it linked as link_pull_request would. const linkCreatedChangeRequests = Effect.fn("PullRequestSyncReactor.linkCreatedChangeRequests")( function* (threadId: ThreadId, item: CommandExecution) { @@ -690,14 +728,11 @@ export const make = Effect.gen(function* () { Stream.runForEach(events, (event) => { switch (event.type) { case "thread.pull-request-link-updated": - return event.payload.snapshot === null ? requestSync(event.payload) : Effect.void; + return requestUnsyncedLinks(event.threadId, [event.payload]); case "thread.pull-request-synced": - return Effect.forEach( - visibleThreadPullRequests(event.payload.pullRequests ?? []).filter( - (link) => link.snapshot === null, - ), - requestSync, - { discard: true }, + return requestUnsyncedLinks( + event.threadId, + visibleThreadPullRequests(event.payload.pullRequests ?? []), ); // 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 unmerged links fresh once that command