authorclover caruso <git@paperclover.net> 2026-10-14 14:12:29-07:00
committerclover caruso <git@paperclover.net> 2026-10-16 16:52:04-07:00
log28cb2bd6523c6c53c7c801784e5a2724cee49a3f
tree722f93022e92f3d54d74fe4ce641fc9fd90544c2
parente967bf6d83336edcf6c8bf6ea1e22edb45f32a16
signature Signed by SSH key SHA256:cOKiuRFOeSRxne6EWgHtdQQSlBxjOXm2hOCFnCdLQbQ (~clover)

fix: resume PR watches when a pull request merges

Clover's PR watch ended after a merge without starting the promised follow-up turn. Atomically clear the watch and enqueue one wake when a fresh observation confirms the merge. Configured work watches read verified Mergebot checks and comments and react to its live updates. Serialize sweeps and distinguish check reruns without inventing native permissions or falling back to GitHub. Ordinary repositories retain their provider reads. Do not resume settled or archived chats. Preserve guards against stopping or restarting a watch while a read is in flight, and let unavailable work data pause observation. Assisted-by: gpt-6.1-sol

3 files changed, 431 insertions(+), 36 deletions(-)

apps/server/src/orchestration-v2/PullRequestWatchReactor.test.ts created+319
...@@ -0,0 +1,319 @@
1import * as NodeCrypto from "@effect/platform-node/NodeCrypto";
2import { assert, it } from "@effect/vitest";
3import {
4 CommandId,
5 ProjectId,
6 ProviderDriverKind,
7 ProviderInstanceId,
8 SourceControlProviderKind,
9 ThreadId,
10 type PullRequestDetail,
11 type MergebotPullRequestDetails,
12 type ThreadPullRequestSnapshot,
13} from "@t3tools/contracts";
14import type * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter";
15import * as Effect from "effect/Effect";
16import * as Layer from "effect/Layer";
17import * as Ref from "effect/Ref";
18import * as Stream from "effect/Stream";
19import * as MergebotQueue from "../mergebot/MergebotQueue.ts";
20import * as TestClock from "effect/testing/TestClock";
21
22import * as OrchestrationEventStore from "../persistence/OrchestrationEventStore.ts";
23import * as SqlitePersistence from "../persistence/Sqlite.ts";
24import * as PullRequestService from "../pullRequest/PullRequestService.ts";
25import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts";
26import * as Orchestrator from "./Orchestrator.ts";
27import * as ProjectionStore from "./ProjectionStore.ts";
28import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts";
29import * as PullRequestWatchReactor from "./PullRequestWatchReactor.ts";
30import * as ProviderReplayHarness from "./testkit/ProviderReplayHarness.ts";
31
32const threadId = ThreadId.make("watch-merge");
33const projectId = ProjectId.make("watch-project");
34const instanceId = ProviderInstanceId.make("codex");
35const key = { host: "github.com", repository: "owner/repository", number: 42 };
36const url = "https://github.com/owner/repository/pull/42";
37const timestamp = "2026-10-14T12:00:00.000Z";
38const adapter = {
39 instanceId,
40 driver: ProviderDriverKind.make("codex"),
41 getCapabilities: () => Effect.succeed(CodexProviderCapabilitiesV2),
42 planSelectionTransition: () => Effect.succeed({ type: "apply_on_next_turn" as const }),
43 openSession: () => Effect.die("Merge-watch tests do not start providers"),
44} as ProviderAdapter.ProviderAdapterV2["Service"];
45const database = SqlitePersistence.layerMemory;
46const layerTest = Layer.mergeAll(
47 NodeCrypto.layer,
48 ProjectionStore.layer.pipe(Layer.provide(database)),
49 OrchestrationEventStore.layer.pipe(Layer.provide(database)),
50 ProviderReplayHarness.layerWithRegistry(
51 { name: "pull-request-watch" },
52 ProviderAdapterRegistry.layerFromAdapters([adapter]),
53 { databaseLayer: database, runEffectWorker: false },
54 ),
55);
56
57const detail: PullRequestDetail = {
58 provider: SourceControlProviderKind.make("github"),
59 capabilities: {
60 diff: false,
61 comment: false,
62 actions: [],
63 mergeMethods: [],
64 search: false,
65 review: { inlineComment: false, reply: false, resolve: false, verdicts: [] },
66 reviewers: { request: false, listCandidates: false },
67 },
68 viewerPermissions: {
69 actions: [],
70 comment: false,
71 resolve: false,
72 verdicts: [],
73 requestReviewers: false,
74 },
75 projectId,
76 projectTitle: "Project",
77 workspaceRoot: "/workspace",
78 repository: key.repository,
79 number: key.number,
80 title: "Change",
81 body: "",
82 url,
83 author: null,
84 state: "merged",
85 isDraft: false,
86 mergeability: "mergeable",
87 additions: 0,
88 deletions: 0,
89 changedFiles: 0,
90 headBranch: "feature",
91 baseBranch: "main",
92 createdAt: timestamp,
93 updatedAt: timestamp,
94 mergedAt: timestamp,
95 closedAt: null,
96 reviewers: [],
97 labels: [],
98 checks: [],
99 mergeCapabilities: { merge: false, squash: false, rebase: false },
100};
101
102const makeHarness = Effect.fn("makeMergeWatchHarness")(function* (options: {
103 readonly mergebot?: MergebotQueue.MergebotQueue["Service"];
104 readonly snapshotState: ThreadPullRequestSnapshot["state"];
105 readonly beforeRead?: (engine: Orchestrator.OrchestratorV2["Service"]) => Effect.Effect<void>;
106}) {
107 const engine = yield* Orchestrator.OrchestratorV2;
108 yield* engine.dispatch({
109 type: "thread.create",
110 commandId: CommandId.make("create"),
111 threadId,
112 projectId,
113 title: "Wait for merge",
114 modelSelection: { instanceId, model: "gpt-5.1-codex" },
115 runtimeMode: "full-access",
116 interactionMode: "default",
117 branch: null,
118 worktreePath: null,
119 createdBy: "user",
120 creationSource: "web",
121 });
122 yield* engine.dispatch({
123 type: "thread.pull-request.watch",
124 commandId: CommandId.make("watch"),
125 threadId,
126 ...key,
127 watching: true,
128 link: { url, source: "agent" },
129 });
130 yield* engine.dispatch({
131 type: "thread.pull-request-link.sync",
132 commandId: CommandId.make("snapshot"),
133 threadId,
134 ...key,
135 snapshot: {
136 state: options.snapshotState,
137 title: "Change",
138 headBranch: "feature",
139 baseBranch: "main",
140 isDraft: false,
141 updatedAt: timestamp,
142 syncedAt: timestamp,
143 },
144 stack: null,
145 });
146 const reads = yield* Ref.make(0);
147 const reactor = yield* PullRequestWatchReactor.make.pipe(
148 Effect.provide(
149 Layer.merge(
150 options.mergebot
151 ? Layer.succeed(MergebotQueue.MergebotQueue, options.mergebot)
152 : Layer.mock(MergebotQueue.MergebotQueue)({
153 configuration: null,
154 streamChanges: Stream.empty,
155 }),
156 Layer.mock(PullRequestService.PullRequestService)({
157 watchFingerprint: () => Effect.succeed(null),
158 detail: () =>
159 Ref.update(reads, (count) => count + 1).pipe(
160 Effect.andThen(options.beforeRead?.(engine) ?? Effect.void),
161 Effect.as(detail),
162 ),
163 activity: () =>
164 Effect.succeed({
165 comments: [],
166 commentCount: 0,
167 commentsTruncated: false,
168 reviewThreads: [],
169 commits: [],
170 }),
171 }),
172 ),
173 ),
174 );
175 return { engine, reactor, reads };
176});
177
178it.effect.each(["merged", "open"] as const)(
179 "wakes once and ends the watch when the %s snapshot leads to a merged PR",
180 (snapshotState) =>
181 Effect.gen(function* () {
182 const { engine, reactor, reads } = yield* makeHarness({ snapshotState });
183 yield* reactor.sweep;
184 const first = yield* engine.getThreadRecords(threadId, ["runs", "turnItems"]);
185 assert.isUndefined(first.thread.pullRequests?.[0]?.watch);
186 assert.equal(first.runs.length, 1);
187 const notice = first.turnItems.find((item) => item.type === "notification");
188 assert.ok(notice?.type === "notification");
189 assert.equal(notice.summary, "Pull request #42 merged");
190 assert.equal(notice.outcome, "completed");
191 assert.deepEqual(notice.source, { kind: "monitor" });
192 assert.equal(yield* Ref.get(reads), snapshotState === "merged" ? 0 : 1);
193 yield* reactor.sweep;
194 const second = yield* engine.getThreadRecords(threadId, ["runs", "turnItems"]);
195 assert.equal(second.runs.length, 1);
196 assert.equal(second.turnItems.filter((item) => item.type === "notification").length, 1);
197 }).pipe(Effect.provide(layerTest)),
198);
199
200it.effect.each(["stop", "settle", "archive", "restart"] as const)(
201 "a %s during the host read prevents a stale merged wake",
202 (action) =>
203 Effect.gen(function* () {
204 const { engine, reactor } = yield* makeHarness({
205 snapshotState: "open",
206 beforeRead: (engine) =>
207 Effect.gen(function* () {
208 if (action === "settle" || action === "archive") {
209 yield* engine.dispatch({
210 type: action === "settle" ? "thread.settle" : "thread.archive",
211 commandId: CommandId.make(action),
212 threadId,
213 });
214 return;
215 }
216 yield* engine.dispatch({
217 type: "thread.pull-request.watch",
218 commandId: CommandId.make("stop"),
219 threadId,
220 ...key,
221 watching: false,
222 });
223 if (action === "restart") {
224 yield* TestClock.adjust(1);
225 yield* engine.dispatch({
226 type: "thread.pull-request.watch",
227 commandId: CommandId.make("restart"),
228 threadId,
229 ...key,
230 watching: true,
231 });
232 }
233 }).pipe(Effect.orDie),
234 });
235 yield* reactor.sweep;
236 const records = yield* engine.getThreadRecords(threadId, ["runs", "turnItems"]);
237 assert.equal(records.runs.length, 0);
238 assert.equal(records.turnItems.filter((item) => item.type === "notification").length, 0);
239 if (action === "restart") assert.isDefined(records.thread.pullRequests?.[0]?.watch);
240 else assert.isUndefined(records.thread.pullRequests?.[0]?.watch);
241 if (action === "settle") assert.equal(records.thread.settledOverride, "settled");
242 if (action === "archive") assert.isNotNull(records.thread.archivedAt);
243 }).pipe(Effect.provide(layerTest)),
244);
245
246it.effect.each(["settle", "archive"] as const)(
247 "a thread already %s does not wake on a merged snapshot",
248 (action) =>
249 Effect.gen(function* () {
250 const { engine, reactor, reads } = yield* makeHarness({ snapshotState: "merged" });
251 yield* engine.dispatch({
252 type: action === "settle" ? "thread.settle" : "thread.archive",
253 commandId: CommandId.make(action),
254 threadId,
255 });
256 yield* reactor.sweep;
257 const records = yield* engine.getThreadRecords(threadId, ["runs", "turnItems"]);
258 assert.equal(records.runs.length, 0);
259 assert.equal(records.turnItems.filter((item) => item.type === "notification").length, 0);
260 assert.equal(yield* Ref.get(reads), 0);
261 }).pipe(Effect.provide(layerTest)),
262);
263
264it.effect(
265 "work PR watches verify Mergebot and never call GitHub detail, activity or fingerprint reads",
266 () =>
267 Effect.gen(function* () {
268 let nativeReads = 0;
269 const value: MergebotPullRequestDetails = {
270 number: key.number,
271 headSha: "head",
272 lifecycle: "merged",
273 queueStatus: "not_queued",
274 position: null,
275 mergeEtaAt: null,
276 title: "Change",
277 body: null,
278 url: "https://mergebot.example/owner/repository/pull/42",
279 createdAt: timestamp,
280 updatedAt: timestamp,
281 mergedAt: timestamp,
282 closedAt: timestamp,
283 author: null,
284 additions: 3,
285 deletions: 1,
286 changedFilesCount: 1,
287 labels: [],
288 comments: [],
289 observedAt: Date.parse(timestamp),
290 draft: false,
291 baseBranch: "main",
292 headBranch: "feature",
293 ciStatus: "passed",
294 hasConflicts: false,
295 checks: [],
296 };
297 const { engine, reactor, reads } = yield* makeHarness({
298 snapshotState: "merged",
299 mergebot: {
300 configuration: { origin: "https://mergebot.example", repositories: [key.repository] },
301 read: () => Effect.succeed(null),
302 watchQueue: () => Stream.empty,
303 findPullRequest: () => Effect.succeed(null),
304 readPullRequest: () =>
305 Effect.sync(() => {
306 nativeReads++;
307 return value;
308 }),
309 streamChanges: Stream.empty,
310 },
311 });
312 yield* reactor.sweep;
313 const records = yield* engine.getThreadRecords(threadId, ["runs", "turnItems"]);
314 assert.equal(records.runs.length, 1);
315 assert.isUndefined(records.thread.pullRequests?.[0]?.watch);
316 assert.equal(yield* Ref.get(reads), 0);
317 assert.isAbove(nativeReads, 0);
318 }).pipe(Effect.provide(layerTest)),
319);
apps/server/src/orchestration-v2/PullRequestWatchReactor.ts+103-34
...@@ -1,8 +1,10 @@...@@ -1,8 +1,10 @@
1import {1import {
2 CommandId,2 CommandId,
3 isMergebotRepository,
3 MessageId,4 MessageId,
4 type OrchestrationV2Notification,5 type OrchestrationV2Notification,
5 type PullRequestActivity,6 type PullRequestActivity,
7 type PullRequestDetail,
6 type PullRequestComment,8 type PullRequestComment,
7 type PullRequestRef,9 type PullRequestRef,
8 type PullRequestThreadCommentsResult,10 type PullRequestThreadCommentsResult,
...@@ -23,6 +25,8 @@ import * as Exit from "effect/Exit";...@@ -23,6 +25,8 @@ import * as Exit from "effect/Exit";
23import * as Layer from "effect/Layer";25import * as Layer from "effect/Layer";
24import * as Option from "effect/Option";26import * as Option from "effect/Option";
25import * as Schedule from "effect/Schedule";27import * as Schedule from "effect/Schedule";
28import * as Semaphore from "effect/Semaphore";
29import * as Stream from "effect/Stream";
26import * as Schema from "effect/Schema";30import * as Schema from "effect/Schema";
27import type * as Scope from "effect/Scope";31import type * as Scope from "effect/Scope";
2832
...@@ -30,6 +34,12 @@ import {...@@ -30,6 +34,12 @@ import {
30 type ProviderChangeRequestWatchFingerprint,34 type ProviderChangeRequestWatchFingerprint,
31 PullRequestProviderError,35 PullRequestProviderError,
32} from "@t3tools/source-control-core/server/PullRequestProvider";36} from "@t3tools/source-control-core/server/PullRequestProvider";
37import * as MergebotQueue from "../mergebot/MergebotQueue.ts";
38import {
39 mergebotChecks,
40 mergebotMergeability,
41 mergebotWatchFingerprint,
42} from "../mergebot/pullRequestRead.ts";
33import * as PullRequestService from "../pullRequest/PullRequestService.ts";43import * as PullRequestService from "../pullRequest/PullRequestService.ts";
34import { forkParked } from "../serverActivation.ts";44import { forkParked } from "../serverActivation.ts";
35import * as Orchestrator from "./Orchestrator.ts";45import * as Orchestrator from "./Orchestrator.ts";
...@@ -217,9 +227,9 @@ function watchesEqual(left: ThreadPullRequestWatch, right: ThreadPullRequestWatc...@@ -217,9 +227,9 @@ function watchesEqual(left: ThreadPullRequestWatch, right: ThreadPullRequestWatc
217 * of a project that watch it. Where the host has a fingerprint, one batched read of those tells227 * of a project that watch it. Where the host has a fingerprint, one batched read of those tells
218 * the pass which pull requests moved, and only those are read; elsewhere a pull request is read228 * the pass which pull requests moved, and only those are read; elsewhere a pull request is read
219 * unless its sync snapshot has not moved while nothing is in flight.229 * unless its sync snapshot has not moved while nothing is in flight.
220 * Settling or archiving a thread ends its watches, and a merged or closed pull request ends230 * Settling or archiving a thread ends its watches. A merged or closed pull request wakes its
221 * its watch. Each ended watch is logged once with why it ended and how long it went without231 * active thread once and ends its watch. Each ended watch is logged once with why it ended
222 * a push, for debugging watches that live too long.232 * and how long it went without a push, for debugging watches that live too long.
223 */233 */
224export class PullRequestWatchReactor extends Context.Service<234export class PullRequestWatchReactor extends Context.Service<
225 PullRequestWatchReactor,235 PullRequestWatchReactor,
...@@ -235,6 +245,8 @@ export const make = Effect.gen(function* () {...@@ -235,6 +245,8 @@ export const make = Effect.gen(function* () {
235 const engine = yield* Orchestrator.OrchestratorV2;245 const engine = yield* Orchestrator.OrchestratorV2;
236 const projections = yield* ProjectionStore.ProjectionStoreV2;246 const projections = yield* ProjectionStore.ProjectionStoreV2;
237 const pullRequests = yield* PullRequestService.PullRequestService;247 const pullRequests = yield* PullRequestService.PullRequestService;
248 const mergebot = yield* MergebotQueue.MergebotQueue;
249 const sweepLock = yield* Semaphore.make(1);
238 const crypto = yield* Crypto.Crypto;250 const crypto = yield* Crypto.Crypto;
239 const bootedAt = yield* Clock.currentTimeMillis;251 const bootedAt = yield* Clock.currentTimeMillis;
240 const lives = new Map<string, WatchLife>();252 const lives = new Map<string, WatchLife>();
...@@ -406,16 +418,32 @@ export const make = Effect.gen(function* () {...@@ -406,16 +418,32 @@ export const make = Effect.gen(function* () {
406 Effect.tap(() => ended(target, "closed")),418 Effect.tap(() => ended(target, "closed")),
407 );419 );
408420
421 const merged = (target: WatchTarget) =>
422 record(target, null, {
423 text: `Pull request #${target.link.number} (${target.link.url}) merged. T3 Code stopped watching it. Continue any work you were waiting to do after this merge.`,
424 notification: {
425 source: { kind: "monitor" },
426 outcome: "completed",
427 summary: `Pull request #${target.link.number} merged`,
428 },
429 }).pipe(
430 Effect.tap(() => woke(target)),
431 Effect.tap(() => ended(target, "merged")),
432 );
433
409 /** Why a watch ends without a host read; the rest are read once per pull request. */434 /** Why a watch ends without a host read; the rest are read once per pull request. */
435 const isWorkLink = (link: ThreadPullRequestLink) =>
436 normalizeThreadPullRequestKey(link).host === "github.com" &&
437 isMergebotRepository(mergebot.configuration ?? undefined, link.repository);
438
410 const endsWithoutRead = ({ thread, link }: WatchTarget): WatchEndReason | undefined =>439 const endsWithoutRead = ({ thread, link }: WatchTarget): WatchEndReason | undefined =>
411 // A merged pull request cannot reopen. Settling and archiving end watches, and a subagent440 // Inactive threads discard even a merged watch without waking.
412 // cannot start one; a watch left from before those rules ends here.441 thread.settledOverride === "settled" || thread.settledAt !== null
413 link.snapshot?.state === "merged"442 ? "settled"
414 ? "merged"443 : thread.lineage.relationshipToParent === "subagent"
415 : thread.settledOverride === "settled" || thread.settledAt !== null444 ? "subagent"
416 ? "settled"445 : link.snapshot?.state === "merged" && !isWorkLink(link)
417 : thread.lineage.relationshipToParent === "subagent"446 ? "merged"
418 ? "subagent"
419 : undefined;447 : undefined;
420448
421 /**449 /**
...@@ -451,8 +479,14 @@ export const make = Effect.gen(function* () {...@@ -451,8 +479,14 @@ export const make = Effect.gen(function* () {
451 * sync-snapshot gating; "paused" skips the pass while the host is rate limited, since every479 * sync-snapshot gating; "paused" skips the pass while the host is rate limited, since every
452 * read would be refused.480 * read would be refused.
453 */481 */
454 const fingerprintOf = (group: WatchGroup) =>482 const fingerprintOf = Effect.fnUntraced(function* (
455 pullRequests.watchFingerprint(referenceOf(group)).pipe(483 group: WatchGroup,
484 ): Effect.fn.Return<ProviderChangeRequestWatchFingerprint | null | "paused"> {
485 if (isWorkLink(group.targets[0]!.link)) {
486 const value = yield* mergebot.readPullRequest(referenceOf(group));
487 return value === null ? "paused" : mergebotWatchFingerprint(value);
488 }
489 return yield* pullRequests.watchFingerprint(referenceOf(group)).pipe(
456 Effect.catchCause(490 Effect.catchCause(
457 (cause): Effect.Effect<ProviderChangeRequestWatchFingerprint | null | "paused"> =>491 (cause): Effect.Effect<ProviderChangeRequestWatchFingerprint | null | "paused"> =>
458 Cause.hasInterruptsOnly(cause)492 Cause.hasInterruptsOnly(cause)
...@@ -467,6 +501,7 @@ export const make = Effect.gen(function* () {...@@ -467,6 +501,7 @@ export const make = Effect.gen(function* () {
467 }).pipe(Effect.as(null)),501 }).pipe(Effect.as(null)),
468 ),502 ),
469 );503 );
504 });
470505
471 const readGroup = Effect.fn("PullRequestWatchReactor.readGroup")(function* (506 const readGroup = Effect.fn("PullRequestWatchReactor.readGroup")(function* (
472 group: WatchGroup,507 group: WatchGroup,
...@@ -479,18 +514,42 @@ export const make = Effect.gen(function* () {...@@ -479,18 +514,42 @@ export const make = Effect.gen(function* () {
479 const reference = referenceOf(group);514 const reference = referenceOf(group);
480 // The reads are recorded against this fingerprint, so a cached answer older than it must not515 // The reads are recorded against this fingerprint, so a cached answer older than it must not
481 // stand in for them. Clients are not told: this only drops this pull request's cached reads.516 // stand in for them. Clients are not told: this only drops this pull request's cached reads.
482 if (fingerprint !== null) yield* pullRequests.invalidate({ reference });517 const work = isWorkLink(group.targets[0]!.link);
483 const read = yield* Effect.exit(518 const native = work ? yield* mergebot.readPullRequest(reference) : null;
484 Effect.all(519 if (work && native === null) return;
485 [520 if (!work && fingerprint !== null) yield* pullRequests.invalidate({ reference });
486 pullRequests.detail({ ...reference, allowStale: false }),521 const reading: Effect.Effect<
487 plan.activity522 readonly [
488 ? pullRequests.activity(reference).pipe(Effect.map(Option.some))523 Pick<
489 : Effect.succeed(Option.none<PullRequestActivity>()),524 PullRequestDetail,
490 ],525 "state" | "headSha" | "baseBranch" | "mergeability" | "checks" | "author" | "viewer"
491 { concurrency: 2 },526 >,
492 ),527 Option.Option<PullRequestActivity>,
493 );528 ],
529 PullRequestService.PullRequestError
530 > =
531 native !== null
532 ? Effect.succeed([
533 {
534 state: native.lifecycle,
535 headSha: native.headSha,
536 baseBranch: native.baseBranch,
537 mergeability: mergebotMergeability(native),
538 checks: mergebotChecks(native),
539 author: native.author,
540 },
541 Option.none<PullRequestActivity>(),
542 ] as const)
543 : Effect.all(
544 [
545 pullRequests.detail({ ...reference, allowStale: false }),
546 plan.activity
547 ? pullRequests.activity(reference).pipe(Effect.map(Option.some))
548 : Effect.succeed(Option.none<PullRequestActivity>()),
549 ],
550 { concurrency: 2 },
551 );
552 const read = yield* Effect.exit(reading);
494 if (Exit.isFailure(read)) {553 if (Exit.isFailure(read)) {
495 if (Cause.hasInterruptsOnly(read.cause)) return yield* Effect.failCause(read.cause);554 if (Cause.hasInterruptsOnly(read.cause)) return yield* Effect.failCause(read.cause);
496 lastReads.delete(group.key);555 lastReads.delete(group.key);
...@@ -524,17 +583,18 @@ export const make = Effect.gen(function* () {...@@ -524,17 +583,18 @@ export const make = Effect.gen(function* () {
524 if (detail.state !== "open") {583 if (detail.state !== "open") {
525 lastReads.delete(group.key);584 lastReads.delete(group.key);
526 return yield* eachTarget(group, (target) =>585 return yield* eachTarget(group, (target) =>
527 detail.state === "closed"586 detail.state === "closed" ? closed(target) : merged(target),
528 ? closed(target)
529 : record(target, null).pipe(Effect.tap(() => ended(target, "merged"))),
530 );587 );
531 }588 }
532589
533 // Never advance the remark watermark past comments an incomplete read could have missed.590 // Never advance the remark watermark past comments an incomplete read could have missed.
534 // A pass that did not read the activity has no remarks to report.591 // A pass that did not read the activity has no remarks to report.
535 const remarks = Option.isSome(activity)592 const remarks =
536 ? yield* readRemarks(group.key, reference, activity.value)593 native !== null
537 : null;594 ? native.comments
595 : Option.isSome(activity)
596 ? yield* readRemarks(group.key, reference, activity.value)
597 : null;
538 lastReads.set(group.key, {598 lastReads.set(group.key, {
539 at: plan.activity || last === undefined ? now : last.at,599 at: plan.activity || last === undefined ? now : last.at,
540 fingerprint: group.fingerprint,600 fingerprint: group.fingerprint,
...@@ -608,8 +668,10 @@ export const make = Effect.gen(function* () {...@@ -608,8 +668,10 @@ export const make = Effect.gen(function* () {
608 yield* Effect.forEach(668 yield* Effect.forEach(
609 ending,669 ending,
610 ([target, reason]) =>670 ([target, reason]) =>
611 record(target, null).pipe(671 (reason === "merged"
612 Effect.tap(() => ended(target, reason)),672 ? merged(target)
673 : record(target, null).pipe(Effect.tap(() => ended(target, reason)))
674 ).pipe(
613 Effect.catchCause(675 Effect.catchCause(
614 logFailure("pull request watch stop failed", {676 logFailure("pull request watch stop failed", {
615 threadId: target.thread.id,677 threadId: target.thread.id,
...@@ -639,11 +701,18 @@ export const make = Effect.gen(function* () {...@@ -639,11 +701,18 @@ export const make = Effect.gen(function* () {
639 }).pipe(701 }).pipe(
640 Effect.catchCause(logFailure("pull request watch sweep failed", {})),702 Effect.catchCause(logFailure("pull request watch sweep failed", {})),
641 Effect.withSpan("PullRequestWatchReactor.sweep"),703 Effect.withSpan("PullRequestWatchReactor.sweep"),
704 sweepLock.withPermit,
642 );705 );
643706
644 const start: PullRequestWatchReactor["Service"]["start"] = () =>707 const start: PullRequestWatchReactor["Service"]["start"] = () =>
645 forkParked(708 forkParked(
646 sweep.pipe(Effect.repeat(Schedule.spaced(`${SWEEP_MINUTES} minutes`)), Effect.asVoid),709 Effect.all(
710 [
711 sweep.pipe(Effect.repeat(Schedule.spaced(`${SWEEP_MINUTES} minutes`)), Effect.asVoid),
712 mergebot.streamChanges.pipe(Stream.runForEach(() => sweep)),
713 ],
714 { concurrency: "unbounded", discard: true },
715 ),
647 );716 );
648717
649 return { start, sweep } satisfies PullRequestWatchReactor["Service"];718 return { start, sweep } satisfies PullRequestWatchReactor["Service"];
docs/user/source-control.md+9-2
...@@ -153,6 +153,12 @@ Open **Pull requests** to review changes and comments, request reviewers, check...@@ -153,6 +153,12 @@ Open **Pull requests** to review changes and comments, request reviewers, check
153or merge. You can edit review titles and descriptions and your own comments where the host allows it.153or merge. You can edit review titles and descriptions and your own comments where the host allows it.
154GitLab calls these merge requests.154GitLab calls these merge requests.
155155
156Configured work repositories use Mergebot for PR status, checks, and live updates.
157Their PR links open Mergebot inside T3 Code's Browser, with its queue and review actions.
158Open **Mergebot** from the sidebar or Pull requests page to see the queue without a chat.
159On mobile, these links use the environment's server Browser and require browser access.
160If Mergebot disconnects, PR decisions wait for a verified update after it reconnects.
161
156On mobile, **Pull requests** lists reviews across your connected environments. Open a review162On mobile, **Pull requests** lists reviews across your connected environments. Open a review
157to read its description, activity, commits, and changed files. Line comments stay as drafts163to read its description, activity, commits, and changed files. Line comments stay as drafts
158until you submit the review. **Work on PR** opens a chat with the review attached; sending164until you submit the review. **Work on PR** opens a chat with the review attached; sending
...@@ -279,8 +285,9 @@ the thread is active, the server checks the pull request every two minutes and w...@@ -279,8 +285,9 @@ the thread is active, the server checks the pull request every two minutes and w
279check fails, the required checks pass, someone else comments or reviews, or the branch starts to285check fails, the required checks pass, someone else comments or reviews, or the branch starts to
280conflict. Threads in a project that watch the same pull request share one check. On GitHub, a check286conflict. Threads in a project that watch the same pull request share one check. On GitHub, a check
281first asks whether anything changed and reads the pull request only when it did, which keeps287first asks whether anything changed and reads the pull request only when it did, which keeps
282watching inside GitHub's rate limit. Comments from your own account do not wake it. Watching ends288watching inside GitHub's rate limit. Comments from your own account do not wake it. A merge wakes
283when the pull request merges or closes, after 10 wakes in a row that bring only comments, after 8289the agent once to finish any follow-up work and ends the watch. Watching also ends
290when the pull request closes, after 10 wakes in a row that bring only comments, after 8
284failed reads in a row, or when you press Stop on the thread. A rate limit only pauses watching.291failed reads in a row, or when you press Stop on the thread. A rate limit only pauses watching.
285Settling or archiving a thread also ends all its watches. Unsettle the thread before starting a new292Settling or archiving a thread also ends all its watches. Unsettle the thread before starting a new
286watch. Subagents cannot watch pull requests; the thread that delegated to them does. To start or stop293watch. Subagents cannot watch pull requests; the thread that delegated to them does. To start or stop