| ... | @@ -91,6 +91,44 @@ interface LastRead { | ... | @@ -91,6 +91,44 @@ interface LastRead { |
| 91 | | 91 | |
| 92 | const watchKey = ({ thread, watch }: WatchTarget) => `${thread.id} ${watch.startedAt}`; | 92 | const watchKey = ({ thread, watch }: WatchTarget) => `${thread.id} ${watch.startedAt}`; |
| 93 | | 93 | |
| | 94 | /** |
| | 95 | * Why a watch ended. `stopped` is anything outside this reactor: the agent unwatched, the user |
| | 96 | * pressed Stop, or the thread settled, archived, or was deleted between passes. |
| | 97 | */ |
| | 98 | type WatchEndReason = |
| | 99 | | "merged" |
| | 100 | | "closed" |
| | 101 | | "unreadable" |
| | 102 | | "comment-limit" |
| | 103 | | "settled" |
| | 104 | | "subagent" |
| | 105 | | "stopped"; |
| | 106 | |
| | 107 | /** |
| | 108 | * What one watch did while this server ran, logged once when it ends so we can see how long |
| | 109 | * watches stay quiet. Kept in memory: a watch older than the server process only has partial |
| | 110 | * numbers, and one that ends while the server is down, or starts and ends between two passes, |
| | 111 | * is not logged. |
| | 112 | */ |
| | 113 | interface WatchLife { |
| | 114 | readonly threadId: string; |
| | 115 | readonly pullRequest: string; |
| | 116 | readonly startedAt: number; |
| | 117 | /** The head commit the last successful read saw. */ |
| | 118 | readonly headSha: string | null; |
| | 119 | /** When a pass last saw the head commit move, or the start. */ |
| | 120 | readonly pushedAt: number; |
| | 121 | /** Longest time between pushes, not counting the time since the last one. */ |
| | 122 | readonly longestQuietMs: number; |
| | 123 | readonly wakes: number; |
| | 124 | readonly reads: number; |
| | 125 | } |
| | 126 | |
| | 127 | const lifeKey = ({ thread, link, watch }: WatchTarget) => |
| | 128 | `${thread.id} ${threadPullRequestKeyOf(link)} ${watch.startedAt}`; |
| | 129 | |
| | 130 | const minutes = (ms: number) => Math.max(0, Math.round(ms / 60_000)); |
| | 131 | |
| 94 | const snapshotFingerprint = ({ link }: WatchTarget) => { | 132 | const snapshotFingerprint = ({ link }: WatchTarget) => { |
| 95 | const snapshot = link.snapshot; | 133 | const snapshot = link.snapshot; |
| 96 | return snapshot === null | 134 | return snapshot === null |
| ... | @@ -136,7 +174,8 @@ function watchesEqual(left: ThreadPullRequestWatch, right: ThreadPullRequestWatc | ... | @@ -136,7 +174,8 @@ function watchesEqual(left: ThreadPullRequestWatch, right: ThreadPullRequestWatc |
| 136 | * of a project that watch it, and skips one whose sync snapshot has not moved while nothing is | 174 | * of a project that watch it, and skips one whose sync snapshot has not moved while nothing is |
| 137 | * in flight. | 175 | * in flight. |
| 138 | * Settling or archiving a thread ends its watches, and a merged or closed pull request ends | 176 | * Settling or archiving a thread ends its watches, and a merged or closed pull request ends |
| 139 | * its watch. | 177 | * its watch. Each ended watch is logged once with why it ended and how long it went without |
| | 178 | * a push, for debugging watches that live too long. |
| 140 | */ | 179 | */ |
| 141 | export class PullRequestWatchReactor extends Context.Service< | 180 | export class PullRequestWatchReactor extends Context.Service< |
| 142 | PullRequestWatchReactor, | 181 | PullRequestWatchReactor, |
| ... | @@ -153,6 +192,51 @@ export const make = Effect.gen(function* () { | ... | @@ -153,6 +192,51 @@ export const make = Effect.gen(function* () { |
| 153 | const projections = yield* ProjectionStore.ProjectionStoreV2; | 192 | const projections = yield* ProjectionStore.ProjectionStoreV2; |
| 154 | const pullRequests = yield* PullRequestService.PullRequestService; | 193 | const pullRequests = yield* PullRequestService.PullRequestService; |
| 155 | const crypto = yield* Crypto.Crypto; | 194 | const crypto = yield* Crypto.Crypto; |
| | 195 | const bootedAt = yield* Clock.currentTimeMillis; |
| | 196 | const lives = new Map<string, WatchLife>(); |
| | 197 | |
| | 198 | const lifeOf = (target: WatchTarget): WatchLife => { |
| | 199 | const existing = lives.get(lifeKey(target)); |
| | 200 | if (existing !== undefined) return existing; |
| | 201 | const startedAt = Date.parse(target.watch.startedAt); |
| | 202 | const life: WatchLife = { |
| | 203 | threadId: target.thread.id, |
| | 204 | pullRequest: threadPullRequestKeyOf(target.link), |
| | 205 | startedAt, |
| | 206 | headSha: target.watch.headSha, |
| | 207 | pushedAt: startedAt, |
| | 208 | longestQuietMs: 0, |
| | 209 | wakes: 0, |
| | 210 | reads: 0, |
| | 211 | }; |
| | 212 | lives.set(lifeKey(target), life); |
| | 213 | return life; |
| | 214 | }; |
| | 215 | const woke = (target: WatchTarget) => |
| | 216 | Effect.sync(() => { |
| | 217 | const life = lifeOf(target); |
| | 218 | lives.set(lifeKey(target), { ...life, wakes: life.wakes + 1 }); |
| | 219 | }); |
| | 220 | |
| | 221 | const reportEnd = (key: string, life: WatchLife, reason: WatchEndReason) => |
| | 222 | Effect.gen(function* () { |
| | 223 | lives.delete(key); |
| | 224 | const now = yield* Clock.currentTimeMillis; |
| | 225 | const quietMs = now - life.pushedAt; |
| | 226 | yield* Effect.logInfo("pull request watch ended", { |
| | 227 | threadId: life.threadId, |
| | 228 | pullRequest: life.pullRequest, |
| | 229 | reason, |
| | 230 | minutes: minutes(now - life.startedAt), |
| | 231 | quietMinutes: minutes(quietMs), |
| | 232 | longestQuietMinutes: minutes(Math.max(life.longestQuietMs, quietMs)), |
| | 233 | wakes: life.wakes, |
| | 234 | reads: life.reads, |
| | 235 | partial: life.startedAt < bootedAt, |
| | 236 | }); |
| | 237 | }); |
| | 238 | const ended = (target: WatchTarget, reason: WatchEndReason) => |
| | 239 | Effect.suspend(() => reportEnd(lifeKey(target), lifeOf(target), reason)); |
| 156 | | 240 | |
| 157 | // Reads in a row that failed, per pull request. Kept in memory: a restart only delays the stop. | 241 | // Reads in a row that failed, per pull request. Kept in memory: a restart only delays the stop. |
| 158 | const readFailures = new Map<string, number>(); | 242 | const readFailures = new Map<string, number>(); |
| ... | @@ -205,7 +289,11 @@ export const make = Effect.gen(function* () { | ... | @@ -205,7 +289,11 @@ export const make = Effect.gen(function* () { |
| 205 | outcome: "failed", | 289 | outcome: "failed", |
| 206 | summary: `#${target.link.number}: stopped watching, could not read it`, | 290 | summary: `#${target.link.number}: stopped watching, could not read it`, |
| 207 | }, | 291 | }, |
| 208 | }).pipe(Effect.catch(() => record(target, null))); | 292 | }).pipe( |
| | 293 | Effect.tap(() => woke(target)), |
| | 294 | Effect.catch(() => record(target, null)), |
| | 295 | Effect.tap(() => ended(target, "unreadable")), |
| | 296 | ); |
| 209 | | 297 | |
| 210 | const readRemarks = Effect.fn("PullRequestWatchReactor.readRemarks")( | 298 | const readRemarks = Effect.fn("PullRequestWatchReactor.readRemarks")( |
| 211 | function* (key: string, reference: PullRequestRef, activity: PullRequestActivity) { | 299 | function* (key: string, reference: PullRequestRef, activity: PullRequestActivity) { |
| ... | @@ -269,16 +357,22 @@ export const make = Effect.gen(function* () { | ... | @@ -269,16 +357,22 @@ export const make = Effect.gen(function* () { |
| 269 | outcome: "updated", | 357 | outcome: "updated", |
| 270 | summary: `#${target.link.number}: closed, stopped watching`, | 358 | summary: `#${target.link.number}: closed, stopped watching`, |
| 271 | }, | 359 | }, |
| 272 | }); | 360 | }).pipe( |
| | 361 | Effect.tap(() => woke(target)), |
| | 362 | Effect.tap(() => ended(target, "closed")), |
| | 363 | ); |
| 273 | | 364 | |
| 274 | /** Watches that end without a host read; the rest are read once per pull request. */ | 365 | /** Why a watch ends without a host read; the rest are read once per pull request. */ |
| 275 | const endsWithoutRead = ({ thread, link }: WatchTarget) => | 366 | const endsWithoutRead = ({ thread, link }: WatchTarget): WatchEndReason | undefined => |
| 276 | // A merged pull request cannot reopen. Settling and archiving end watches, and a subagent | 367 | // 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. | 368 | // cannot start one; a watch left from before those rules ends here. |
| 278 | link.snapshot?.state === "merged" || | 369 | link.snapshot?.state === "merged" |
| 279 | thread.settledOverride === "settled" || | 370 | ? "merged" |
| 280 | thread.settledAt !== null || | 371 | : thread.settledOverride === "settled" || thread.settledAt !== null |
| 281 | thread.lineage.relationshipToParent === "subagent"; | 372 | ? "settled" |
| | 373 | : thread.lineage.relationshipToParent === "subagent" |
| | 374 | ? "subagent" |
| | 375 | : undefined; |
| 282 | | 376 | |
| 283 | /** | 377 | /** |
| 284 | * Runs one thread's step for each thread in a group, so one refusal does not skip the rest. | 378 | * Runs one thread's step for each thread in a group, so one refusal does not skip the rest. |
| ... | @@ -332,10 +426,26 @@ export const make = Effect.gen(function* () { | ... | @@ -332,10 +426,26 @@ export const make = Effect.gen(function* () { |
| 332 | } | 426 | } |
| 333 | readFailures.delete(group.key); | 427 | readFailures.delete(group.key); |
| 334 | const [detail, activity] = read.value; | 428 | const [detail, activity] = read.value; |
| | 429 | const headSha = detail.headSha ?? null; |
| | 430 | for (const target of group.targets) { |
| | 431 | const life = lifeOf(target); |
| | 432 | // The first read only learns the head, so it is not a push. |
| | 433 | const pushed = life.headSha !== null && headSha !== life.headSha; |
| | 434 | lives.set(lifeKey(target), { |
| | 435 | ...life, |
| | 436 | headSha, |
| | 437 | reads: life.reads + 1, |
| | 438 | ...(pushed |
| | 439 | ? { pushedAt: now, longestQuietMs: Math.max(life.longestQuietMs, now - life.pushedAt) } |
| | 440 | : {}), |
| | 441 | }); |
| | 442 | } |
| 335 | if (detail.state !== "open") { | 443 | if (detail.state !== "open") { |
| 336 | lastReads.delete(group.key); | 444 | lastReads.delete(group.key); |
| 337 | return yield* eachTarget(group, (target) => | 445 | return yield* eachTarget(group, (target) => |
| 338 | detail.state === "closed" ? closed(target) : record(target, null), | 446 | detail.state === "closed" |
| | 447 | ? closed(target) |
| | 448 | : record(target, null).pipe(Effect.tap(() => ended(target, "merged"))), |
| 339 | ); | 449 | ); |
| 340 | } | 450 | } |
| 341 | | 451 | |
| ... | @@ -364,6 +474,9 @@ export const make = Effect.gen(function* () { | ... | @@ -364,6 +474,9 @@ export const make = Effect.gen(function* () { |
| 364 | headSha: report.next.headSha, | 474 | headSha: report.next.headSha, |
| 365 | report, | 475 | report, |
| 366 | }), | 476 | }), |
| | 477 | ).pipe( |
| | 478 | Effect.tap(() => woke(target)), |
| | 479 | Effect.tap(() => (report.exhausted ? ended(target, "comment-limit") : Effect.void)), |
| 367 | ); | 480 | ); |
| 368 | } | 481 | } |
| 369 | return watchesEqual(report.next, target.watch) ? Effect.void : record(target, report.next); | 482 | return watchesEqual(report.next, target.watch) ? Effect.void : record(target, report.next); |
| ... | @@ -377,11 +490,20 @@ export const make = Effect.gen(function* () { | ... | @@ -377,11 +490,20 @@ export const make = Effect.gen(function* () { |
| 377 | link.watch === undefined ? [] : [{ thread, link, watch: link.watch }], | 490 | link.watch === undefined ? [] : [{ thread, link, watch: link.watch }], |
| 378 | ), | 491 | ), |
| 379 | ); | 492 | ); |
| | 493 | // A watch seen last pass and gone now was ended outside this reactor. |
| | 494 | const present = new Set(targets.map(lifeKey)); |
| | 495 | yield* Effect.forEach( |
| | 496 | [...lives].filter(([key]) => !present.has(key)), |
| | 497 | ([key, life]) => reportEnd(key, life, "stopped"), |
| | 498 | { discard: true }, |
| | 499 | ); |
| | 500 | for (const target of targets) lifeOf(target); |
| 380 | const byPullRequest = new Map<string, Array<WatchTarget>>(); | 501 | const byPullRequest = new Map<string, Array<WatchTarget>>(); |
| 381 | const ending: Array<WatchTarget> = []; | 502 | const ending: Array<readonly [WatchTarget, WatchEndReason]> = []; |
| 382 | for (const target of targets) { | 503 | for (const target of targets) { |
| 383 | if (endsWithoutRead(target)) { | 504 | const reason = endsWithoutRead(target); |
| 384 | ending.push(target); | 505 | if (reason !== undefined) { |
| | 506 | ending.push([target, reason]); |
| 385 | continue; | 507 | continue; |
| 386 | } | 508 | } |
| 387 | // Grouped per project too: each project reads through its own checkout, so one that cannot | 509 | // Grouped per project too: each project reads through its own checkout, so one that cannot |
| ... | @@ -399,8 +521,9 @@ export const make = Effect.gen(function* () { | ... | @@ -399,8 +521,9 @@ export const make = Effect.gen(function* () { |
| 399 | } | 521 | } |
| 400 | yield* Effect.forEach( | 522 | yield* Effect.forEach( |
| 401 | ending, | 523 | ending, |
| 402 | (target) => | 524 | ([target, reason]) => |
| 403 | record(target, null).pipe( | 525 | record(target, null).pipe( |
| | 526 | Effect.tap(() => ended(target, reason)), |
| 404 | Effect.catchCause( | 527 | Effect.catchCause( |
| 405 | logFailure("pull request watch stop failed", { | 528 | logFailure("pull request watch stop failed", { |
| 406 | threadId: target.thread.id, | 529 | threadId: target.thread.id, |
Theo Browne <me@t3.gg> 2026-10-05 17:15:34-07:00
GitHub <noreply@github.com> 2026-10-05 17:15:34-07:00