authorgravatar for me@t3.ggTheo Browne <me@t3.gg> 2026-10-02 22:02:59-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-10-02 22:02:59-07:00
logca4f84e702428656665d53635e16251112091657
tree761e93a9c9c18607b79cea00bf1b2d99147ee2f4
parent736130c2fd39b85cd515fa2a41cd10c8a8a52630
signature Signed by PGP key B5690EEEBB952194

fix(server): threads settle as soon as an agent merges their PR (#15024)

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

5 files changed, 202 insertions(+), 19 deletions(-)

apps/server/src/orchestration-v2/ProjectionSettlement.test.ts+4
......@@ -335,6 +335,10 @@ it.effect.each([
335335 [open, null, 2],
336336 ],
337337 );
338 assert.deepEqual(
339 (yield* store.getThreadsWithPullRequests(open)).map((thread) => thread.id),
340 [open],
341 );
338342 }).pipe(Effect.provide(testLayer)),
339343);
340344
apps/server/src/orchestration-v2/ProjectionStore.ts+11-8
......@@ -349,12 +349,12 @@ export interface ProjectionStoreV2Shape {
349349 ) => Effect.Effect<ReadonlyArray<ProjectionSettlementCandidate>, ProjectionStoreV2Error>;
350350 /**
351351 * Active (not deleted, not archived) threads with at least one pull request
352 * link, in shell snapshot order. Skips run, message and item reads.
352 * link, in shell snapshot order, or only `threadId` when given. Skips run,
353 * message and item reads.
353354 */
354 readonly getThreadsWithPullRequests: () => Effect.Effect<
355 ReadonlyArray<ProjectionThreadPullRequests>,
356 ProjectionStoreV2Error
357 >;
355 readonly getThreadsWithPullRequests: (
356 threadId?: ThreadId,
357 ) => Effect.Effect<ReadonlyArray<ProjectionThreadPullRequests>, ProjectionStoreV2Error>;
358358 readonly getTurnStartContext: (
359359 threadId: ThreadId,
360360 runId: RunId,
......@@ -5133,12 +5133,14 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
51335133 )
51345134 .pipe(Effect.mapError((cause) => new ProjectionStoreSetupError({ cause })));
51355135
5136 const getThreadsWithPullRequests: ProjectionStoreV2Shape["getThreadsWithPullRequests"] = () =>
5136 const getThreadsWithPullRequests: ProjectionStoreV2Shape["getThreadsWithPullRequests"] = (
5137 threadId,
5138 ) =>
51375139 Effect.gen(function* () {
51385140 const rows = yield* sql<PayloadRow>`
51395141 SELECT payload_json
51405142 FROM orchestration_v2_projection_threads
5141 WHERE deleted_at IS NULL
5143 WHERE deleted_at IS NULL${threadId === undefined ? sql`` : sql` AND thread_id = ${threadId}`}
51425144 AND json_extract(payload_json, '$.archivedAt') IS NULL
51435145 AND json_array_length(payload_json, '$.pullRequests') > 0
51445146 ORDER BY updated_at ASC, thread_id ASC
......@@ -5583,13 +5585,14 @@ export const layerMemory: Layer.Layer<ProjectionStoreV2> = Layer.effect(
55835585 left.id.localeCompare(right.id),
55845586 );
55855587 }),
5586 getThreadsWithPullRequests: () =>
5588 getThreadsWithPullRequests: (threadId) =>
55875589 Ref.get(replayState).pipe(
55885590 Effect.map((state) =>
55895591 [...state.projections.values()]
55905592 .map(({ thread }) => thread)
55915593 .filter(
55925594 (thread) =>
5595 (threadId === undefined || thread.id === threadId) &&
55935596 thread.deletedAt === null &&
55945597 thread.archivedAt === null &&
55955598 (thread.pullRequests ?? []).length > 0,
apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts+134-3
......@@ -1,10 +1,16 @@
11import * as Stream from "effect/Stream";
22import {
3 EventId,
4 MessageId,
35 ProjectId,
46 ProviderInstanceId,
57 PullRequestOperationError,
8 RunId,
69 ThreadId,
10 TurnItemId,
711 type OrchestrationV2Command as OrchestrationCommand,
12 type OrchestrationV2DomainEvent,
13 type OrchestrationV2Run,
814 type OrchestrationProjectShell,
915 type PullRequestRef,
1016 type PullRequestStack,
......@@ -171,6 +177,7 @@ const makeHarness = Effect.fn("makePullRequestSyncHarness")(function* (options:
171177 const linkCommands = yield* Ref.make<ReadonlyArray<LinkCommand>>([]);
172178 const summaryCalls = yield* Ref.make<ReadonlyArray<PullRequestRef>>([]);
173179 const stackCalls = yield* Ref.make<ReadonlyArray<PullRequestRef>>([]);
180 const domainEvents = yield* Queue.unbounded<OrchestrationV2DomainEvent>();
174181
175182 const summary: PullRequestService.PullRequestService["Service"]["summary"] = (
176183 input,
......@@ -211,12 +218,17 @@ const makeHarness = Effect.fn("makePullRequestSyncHarness")(function* (options:
211218 }),
212219 Layer.mock(ProjectionStore.ProjectionStoreV2)({
213220 // Mirrors the store's filter: active threads that have at least one link.
214 getThreadsWithPullRequests: () =>
221 getThreadsWithPullRequests: (threadId) =>
215222 Queue.offer(snapshotReads, undefined).pipe(
216223 Effect.andThen(Ref.get(snapshots)),
217224 Effect.map((snapshot) =>
218225 snapshot.threads
219 .filter((thread) => thread.archivedAt === null && thread.pullRequests.length > 0)
226 .filter(
227 (thread) =>
228 (threadId === undefined || thread.id === threadId) &&
229 thread.archivedAt === null &&
230 thread.pullRequests.length > 0,
231 )
220232 .map((thread) => ({
221233 id: thread.id,
222234 projectId: thread.projectId,
......@@ -233,7 +245,7 @@ const makeHarness = Effect.fn("makePullRequestSyncHarness")(function* (options:
233245 Effect.andThen(Effect.die(new Error("pull request sync must not read the shell"))),
234246 ),
235247 dispatch,
236 streamDomainEvents: Stream.empty,
248 streamDomainEvents: Stream.fromQueue(domainEvents),
237249 }),
238250 Layer.succeed(ServerActivation.ServerActivation, Deferred.await(activation)),
239251 Layer.succeed(Crypto.Crypto, testCrypto),
......@@ -248,6 +260,7 @@ const makeHarness = Effect.fn("makePullRequestSyncHarness")(function* (options:
248260 linkCommands,
249261 summaryCalls,
250262 stackCalls,
263 domainEvents,
251264 layer: PullRequestSyncReactor.layer.pipe(Layer.provide(dependencies)),
252265 };
253266});
......@@ -272,6 +285,68 @@ const sweepAgain = Effect.fn("sweepPullRequestSyncHarness")(function* (
272285 yield* reactor.drain;
273286});
274287
288function runUpdated(
289 threadId: ThreadId,
290 status: OrchestrationV2Run["status"],
291): OrchestrationV2DomainEvent {
292 const runId = RunId.make(`run:${threadId}:1`);
293 const at = DateTime.makeUnsafe(NOW);
294 return {
295 type: "run.updated",
296 id: EventId.make(`event:${threadId}:${status}`),
297 threadId,
298 runId,
299 occurredAt: at,
300 payload: {
301 id: runId,
302 threadId,
303 ordinal: 1,
304 providerInstanceId: ProviderInstanceId.make("codex"),
305 modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5" },
306 providerThreadId: null,
307 userMessageId: MessageId.make(`message:${threadId}:1`),
308 rootNodeId: null,
309 activeAttemptId: null,
310 status,
311 requestedAt: at,
312 startedAt: at,
313 completedAt: at,
314 checkpointId: null,
315 contextHandoffId: null,
316 },
317 };
318}
319
320function commandRan(threadId: ThreadId, input: string): OrchestrationV2DomainEvent {
321 const runId = RunId.make(`run:${threadId}:1`);
322 const at = DateTime.makeUnsafe(NOW);
323 return {
324 type: "turn-item.updated",
325 id: EventId.make(`event:${threadId}:command`),
326 threadId,
327 runId,
328 occurredAt: at,
329 payload: {
330 id: TurnItemId.make(`item:${threadId}:command`),
331 threadId,
332 runId,
333 nodeId: null,
334 providerThreadId: null,
335 providerTurnId: null,
336 nativeItemRef: null,
337 parentItemId: null,
338 ordinal: 1,
339 status: "completed",
340 title: null,
341 startedAt: at,
342 completedAt: at,
343 updatedAt: at,
344 type: "command_execution",
345 input,
346 },
347 };
348}
349
275350/** What the reactor would have persisted, so the next sweep sees its own writes. */
276351function applySync(
277352 snapshot: TestShellSnapshot,
......@@ -832,4 +907,60 @@ describe("PullRequestSyncReactor", () => {
832907 }),
833908 ),
834909 );
910
911 it.effect("reads open links fresh only when a run that ran a merge command ends", () =>
912 Effect.scoped(
913 Effect.gen(function* () {
914 yield* TestClock.setTime(Date.parse(NOW));
915 const state = yield* Ref.make<PullRequestSummary["state"]>("open");
916 const invalidated = yield* Ref.make<ReadonlyArray<number>>([]);
917 const fixture = yield* makeHarness({
918 snapshot: makeSnapshot([
919 makeThread("agent", { pullRequests: [makeLink(7, { state: "open" })] }),
920 makeThread("other", { pullRequests: [makeLink(9, { state: "open" })] }),
921 ]),
922 summary: (input) =>
923 Ref.get(state).pipe(
924 Effect.map((current) =>
925 makeSummary(input, current === "merged" ? { state: current, mergedAt: NOW } : {}),
926 ),
927 ),
928 invalidate: ({ reference }) =>
929 Ref.update(invalidated, (numbers) => [...numbers, reference?.number ?? -1]),
930 });
931
932 yield* Effect.gen(function* () {
933 const reactor = yield* startAndSweep(fixture);
934 // The agent merges from a shell during its turn, inside the summary cache window.
935 yield* Ref.set(state, "merged");
936 yield* Ref.set(fixture.summaryCalls, []);
937
938 yield* Queue.offerAll(fixture.domainEvents, [
939 // A run that only reads its pull request costs no host read when it ends.
940 commandRan(ThreadId.make("other"), "gh pr view 9"),
941 runUpdated(ThreadId.make("other"), "completed"),
942 commandRan(ThreadId.make("agent"), "gh pr merge 7 --squash 2>&1 | tail -3"),
943 runUpdated(ThreadId.make("agent"), "completed"),
944 ]);
945 // The agent thread's lookup, then the requested sweep.
946 yield* Queue.take(fixture.snapshotReads);
947 yield* Queue.take(fixture.snapshotReads);
948 yield* reactor.drain;
949
950 assert.deepStrictEqual(yield* Ref.get(invalidated), [7]);
951 assert.deepStrictEqual(
952 (yield* Ref.get(fixture.summaryCalls)).map((call) => call.number),
953 [7],
954 );
955 assert.deepStrictEqual(
956 (yield* Ref.get(fixture.syncCommands)).map((command) => [
957 command.number,
958 command.snapshot.state,
959 ]),
960 [[7, "merged"]],
961 );
962 }).pipe(Effect.provide(fixture.layer));
963 }),
964 ),
965 );
835966});
apps/server/src/orchestration-v2/PullRequestSyncReactor.ts+48-6
......@@ -2,6 +2,7 @@ import { siblingPullRequestUrl } from "@t3tools/shared/changeRequestUrl";
22import {
33 CommandId,
44 type PullRequestSummary,
5 type ThreadId,
56 type ThreadPullRequestKey,
67 type ThreadPullRequestLink,
78 type ThreadPullRequestSnapshot,
......@@ -29,8 +30,11 @@ import * as PullRequestService from "../pullRequest/PullRequestService.ts";
2930import { forkParked } from "../serverActivation.ts";
3031import * as Orchestrator from "./Orchestrator.ts";
3132import * as ProjectionStore from "./ProjectionStore.ts";
33import { isTerminalRunStatus } from "./ThreadManagementService.ts";
3234
3335const SLOW_SYNC_INTERVAL_MS = 15 * 60 * 1_000;
36/** Shell commands that can merge or close a pull request without a merge notification. */
37const PULL_REQUEST_CLOSE_COMMAND = /\b(?:gh\s+pr|glab\s+mr)\s+(?:merge|close)\b/u;
3438
3539type SnapshotFields = Omit<ThreadPullRequestSnapshot, "syncedAt">;
3640
......@@ -330,22 +334,60 @@ export const make = Effect.gen(function* () {
330334 }).pipe(Effect.catchCause(logSkipped("pull request sync sweep failed", {}))),
331335 );
332336
337 // Threads whose current run ran a merge or close command, until that run ends.
338 const closeCommandThreads = new Set<ThreadId>();
339 const refreshOpenLinks = (threadId: ThreadId) =>
340 projections.getThreadsWithPullRequests(threadId).pipe(
341 Effect.flatMap((threads) =>
342 Effect.forEach(
343 threads.flatMap((thread) =>
344 visibleThreadPullRequests(thread.pullRequests ?? []).filter(
345 (link) => link.snapshot?.state === "open",
346 ),
347 ),
348 requestSync,
349 { discard: true },
350 ),
351 ),
352 Effect.catchCause(logSkipped("pull request refresh after run skipped", { threadId })),
353 );
354
333355 const start: PullRequestSyncReactor["Service"]["start"] = Effect.fn(
334356 "PullRequestSyncReactor.start",
335357 )(function* () {
336358 const events = engine.streamDomainEvents;
337359 yield* forkParked(
338 Stream.runForEach(events, (event) =>
339 event.type === "thread.pull-request-synced"
340 ? Effect.forEach(
360 Stream.runForEach(events, (event) => {
361 switch (event.type) {
362 case "thread.pull-request-synced":
363 return Effect.forEach(
341364 visibleThreadPullRequests(event.payload.pullRequests ?? []).filter(
342365 (link) => link.snapshot === null,
343366 ),
344367 requestSync,
345368 { discard: true },
346 )
347 : Effect.void,
348 ).pipe(Effect.catchCause(logSkipped("pull request sync event stream failed", {}))),
369 );
370 // An agent can merge or close its pull request from a shell (`gh pr merge`), which
371 // sends no merge notification. When a run that ran such a command ends, read the
372 // thread's open links fresh, so settlement does not wait for the next sweep and the
373 // cached summary. Other runs add no host reads.
374 case "turn-item.updated":
375 if (
376 event.payload.type === "command_execution" &&
377 PULL_REQUEST_CLOSE_COMMAND.test(event.payload.input)
378 ) {
379 closeCommandThreads.add(event.threadId);
380 }
381 return Effect.void;
382 case "run.updated":
383 return isTerminalRunStatus(event.payload.status) &&
384 closeCommandThreads.delete(event.threadId)
385 ? refreshOpenLinks(event.threadId)
386 : Effect.void;
387 default:
388 return Effect.void;
389 }
390 }).pipe(Effect.catchCause(logSkipped("pull request sync event stream failed", {}))),
349391 );
350392 yield* forkParked(
351393 Effect.gen(function* () {
docs/internals/overview.md+5-2
......@@ -87,8 +87,11 @@ must reject that operation before changing the filesystem.
8787Thread settlement is server-owned. The
8888[settlement service](../../apps/server/src/orchestration-v2/ThreadSettlementService.ts) evaluates PR
8989and inactivity settings without a connected client. Merge notifications invalidate cached PR state
90and trigger a check. The guarded `thread.auto-settle` command rejects newer activity, explicit
91settlement overrides, and live or blocked work. It records the activity timestamp for stable
90and trigger a check. A merge outside T3, such as an agent running `gh pr merge`, sends no
91notification, so the [PR sync reactor](../../apps/server/src/orchestration-v2/PullRequestSyncReactor.ts)
92re-reads a thread's open links when a run that ran a merge or close command ends. The guarded
93`thread.auto-settle` command rejects newer activity, explicit settlement overrides, and live or
94blocked work. It records the activity timestamp for stable
9295sorting and detaches idle provider sessions. Clients render the persisted result; they do not
9396derive settlement from their own clocks or PR caches.
9497