From 9e5229b47f7af66e2f88447a49f8ea7f00c734c6 Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Mon, 5 Oct 2026 15:22:14 -0700 Subject: [PATCH] fix(server): PR watches stop burning GitHub's rate limit and giving up (#16208) Co-authored-by: Claude Opus 5.5 (1M context) --- .../toolkits/pullRequests/handlers.test.ts | 35 ++ .../src/mcp/toolkits/pullRequests/handlers.ts | 11 +- .../src/mcp/toolkits/pullRequests/tools.ts | 12 +- .../src/orchestration-v2/Orchestrator.ts | 28 +- .../src/orchestration-v2/ProjectionStore.ts | 4 +- .../PullRequestSyncReactor.test.ts | 5 + .../PullRequestWatchReactor.ts | 264 +++++++++++---- .../src/orchestration-v2/ThreadStop.test.ts | 25 +- .../orchestration-v2/pullRequestWatch.test.ts | 6 +- .../src/orchestration-v2/pullRequestWatch.ts | 5 +- .../src/orchestration-v2/runtimeLayer.test.ts | 305 +++++++++++++++--- .../testkit/pullRequestFixtures.ts | 7 +- docs/user/source-control.md | 17 +- 13 files changed, 577 insertions(+), 147 deletions(-) diff --git a/apps/server/src/mcp/toolkits/pullRequests/handlers.test.ts b/apps/server/src/mcp/toolkits/pullRequests/handlers.test.ts index d940916f904c5c1ff2e278bd6a16a88ce306fbbc..53af795545396d1d360a959793d80a58106dd30a 100644 --- a/apps/server/src/mcp/toolkits/pullRequests/handlers.test.ts +++ b/apps/server/src/mcp/toolkits/pullRequests/handlers.test.ts @@ -279,6 +279,41 @@ describe("pull request toolkit handlers", () => { }), ); + it.effect("watches a pull request saved as closed, since it may have reopened", () => + Effect.gen(function* () { + const closed = makeLink(1, { headBranch: "closed" }); + const harness = yield* makeHarness({ + thread: makeThread([ + { ...closed, snapshot: closed.snapshot && { ...closed.snapshot, state: "closed" } }, + ]), + }); + yield* harness.call("watch_pull_request", { repository: "t3tools/t3code", number: 1 }); + expect(yield* Ref.get(harness.commands)).toMatchObject([ + { type: "thread.pull-request.watch", number: 1, watching: true }, + ]); + }), + ); + + it.effect("refuses a watch from a subagent thread, whose parent owns the pull request", () => + Effect.gen(function* () { + const harness = yield* makeHarness({ + thread: { + ...makeThread([makeLink(1, { headBranch: "feature" })]), + lineage: { + rootThreadId: ThreadId.make("parent"), + parentThreadId: ThreadId.make("parent"), + relationshipToParent: "subagent", + }, + }, + }); + const error = yield* harness + .call("watch_pull_request", { repository: "t3tools/t3code", number: 1 }) + .pipe(Effect.flip); + expect(error).toMatchObject({ _tag: "PullRequestWatchFromSubagentError" }); + expect(yield* Ref.get(harness.commands)).toEqual([]); + }), + ); + it.effect("links by repository and number, defaulting the host to the project's", () => Effect.gen(function* () { const harness = yield* makeHarness(); diff --git a/apps/server/src/mcp/toolkits/pullRequests/handlers.ts b/apps/server/src/mcp/toolkits/pullRequests/handlers.ts index e43208256ea9590a6dcd324987d4df55b4fa0382..c81e05b2c0228bada15c74c8f56d295750df3426 100644 --- a/apps/server/src/mcp/toolkits/pullRequests/handlers.ts +++ b/apps/server/src/mcp/toolkits/pullRequests/handlers.ts @@ -38,6 +38,7 @@ import { PullRequestNotOpenError, type PullRequestTargetInput, PullRequestWatchFailedError, + PullRequestWatchFromSubagentError, PullRequestThreadNotFoundError, PullRequestThreadAboveLimitsError, PullRequestThreadRequiredError, @@ -273,10 +274,14 @@ const make = Effect.gen(function* () { threadPullRequestsOf(shell).find( (link) => link.source !== "stack-dismissed" && threadPullRequestKeysEqual(link, target), ); + if (watching && thread.lineage.relationshipToParent === "subagent") { + return yield* new PullRequestWatchFromSubagentError(); + } const before = watchedLink(thread); - const state = before?.snapshot?.state; - if (watching && state !== undefined && state !== "open") { - return yield* new PullRequestNotOpenError({ state }); + // A merged pull request cannot reopen. A closed one can, and its saved state may be stale, + // so the watch starts and its first read ends it if the host still says closed. + if (watching && before?.snapshot?.state === "merged") { + return yield* new PullRequestNotOpenError({ state: "merged" }); } yield* engine .dispatch({ diff --git a/apps/server/src/mcp/toolkits/pullRequests/tools.ts b/apps/server/src/mcp/toolkits/pullRequests/tools.ts index 1894014ebf2ae08d23326666c781b4afaa5d9d62..2a150617a4059e0a693f8ea264bc22d5b8c3387f 100644 --- a/apps/server/src/mcp/toolkits/pullRequests/tools.ts +++ b/apps/server/src/mcp/toolkits/pullRequests/tools.ts @@ -150,6 +150,15 @@ export class PullRequestNotOpenError extends Schema.TaggedError()( + "PullRequestWatchFromSubagentError", + {}, +) { + override get message(): string { + return "This thread is a subagent, so it cannot watch pull requests. Its parent thread owns the pull request: finish your task and report back instead."; + } +} + export class PullRequestListFailedError extends Schema.TaggedError()( "PullRequestListFailedError", { cause: Schema.Defect() }, @@ -172,6 +181,7 @@ export const PullRequestToolError = Schema.Union([ PullRequestListFailedError, PullRequestWatchFailedError, PullRequestNotOpenError, + PullRequestWatchFromSubagentError, ]); export type PullRequestToolError = typeof PullRequestToolError.Type; @@ -289,7 +299,7 @@ const ListThreadPullRequestsTool = Tool.make("list_thread_pull_requests", { const WatchPullRequestTool = Tool.make("watch_pull_request", { description: - "Have T3 Code watch an open pull request for this thread, linking it first if needed. T3 Code checks it every minute and wakes you with a message when a check fails, the required checks pass, someone else comments or reviews, or the branch starts to conflict with its base. Use this to monitor or babysit a pull request instead of polling, sleeping, or running a watcher. Only comments posted after this call wake you, so handle the existing ones first, then end your turn. A wake is news, not a merge decision: check readiness yourself before merging. While T3 Code watches, the thread stays in the user's Working list, not their inbox. When you hand the work back to the user, call unwatch_pull_request first so the thread returns to their inbox. Watching ends when the pull request merges or closes, when its thread settles, when T3 Code cannot read it for 15 minutes, when the user stops this thread, or when you call unwatch_pull_request. Unsettle the thread before starting a new watch.", + "Have T3 Code watch an open pull request for this thread, linking it first if needed. T3 Code checks it every two minutes and wakes you with a message when a check fails, the required checks pass, someone else comments or reviews, or the branch starts to conflict with its base. Use this to monitor or babysit a pull request instead of polling, sleeping, or running a watcher. Only comments posted after this call wake you, so handle the existing ones first, then end your turn. A wake is news, not a merge decision: check readiness yourself before merging. While T3 Code watches, the thread stays in the user's Working list, not their inbox. When you hand the work back to the user, call unwatch_pull_request first so the thread returns to their inbox. Watching ends when the pull request merges or closes, when its thread settles or is archived, when T3 Code fails to read it 8 times in a row (a host rate limit only delays it), when the user stops this thread, or when you call unwatch_pull_request. Unsettle the thread before starting a new watch. A subagent cannot watch: its parent thread owns the pull request.", parameters: PullRequestTargetInput, success: WatchPullRequestResult, failure: PullRequestToolError, diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 01c5057d464c9bb4312ea59c33298e327c6098c2..ca772485b1af459b7cb692f7b0b03f8d464a7a3e 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -2382,12 +2382,27 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio if ( command.type === "thread.pull-request.watch" && command.watching && - (thread.settledOverride === "settled" || thread.settledAt !== null) + (thread.settledOverride === "settled" || + thread.settledAt !== null || + thread.archivedAt !== null) ) { return yield* new OrchestratorDispatchError({ commandId: command.commandId, commandType: command.type, - cause: `Thread ${command.threadId} is settled and cannot watch pull requests.`, + cause: `Thread ${command.threadId} is settled or archived and cannot watch pull requests.`, + }); + } + // A subagent or delegated task reports to its parent thread, which owns the pull request. + // Its own watch would wake it on every change of a PR it was not asked to babysit. + if ( + command.type === "thread.pull-request.watch" && + command.watching && + thread.lineage.relationshipToParent === "subagent" + ) { + return yield* new OrchestratorDispatchError({ + commandId: command.commandId, + commandType: command.type, + cause: `Thread ${command.threadId} is a subagent; its parent thread watches pull requests.`, }); } // Only the agent's watch_pull_request links as "agent". A call that raced a Stop may land @@ -2713,7 +2728,14 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio const updatedThread: OrchestrationV2AppThread = (() => { switch (command.type) { case "thread.archive": - return { ...thread, archivedAt: now, titleRegeneration: null, updatedAt: now }; + // An archived thread takes no wakes, so its watches end like a settled thread's. + return { + ...thread, + archivedAt: now, + titleRegeneration: null, + pullRequests: thread.pullRequests?.map((link) => withPullRequestWatch(link, undefined)), + updatedAt: now, + }; case "thread.unarchive": return { ...thread, archivedAt: null, updatedAt: now }; case "thread.settle": { diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index bec60cb4d79d9cd9ca7f1e525a7c8da22645adec..23563a2d69ff688d4ca7363d877ffd2150d08d49 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -156,7 +156,7 @@ export type ProjectionLimitRecoveryCandidate = Pick< /** The thread fields pull request sync reads, for a thread with at least one link. */ export type ProjectionThreadPullRequests = Pick< OrchestrationV2AppThread, - "id" | "projectId" | "settledOverride" | "settledAt" | "pullRequests" + "id" | "projectId" | "lineage" | "settledOverride" | "settledAt" | "pullRequests" >; /** @@ -5247,6 +5247,7 @@ export const layer: Layer.Layer = Effect.map((thread): ProjectionThreadPullRequests => ({ id: thread.id, projectId: thread.projectId, + lineage: thread.lineage, settledOverride: thread.settledOverride, settledAt: thread.settledAt, pullRequests: thread.pullRequests ?? [], @@ -5718,6 +5719,7 @@ export const layerMemory: Layer.Layer = Layer.effect( .map((thread): ProjectionThreadPullRequests => ({ id: thread.id, projectId: thread.projectId, + lineage: thread.lineage, settledOverride: thread.settledOverride, settledAt: thread.settledAt, pullRequests: thread.pullRequests ?? [], diff --git a/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts b/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts index 256544e993121dff03c5533a363b719c2db43361..f615050c34095beea156284177c71b97faf1e97c 100644 --- a/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts +++ b/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts @@ -235,6 +235,11 @@ const makeHarness = Effect.fn("makePullRequestSyncHarness")(function* (options: .map((thread) => ({ id: thread.id, projectId: thread.projectId, + lineage: { + parentThreadId: null, + relationshipToParent: null, + rootThreadId: thread.id, + }, settledOverride: thread.settledOverride, settledAt: thread.settledAt === null ? null : DateTime.makeUnsafe(thread.settledAt), pullRequests: thread.pullRequests, diff --git a/apps/server/src/orchestration-v2/PullRequestWatchReactor.ts b/apps/server/src/orchestration-v2/PullRequestWatchReactor.ts index 5300742706380e2e0ddc5466c27b4913bee3113e..a0bffbf34002612527868c6fb9bd1182656a0282 100644 --- a/apps/server/src/orchestration-v2/PullRequestWatchReactor.ts +++ b/apps/server/src/orchestration-v2/PullRequestWatchReactor.ts @@ -15,22 +15,48 @@ import { visibleThreadPullRequests, } from "@t3tools/shared/threadPullRequests"; import * as Cause from "effect/Cause"; +import * as Clock from "effect/Clock"; import * as Context from "effect/Context"; import * as Crypto from "effect/Crypto"; import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; 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 * as PullRequestService from "../pullRequest/PullRequestService.ts"; import { forkParked } from "../serverActivation.ts"; import * as Orchestrator from "./Orchestrator.ts"; import * as ProjectionStore from "./ProjectionStore.ts"; import { evaluatePullRequestWatch, pullRequestWatchMessage } from "./pullRequestWatch.ts"; -/** Passes in a row that could not read a pull request before its watch ends (one a minute). */ -const READ_FAILURE_LIMIT = 15; +/** + * Minutes between passes. Checks take minutes, so a faster pass mostly spends the host's rate + * limit, which every machine on the same account shares. + */ +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. + */ +const QUIET_REREAD_MS = 10 * 60_000; + +const isProviderError = Schema.is(PullRequestProviderError); + +/** A rate limit is the host asking us to wait, not a sign the pull request cannot be read. */ +const isRateLimited = (cause: Cause.Cause) => + Option.match(Cause.findErrorOption(cause), { + onNone: () => false, + onSome: (error) => + error._tag === "PullRequestOperationError" && + isProviderError(error.cause) && + error.cause.reason === "rate-limited", + }); const logFailure = (message: string, fields: Record) => @@ -45,8 +71,49 @@ interface WatchTarget { readonly watch: ThreadPullRequestWatch; } -const failureKey = ({ thread, link, watch }: WatchTarget) => - `${thread.id} ${threadPullRequestKeyOf(link)} ${watch.startedAt}`; +/** Every thread of one project that watches one pull request, read once per pass. */ +interface WatchGroup { + readonly key: string; + readonly targets: ReadonlyArray; + /** What sync last saw of the pull request, to skip a read when nothing moved. */ + readonly fingerprint: string; +} + +/** The last successful read of a pull request, kept in memory: a restart reads each once. */ +interface LastRead { + readonly at: number; + readonly fingerprint: string; + /** Nothing is in flight or unread, so only a moved snapshot or the reread brings news. */ + readonly quiet: boolean; + /** The watches this read evaluated; a watch started since takes its first look next pass. */ + readonly watches: ReadonlySet; +} + +const watchKey = ({ thread, watch }: WatchTarget) => `${thread.id} ${watch.startedAt}`; + +const snapshotFingerprint = ({ link }: WatchTarget) => { + const snapshot = link.snapshot; + return snapshot === null + ? "" + : [ + snapshot.state, + snapshot.updatedAt, + snapshot.checksState, + snapshot.mergeability, + snapshot.reviewDecision, + snapshot.isDraft, + ].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))) + ); +} function watchesEqual(left: ThreadPullRequestWatch, right: ThreadPullRequestWatch): boolean { return ( @@ -65,8 +132,11 @@ 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. One pass a minute reads each watched pull request; settling a thread ends its - * watches, and a merged or closed pull request ends its watch. + * 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. + * Settling or archiving a thread ends its watches, and a merged or closed pull request ends + * its watch. */ export class PullRequestWatchReactor extends Context.Service< PullRequestWatchReactor, @@ -84,10 +154,11 @@ export const make = Effect.gen(function* () { const pullRequests = yield* PullRequestService.PullRequestService; const crypto = yield* Crypto.Crypto; - // Passes in a row that failed, per watch. Kept in memory: a restart only delays the stop. + // Reads in a row that failed, per pull request. Kept in memory: a restart only delays the stop. const readFailures = new Map(); - // Replies past each long thread's first page, per watch, so a pass pages a thread again only - // when the host's count of it moves. Kept in memory: a restart pages each thread once more. + const lastReads = new Map(); + // Replies past each long thread's first page, per pull request, so a pass pages a thread + // again only when the host's count of it moves. A restart pages each thread once more. const threadTails = new Map< string, Map }> @@ -128,7 +199,7 @@ export const make = Effect.gen(function* () { // "Watching" while it learns nothing. const giveUp = (target: WatchTarget) => record(target, null, { - text: `T3 Code stopped watching pull request #${target.link.number} (${target.link.url}) because it could not read it from the host for ${READ_FAILURE_LIMIT} minutes. Check it yourself, and call watch_pull_request to watch it again.`, + text: `T3 Code stopped watching pull request #${target.link.number} (${target.link.url}) because it failed to read it from the host ${READ_FAILURE_LIMIT} times in a row. Check it yourself, and call watch_pull_request to watch it again.`, notification: { source: { kind: "monitor" }, outcome: "failed", @@ -137,12 +208,11 @@ export const make = Effect.gen(function* () { }).pipe(Effect.catch(() => record(target, null))); const readRemarks = Effect.fn("PullRequestWatchReactor.readRemarks")( - function* (target: WatchTarget, reference: PullRequestRef, activity: PullRequestActivity) { + function* (key: string, reference: PullRequestRef, activity: PullRequestActivity) { // Comment cursors cannot account for missing threads. Only finish a truncated read // when the host confirms that every thread was listed. if (activity.commentsTruncated && activity.reviewThreadsTruncated !== false) return null; - const key = failureKey(target); let tails = threadTails.get(key); if (tails === undefined) { tails = new Map(); @@ -190,18 +260,53 @@ export const make = Effect.gen(function* () { ), ); - const check = Effect.fn("PullRequestWatchReactor.check")(function* (target: WatchTarget) { - const { thread, link, watch } = target; - const pullRequest = identityOf(link); - // A merged pull request cannot reopen, so its watch ends without a host read. A closed one - // can, so the host decides below. Settling ends watches; one left from before that rule - // ends here, so the thread does not show as working while settled. - if (link.snapshot?.state === "merged") return yield* record(target, null); - if (thread.settledOverride === "settled" || thread.settledAt !== null) { - return yield* record(target, null); - } + // A closed pull request can reopen, but the watch has nothing to report until then. + const closed = (target: WatchTarget) => + record(target, null, { + text: `Pull request #${target.link.number} (${target.link.url}) was closed, so T3 Code stopped watching it. Call watch_pull_request if it reopens.`, + notification: { + source: { kind: "monitor" }, + outcome: "updated", + summary: `#${target.link.number}: closed, stopped watching`, + }, + }); - const reference = { projectId: thread.projectId, ...pullRequest }; + /** Watches that end without a host read; the rest are read once per pull request. */ + const endsWithoutRead = ({ thread, link }: WatchTarget) => + // A merged pull request cannot reopen. Settling and archiving end watches, and a subagent + // cannot start one; a watch left from before those rules ends here. + link.snapshot?.state === "merged" || + thread.settledOverride === "settled" || + thread.settledAt !== null || + thread.lineage.relationshipToParent === "subagent"; + + /** + * Runs one thread's step for each thread in a group, so one refusal does not skip the rest. + * Succeeds with whether every step landed. + */ + const eachTarget = ( + group: WatchGroup, + step: (target: WatchTarget) => Effect.Effect, + ) => + Effect.forEach(group.targets, (target) => + step(target).pipe( + Effect.as(true), + Effect.catchCause((cause) => { + // A thread that did not get its update must not wait for the quiet reread. + lastReads.delete(group.key); + return logFailure("pull request watch update failed", { + threadId: target.thread.id, + pullRequest: group.key, + })(cause).pipe(Effect.as(false)); + }), + ), + ).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 first = group.targets[0]!; + const reference = { projectId: first.thread.projectId, ...identityOf(first.link) }; const read = yield* Effect.exit( Effect.all( [ @@ -211,40 +316,58 @@ export const make = Effect.gen(function* () { { concurrency: 2 }, ), ); - // Only host reads count towards giving up; a refused wake is not the host's fault. - const key = failureKey(target); if (Exit.isFailure(read)) { if (Cause.hasInterruptsOnly(read.cause)) return yield* Effect.failCause(read.cause); - const failures = (readFailures.get(key) ?? 0) + 1; - readFailures.set(key, failures); - // The count stays until the stop lands, so a failed stop is tried again next pass. - if (failures >= READ_FAILURE_LIMIT) { - yield* giveUp(target); - readFailures.delete(key); + lastReads.delete(group.key); + // The host's pause refuses later reads without a request, so waiting it out is free. + if (isRateLimited(read.cause)) return; + const failures = (readFailures.get(group.key) ?? 0) + 1; + readFailures.set(group.key, failures); + // The count stays until every stop lands, so a failed stop is tried again on the next + // failed read, and a watch started after the stops begins from zero. + if (failures >= READ_FAILURE_LIMIT && (yield* eachTarget(group, giveUp))) { + readFailures.delete(group.key); } return yield* Effect.failCause(read.cause); } - readFailures.delete(key); + readFailures.delete(group.key); const [detail, activity] = read.value; - if (detail.state !== "open") return yield* record(target, null); - - // Never advance the remark watermark past comments an incomplete read could have missed. - const remarks = yield* readRemarks(target, reference, activity); - const report = evaluatePullRequestWatch(watch, detail, remarks); - if (report.changes.length > 0) { - return yield* record( - target, - report.exhausted ? null : report.next, - pullRequestWatchMessage({ - number: link.number, - url: link.url, - baseBranch: detail.baseBranch, - headSha: report.next.headSha, - report, - }), + if (detail.state !== "open") { + lastReads.delete(group.key); + return yield* eachTarget(group, (target) => + detail.state === "closed" ? closed(target) : record(target, null), ); } - if (!watchesEqual(report.next, watch)) yield* record(target, report.next); + + // Never advance the remark watermark past comments an incomplete read could have missed. + const remarks = yield* readRemarks(group.key, reference, activity); + lastReads.set(group.key, { + at: now, + fingerprint: group.fingerprint, + // 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"), + watches: new Set(group.targets.map(watchKey)), + }); + yield* eachTarget(group, (target) => { + const report = evaluatePullRequestWatch(target.watch, detail, remarks); + if (report.changes.length > 0) { + return record( + target, + report.exhausted ? null : report.next, + pullRequestWatchMessage({ + number: target.link.number, + url: target.link.url, + baseBranch: detail.baseBranch, + headSha: report.next.headSha, + report, + }), + ); + } + return watchesEqual(report.next, target.watch) ? Effect.void : record(target, report.next); + }); }); const sweep = Effect.gen(function* () { @@ -254,20 +377,47 @@ export const make = Effect.gen(function* () { link.watch === undefined ? [] : [{ thread, link, watch: link.watch }], ), ); - const keys = new Set(targets.map(failureKey)); - for (const key of readFailures.keys()) if (!keys.has(key)) readFailures.delete(key); - for (const key of threadTails.keys()) if (!keys.has(key)) threadTails.delete(key); + const byPullRequest = new Map>(); + const ending: Array = []; + for (const target of targets) { + if (endsWithoutRead(target)) { + ending.push(target); + continue; + } + // Grouped per project too: each project reads through its own checkout, so one that cannot + // read the pull request must not end another project's watches. + const key = `${target.thread.projectId} ${threadPullRequestKeyOf(target.link)}`; + byPullRequest.set(key, [...(byPullRequest.get(key) ?? []), target]); + } + const groups = [...byPullRequest].map(([key, members]): WatchGroup => ({ + key, + targets: members, + fingerprint: [...new Set(members.map(snapshotFingerprint))].toSorted().join("\n"), + })); + for (const cache of [readFailures, lastReads, threadTails]) { + for (const key of cache.keys()) if (!byPullRequest.has(key)) cache.delete(key); + } yield* Effect.forEach( - targets, + ending, (target) => - check(target).pipe( + record(target, null).pipe( Effect.catchCause( - logFailure("pull request watch check failed", { + logFailure("pull request watch stop failed", { threadId: target.thread.id, pullRequest: threadPullRequestKeyOf(target.link), }), ), ), + { discard: true }, + ); + yield* Effect.forEach( + groups, + (group) => + readGroup(group).pipe( + Effect.catchCause( + logFailure("pull request watch check failed", { pullRequest: group.key }), + ), + ), { concurrency: 4, discard: true }, ); }).pipe( @@ -276,7 +426,9 @@ export const make = Effect.gen(function* () { ); const start: PullRequestWatchReactor["Service"]["start"] = () => - forkParked(sweep.pipe(Effect.repeat(Schedule.spaced("1 minute")), Effect.asVoid)); + forkParked( + sweep.pipe(Effect.repeat(Schedule.spaced(`${SWEEP_MINUTES} minutes`)), Effect.asVoid), + ); return { start, sweep } satisfies PullRequestWatchReactor["Service"]; }); diff --git a/apps/server/src/orchestration-v2/ThreadStop.test.ts b/apps/server/src/orchestration-v2/ThreadStop.test.ts index 3320830e409bb68df2b2f873e5b8029424746ec8..9b465b1c6a55926be9446814463fbf3dd49760a0 100644 --- a/apps/server/src/orchestration-v2/ThreadStop.test.ts +++ b/apps/server/src/orchestration-v2/ThreadStop.test.ts @@ -156,7 +156,8 @@ it.effect("Stop ends watches, holds queues, and stops the delegated tasks under yield* createWatchingThread(parentThreadId, 1); yield* send(parentThreadId, "first", "start_immediately"); const childThreadId = yield* delegate(parentThreadId, "child task"); - yield* watch(childThreadId, 2); + // The parent owns its pull requests; a delegated task cannot watch one. + assert.isTrue(Exit.isFailure(yield* Effect.exit(watch(childThreadId, 2)))); const grandchildThreadId = yield* delegate(childThreadId, "grandchild task"); yield* send(childThreadId, "child follow-up", "queue_after_active"); @@ -174,9 +175,6 @@ it.effect("Stop ends watches, holds queues, and stops the delegated tasks under }); yield* send(parentThreadId, "second", "start_immediately"); const secondRun = (yield* orchestrator.getThreadProjection(parentThreadId)).runs.at(-1)!; - const childWatch = (yield* orchestrator.getThreadProjection(childThreadId)).thread - .pullRequests?.[0]?.watch; - assert.isDefined(childWatch); const stopCommandId = CommandId.make("stop-parent"); yield* orchestrator.dispatch({ @@ -209,25 +207,6 @@ it.effect("Stop ends watches, holds queues, and stops the delegated tasks under }); assert.deepEqual((yield* threadState(grandchildThreadId)).runs, ["interrupted"]); - // A watch read that raced the Stop cannot wake the stopped child. - const lateWake = yield* Effect.exit( - orchestrator.dispatch({ - type: "thread.pull-request-watch.sync", - commandId: CommandId.make("late-watch-wake"), - threadId: childThreadId, - ...pullRequest(2), - startedAt: childWatch!.startedAt, - watch: null, - wake: { - messageId: MessageId.make("message:late-watch-wake"), - text: "Checks passed.", - notification: { source: { kind: "monitor" }, outcome: "completed", summary: "#2" }, - }, - }), - ); - assert.isTrue(Exit.isFailure(lateWake)); - assert.deepEqual((yield* threadState(childThreadId)).runs, ["interrupted", "queued:held"]); - // A retried effect stops nothing twice. yield* threads.stopDelegatedTasks({ threadId: parentThreadId, commandId: stopCommandId }); assert.deepEqual((yield* threadState(childThreadId)).runs, ["interrupted", "queued:held"]); diff --git a/apps/server/src/orchestration-v2/pullRequestWatch.test.ts b/apps/server/src/orchestration-v2/pullRequestWatch.test.ts index f69f2c7b4206a7f78f6f540fc05885a43c0a0687..ddd329d411730ef9086572a2b2d2c04b05d7c345 100644 --- a/apps/server/src/orchestration-v2/pullRequestWatch.test.ts +++ b/apps/server/src/orchestration-v2/pullRequestWatch.test.ts @@ -134,7 +134,7 @@ describe("evaluatePullRequestWatch", () => { ]); }); - it("reports passed again when a check shows up already passed where none is required", () => { + it("does not report passed again for a new passed check where none is required", () => { const first = evaluatePullRequestWatch( watch(), detail({ checks: [check("test", "success")] }), @@ -142,9 +142,7 @@ describe("evaluatePullRequestWatch", () => { ); assert.deepEqual(first.changes, [{ kind: "checks-passed", count: 1, required: false }]); const both = detail({ checks: [check("test", "success"), check("lint", "success")] }); - assert.deepEqual(evaluatePullRequestWatch(first.next, both, noRemarks).changes, [ - { kind: "checks-passed", count: 2, required: false }, - ]); + assert.deepEqual(evaluatePullRequestWatch(first.next, both, noRemarks).changes, []); }); it("does not wake a watch saved before passed checks were recorded", () => { diff --git a/apps/server/src/orchestration-v2/pullRequestWatch.ts b/apps/server/src/orchestration-v2/pullRequestWatch.ts index 57b93f7847d288d6ab949a3a83fcb46e3d37aab8..2d0fd15a98895625f0e3bbc9d469debee47d8c03 100644 --- a/apps/server/src/orchestration-v2/pullRequestWatch.ts +++ b/apps/server/src/orchestration-v2/pullRequestWatch.ts @@ -67,8 +67,9 @@ export function evaluatePullRequestWatch( const gateNames = gate.map((check) => check.name); // A watch saved before passedChecks existed takes the current names, so it does not wake. const told = passed && passedChecks.length === 0 ? gateNames : passedChecks; - // A required job created and finished between two passes is never seen pending. - const gateGrew = gateNames.some((name) => !told.includes(name)); + // A required job created and finished between two passes is never seen pending. Without + // required checks, any check counts, and advisory bots keep adding passed ones: no wake. + const gateGrew = required.length > 0 && gateNames.some((name) => !told.includes(name)); if (passedNow && (!passed || gateGrew)) { changes.push({ kind: "checks-passed", count: gate.length, required: required.length > 0 }); } diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index ac49f09454488fb115c4763fcc56424d46f0dcf9..77f22d4600a3de8f044b6ad203e6b71c70cc6898 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -63,6 +63,7 @@ import * as ProviderRuntimeRecoveryService from "./ProviderRuntimeRecoveryServic import * as ProjectionMaintenance from "./ProjectionMaintenance.ts"; import * as ProjectionStore from "./ProjectionStore.ts"; import * as PullRequestWatchReactor from "./PullRequestWatchReactor.ts"; +import { PullRequestProviderError } from "../pullRequest/PullRequestProvider.ts"; import * as PullRequestService from "../pullRequest/PullRequestService.ts"; import * as ProjectStore from "./ProjectStore.ts"; import type { ProviderAdapterV2SessionRuntime, ProviderAdapterV2Shape } from "./ProviderAdapter.ts"; @@ -165,6 +166,58 @@ const TestProviderInstanceRegistry = Layer.succeed( ); /** Seed a project row the way a committed `project.created` event folds into it. */ +/** A watched pull request as the host reports it: open, mergeable, with a failing lint check. */ +const watchedPullRequestDetail = (input: { + readonly projectId: ProjectId; + readonly number: number; + readonly at: string; +}): PullRequestDetail => ({ + provider: "github", + capabilities: { + diff: true, + comment: true, + actions: [], + mergeMethods: [], + search: false, + review: { inlineComment: false, reply: false, resolve: false, verdicts: [] }, + reviewers: { request: false, listCandidates: false }, + }, + viewerPermissions: { + actions: [], + comment: true, + resolve: true, + verdicts: [], + requestReviewers: false, + }, + projectId: input.projectId, + projectTitle: "Watch wake", + workspaceRoot: "/workspace/watch", + repository: "pingdotgg/t3code", + number: input.number, + title: "Watched pull request", + body: "", + url: `https://github.com/pingdotgg/t3code/pull/${input.number}`, + author: { login: "agent-user", name: null, avatarUrl: null }, + state: "open", + isDraft: false, + mergeability: "mergeable", + additions: 1, + deletions: 0, + changedFiles: 1, + headBranch: "feature", + headSha: "abc1234def", + baseBranch: "main", + createdAt: input.at, + updatedAt: input.at, + mergedAt: null, + closedAt: null, + reviewers: [], + labels: [], + checks: [{ name: "lint", status: "failure", description: null, url: null }], + mergeCapabilities: { merge: true, squash: true, rebase: true }, + viewer: "agent-user", +}); + const seedProject = (input: { readonly projectId: ProjectId; readonly title: string; @@ -2352,6 +2405,63 @@ it.layer(TestLayer)("OrchestrationV2LayerLive lifecycle", (it) => { }), ); + it.effect("archiving ends watches, and an archived thread cannot start one", () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const threadId = ThreadId.make("runtime-pull-request-watch-archive"); + yield* orchestrator.dispatch({ + type: "thread.create", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make("pr-watch-archive-create"), + threadId, + projectId: ProjectId.make("pr-watch-archive-project"), + title: "Archive watched PR", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + }); + const watch = { + type: "thread.pull-request.watch" as const, + threadId, + host: "github.com", + repository: "pingdotgg/t3code", + number: 7, + watching: true, + link: { url: "https://github.com/pingdotgg/t3code/pull/7", source: "agent" as const }, + }; + const watched = Effect.map( + orchestrator.getThreadShell(threadId), + (thread) => thread?.pullRequests?.[0]?.watch !== undefined, + ); + yield* orchestrator.dispatch({ + ...watch, + commandId: CommandId.make("pr-watch-archive-start"), + }); + assert.isTrue(yield* watched); + + yield* orchestrator.dispatch({ + type: "thread.archive", + commandId: CommandId.make("pr-watch-archive"), + threadId, + }); + assert.isFalse(yield* watched); + const refused = yield* orchestrator + .dispatch({ ...watch, commandId: CommandId.make("pr-watch-archive-restart") }) + .pipe(Effect.flip); + assert.equal(refused._tag, "OrchestratorDispatchError"); + + yield* orchestrator.dispatch({ + type: "thread.unarchive", + commandId: CommandId.make("pr-watch-unarchive"), + threadId, + }); + assert.isFalse(yield* watched); + }), + ); + it.effect("ends a watch it cannot read, and tells the agent", () => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; @@ -2411,6 +2521,154 @@ it.layer(TestLayer)("OrchestrationV2LayerLive lifecycle", (it) => { }), ); + it.effect( + "reads a watched pull request once a pass, waits out rate limits, and skips quiet passes", + () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const projectId = ProjectId.make("pr-watch-shared-project"); + yield* seedProject({ + projectId, + title: "Watch shared", + workspaceRoot: "/workspace/watch-shared", + defaultModelSelection: null, + createdAt: "2026-10-01T00:00:00.000Z", + }); + const threadIds = ["one", "two"].map((name) => + ThreadId.make(`runtime-pull-request-watch-shared-${name}`), + ); + const watchFrom = (threadId: ThreadId) => + Effect.gen(function* () { + yield* orchestrator.dispatch({ + type: "thread.create", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make(`create:${threadId}`), + threadId, + projectId, + title: "Watch shared", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + }); + yield* orchestrator.dispatch({ + type: "thread.pull-request.watch", + commandId: CommandId.make(`watch:${threadId}`), + threadId, + host: "github.com", + repository: "pingdotgg/t3code", + number: 9, + watching: true, + link: { url: "https://github.com/pingdotgg/t3code/pull/9", source: "agent" }, + }); + }); + for (const threadId of threadIds) yield* watchFrom(threadId); + const rateLimited = new PullRequestOperationError({ + operation: "getChangeRequest", + detail: "github requests are paused until the rate limit resets", + cause: new PullRequestProviderError({ + provider: "github", + operation: "getChangeRequest", + reason: "rate-limited", + detail: "paused", + }), + }); + let host: "rate-limited" | "open" | "closed" = "rate-limited"; + let reads = 0; + const reactor = yield* PullRequestWatchReactor.make.pipe( + Effect.provide( + Layer.mergeAll( + NodeServices.layer, + Layer.mock(PullRequestService.PullRequestService)({ + detail: () => + Effect.suspend(() => { + reads += 1; + return host === "rate-limited" + ? Effect.fail(rateLimited) + : Effect.succeed({ + ...watchedPullRequestDetail({ + projectId, + number: 9, + at: "2026-10-02T12:00:00.000Z", + }), + state: host, + checks: [ + { name: "lint", status: "success", description: null, url: null }, + ], + }); + }), + activity: () => + host === "rate-limited" + ? Effect.fail(rateLimited) + : Effect.succeed({ + comments: [], + commentCount: 0, + commentsTruncated: false, + reviewThreads: [], + commits: [], + }), + }), + ), + ), + ); + const summaries = (threadId: ThreadId) => + Effect.map(orchestrator.getThreadRecords(threadId, ["messages"]), ({ messages }) => + messages.flatMap((message) => message.notification?.summary ?? []), + ); + const watching = (threadId: ThreadId) => + Effect.map( + orchestrator.getThreadShell(threadId), + (thread) => thread?.pullRequests?.[0]?.watch !== undefined, + ); + + // However long the host stays rate limited, the watch waits instead of giving up, and + // both threads share one read a pass. + for (let pass = 0; pass < 20; pass += 1) yield* reactor.sweep; + assert.equal(reads, 20); + for (const threadId of threadIds) { + assert.isTrue(yield* watching(threadId)); + assert.deepEqual(yield* summaries(threadId), []); + } + + host = "open"; + yield* reactor.sweep; + assert.equal(reads, 21); + for (const threadId of threadIds) { + assert.deepEqual(yield* summaries(threadId), ["#9: checks passed"]); + } + + // Nothing in flight and nothing moved, so the next pass spends no request. + yield* reactor.sweep; + assert.equal(reads, 21); + + // A new watch on the quiet pull request takes its first look on the next pass. + const late = ThreadId.make("runtime-pull-request-watch-shared-late"); + yield* watchFrom(late); + yield* reactor.sweep; + assert.equal(reads, 22); + assert.deepEqual(yield* summaries(late), ["#9: checks passed"]); + for (const threadId of threadIds) { + assert.deepEqual(yield* summaries(threadId), ["#9: checks passed"]); + } + threadIds.push(late); + + // The quiet reread finds the pull request closed, ends both watches, and says so. + host = "closed"; + yield* TestClock.adjust("10 minutes"); + yield* reactor.sweep; + assert.equal(reads, 23); + for (const threadId of threadIds) { + assert.isFalse(yield* watching(threadId)); + assert.deepEqual(yield* summaries(threadId), [ + "#9: checks passed", + "#9: closed, stopped watching", + ]); + } + }), + ); + it.effect.each([ "single page", "paginated", @@ -2463,52 +2721,7 @@ it.layer(TestLayer)("OrchestrationV2LayerLive lifecycle", (it) => { }); const at = "2026-10-02T12:00:00.000Z"; - const detail: PullRequestDetail = { - provider: "github", - capabilities: { - diff: true, - comment: true, - actions: [], - mergeMethods: [], - search: false, - review: { inlineComment: false, reply: false, resolve: false, verdicts: [] }, - reviewers: { request: false, listCandidates: false }, - }, - viewerPermissions: { - actions: [], - comment: true, - resolve: true, - verdicts: [], - requestReviewers: false, - }, - projectId, - projectTitle: "Watch wake", - workspaceRoot: "/workspace/watch", - repository: key.repository, - number: key.number, - title: "Watched pull request", - body: "", - url, - author: { login: "agent-user", name: null, avatarUrl: null }, - state: "open", - isDraft: false, - mergeability: "mergeable", - additions: 1, - deletions: 0, - changedFiles: 1, - headBranch: "feature", - headSha: "abc1234def", - baseBranch: "main", - createdAt: at, - updatedAt: at, - mergedAt: null, - closedAt: null, - reviewers: [], - labels: [], - checks: [{ name: "lint", status: "failure", description: null, url: null }], - mergeCapabilities: { merge: true, squash: true, rebase: true }, - viewer: "agent-user", - }; + const detail = watchedPullRequestDetail({ projectId, number: key.number, at }); const initialWatch = (yield* orchestrator.getThreadShell(threadId))?.pullRequests?.[0]?.watch; assert.isDefined(initialWatch); const remark: PullRequestComment = { diff --git a/apps/server/src/orchestration-v2/testkit/pullRequestFixtures.ts b/apps/server/src/orchestration-v2/testkit/pullRequestFixtures.ts index 98b6d1421ff2b86052efe616676747ded09cc4c5..74273af6a73db28fffa0ff039623845484e4e095 100644 --- a/apps/server/src/orchestration-v2/testkit/pullRequestFixtures.ts +++ b/apps/server/src/orchestration-v2/testkit/pullRequestFixtures.ts @@ -23,6 +23,7 @@ export interface PullRequestTestThread { readonly linkedPullRequest?: ThreadLinkedPullRequest | null; readonly branchPullRequest?: ThreadLinkedPullRequest | null; readonly latestUserMessageAt: string | null; + readonly lineage?: OrchestrationV2ThreadShell["lineage"]; readonly createdAt: string; readonly updatedAt: string; readonly archivedAt: string | null; @@ -47,7 +48,11 @@ export function v2PullRequestThread(thread: PullRequestTestThread): Orchestratio createdBy: "user", creationSource: "web", activeProviderThreadId: null, - lineage: { rootThreadId: thread.id, parentThreadId: null, relationshipToParent: null }, + lineage: thread.lineage ?? { + rootThreadId: thread.id, + parentThreadId: null, + relationshipToParent: null, + }, forkedFrom: null, latestRunId: null, activeRunId: null, diff --git a/docs/user/source-control.md b/docs/user/source-control.md index b849fcdf2b89aa8e38972efcb5bac25ac745a789..e84d16f90e3f0177efe15057c47f8d96d88db0c6 100644 --- a/docs/user/source-control.md +++ b/docs/user/source-control.md @@ -192,13 +192,16 @@ when requested. With **Auto-settle merged threads** enabled, a thread can settle review is terminal. An open or unsynced link keeps it active. Ask the agent to watch, monitor, or babysit a pull request and it calls `watch_pull_request`. While -the thread is active, the server checks the pull request every minute and wakes the agent when a check -fails, the required checks pass, someone else comments or reviews, or the branch starts to conflict. -Comments from your own account do not wake it. Watching ends when the pull request merges or closes, -after 10 wakes in a row that bring only comments, when the server cannot read the pull request for -15 minutes, or when you press Stop on the thread. Settling a thread also ends all its watches. -Unsettle the thread before starting a new watch. To start or stop it yourself, use the row menu in -the **Linked pull requests** panel. +the thread is active, the server checks the pull request every two minutes and wakes the agent when a +check fails, the required checks pass, someone else comments or reviews, or the branch starts to +conflict. Threads in a project that watch the same pull request share one check, and a pull request +with nothing in progress is checked again when something changes or every 10 minutes, which keeps +watching inside GitHub's rate limit. Comments from your own account do not wake it. Watching ends when the pull request merges or closes, +after 10 wakes in a row that bring only comments, after 8 failed reads in a row, or when you press +Stop on the thread. A rate limit only pauses watching. Settling or archiving a thread also ends all +its watches. Unsettle the thread before starting a new watch. Subagents cannot watch pull requests; +the thread that delegated to them does. To start or stop it yourself, use the row menu in the +**Linked pull requests** panel. A watched thread counts as working between wakes, so it stays in the **Working** section and does not auto-settle. Agents stop watching when they hand the work back to you, and the thread then -- 2.54.0