authorgravatar for me@t3.ggTheo Browne <me@t3.gg> 2026-10-05 15:22:14-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-10-05 15:22:14-07:00
log9e5229b47f7af66e2f88447a49f8ea7f00c734c6
treed0dc41ff9de7cf1f1cb5b30f27cb55a205a51d17
parentf33b060cafc96d7c3229744315c230a3b8d1f631
signature Signed by PGP key B5690EEEBB952194

fix(server): PR watches stop burning GitHub's rate limit and giving up (#16208)

Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

13 files changed, 577 insertions(+), 147 deletions(-)

apps/server/src/mcp/toolkits/pullRequests/handlers.test.ts+35
......@@ -279,6 +279,41 @@ describe("pull request toolkit handlers", () => {
279279 }),
280280 );
281281
282 it.effect("watches a pull request saved as closed, since it may have reopened", () =>
283 Effect.gen(function* () {
284 const closed = makeLink(1, { headBranch: "closed" });
285 const harness = yield* makeHarness({
286 thread: makeThread([
287 { ...closed, snapshot: closed.snapshot && { ...closed.snapshot, state: "closed" } },
288 ]),
289 });
290 yield* harness.call("watch_pull_request", { repository: "t3tools/t3code", number: 1 });
291 expect(yield* Ref.get(harness.commands)).toMatchObject([
292 { type: "thread.pull-request.watch", number: 1, watching: true },
293 ]);
294 }),
295 );
296
297 it.effect("refuses a watch from a subagent thread, whose parent owns the pull request", () =>
298 Effect.gen(function* () {
299 const harness = yield* makeHarness({
300 thread: {
301 ...makeThread([makeLink(1, { headBranch: "feature" })]),
302 lineage: {
303 rootThreadId: ThreadId.make("parent"),
304 parentThreadId: ThreadId.make("parent"),
305 relationshipToParent: "subagent",
306 },
307 },
308 });
309 const error = yield* harness
310 .call("watch_pull_request", { repository: "t3tools/t3code", number: 1 })
311 .pipe(Effect.flip);
312 expect(error).toMatchObject({ _tag: "PullRequestWatchFromSubagentError" });
313 expect(yield* Ref.get(harness.commands)).toEqual([]);
314 }),
315 );
316
282317 it.effect("links by repository and number, defaulting the host to the project's", () =>
283318 Effect.gen(function* () {
284319 const harness = yield* makeHarness();
apps/server/src/mcp/toolkits/pullRequests/handlers.ts+8-3
......@@ -38,6 +38,7 @@ import {
3838 PullRequestNotOpenError,
3939 type PullRequestTargetInput,
4040 PullRequestWatchFailedError,
41 PullRequestWatchFromSubagentError,
4142 PullRequestThreadNotFoundError,
4243 PullRequestThreadAboveLimitsError,
4344 PullRequestThreadRequiredError,
......@@ -273,10 +274,14 @@ const make = Effect.gen(function* () {
273274 threadPullRequestsOf(shell).find(
274275 (link) => link.source !== "stack-dismissed" && threadPullRequestKeysEqual(link, target),
275276 );
277 if (watching && thread.lineage.relationshipToParent === "subagent") {
278 return yield* new PullRequestWatchFromSubagentError();
279 }
276280 const before = watchedLink(thread);
277 const state = before?.snapshot?.state;
278 if (watching && state !== undefined && state !== "open") {
279 return yield* new PullRequestNotOpenError({ state });
281 // A merged pull request cannot reopen. A closed one can, and its saved state may be stale,
282 // so the watch starts and its first read ends it if the host still says closed.
283 if (watching && before?.snapshot?.state === "merged") {
284 return yield* new PullRequestNotOpenError({ state: "merged" });
280285 }
281286 yield* engine
282287 .dispatch({
apps/server/src/mcp/toolkits/pullRequests/tools.ts+11-1
......@@ -150,6 +150,15 @@ export class PullRequestNotOpenError extends Schema.TaggedError<PullRequestNotOp
150150 }
151151}
152152
153export class PullRequestWatchFromSubagentError extends Schema.TaggedError<PullRequestWatchFromSubagentError>()(
154 "PullRequestWatchFromSubagentError",
155 {},
156) {
157 override get message(): string {
158 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.";
159 }
160}
161
153162export class PullRequestListFailedError extends Schema.TaggedError<PullRequestListFailedError>()(
154163 "PullRequestListFailedError",
155164 { cause: Schema.Defect() },
......@@ -172,6 +181,7 @@ export const PullRequestToolError = Schema.Union([
172181 PullRequestListFailedError,
173182 PullRequestWatchFailedError,
174183 PullRequestNotOpenError,
184 PullRequestWatchFromSubagentError,
175185]);
176186export type PullRequestToolError = typeof PullRequestToolError.Type;
177187
......@@ -289,7 +299,7 @@ const ListThreadPullRequestsTool = Tool.make("list_thread_pull_requests", {
289299
290300const WatchPullRequestTool = Tool.make("watch_pull_request", {
291301 description:
292 "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.",
302 "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.",
293303 parameters: PullRequestTargetInput,
294304 success: WatchPullRequestResult,
295305 failure: PullRequestToolError,
apps/server/src/orchestration-v2/Orchestrator.ts+25-3
......@@ -2382,12 +2382,27 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
23822382 if (
23832383 command.type === "thread.pull-request.watch" &&
23842384 command.watching &&
2385 (thread.settledOverride === "settled" || thread.settledAt !== null)
2385 (thread.settledOverride === "settled" ||
2386 thread.settledAt !== null ||
2387 thread.archivedAt !== null)
23862388 ) {
23872389 return yield* new OrchestratorDispatchError({
23882390 commandId: command.commandId,
23892391 commandType: command.type,
2390 cause: `Thread ${command.threadId} is settled and cannot watch pull requests.`,
2392 cause: `Thread ${command.threadId} is settled or archived and cannot watch pull requests.`,
2393 });
2394 }
2395 // A subagent or delegated task reports to its parent thread, which owns the pull request.
2396 // Its own watch would wake it on every change of a PR it was not asked to babysit.
2397 if (
2398 command.type === "thread.pull-request.watch" &&
2399 command.watching &&
2400 thread.lineage.relationshipToParent === "subagent"
2401 ) {
2402 return yield* new OrchestratorDispatchError({
2403 commandId: command.commandId,
2404 commandType: command.type,
2405 cause: `Thread ${command.threadId} is a subagent; its parent thread watches pull requests.`,
23912406 });
23922407 }
23932408 // 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
27132728 const updatedThread: OrchestrationV2AppThread = (() => {
27142729 switch (command.type) {
27152730 case "thread.archive":
2716 return { ...thread, archivedAt: now, titleRegeneration: null, updatedAt: now };
2731 // An archived thread takes no wakes, so its watches end like a settled thread's.
2732 return {
2733 ...thread,
2734 archivedAt: now,
2735 titleRegeneration: null,
2736 pullRequests: thread.pullRequests?.map((link) => withPullRequestWatch(link, undefined)),
2737 updatedAt: now,
2738 };
27172739 case "thread.unarchive":
27182740 return { ...thread, archivedAt: null, updatedAt: now };
27192741 case "thread.settle": {
apps/server/src/orchestration-v2/ProjectionStore.ts+3-1
......@@ -156,7 +156,7 @@ export type ProjectionLimitRecoveryCandidate = Pick<
156156/** The thread fields pull request sync reads, for a thread with at least one link. */
157157export type ProjectionThreadPullRequests = Pick<
158158 OrchestrationV2AppThread,
159 "id" | "projectId" | "settledOverride" | "settledAt" | "pullRequests"
159 "id" | "projectId" | "lineage" | "settledOverride" | "settledAt" | "pullRequests"
160160>;
161161
162162/**
......@@ -5247,6 +5247,7 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
52475247 Effect.map((thread): ProjectionThreadPullRequests => ({
52485248 id: thread.id,
52495249 projectId: thread.projectId,
5250 lineage: thread.lineage,
52505251 settledOverride: thread.settledOverride,
52515252 settledAt: thread.settledAt,
52525253 pullRequests: thread.pullRequests ?? [],
......@@ -5718,6 +5719,7 @@ export const layerMemory: Layer.Layer<ProjectionStoreV2> = Layer.effect(
57185719 .map((thread): ProjectionThreadPullRequests => ({
57195720 id: thread.id,
57205721 projectId: thread.projectId,
5722 lineage: thread.lineage,
57215723 settledOverride: thread.settledOverride,
57225724 settledAt: thread.settledAt,
57235725 pullRequests: thread.pullRequests ?? [],
apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts+5
......@@ -235,6 +235,11 @@ const makeHarness = Effect.fn("makePullRequestSyncHarness")(function* (options:
235235 .map((thread) => ({
236236 id: thread.id,
237237 projectId: thread.projectId,
238 lineage: {
239 parentThreadId: null,
240 relationshipToParent: null,
241 rootThreadId: thread.id,
242 },
238243 settledOverride: thread.settledOverride,
239244 settledAt: thread.settledAt === null ? null : DateTime.makeUnsafe(thread.settledAt),
240245 pullRequests: thread.pullRequests,
apps/server/src/orchestration-v2/PullRequestWatchReactor.ts+208-56
......@@ -15,22 +15,48 @@ import {
1515 visibleThreadPullRequests,
1616} from "@t3tools/shared/threadPullRequests";
1717import * as Cause from "effect/Cause";
18import * as Clock from "effect/Clock";
1819import * as Context from "effect/Context";
1920import * as Crypto from "effect/Crypto";
2021import * as Effect from "effect/Effect";
2122import * as Exit from "effect/Exit";
2223import * as Layer from "effect/Layer";
24import * as Option from "effect/Option";
2325import * as Schedule from "effect/Schedule";
26import * as Schema from "effect/Schema";
2427import type * as Scope from "effect/Scope";
2528
29import { PullRequestProviderError } from "../pullRequest/PullRequestProvider.ts";
2630import * as PullRequestService from "../pullRequest/PullRequestService.ts";
2731import { forkParked } from "../serverActivation.ts";
2832import * as Orchestrator from "./Orchestrator.ts";
2933import * as ProjectionStore from "./ProjectionStore.ts";
3034import { evaluatePullRequestWatch, pullRequestWatchMessage } from "./pullRequestWatch.ts";
3135
32/** Passes in a row that could not read a pull request before its watch ends (one a minute). */
33const READ_FAILURE_LIMIT = 15;
36/**
37 * Minutes between passes. Checks take minutes, so a faster pass mostly spends the host's rate
38 * limit, which every machine on the same account shares.
39 */
40const SWEEP_MINUTES = 2;
41/** Reads in a row that failed for a reason other than a rate limit before the watch ends. */
42const READ_FAILURE_LIMIT = 8;
43/**
44 * A pull request with nothing in flight is read again only when its sync snapshot moves, or
45 * after this long, for news the snapshot cannot show, such as a bot editing its review.
46 */
47const QUIET_REREAD_MS = 10 * 60_000;
48
49const isProviderError = Schema.is(PullRequestProviderError);
50
51/** A rate limit is the host asking us to wait, not a sign the pull request cannot be read. */
52const isRateLimited = (cause: Cause.Cause<PullRequestService.PullRequestError>) =>
53 Option.match(Cause.findErrorOption(cause), {
54 onNone: () => false,
55 onSome: (error) =>
56 error._tag === "PullRequestOperationError" &&
57 isProviderError(error.cause) &&
58 error.cause.reason === "rate-limited",
59 });
3460
3561const logFailure =
3662 (message: string, fields: Record<string, unknown>) =>
......@@ -45,8 +71,49 @@ interface WatchTarget {
4571 readonly watch: ThreadPullRequestWatch;
4672}
4773
48const failureKey = ({ thread, link, watch }: WatchTarget) =>
49 `${thread.id} ${threadPullRequestKeyOf(link)} ${watch.startedAt}`;
74/** Every thread of one project that watches one pull request, read once per pass. */
75interface WatchGroup {
76 readonly key: string;
77 readonly targets: ReadonlyArray<WatchTarget>;
78 /** What sync last saw of the pull request, to skip a read when nothing moved. */
79 readonly fingerprint: string;
80}
81
82/** The last successful read of a pull request, kept in memory: a restart reads each once. */
83interface LastRead {
84 readonly at: number;
85 readonly fingerprint: string;
86 /** Nothing is in flight or unread, so only a moved snapshot or the reread brings news. */
87 readonly quiet: boolean;
88 /** The watches this read evaluated; a watch started since takes its first look next pass. */
89 readonly watches: ReadonlySet<string>;
90}
91
92const watchKey = ({ thread, watch }: WatchTarget) => `${thread.id} ${watch.startedAt}`;
93
94const snapshotFingerprint = ({ link }: WatchTarget) => {
95 const snapshot = link.snapshot;
96 return snapshot === null
97 ? ""
98 : [
99 snapshot.state,
100 snapshot.updatedAt,
101 snapshot.checksState,
102 snapshot.mergeability,
103 snapshot.reviewDecision,
104 snapshot.isDraft,
105 ].join(" ");
106};
107
108function needsRead(group: WatchGroup, last: LastRead | undefined, now: number): boolean {
109 return (
110 last === undefined ||
111 !last.quiet ||
112 last.fingerprint !== group.fingerprint ||
113 now - last.at >= QUIET_REREAD_MS ||
114 group.targets.some((target) => !last.watches.has(watchKey(target)))
115 );
116}
50117
51118function watchesEqual(left: ThreadPullRequestWatch, right: ThreadPullRequestWatch): boolean {
52119 return (
......@@ -65,8 +132,11 @@ function watchesEqual(left: ThreadPullRequestWatch, right: ThreadPullRequestWatc
65132/**
66133 * Wakes a thread's agent when a pull request it watches (`watch_pull_request`) needs a look:
67134 * checks finished on the head commit, someone else commented, or the branch started to
68 * conflict. One pass a minute reads each watched pull request; settling a thread ends its
69 * watches, and a merged or closed pull request ends its watch.
135 * conflict. A pass every two minutes reads each watched pull request once for all the threads
136 * of a project that watch it, and skips one whose sync snapshot has not moved while nothing is
137 * in flight.
138 * Settling or archiving a thread ends its watches, and a merged or closed pull request ends
139 * its watch.
70140 */
71141export class PullRequestWatchReactor extends Context.Service<
72142 PullRequestWatchReactor,
......@@ -84,10 +154,11 @@ export const make = Effect.gen(function* () {
84154 const pullRequests = yield* PullRequestService.PullRequestService;
85155 const crypto = yield* Crypto.Crypto;
86156
87 // Passes in a row that failed, per watch. Kept in memory: a restart only delays the stop.
157 // Reads in a row that failed, per pull request. Kept in memory: a restart only delays the stop.
88158 const readFailures = new Map<string, number>();
89 // Replies past each long thread's first page, per watch, so a pass pages a thread again only
90 // when the host's count of it moves. Kept in memory: a restart pages each thread once more.
159 const lastReads = new Map<string, LastRead>();
160 // Replies past each long thread's first page, per pull request, so a pass pages a thread
161 // again only when the host's count of it moves. A restart pages each thread once more.
91162 const threadTails = new Map<
92163 string,
93164 Map<string, { readonly count: number; readonly comments: ReadonlyArray<PullRequestComment> }>
......@@ -128,7 +199,7 @@ export const make = Effect.gen(function* () {
128199 // "Watching" while it learns nothing.
129200 const giveUp = (target: WatchTarget) =>
130201 record(target, null, {
131 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.`,
202 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.`,
132203 notification: {
133204 source: { kind: "monitor" },
134205 outcome: "failed",
......@@ -137,12 +208,11 @@ export const make = Effect.gen(function* () {
137208 }).pipe(Effect.catch(() => record(target, null)));
138209
139210 const readRemarks = Effect.fn("PullRequestWatchReactor.readRemarks")(
140 function* (target: WatchTarget, reference: PullRequestRef, activity: PullRequestActivity) {
211 function* (key: string, reference: PullRequestRef, activity: PullRequestActivity) {
141212 // Comment cursors cannot account for missing threads. Only finish a truncated read
142213 // when the host confirms that every thread was listed.
143214 if (activity.commentsTruncated && activity.reviewThreadsTruncated !== false) return null;
144215
145 const key = failureKey(target);
146216 let tails = threadTails.get(key);
147217 if (tails === undefined) {
148218 tails = new Map();
......@@ -190,18 +260,53 @@ export const make = Effect.gen(function* () {
190260 ),
191261 );
192262
193 const check = Effect.fn("PullRequestWatchReactor.check")(function* (target: WatchTarget) {
194 const { thread, link, watch } = target;
195 const pullRequest = identityOf(link);
196 // A merged pull request cannot reopen, so its watch ends without a host read. A closed one
197 // can, so the host decides below. Settling ends watches; one left from before that rule
198 // ends here, so the thread does not show as working while settled.
199 if (link.snapshot?.state === "merged") return yield* record(target, null);
200 if (thread.settledOverride === "settled" || thread.settledAt !== null) {
201 return yield* record(target, null);
202 }
263 // A closed pull request can reopen, but the watch has nothing to report until then.
264 const closed = (target: WatchTarget) =>
265 record(target, null, {
266 text: `Pull request #${target.link.number} (${target.link.url}) was closed, so T3 Code stopped watching it. Call watch_pull_request if it reopens.`,
267 notification: {
268 source: { kind: "monitor" },
269 outcome: "updated",
270 summary: `#${target.link.number}: closed, stopped watching`,
271 },
272 });
273
274 /** Watches that end without a host read; the rest are read once per pull request. */
275 const endsWithoutRead = ({ thread, link }: WatchTarget) =>
276 // A merged pull request cannot reopen. Settling and archiving end watches, and a subagent
277 // cannot start one; a watch left from before those rules ends here.
278 link.snapshot?.state === "merged" ||
279 thread.settledOverride === "settled" ||
280 thread.settledAt !== null ||
281 thread.lineage.relationshipToParent === "subagent";
203282
204 const reference = { projectId: thread.projectId, ...pullRequest };
283 /**
284 * Runs one thread's step for each thread in a group, so one refusal does not skip the rest.
285 * Succeeds with whether every step landed.
286 */
287 const eachTarget = <E>(
288 group: WatchGroup,
289 step: (target: WatchTarget) => Effect.Effect<void, E>,
290 ) =>
291 Effect.forEach(group.targets, (target) =>
292 step(target).pipe(
293 Effect.as(true),
294 Effect.catchCause((cause) => {
295 // A thread that did not get its update must not wait for the quiet reread.
296 lastReads.delete(group.key);
297 return logFailure("pull request watch update failed", {
298 threadId: target.thread.id,
299 pullRequest: group.key,
300 })(cause).pipe(Effect.as(false));
301 }),
302 ),
303 ).pipe(Effect.map((landed) => landed.every(Boolean)));
304
305 const readGroup = Effect.fn("PullRequestWatchReactor.readGroup")(function* (group: WatchGroup) {
306 const now = yield* Clock.currentTimeMillis;
307 if (!needsRead(group, lastReads.get(group.key), now)) return;
308 const first = group.targets[0]!;
309 const reference = { projectId: first.thread.projectId, ...identityOf(first.link) };
205310 const read = yield* Effect.exit(
206311 Effect.all(
207312 [
......@@ -211,40 +316,58 @@ export const make = Effect.gen(function* () {
211316 { concurrency: 2 },
212317 ),
213318 );
214 // Only host reads count towards giving up; a refused wake is not the host's fault.
215 const key = failureKey(target);
216319 if (Exit.isFailure(read)) {
217320 if (Cause.hasInterruptsOnly(read.cause)) return yield* Effect.failCause(read.cause);
218 const failures = (readFailures.get(key) ?? 0) + 1;
219 readFailures.set(key, failures);
220 // The count stays until the stop lands, so a failed stop is tried again next pass.
221 if (failures >= READ_FAILURE_LIMIT) {
222 yield* giveUp(target);
223 readFailures.delete(key);
321 lastReads.delete(group.key);
322 // The host's pause refuses later reads without a request, so waiting it out is free.
323 if (isRateLimited(read.cause)) return;
324 const failures = (readFailures.get(group.key) ?? 0) + 1;
325 readFailures.set(group.key, failures);
326 // The count stays until every stop lands, so a failed stop is tried again on the next
327 // failed read, and a watch started after the stops begins from zero.
328 if (failures >= READ_FAILURE_LIMIT && (yield* eachTarget(group, giveUp))) {
329 readFailures.delete(group.key);
224330 }
225331 return yield* Effect.failCause(read.cause);
226332 }
227 readFailures.delete(key);
333 readFailures.delete(group.key);
228334 const [detail, activity] = read.value;
229 if (detail.state !== "open") return yield* record(target, null);
230
231 // Never advance the remark watermark past comments an incomplete read could have missed.
232 const remarks = yield* readRemarks(target, reference, activity);
233 const report = evaluatePullRequestWatch(watch, detail, remarks);
234 if (report.changes.length > 0) {
235 return yield* record(
236 target,
237 report.exhausted ? null : report.next,
238 pullRequestWatchMessage({
239 number: link.number,
240 url: link.url,
241 baseBranch: detail.baseBranch,
242 headSha: report.next.headSha,
243 report,
244 }),
335 if (detail.state !== "open") {
336 lastReads.delete(group.key);
337 return yield* eachTarget(group, (target) =>
338 detail.state === "closed" ? closed(target) : record(target, null),
245339 );
246340 }
247 if (!watchesEqual(report.next, watch)) yield* record(target, report.next);
341
342 // Never advance the remark watermark past comments an incomplete read could have missed.
343 const remarks = yield* readRemarks(group.key, reference, activity);
344 lastReads.set(group.key, {
345 at: now,
346 fingerprint: group.fingerprint,
347 // Comments a partial read could not see are read again next pass.
348 quiet:
349 remarks !== null &&
350 detail.mergeability !== "unknown" &&
351 detail.checks.every((check) => check.status !== "pending"),
352 watches: new Set(group.targets.map(watchKey)),
353 });
354 yield* eachTarget(group, (target) => {
355 const report = evaluatePullRequestWatch(target.watch, detail, remarks);
356 if (report.changes.length > 0) {
357 return record(
358 target,
359 report.exhausted ? null : report.next,
360 pullRequestWatchMessage({
361 number: target.link.number,
362 url: target.link.url,
363 baseBranch: detail.baseBranch,
364 headSha: report.next.headSha,
365 report,
366 }),
367 );
368 }
369 return watchesEqual(report.next, target.watch) ? Effect.void : record(target, report.next);
370 });
248371 });
249372
250373 const sweep = Effect.gen(function* () {
......@@ -254,20 +377,47 @@ export const make = Effect.gen(function* () {
254377 link.watch === undefined ? [] : [{ thread, link, watch: link.watch }],
255378 ),
256379 );
257 const keys = new Set(targets.map(failureKey));
258 for (const key of readFailures.keys()) if (!keys.has(key)) readFailures.delete(key);
259 for (const key of threadTails.keys()) if (!keys.has(key)) threadTails.delete(key);
380 const byPullRequest = new Map<string, Array<WatchTarget>>();
381 const ending: Array<WatchTarget> = [];
382 for (const target of targets) {
383 if (endsWithoutRead(target)) {
384 ending.push(target);
385 continue;
386 }
387 // Grouped per project too: each project reads through its own checkout, so one that cannot
388 // read the pull request must not end another project's watches.
389 const key = `${target.thread.projectId} ${threadPullRequestKeyOf(target.link)}`;
390 byPullRequest.set(key, [...(byPullRequest.get(key) ?? []), target]);
391 }
392 const groups = [...byPullRequest].map(([key, members]): WatchGroup => ({
393 key,
394 targets: members,
395 fingerprint: [...new Set(members.map(snapshotFingerprint))].toSorted().join("\n"),
396 }));
397 for (const cache of [readFailures, lastReads, threadTails]) {
398 for (const key of cache.keys()) if (!byPullRequest.has(key)) cache.delete(key);
399 }
260400 yield* Effect.forEach(
261 targets,
401 ending,
262402 (target) =>
263 check(target).pipe(
403 record(target, null).pipe(
264404 Effect.catchCause(
265 logFailure("pull request watch check failed", {
405 logFailure("pull request watch stop failed", {
266406 threadId: target.thread.id,
267407 pullRequest: threadPullRequestKeyOf(target.link),
268408 }),
269409 ),
270410 ),
411 { discard: true },
412 );
413 yield* Effect.forEach(
414 groups,
415 (group) =>
416 readGroup(group).pipe(
417 Effect.catchCause(
418 logFailure("pull request watch check failed", { pullRequest: group.key }),
419 ),
420 ),
271421 { concurrency: 4, discard: true },
272422 );
273423 }).pipe(
......@@ -276,7 +426,9 @@ export const make = Effect.gen(function* () {
276426 );
277427
278428 const start: PullRequestWatchReactor["Service"]["start"] = () =>
279 forkParked(sweep.pipe(Effect.repeat(Schedule.spaced("1 minute")), Effect.asVoid));
429 forkParked(
430 sweep.pipe(Effect.repeat(Schedule.spaced(`${SWEEP_MINUTES} minutes`)), Effect.asVoid),
431 );
280432
281433 return { start, sweep } satisfies PullRequestWatchReactor["Service"];
282434});
apps/server/src/orchestration-v2/ThreadStop.test.ts+2-23
......@@ -156,7 +156,8 @@ it.effect("Stop ends watches, holds queues, and stops the delegated tasks under
156156 yield* createWatchingThread(parentThreadId, 1);
157157 yield* send(parentThreadId, "first", "start_immediately");
158158 const childThreadId = yield* delegate(parentThreadId, "child task");
159 yield* watch(childThreadId, 2);
159 // The parent owns its pull requests; a delegated task cannot watch one.
160 assert.isTrue(Exit.isFailure(yield* Effect.exit(watch(childThreadId, 2))));
160161 const grandchildThreadId = yield* delegate(childThreadId, "grandchild task");
161162 yield* send(childThreadId, "child follow-up", "queue_after_active");
162163
......@@ -174,9 +175,6 @@ it.effect("Stop ends watches, holds queues, and stops the delegated tasks under
174175 });
175176 yield* send(parentThreadId, "second", "start_immediately");
176177 const secondRun = (yield* orchestrator.getThreadProjection(parentThreadId)).runs.at(-1)!;
177 const childWatch = (yield* orchestrator.getThreadProjection(childThreadId)).thread
178 .pullRequests?.[0]?.watch;
179 assert.isDefined(childWatch);
180178
181179 const stopCommandId = CommandId.make("stop-parent");
182180 yield* orchestrator.dispatch({
......@@ -209,25 +207,6 @@ it.effect("Stop ends watches, holds queues, and stops the delegated tasks under
209207 });
210208 assert.deepEqual((yield* threadState(grandchildThreadId)).runs, ["interrupted"]);
211209
212 // A watch read that raced the Stop cannot wake the stopped child.
213 const lateWake = yield* Effect.exit(
214 orchestrator.dispatch({
215 type: "thread.pull-request-watch.sync",
216 commandId: CommandId.make("late-watch-wake"),
217 threadId: childThreadId,
218 ...pullRequest(2),
219 startedAt: childWatch!.startedAt,
220 watch: null,
221 wake: {
222 messageId: MessageId.make("message:late-watch-wake"),
223 text: "Checks passed.",
224 notification: { source: { kind: "monitor" }, outcome: "completed", summary: "#2" },
225 },
226 }),
227 );
228 assert.isTrue(Exit.isFailure(lateWake));
229 assert.deepEqual((yield* threadState(childThreadId)).runs, ["interrupted", "queued:held"]);
230
231210 // A retried effect stops nothing twice.
232211 yield* threads.stopDelegatedTasks({ threadId: parentThreadId, commandId: stopCommandId });
233212 assert.deepEqual((yield* threadState(childThreadId)).runs, ["interrupted", "queued:held"]);
apps/server/src/orchestration-v2/pullRequestWatch.test.ts+2-4
......@@ -134,7 +134,7 @@ describe("evaluatePullRequestWatch", () => {
134134 ]);
135135 });
136136
137 it("reports passed again when a check shows up already passed where none is required", () => {
137 it("does not report passed again for a new passed check where none is required", () => {
138138 const first = evaluatePullRequestWatch(
139139 watch(),
140140 detail({ checks: [check("test", "success")] }),
......@@ -142,9 +142,7 @@ describe("evaluatePullRequestWatch", () => {
142142 );
143143 assert.deepEqual(first.changes, [{ kind: "checks-passed", count: 1, required: false }]);
144144 const both = detail({ checks: [check("test", "success"), check("lint", "success")] });
145 assert.deepEqual(evaluatePullRequestWatch(first.next, both, noRemarks).changes, [
146 { kind: "checks-passed", count: 2, required: false },
147 ]);
145 assert.deepEqual(evaluatePullRequestWatch(first.next, both, noRemarks).changes, []);
148146 });
149147
150148 it("does not wake a watch saved before passed checks were recorded", () => {
apps/server/src/orchestration-v2/pullRequestWatch.ts+3-2
......@@ -67,8 +67,9 @@ export function evaluatePullRequestWatch(
6767 const gateNames = gate.map((check) => check.name);
6868 // A watch saved before passedChecks existed takes the current names, so it does not wake.
6969 const told = passed && passedChecks.length === 0 ? gateNames : passedChecks;
70 // A required job created and finished between two passes is never seen pending.
71 const gateGrew = gateNames.some((name) => !told.includes(name));
70 // A required job created and finished between two passes is never seen pending. Without
71 // required checks, any check counts, and advisory bots keep adding passed ones: no wake.
72 const gateGrew = required.length > 0 && gateNames.some((name) => !told.includes(name));
7273 if (passedNow && (!passed || gateGrew)) {
7374 changes.push({ kind: "checks-passed", count: gate.length, required: required.length > 0 });
7475 }
apps/server/src/orchestration-v2/runtimeLayer.test.ts+259-46
......@@ -63,6 +63,7 @@ import * as ProviderRuntimeRecoveryService from "./ProviderRuntimeRecoveryServic
6363import * as ProjectionMaintenance from "./ProjectionMaintenance.ts";
6464import * as ProjectionStore from "./ProjectionStore.ts";
6565import * as PullRequestWatchReactor from "./PullRequestWatchReactor.ts";
66import { PullRequestProviderError } from "../pullRequest/PullRequestProvider.ts";
6667import * as PullRequestService from "../pullRequest/PullRequestService.ts";
6768import * as ProjectStore from "./ProjectStore.ts";
6869import type { ProviderAdapterV2SessionRuntime, ProviderAdapterV2Shape } from "./ProviderAdapter.ts";
......@@ -165,6 +166,58 @@ const TestProviderInstanceRegistry = Layer.succeed(
165166);
166167
167168/** Seed a project row the way a committed `project.created` event folds into it. */
169/** A watched pull request as the host reports it: open, mergeable, with a failing lint check. */
170const watchedPullRequestDetail = (input: {
171 readonly projectId: ProjectId;
172 readonly number: number;
173 readonly at: string;
174}): PullRequestDetail => ({
175 provider: "github",
176 capabilities: {
177 diff: true,
178 comment: true,
179 actions: [],
180 mergeMethods: [],
181 search: false,
182 review: { inlineComment: false, reply: false, resolve: false, verdicts: [] },
183 reviewers: { request: false, listCandidates: false },
184 },
185 viewerPermissions: {
186 actions: [],
187 comment: true,
188 resolve: true,
189 verdicts: [],
190 requestReviewers: false,
191 },
192 projectId: input.projectId,
193 projectTitle: "Watch wake",
194 workspaceRoot: "/workspace/watch",
195 repository: "pingdotgg/t3code",
196 number: input.number,
197 title: "Watched pull request",
198 body: "",
199 url: `https://github.com/pingdotgg/t3code/pull/${input.number}`,
200 author: { login: "agent-user", name: null, avatarUrl: null },
201 state: "open",
202 isDraft: false,
203 mergeability: "mergeable",
204 additions: 1,
205 deletions: 0,
206 changedFiles: 1,
207 headBranch: "feature",
208 headSha: "abc1234def",
209 baseBranch: "main",
210 createdAt: input.at,
211 updatedAt: input.at,
212 mergedAt: null,
213 closedAt: null,
214 reviewers: [],
215 labels: [],
216 checks: [{ name: "lint", status: "failure", description: null, url: null }],
217 mergeCapabilities: { merge: true, squash: true, rebase: true },
218 viewer: "agent-user",
219});
220
168221const seedProject = (input: {
169222 readonly projectId: ProjectId;
170223 readonly title: string;
......@@ -2352,6 +2405,63 @@ it.layer(TestLayer)("OrchestrationV2LayerLive lifecycle", (it) => {
23522405 }),
23532406 );
23542407
2408 it.effect("archiving ends watches, and an archived thread cannot start one", () =>
2409 Effect.gen(function* () {
2410 const orchestrator = yield* Orchestrator.OrchestratorV2;
2411 const threadId = ThreadId.make("runtime-pull-request-watch-archive");
2412 yield* orchestrator.dispatch({
2413 type: "thread.create",
2414 createdBy: "user",
2415 creationSource: "web",
2416 commandId: CommandId.make("pr-watch-archive-create"),
2417 threadId,
2418 projectId: ProjectId.make("pr-watch-archive-project"),
2419 title: "Archive watched PR",
2420 modelSelection,
2421 runtimeMode: "full-access",
2422 interactionMode: "default",
2423 branch: null,
2424 worktreePath: null,
2425 });
2426 const watch = {
2427 type: "thread.pull-request.watch" as const,
2428 threadId,
2429 host: "github.com",
2430 repository: "pingdotgg/t3code",
2431 number: 7,
2432 watching: true,
2433 link: { url: "https://github.com/pingdotgg/t3code/pull/7", source: "agent" as const },
2434 };
2435 const watched = Effect.map(
2436 orchestrator.getThreadShell(threadId),
2437 (thread) => thread?.pullRequests?.[0]?.watch !== undefined,
2438 );
2439 yield* orchestrator.dispatch({
2440 ...watch,
2441 commandId: CommandId.make("pr-watch-archive-start"),
2442 });
2443 assert.isTrue(yield* watched);
2444
2445 yield* orchestrator.dispatch({
2446 type: "thread.archive",
2447 commandId: CommandId.make("pr-watch-archive"),
2448 threadId,
2449 });
2450 assert.isFalse(yield* watched);
2451 const refused = yield* orchestrator
2452 .dispatch({ ...watch, commandId: CommandId.make("pr-watch-archive-restart") })
2453 .pipe(Effect.flip);
2454 assert.equal(refused._tag, "OrchestratorDispatchError");
2455
2456 yield* orchestrator.dispatch({
2457 type: "thread.unarchive",
2458 commandId: CommandId.make("pr-watch-unarchive"),
2459 threadId,
2460 });
2461 assert.isFalse(yield* watched);
2462 }),
2463 );
2464
23552465 it.effect("ends a watch it cannot read, and tells the agent", () =>
23562466 Effect.gen(function* () {
23572467 const orchestrator = yield* Orchestrator.OrchestratorV2;
......@@ -2411,6 +2521,154 @@ it.layer(TestLayer)("OrchestrationV2LayerLive lifecycle", (it) => {
24112521 }),
24122522 );
24132523
2524 it.effect(
2525 "reads a watched pull request once a pass, waits out rate limits, and skips quiet passes",
2526 () =>
2527 Effect.gen(function* () {
2528 const orchestrator = yield* Orchestrator.OrchestratorV2;
2529 const projectId = ProjectId.make("pr-watch-shared-project");
2530 yield* seedProject({
2531 projectId,
2532 title: "Watch shared",
2533 workspaceRoot: "/workspace/watch-shared",
2534 defaultModelSelection: null,
2535 createdAt: "2026-10-01T00:00:00.000Z",
2536 });
2537 const threadIds = ["one", "two"].map((name) =>
2538 ThreadId.make(`runtime-pull-request-watch-shared-${name}`),
2539 );
2540 const watchFrom = (threadId: ThreadId) =>
2541 Effect.gen(function* () {
2542 yield* orchestrator.dispatch({
2543 type: "thread.create",
2544 createdBy: "user",
2545 creationSource: "web",
2546 commandId: CommandId.make(`create:${threadId}`),
2547 threadId,
2548 projectId,
2549 title: "Watch shared",
2550 modelSelection,
2551 runtimeMode: "full-access",
2552 interactionMode: "default",
2553 branch: null,
2554 worktreePath: null,
2555 });
2556 yield* orchestrator.dispatch({
2557 type: "thread.pull-request.watch",
2558 commandId: CommandId.make(`watch:${threadId}`),
2559 threadId,
2560 host: "github.com",
2561 repository: "pingdotgg/t3code",
2562 number: 9,
2563 watching: true,
2564 link: { url: "https://github.com/pingdotgg/t3code/pull/9", source: "agent" },
2565 });
2566 });
2567 for (const threadId of threadIds) yield* watchFrom(threadId);
2568 const rateLimited = new PullRequestOperationError({
2569 operation: "getChangeRequest",
2570 detail: "github requests are paused until the rate limit resets",
2571 cause: new PullRequestProviderError({
2572 provider: "github",
2573 operation: "getChangeRequest",
2574 reason: "rate-limited",
2575 detail: "paused",
2576 }),
2577 });
2578 let host: "rate-limited" | "open" | "closed" = "rate-limited";
2579 let reads = 0;
2580 const reactor = yield* PullRequestWatchReactor.make.pipe(
2581 Effect.provide(
2582 Layer.mergeAll(
2583 NodeServices.layer,
2584 Layer.mock(PullRequestService.PullRequestService)({
2585 detail: () =>
2586 Effect.suspend(() => {
2587 reads += 1;
2588 return host === "rate-limited"
2589 ? Effect.fail(rateLimited)
2590 : Effect.succeed({
2591 ...watchedPullRequestDetail({
2592 projectId,
2593 number: 9,
2594 at: "2026-10-02T12:00:00.000Z",
2595 }),
2596 state: host,
2597 checks: [
2598 { name: "lint", status: "success", description: null, url: null },
2599 ],
2600 });
2601 }),
2602 activity: () =>
2603 host === "rate-limited"
2604 ? Effect.fail(rateLimited)
2605 : Effect.succeed({
2606 comments: [],
2607 commentCount: 0,
2608 commentsTruncated: false,
2609 reviewThreads: [],
2610 commits: [],
2611 }),
2612 }),
2613 ),
2614 ),
2615 );
2616 const summaries = (threadId: ThreadId) =>
2617 Effect.map(orchestrator.getThreadRecords(threadId, ["messages"]), ({ messages }) =>
2618 messages.flatMap((message) => message.notification?.summary ?? []),
2619 );
2620 const watching = (threadId: ThreadId) =>
2621 Effect.map(
2622 orchestrator.getThreadShell(threadId),
2623 (thread) => thread?.pullRequests?.[0]?.watch !== undefined,
2624 );
2625
2626 // However long the host stays rate limited, the watch waits instead of giving up, and
2627 // both threads share one read a pass.
2628 for (let pass = 0; pass < 20; pass += 1) yield* reactor.sweep;
2629 assert.equal(reads, 20);
2630 for (const threadId of threadIds) {
2631 assert.isTrue(yield* watching(threadId));
2632 assert.deepEqual(yield* summaries(threadId), []);
2633 }
2634
2635 host = "open";
2636 yield* reactor.sweep;
2637 assert.equal(reads, 21);
2638 for (const threadId of threadIds) {
2639 assert.deepEqual(yield* summaries(threadId), ["#9: checks passed"]);
2640 }
2641
2642 // Nothing in flight and nothing moved, so the next pass spends no request.
2643 yield* reactor.sweep;
2644 assert.equal(reads, 21);
2645
2646 // A new watch on the quiet pull request takes its first look on the next pass.
2647 const late = ThreadId.make("runtime-pull-request-watch-shared-late");
2648 yield* watchFrom(late);
2649 yield* reactor.sweep;
2650 assert.equal(reads, 22);
2651 assert.deepEqual(yield* summaries(late), ["#9: checks passed"]);
2652 for (const threadId of threadIds) {
2653 assert.deepEqual(yield* summaries(threadId), ["#9: checks passed"]);
2654 }
2655 threadIds.push(late);
2656
2657 // The quiet reread finds the pull request closed, ends both watches, and says so.
2658 host = "closed";
2659 yield* TestClock.adjust("10 minutes");
2660 yield* reactor.sweep;
2661 assert.equal(reads, 23);
2662 for (const threadId of threadIds) {
2663 assert.isFalse(yield* watching(threadId));
2664 assert.deepEqual(yield* summaries(threadId), [
2665 "#9: checks passed",
2666 "#9: closed, stopped watching",
2667 ]);
2668 }
2669 }),
2670 );
2671
24142672 it.effect.each([
24152673 "single page",
24162674 "paginated",
......@@ -2463,52 +2721,7 @@ it.layer(TestLayer)("OrchestrationV2LayerLive lifecycle", (it) => {
24632721 });
24642722
24652723 const at = "2026-10-02T12:00:00.000Z";
2466 const detail: PullRequestDetail = {
2467 provider: "github",
2468 capabilities: {
2469 diff: true,
2470 comment: true,
2471 actions: [],
2472 mergeMethods: [],
2473 search: false,
2474 review: { inlineComment: false, reply: false, resolve: false, verdicts: [] },
2475 reviewers: { request: false, listCandidates: false },
2476 },
2477 viewerPermissions: {
2478 actions: [],
2479 comment: true,
2480 resolve: true,
2481 verdicts: [],
2482 requestReviewers: false,
2483 },
2484 projectId,
2485 projectTitle: "Watch wake",
2486 workspaceRoot: "/workspace/watch",
2487 repository: key.repository,
2488 number: key.number,
2489 title: "Watched pull request",
2490 body: "",
2491 url,
2492 author: { login: "agent-user", name: null, avatarUrl: null },
2493 state: "open",
2494 isDraft: false,
2495 mergeability: "mergeable",
2496 additions: 1,
2497 deletions: 0,
2498 changedFiles: 1,
2499 headBranch: "feature",
2500 headSha: "abc1234def",
2501 baseBranch: "main",
2502 createdAt: at,
2503 updatedAt: at,
2504 mergedAt: null,
2505 closedAt: null,
2506 reviewers: [],
2507 labels: [],
2508 checks: [{ name: "lint", status: "failure", description: null, url: null }],
2509 mergeCapabilities: { merge: true, squash: true, rebase: true },
2510 viewer: "agent-user",
2511 };
2724 const detail = watchedPullRequestDetail({ projectId, number: key.number, at });
25122725 const initialWatch = (yield* orchestrator.getThreadShell(threadId))?.pullRequests?.[0]?.watch;
25132726 assert.isDefined(initialWatch);
25142727 const remark: PullRequestComment = {
apps/server/src/orchestration-v2/testkit/pullRequestFixtures.ts+6-1
......@@ -23,6 +23,7 @@ export interface PullRequestTestThread {
2323 readonly linkedPullRequest?: ThreadLinkedPullRequest | null;
2424 readonly branchPullRequest?: ThreadLinkedPullRequest | null;
2525 readonly latestUserMessageAt: string | null;
26 readonly lineage?: OrchestrationV2ThreadShell["lineage"];
2627 readonly createdAt: string;
2728 readonly updatedAt: string;
2829 readonly archivedAt: string | null;
......@@ -47,7 +48,11 @@ export function v2PullRequestThread(thread: PullRequestTestThread): Orchestratio
4748 createdBy: "user",
4849 creationSource: "web",
4950 activeProviderThreadId: null,
50 lineage: { rootThreadId: thread.id, parentThreadId: null, relationshipToParent: null },
51 lineage: thread.lineage ?? {
52 rootThreadId: thread.id,
53 parentThreadId: null,
54 relationshipToParent: null,
55 },
5156 forkedFrom: null,
5257 latestRunId: null,
5358 activeRunId: null,
docs/user/source-control.md+10-7
......@@ -192,13 +192,16 @@ when requested. With **Auto-settle merged threads** enabled, a thread can settle
192192review is terminal. An open or unsynced link keeps it active.
193193
194194Ask the agent to watch, monitor, or babysit a pull request and it calls `watch_pull_request`. While
195the thread is active, the server checks the pull request every minute and wakes the agent when a check
196fails, the required checks pass, someone else comments or reviews, or the branch starts to conflict.
197Comments from your own account do not wake it. Watching ends when the pull request merges or closes,
198after 10 wakes in a row that bring only comments, when the server cannot read the pull request for
19915 minutes, or when you press Stop on the thread. Settling a thread also ends all its watches.
200Unsettle the thread before starting a new watch. To start or stop it yourself, use the row menu in
201the **Linked pull requests** panel.
195the thread is active, the server checks the pull request every two minutes and wakes the agent when a
196check fails, the required checks pass, someone else comments or reviews, or the branch starts to
197conflict. Threads in a project that watch the same pull request share one check, and a pull request
198with nothing in progress is checked again when something changes or every 10 minutes, which keeps
199watching inside GitHub's rate limit. Comments from your own account do not wake it. Watching ends when the pull request merges or closes,
200after 10 wakes in a row that bring only comments, after 8 failed reads in a row, or when you press
201Stop on the thread. A rate limit only pauses watching. Settling or archiving a thread also ends all
202its watches. Unsettle the thread before starting a new watch. Subagents cannot watch pull requests;
203the thread that delegated to them does. To start or stop it yourself, use the row menu in the
204**Linked pull requests** panel.
202205
203206A watched thread counts as working between wakes, so it stays in the **Working** section and does
204207not auto-settle. Agents stop watching when they hand the work back to you, and the thread then