diff --git a/apps/server/src/git/GitManager.test.ts b/apps/server/src/git/GitManager.test.ts index 730f1bd0a3519cfd5cb42496d54623ba4d72d789..dfd4bed6dea732f48958757493f85904caf90734 100644 --- a/apps/server/src/git/GitManager.test.ts +++ b/apps/server/src/git/GitManager.test.ts @@ -2661,7 +2661,7 @@ it.layer(GitManagerTestLayer)("GitManager", (it) => { providerOperation: "listChangeRequests", providerCommand: "gh", errorDetail: - "GitHub API rate limit exceeded. Run `gh api rate_limit` to inspect the quota and reset time.", + "GitHub API rate limit exceeded. For the GraphQL quota and reset time, run `gh api graphql -f query='{rateLimit{remaining resetAt}}'`; `gh api rate_limit` reports REST.", }); const loggedText = [ warning?.message ?? "", diff --git a/apps/server/src/orchestration-v2/PullRequestWatchReactor.ts b/apps/server/src/orchestration-v2/PullRequestWatchReactor.ts index db7e99fc3b4ba77d85e7aadec4959b4b83f27ce5..7005534f89edee4aa792019bf50dde60896be58e 100644 --- a/apps/server/src/orchestration-v2/PullRequestWatchReactor.ts +++ b/apps/server/src/orchestration-v2/PullRequestWatchReactor.ts @@ -26,7 +26,10 @@ import * as Schedule from "effect/Schedule"; import * as Schema from "effect/Schema"; import type * as Scope from "effect/Scope"; -import { PullRequestProviderError } from "../pullRequest/PullRequestProvider.ts"; +import { + type ProviderChangeRequestWatchFingerprint, + PullRequestProviderError, +} from "../pullRequest/PullRequestProvider.ts"; import * as PullRequestService from "../pullRequest/PullRequestService.ts"; import { forkParked } from "../serverActivation.ts"; import * as Orchestrator from "./Orchestrator.ts"; @@ -41,10 +44,16 @@ const SWEEP_MINUTES = 2; /** Reads in a row that failed for a reason other than a rate limit before the watch ends. */ const READ_FAILURE_LIMIT = 8; /** - * A pull request with nothing in flight is read again only when its sync snapshot moves, or - * after this long, for news the snapshot cannot show, such as a bot editing its review. + * Without a host fingerprint, a pull request with nothing in flight is read again only when its + * sync snapshot moves, or after this long, for news the snapshot cannot show, such as a bot + * editing its review. */ const QUIET_REREAD_MS = 10 * 60_000; +/** + * With a host fingerprint, how long until the activity is read again anyway: the fingerprint + * does not see edits to comments inside review threads. + */ +const FINGERPRINT_REREAD_MS = 30 * 60_000; const isProviderError = Schema.is(PullRequestProviderError); @@ -81,12 +90,25 @@ interface WatchGroup { /** The last successful read of a pull request, kept in memory: a restart reads each once. */ interface LastRead { + /** When the activity was last read. */ readonly at: number; + /** The sync snapshot this read saw. */ readonly fingerprint: string; - /** Nothing is in flight or unread, so only a moved snapshot or the reread brings news. */ - readonly quiet: boolean; + /** A check was still running or mergeability unknown, so the detail can move unannounced. */ + readonly inFlight: boolean; + /** The last activity read listed every remark; an incomplete one is read again. */ + readonly remarksComplete: boolean; /** The watches this read evaluated; a watch started since takes its first look next pass. */ readonly watches: ReadonlySet; + /** The host fingerprint parts these reads answer, when the host gives one. */ + readonly status: string | null; + readonly remarks: string | null; +} + +/** Which reads a pass makes for one pull request. */ +interface ReadPlan { + readonly detail: boolean; + readonly activity: boolean; } const watchKey = ({ thread, watch }: WatchTarget) => `${thread.id} ${watch.startedAt}`; @@ -143,14 +165,35 @@ const snapshotFingerprint = ({ link }: WatchTarget) => { ].join(" "); }; -function needsRead(group: WatchGroup, last: LastRead | undefined, now: number): boolean { - return ( - last === undefined || - !last.quiet || - last.fingerprint !== group.fingerprint || - now - last.at >= QUIET_REREAD_MS || - group.targets.some((target) => !last.watches.has(watchKey(target))) - ); +/** + * With a host fingerprint, the detail (1 point on GitHub) is read when anything moved or a check + * is still running, since check counts by state cannot tell which check finished, and the + * activity (15 points) only when the remarks moved. Without one, both are read whenever the sync + * snapshot moved or something is in flight. + */ +function planRead( + group: WatchGroup, + last: LastRead | undefined, + fingerprint: ProviderChangeRequestWatchFingerprint | null, + now: number, +): ReadPlan { + // A watch takes its first look with everything. + if (last === undefined || group.targets.some((target) => !last.watches.has(watchKey(target)))) { + return { detail: true, activity: true }; + } + if (fingerprint === null) { + const read = + last.inFlight || + !last.remarksComplete || + last.fingerprint !== group.fingerprint || + now - last.at >= QUIET_REREAD_MS; + return { detail: read, activity: read }; + } + const activity = + last.remarks !== fingerprint.remarks || + !last.remarksComplete || + now - last.at >= FINGERPRINT_REREAD_MS; + return { detail: activity || last.status !== fingerprint.status || last.inFlight, activity }; } function watchesEqual(left: ThreadPullRequestWatch, right: ThreadPullRequestWatch): boolean { @@ -170,9 +213,10 @@ function watchesEqual(left: ThreadPullRequestWatch, right: ThreadPullRequestWatc /** * Wakes a thread's agent when a pull request it watches (`watch_pull_request`) needs a look: * checks finished on the head commit, someone else commented, or the branch started to - * conflict. A pass every two minutes reads each watched pull request once for all the threads - * of a project that watch it, and skips one whose sync snapshot has not moved while nothing is - * in flight. + * conflict. A pass every two minutes looks at each watched pull request once for all the threads + * of a project that watch it. Where the host has a fingerprint, one batched read of those tells + * the pass which pull requests moved, and only those are read; elsewhere a pull request is read + * unless its sync snapshot has not moved while nothing is in flight. * Settling or archiving a thread ends its watches, and a merged or closed pull request ends * its watch. Each ended watch is logged once with why it ended and how long it went without * a push, for debugging watches that live too long. @@ -396,16 +440,53 @@ export const make = Effect.gen(function* () { ), ).pipe(Effect.map((landed) => landed.every(Boolean))); - const readGroup = Effect.fn("PullRequestWatchReactor.readGroup")(function* (group: WatchGroup) { - const now = yield* Clock.currentTimeMillis; - if (!needsRead(group, lastReads.get(group.key), now)) return; + const referenceOf = (group: WatchGroup) => { const first = group.targets[0]!; - const reference = { projectId: first.thread.projectId, ...identityOf(first.link) }; + return { projectId: first.thread.projectId, ...identityOf(first.link) }; + }; + + /** + * The host fingerprint of a watched pull request, read for every group at once so they share + * one batched request. Null, for a host without one or one that gave no answer, keeps the + * sync-snapshot gating; "paused" skips the pass while the host is rate limited, since every + * read would be refused. + */ + const fingerprintOf = (group: WatchGroup) => + pullRequests.watchFingerprint(referenceOf(group)).pipe( + Effect.catchCause( + (cause): Effect.Effect => + Cause.hasInterruptsOnly(cause) + ? Effect.interrupt + : isRateLimited(cause) + ? Effect.succeed("paused") + : // The snapshot-gated reads report a host that cannot be read, and count toward + // giving up. + Effect.logDebug("pull request watch fingerprint failed", { + pullRequest: group.key, + cause, + }).pipe(Effect.as(null)), + ), + ); + + const readGroup = Effect.fn("PullRequestWatchReactor.readGroup")(function* ( + group: WatchGroup, + fingerprint: ProviderChangeRequestWatchFingerprint | null, + ) { + const now = yield* Clock.currentTimeMillis; + const last = lastReads.get(group.key); + const plan = planRead(group, last, fingerprint, now); + if (!plan.detail) return; + const reference = referenceOf(group); + // The reads are recorded against this fingerprint, so a cached answer older than it must not + // stand in for them. Clients are not told: this only drops this pull request's cached reads. + if (fingerprint !== null) yield* pullRequests.invalidate({ reference }); const read = yield* Effect.exit( Effect.all( [ pullRequests.detail({ ...reference, allowStale: false }), - pullRequests.activity(reference), + plan.activity + ? pullRequests.activity(reference).pipe(Effect.map(Option.some)) + : Effect.succeed(Option.none()), ], { concurrency: 2 }, ), @@ -450,16 +531,21 @@ export const make = Effect.gen(function* () { } // Never advance the remark watermark past comments an incomplete read could have missed. - const remarks = yield* readRemarks(group.key, reference, activity); + // A pass that did not read the activity has no remarks to report. + const remarks = Option.isSome(activity) + ? yield* readRemarks(group.key, reference, activity.value) + : null; lastReads.set(group.key, { - at: now, + at: plan.activity || last === undefined ? now : last.at, fingerprint: group.fingerprint, + inFlight: + detail.mergeability === "unknown" || + detail.checks.some((check) => check.status === "pending"), // Comments a partial read could not see are read again next pass. - quiet: - remarks !== null && - detail.mergeability !== "unknown" && - detail.checks.every((check) => check.status !== "pending"), + remarksComplete: plan.activity ? remarks !== null : (last?.remarksComplete ?? false), watches: new Set(group.targets.map(watchKey)), + status: fingerprint?.status ?? null, + remarks: fingerprint?.remarks ?? null, }); yield* eachTarget(group, (target) => { const report = evaluatePullRequestWatch(target.watch, detail, remarks); @@ -533,14 +619,21 @@ export const make = Effect.gen(function* () { ), { discard: true }, ); + const fingerprints = yield* Effect.forEach(groups, fingerprintOf, { + concurrency: "unbounded", + }); yield* Effect.forEach( groups, - (group) => - readGroup(group).pipe( - Effect.catchCause( - logFailure("pull request watch check failed", { pullRequest: group.key }), - ), - ), + (group, index) => { + const fingerprint = fingerprints[index]!; + return fingerprint === "paused" + ? Effect.void + : readGroup(group, fingerprint).pipe( + Effect.catchCause( + logFailure("pull request watch check failed", { pullRequest: group.key }), + ), + ); + }, { concurrency: 4, discard: true }, ); }).pipe( diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index 1ccb4ee89d61ddcb048559756fa7b0d4c6544b2e..248cbe730fa72a4ac93377772c27374e6411134f 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -2767,6 +2767,179 @@ it.layer(TestLayer)("OrchestrationV2LayerLive lifecycle", (it) => { }), ); + it.effect("with a host fingerprint, a watch reads only what moved", () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const projectId = ProjectId.make("pr-watch-fingerprint-project"); + const threadId = ThreadId.make("runtime-pull-request-watch-fingerprint"); + yield* seedProject({ + projectId, + title: "Watch fingerprint", + workspaceRoot: "/workspace/watch-fingerprint", + defaultModelSelection: null, + createdAt: "2026-10-01T00:00:00.000Z", + }); + yield* orchestrator.dispatch({ + type: "thread.create", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make("pr-watch-fingerprint-create"), + threadId, + projectId, + title: "Watch fingerprint", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + }); + yield* orchestrator.dispatch({ + type: "thread.pull-request.watch", + commandId: CommandId.make("pr-watch-fingerprint-start"), + threadId, + host: "github.com", + repository: "pingdotgg/t3code", + number: 11, + watching: true, + link: { url: "https://github.com/pingdotgg/t3code/pull/11", source: "agent" }, + }); + + let fingerprint: { status: string; remarks: string } | "rate-limited" = { + status: "OPEN pending", + remarks: "0", + }; + type CheckStatus = "pending" | "success" | "failure"; + // What the host says now, and what a read answers: a read right after a cache fill can + // still answer from before a change until the cache is invalidated. + let checks: { lint: CheckStatus; test: CheckStatus } = { lint: "pending", test: "pending" }; + let cachedChecks: typeof checks | null = null; + let comments: Array = []; + let detailReads = 0; + let activityReads = 0; + const reactor = yield* PullRequestWatchReactor.make.pipe( + Effect.provide( + Layer.mergeAll( + NodeServices.layer, + Layer.mock(PullRequestService.PullRequestService)({ + watchFingerprint: () => + Effect.suspend(() => + fingerprint === "rate-limited" + ? Effect.fail( + new PullRequestOperationError({ + operation: "watchFingerprint", + detail: "paused", + cause: new PullRequestProviderError({ + provider: "github", + operation: "getChangeRequestWatchFingerprint", + reason: "rate-limited", + detail: "paused", + }), + }), + ) + : Effect.succeed(fingerprint), + ), + invalidate: () => + Effect.sync(() => { + cachedChecks = null; + }), + detail: () => + Effect.sync(() => { + detailReads += 1; + const answer = cachedChecks ?? checks; + cachedChecks = null; + return { + ...watchedPullRequestDetail({ + projectId, + number: 11, + at: "2026-10-02T12:00:00.000Z", + }), + checks: Object.entries(answer).map(([name, status]) => ({ + name, + status, + description: null, + url: null, + })), + }; + }), + activity: () => + Effect.sync(() => { + activityReads += 1; + return { + comments, + commentCount: comments.length, + commentsTruncated: false, + reviewThreads: [], + commits: [], + }; + }), + }), + ), + ), + ); + const reads = () => ({ detail: detailReads, activity: activityReads }); + const summaries = Effect.map( + orchestrator.getThreadRecords(threadId, ["messages"]), + ({ messages }) => messages.flatMap((message) => message.notification?.summary ?? []), + ); + + // The first look reads everything. + yield* reactor.sweep; + assert.deepEqual(reads(), { detail: 1, activity: 1 }); + + // While checks run, the cheap detail is read every pass: check counts by state cannot + // tell which check finished, so a failure can arrive with the fingerprint unchanged. + checks = { lint: "failure", test: "pending" }; + yield* reactor.sweep; + assert.deepEqual(reads(), { detail: 2, activity: 1 }); + assert.deepEqual(yield* summaries, ["#11: checks failed"]); + + // When the fingerprint moves, a cached answer from before the move is not trusted. + yield* TestClock.adjust("1 millis"); + cachedChecks = checks; + checks = { lint: "failure", test: "success" }; + fingerprint = { status: "OPEN settled", remarks: "0" }; + yield* reactor.sweep; + assert.deepEqual(reads(), { detail: 3, activity: 1 }); + + // Nothing in flight and nothing moved, so the passes read nothing. + for (let pass = 0; pass < 5; pass += 1) yield* reactor.sweep; + assert.deepEqual(reads(), { detail: 3, activity: 1 }); + + // A new comment moves the remarks, which take the activity read too. + yield* TestClock.adjust("1 millis"); + comments = [ + { + id: "comment-1", + kind: "issue-comment", + author: { login: "reviewer", name: null, avatarUrl: null }, + body: "Please rename this.", + createdAt: "2999-01-01T00:00:00.000Z", + url: null, + path: null, + reviewState: null, + }, + ]; + fingerprint = { status: "OPEN settled", remarks: "1" }; + yield* reactor.sweep; + assert.deepEqual(reads(), { detail: 4, activity: 2 }); + assert.deepEqual(yield* summaries, ["#11: checks failed", "#11: new comments"]); + + // While the host is rate limited, a pass reads nothing and the watch stays on. + fingerprint = "rate-limited"; + yield* reactor.sweep; + assert.deepEqual(reads(), { detail: 4, activity: 2 }); + assert.isDefined((yield* orchestrator.getThreadShell(threadId))?.pullRequests?.[0]?.watch); + + // Edits inside review threads leave no trace in the fingerprint, so the activity is read + // again after half an hour regardless. + fingerprint = { status: "OPEN settled", remarks: "1" }; + yield* TestClock.adjust("30 minutes"); + yield* reactor.sweep; + assert.deepEqual(reads(), { detail: 5, activity: 3 }); + assert.deepEqual(yield* summaries, ["#11: checks failed", "#11: new comments"]); + }), + ); + it.effect.each([ "single page", "paginated", diff --git a/apps/server/src/pullRequest/GitHubPullRequestCli.test.ts b/apps/server/src/pullRequest/GitHubPullRequestCli.test.ts index 24b84cc4b14b02ed0ca9a61e88e920cb84f9c294..b614cbd4ee702ffa5c9a6a57782bec8c13ee6559 100644 --- a/apps/server/src/pullRequest/GitHubPullRequestCli.test.ts +++ b/apps/server/src/pullRequest/GitHubPullRequestCli.test.ts @@ -532,6 +532,53 @@ layer("GitHubPullRequestCli.layer", (it) => { }), ); + it.effect("fingerprints watched pull requests on one host in one read", () => + Effect.gen(function* () { + const node = (comments: number) => ({ + state: "OPEN", + mergeable: "MERGEABLE", + headRefOid: "abc123", + comments: { totalCount: comments, nodes: [] }, + reviews: { totalCount: 0, nodes: [] }, + reviewThreads: { totalCount: 0 }, + commits: { nodes: [{ commit: { statusCheckRollup: null } }] }, + }); + mockedExecute.mockReturnValueOnce( + Effect.succeed( + output( + // @effect-diagnostics-next-line preferSchemaOverJson:off + JSON.stringify({ + data: { w0: { pullRequest: node(1) }, w1: { pullRequest: null } }, + }), + ), + ), + ); + const cli = yield* GitHubPullRequestCli.GitHubPullRequestCli; + + const reads = yield* Effect.forEach( + [7, 8], + (number) => + cli.getPullRequestWatchFingerprint({ + cwd: "/w", + repository: "acme/web", + host: "github.com", + number, + }), + { concurrency: "unbounded" }, + ).pipe(Effect.forkChild); + yield* TestClock.adjust("10 millis"); + const [seven, eight] = yield* Fiber.join(reads); + + expect(seven?.remarks.startsWith("1 ")).toBe(true); + // GitHub had no answer for #8, so its watch reads it in full. + expect(eight).toBeNull(); + expect(mockedExecute).toHaveBeenCalledOnce(); + expect(callAt(0).args.at(-1) ?? "").toContain( + 'w1: repository(owner: "acme", name: "web") { pullRequest(number: 8)', + ); + }), + ); + it.effect("reads a pull request the batch said nothing about on its own", () => Effect.gen(function* () { mockedExecute diff --git a/apps/server/src/pullRequest/GitHubPullRequestCli.ts b/apps/server/src/pullRequest/GitHubPullRequestCli.ts index 50c7e38a4b6b7972d229c048b6e940e063bc872a..25c379228457edd3e6b7ba9597790812e494c266 100644 --- a/apps/server/src/pullRequest/GitHubPullRequestCli.ts +++ b/apps/server/src/pullRequest/GitHubPullRequestCli.ts @@ -56,6 +56,7 @@ import { pullRequestCoreGraphQlQuery, type GitHubPullRequestCore, type GitHubPullRequestSummary, + type GitHubPullRequestWatchFingerprint, decodePullRequestPreviewJson, PULL_REQUEST_PREVIEW_GRAPHQL_QUERY, decodePullRequestFilesJson, @@ -67,6 +68,7 @@ import { decodePullRequestStacksJson, decodePullRequestStatsJson, decodePullRequestSummariesJson, + decodePullRequestWatchFingerprintsJson, decodeReactionSubjectScopeJson, decodeReviewerCandidatesJson, decodeLabelCandidatesJson, @@ -77,6 +79,7 @@ import { decodeReviewThreadsJson, buildPullRequestStatsGraphQlQuery, buildPullRequestSummariesGraphQlQuery, + buildPullRequestWatchFingerprintsGraphQlQuery, buildPullRequestStackMembershipsGraphQlQuery, decodePullRequestStackMembershipsJson, encodeGraphQlRequestJson, @@ -447,6 +450,17 @@ class PullRequestSummaryRead extends Request.Class< GitHubPullRequestCliError > {} +class PullRequestWatchFingerprintRead extends Request.Class< + { + readonly cwd: string; + readonly repository: string; + readonly host: string; + readonly number: number; + }, + GitHubPullRequestWatchFingerprint | null, + GitHubPullRequestCliError +> {} + export interface GitHubPullRequestSearchBatch { /** Rows across every repository asked for, newest update first, each naming its own. */ readonly items: ReadonlyArray; @@ -543,6 +557,16 @@ export class GitHubPullRequestCli extends Context.Service< readonly host: string; readonly number: number; }) => Effect.Effect; + /** + * What a watch compares between passes, batched like summaries. Null when GitHub gave no + * answer for this pull request, so the watch reads it in full instead. + */ + readonly getPullRequestWatchFingerprint: (input: { + readonly cwd: string; + readonly repository: string; + readonly host: string; + readonly number: number; + }) => Effect.Effect; readonly revalidateChecks: Effect.Success; @@ -1918,6 +1942,60 @@ export const make = Effect.gen(function* () { const getPullRequestSummary: GitHubPullRequestCli["Service"]["getPullRequestSummary"] = (input) => Effect.request(new PullRequestSummaryRead(input), summaryResolver); + // Every watched pull request on a host in one read per pass. A pull request the batch has no + // answer for gets null, and its watch reads it in full; a failed batch fails every entry. + const watchFingerprintResolver = RequestResolver.makeGrouped< + PullRequestWatchFingerprintRead, + string + >({ + key: ({ request, context }) => + JSON.stringify([ + request.host.toLowerCase(), + Context.getOrElse(context, GitHubCli.PinnedGitHubCredential, () => null) + ?.credentialFingerprint ?? null, + Context.getOrElse(context, SourceControlRateLimit.CredentialScope, () => ""), + ]), + resolver: (entries) => { + const [first] = entries; + const batchable = entries.filter( + (entry) => buildPullRequestWatchFingerprintsGraphQlQuery([entry.request]) !== null, + ); + const query = buildPullRequestWatchFingerprintsGraphQlQuery( + batchable.map((entry) => entry.request), + ); + const read = + query === null + ? Effect.succeed(new Map()) + : graphqlRead({ + cwd: first.request.cwd, + host: first.request.host, + operation: "getPullRequestWatchFingerprint", + query, + decode: decodePullRequestWatchFingerprintsJson, + }); + return read.pipe( + Effect.map((fingerprints) => { + for (const entry of entries) { + const index = batchable.indexOf(entry); + entry.completeUnsafe( + Exit.succeed(index === -1 ? null : (fingerprints.get(index) ?? null)), + ); + } + }), + Effect.catchCause((cause) => + Effect.sync(() => { + for (const entry of entries) entry.completeUnsafe(Exit.failCause(cause)); + }), + ), + ); + }, + }).pipe( + RequestResolver.setDelay(SUMMARY_BATCH_WINDOW), + RequestResolver.batchN(STAT_ALIASES_PER_REQUEST), + ); + const getPullRequestWatchFingerprint: GitHubPullRequestCli["Service"]["getPullRequestWatchFingerprint"] = + (input) => Effect.request(new PullRequestWatchFingerprintRead(input), watchFingerprintResolver); + return GitHubPullRequestCli.of({ withVerifiedCredential, revalidateChecks, @@ -2125,6 +2203,7 @@ export const make = Effect.gen(function* () { }, getPullRequestSummary, + getPullRequestWatchFingerprint, getPullRequestDetail, getPullRequestPreview: (input) => { diff --git a/apps/server/src/pullRequest/GitHubPullRequestProvider.ts b/apps/server/src/pullRequest/GitHubPullRequestProvider.ts index f2268a205da340807dde1facbcb8d93340014aca..567dd2925f5cdc2a653e5fd3fd92a3c44b7fae54 100644 --- a/apps/server/src/pullRequest/GitHubPullRequestProvider.ts +++ b/apps/server/src/pullRequest/GitHubPullRequestProvider.ts @@ -360,6 +360,11 @@ export const make = Effect.gen(function* () { getChangeRequestStack: (input) => cli.getPullRequestStack(input).pipe(Effect.mapError(fail("getChangeRequestStack"))), + getChangeRequestWatchFingerprint: (input) => + cli + .getPullRequestWatchFingerprint(input) + .pipe(Effect.mapError(fail("getChangeRequestWatchFingerprint"))), + getChangeRequestPreview: (input) => cli.getPullRequestPreview(input).pipe(Effect.mapError(fail("getChangeRequestPreview"))), diff --git a/apps/server/src/pullRequest/PullRequestProvider.ts b/apps/server/src/pullRequest/PullRequestProvider.ts index c45802549919c33368f33026d83c58c0bab9733d..c3f588a6227749d634d6b5789133afa2a2e8e563 100644 --- a/apps/server/src/pullRequest/PullRequestProvider.ts +++ b/apps/server/src/pullRequest/PullRequestProvider.ts @@ -69,6 +69,15 @@ export class PullRequestProviderError extends Schema.TaggedError Effect.Effect; + /** + * A cheap fingerprint of what a pull request watch reports, so it reads the change request in + * full only when this moves. Null when the host gave no answer for it. Optional: a host + * without one has its watched change requests read in full on every pass. + */ + readonly getChangeRequestWatchFingerprint?: ( + input: ProviderRepositoryRef & { readonly number: number }, + ) => Effect.Effect; + /** * The host-native stack a change request belongs to, or null when it is not stacked. Optional * because most hosts have no such object; the service derives chains from base branches there. diff --git a/apps/server/src/pullRequest/PullRequestService.ts b/apps/server/src/pullRequest/PullRequestService.ts index ad38a220f228ed157310564cdebb1b694c745f68..0ca62e6cd9e9bf3b19e3e967b4ceab9ba1ca9109 100644 --- a/apps/server/src/pullRequest/PullRequestService.ts +++ b/apps/server/src/pullRequest/PullRequestService.ts @@ -85,6 +85,7 @@ import * as SourceControlRateLimit from "../sourceControl/SourceControlRateLimit import { type ProviderChangeRequest, type ProviderListCursor, + type ProviderChangeRequestWatchFingerprint, type PullRequestProviderApi, PullRequestProviderError, } from "./PullRequestProvider.ts"; @@ -218,6 +219,13 @@ export class PullRequestService extends Context.Service< readonly checks: ( input: PullRequestRef, ) => Effect.Effect; + /** + * What a pull request watch compares between passes, so it reads detail and activity only + * when something moved. Null when the host has no fingerprint or gave none for this one. + */ + readonly watchFingerprint: ( + input: PullRequestRef, + ) => Effect.Effect; readonly activity: ( input: PullRequestRef, ) => Effect.Effect; @@ -581,6 +589,14 @@ function withRateLimitBackoff( ...(api.getChangeRequestStack === undefined ? {} : { getChangeRequestStack: wrap("getChangeRequestStack", api.getChangeRequestStack) }), + ...(api.getChangeRequestWatchFingerprint === undefined + ? {} + : { + getChangeRequestWatchFingerprint: wrap( + "getChangeRequestWatchFingerprint", + api.getChangeRequestWatchFingerprint, + ), + }), getChangeRequestActivity: wrap("getChangeRequestActivity", api.getChangeRequestActivity), ...(api.getReviewThreadComments === undefined ? {} @@ -2949,6 +2965,30 @@ export const make = Effect.gen(function* () { }, ); + const watchFingerprintCache = yield* Cache.makeWith( + (key: string) => { + const input = refOfCacheKey(key); + return requireProject(input).pipe( + Effect.flatMap((project) => + project.api.getChangeRequestWatchFingerprint === undefined + ? Effect.succeed(null) + : project.api + .getChangeRequestWatchFingerprint({ + cwd: project.project.workspaceRoot, + repository: project.repository, + host: project.host, + number: input.number, + }) + .pipe(Effect.mapError(toPullRequestError("watchFingerprint"))), + ), + ); + }, + { + capacity: DETAIL_CACHE_CAPACITY, + timeToLive: (exit) => (Exit.isSuccess(exit) ? DETAIL_CACHE_TTL : Duration.zero), + }, + ); + const detailCache = yield* Cache.makeWith( (key: string) => { const statsKey = statsCacheKey(key); @@ -3321,6 +3361,9 @@ export const make = Effect.gen(function* () { refreshAfterTurn, detail: credentialCached(detail), checks: credentialCached((input) => Cache.get(checksCache, refCacheKey(input))), + watchFingerprint: credentialCached((input) => + Cache.get(watchFingerprintCache, refCacheKey(input)), + ), activity: credentialCached(activity), preview: credentialCached(preview), threadComments, diff --git a/apps/server/src/pullRequest/gitHubPullRequestJson.test.ts b/apps/server/src/pullRequest/gitHubPullRequestJson.test.ts index c564e82918d435216a002ff1d76bc220a67a7ca9..b0225545c1f43e4cd6c5aa51769344f2ade88894 100644 --- a/apps/server/src/pullRequest/gitHubPullRequestJson.test.ts +++ b/apps/server/src/pullRequest/gitHubPullRequestJson.test.ts @@ -3,7 +3,9 @@ import { describe, expect, it } from "vite-plus/test"; import { buildPullRequestSummariesGraphQlQuery, + buildPullRequestWatchFingerprintsGraphQlQuery, decodePullRequestSummariesJson, + decodePullRequestWatchFingerprintsJson, buildReviewSubmissionJson, buildPullRequestStackMembershipsGraphQlQuery, decodePullRequestStackMembershipsJson, @@ -2026,3 +2028,81 @@ describe("batched pull request summaries", () => { expect(decoded.get(1)?.stack).toEqual({ number: 3, size: 2, base: "main", position: 2 }); }); }); + +describe("pull request watch fingerprints", () => { + it("asks for every pull request in one aliased read, and refuses an unsafe selector", () => { + const query = buildPullRequestWatchFingerprintsGraphQlQuery([ + { repository: "pingdotgg/t3code", number: 7 }, + { repository: "pingdotgg/lakebed", number: 8 }, + ]); + expect(query).toContain( + 'w0: repository(owner: "pingdotgg", name: "t3code") { pullRequest(number: 7)', + ); + expect(query).toContain( + 'w1: repository(owner: "pingdotgg", name: "lakebed") { pullRequest(number: 8)', + ); + expect( + buildPullRequestWatchFingerprintsGraphQlQuery([{ repository: 'evil") { x', number: 1 }]), + ).toBeNull(); + }); + + it("moves status with checks and remarks with comments, each without the other", () => { + const pullRequest = (input: { + readonly running: number; + readonly failed: number; + readonly commentEditedAt: string | null; + readonly reviews: number; + }) => ({ + state: "OPEN", + mergeable: "MERGEABLE", + headRefOid: "abc123", + comments: { totalCount: 1, nodes: [{ lastEditedAt: input.commentEditedAt }] }, + reviews: { totalCount: input.reviews, nodes: [] }, + reviewThreads: { totalCount: 0 }, + commits: { + nodes: [ + { + commit: { + statusCheckRollup: { + contexts: { + checkRunCountsByState: [ + { state: "IN_PROGRESS", count: input.running }, + { state: "FAILURE", count: input.failed }, + { state: "SUCCESS", count: 0 }, + ], + statusContextCountsByState: null, + }, + }, + }, + }, + ], + }, + }); + const decode = (...pullRequests: ReadonlyArray) => { + const decoded = decodePullRequestWatchFingerprintsJson( + JSON.stringify({ + data: Object.fromEntries( + pullRequests.map((value, index) => [`w${index}`, { pullRequest: value }]), + ), + }), + ); + if (!Result.isSuccess(decoded)) throw new Error("fingerprints did not decode"); + return decoded.success; + }; + const base = { running: 2, failed: 0, commentEditedAt: null, reviews: 1 }; + const fingerprint = (input: Parameters[0]) => + decode(pullRequest(input)).get(0)!; + const before = fingerprint(base); + const checkFinished = fingerprint({ ...base, running: 1, failed: 1 }); + const summaryEdited = fingerprint({ ...base, commentEditedAt: "2026-10-05T23:47:43Z" }); + const replied = fingerprint({ ...base, reviews: 2 }); + + expect(checkFinished.status).not.toBe(before.status); + expect(checkFinished.remarks).toBe(before.remarks); + expect(summaryEdited.remarks).not.toBe(before.remarks); + expect(summaryEdited.status).toBe(before.status); + expect(replied.remarks).not.toBe(before.remarks); + // A pull request GitHub had no answer for is left for a full read. + expect(decode(null, pullRequest(base)).has(0)).toBe(false); + }); +}); diff --git a/apps/server/src/pullRequest/gitHubPullRequestJson.ts b/apps/server/src/pullRequest/gitHubPullRequestJson.ts index 79318ece4a8bc013faa1fbd0ad4bd646c351fb95..0a2f5fe4a60459956f6b411b89dd0ddcddf0dc63 100644 --- a/apps/server/src/pullRequest/gitHubPullRequestJson.ts +++ b/apps/server/src/pullRequest/gitHubPullRequestJson.ts @@ -1993,6 +1993,139 @@ export function decodePullRequestSummariesJson( return Result.succeed(summaries); } +const WATCH_FINGERPRINT_EDITS = "totalCount nodes { lastEditedAt }"; +const WATCH_FINGERPRINT_CHECK_COUNTS = + "checkRunCountsByState { state count } statusContextCountsByState { state count }"; + +/** + * What a pull request watch needs to notice, priced by GitHub at one point for twenty-five pull + * requests, where the detail and activity reads it gates cost sixteen for one. Counts and the + * newest edit cover new comments, reviews (a reply in a review thread is a review), threads, and + * a bot rewriting its summary; check counts by state move whenever a check starts or finishes. + * Comments come most recently updated first, since an edit moves `updatedAt`, so an edit to an + * old comment on a long pull request is still in the page. Reviews have no such order: an edit + * to a review older than the last hundred waits for the watch's periodic full read, as do edits + * to comments inside review threads, which cost a point per pull request to ask for. + */ +const PULL_REQUEST_WATCH_FINGERPRINT_SELECTION = + "state mergeable headRefOid " + + `comments(first: 100, orderBy: { field: UPDATED_AT, direction: DESC }) { ${WATCH_FINGERPRINT_EDITS} } ` + + `reviews(last: 100) { ${WATCH_FINGERPRINT_EDITS} } ` + + "reviewThreads { totalCount } " + + `commits(last: 1) { nodes { commit { statusCheckRollup { contexts { ${WATCH_FINGERPRINT_CHECK_COUNTS} } } } } }`; + +/** Watch fingerprints for pull requests on one host, one aliased lookup each; null when unsafe. */ +export function buildPullRequestWatchFingerprintsGraphQlQuery( + changeRequests: ReadonlyArray<{ readonly repository: string; readonly number: number }>, +): string | null { + if (changeRequests.length === 0) return null; + const selections: string[] = []; + for (const [index, changeRequest] of changeRequests.entries()) { + const [owner, name, ...rest] = changeRequest.repository.trim().split("/"); + if (rest.length > 0 || owner === undefined || name === undefined) return null; + if (!REPOSITORY_PART.test(owner) || !REPOSITORY_PART.test(name)) return null; + if (!Number.isSafeInteger(changeRequest.number) || changeRequest.number <= 0) return null; + selections.push( + ` w${index}: repository(owner: "${owner}", name: "${name}") { pullRequest(number: ${changeRequest.number}) { ${PULL_REQUEST_WATCH_FINGERPRINT_SELECTION} } }`, + ); + } + return `query PullRequestWatchFingerprints {\n${selections.join("\n")}\n}`; +} + +/** + * Two parts, so a watch reads only what moved: `status` (state, mergeability, head, and checks) + * needs the detail read, and `remarks` also needs the far costlier activity read. + */ +export interface GitHubPullRequestWatchFingerprint { + readonly status: string; + readonly remarks: string; +} + +const RawEditedSchema = Schema.Struct({ + totalCount: Schema.Int, + nodes: Schema.Array(Schema.NullOr(Schema.Struct({ lastEditedAt: Schema.NullOr(Schema.String) }))), +}); +const RawStateCountsSchema = Schema.NullOr( + Schema.Array(Schema.Struct({ state: Schema.String, count: Schema.Int })), +); +const decodeWatchFingerprintEntry = Schema.decodeUnknownExit( + Schema.Struct({ + state: Schema.String, + mergeable: Schema.NullOr(Schema.String), + headRefOid: Schema.String, + comments: RawEditedSchema, + reviews: RawEditedSchema, + reviewThreads: Schema.Struct({ totalCount: Schema.Int }), + commits: Schema.Struct({ + nodes: Schema.Array( + Schema.NullOr( + Schema.Struct({ + commit: Schema.Struct({ + statusCheckRollup: Schema.NullOr( + Schema.Struct({ + contexts: Schema.Struct({ + checkRunCountsByState: RawStateCountsSchema, + statusContextCountsByState: RawStateCountsSchema, + }), + }), + ), + }), + }), + ), + ), + }), + }), +); + +// An edit only ever moves `lastEditedAt` forward, so the newest one stands for them all. +const newestEdit = (connection: typeof RawEditedSchema.Type) => + connection.nodes.reduce( + (newest, node) => + node?.lastEditedAt != null && node.lastEditedAt > newest ? node.lastEditedAt : newest, + "", + ); + +const stateCounts = (counts: typeof RawStateCountsSchema.Type) => + (counts ?? []) + .filter(({ count }) => count > 0) + .map(({ state, count }) => `${state}:${count}`) + .toSorted() + .join(","); + +/** Fingerprints by alias index; a pull request GitHub returned nothing for is left out. */ +export function decodePullRequestWatchFingerprintsJson( + raw: string, +): Result.Result, DecodeFailure> { + const decoded = decodeSummaries(raw); + if (!Result.isSuccess(decoded)) return Result.fail(decoded.failure); + const fingerprints = new Map(); + for (const [alias, value] of Object.entries(decoded.success.data ?? {})) { + const index = /^w(\d+)$/.exec(alias)?.[1]; + if (index === undefined || value?.pullRequest == null) continue; + const entry = decodeWatchFingerprintEntry(value.pullRequest); + if (!Exit.isSuccess(entry)) continue; + const pr = entry.value; + const contexts = pr.commits.nodes[0]?.commit.statusCheckRollup?.contexts; + fingerprints.set(Number(index), { + status: [ + pr.state, + pr.mergeable ?? "", + pr.headRefOid, + stateCounts(contexts?.checkRunCountsByState ?? null), + stateCounts(contexts?.statusContextCountsByState ?? null), + ].join(" "), + remarks: [ + pr.comments.totalCount, + newestEdit(pr.comments), + pr.reviews.totalCount, + newestEdit(pr.reviews), + pr.reviewThreads.totalCount, + ].join(" "), + }); + } + return Result.succeed(fingerprints); +} + export interface GitHubPullRequestCore extends GitHubPullRequestDetail { readonly viewerAccess: GitHubViewerAccess & GitHubRepositoryAccess; readonly comparison: GitHubBaseComparison | null; diff --git a/apps/server/src/sourceControl/GitHubCli.ts b/apps/server/src/sourceControl/GitHubCli.ts index 82e2da846881f3d6c3a15c688774eb6528a25d78..3a13ac69e9efbbe0b1c6831512f00966d0ded1fc 100644 --- a/apps/server/src/sourceControl/GitHubCli.ts +++ b/apps/server/src/sourceControl/GitHubCli.ts @@ -115,7 +115,7 @@ export class GitHubCliRateLimitError extends Schema.TaggedError