From 671629c9b5db62b8bf61705cfef72efe5448c0ba Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Fri, 2 Oct 2026 21:45:23 -0700 Subject: [PATCH 1/3] fix(server): threads settle as soon as an agent merges their PR Co-Authored-By: Claude Opus 5.5 (1M context) --- .../ProjectionSettlement.test.ts | 4 + .../src/orchestration-v2/ProjectionStore.ts | 19 ++-- .../PullRequestSyncReactor.test.ts | 100 +++++++++++++++++- .../PullRequestSyncReactor.ts | 40 +++++-- docs/internals/overview.md | 4 +- 5 files changed, 149 insertions(+), 18 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProjectionSettlement.test.ts b/apps/server/src/orchestration-v2/ProjectionSettlement.test.ts index eec1ad1c3f08..60149fe8a292 100644 --- a/apps/server/src/orchestration-v2/ProjectionSettlement.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionSettlement.test.ts @@ -335,6 +335,10 @@ it.effect.each([ [open, null, 2], ], ); + assert.deepEqual( + (yield* store.getThreadsWithPullRequests(open)).map((thread) => thread.id), + [open], + ); }).pipe(Effect.provide(testLayer)), ); diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index b662a528b4de..ac0624b05806 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -349,12 +349,12 @@ export interface ProjectionStoreV2Shape { ) => Effect.Effect, ProjectionStoreV2Error>; /** * Active (not deleted, not archived) threads with at least one pull request - * link, in shell snapshot order. Skips run, message and item reads. + * link, in shell snapshot order, or only `threadId` when given. Skips run, + * message and item reads. */ - readonly getThreadsWithPullRequests: () => Effect.Effect< - ReadonlyArray, - ProjectionStoreV2Error - >; + readonly getThreadsWithPullRequests: ( + threadId?: ThreadId, + ) => Effect.Effect, ProjectionStoreV2Error>; readonly getTurnStartContext: ( threadId: ThreadId, runId: RunId, @@ -5133,12 +5133,14 @@ export const layer: Layer.Layer = ) .pipe(Effect.mapError((cause) => new ProjectionStoreSetupError({ cause }))); - const getThreadsWithPullRequests: ProjectionStoreV2Shape["getThreadsWithPullRequests"] = () => + const getThreadsWithPullRequests: ProjectionStoreV2Shape["getThreadsWithPullRequests"] = ( + threadId, + ) => Effect.gen(function* () { const rows = yield* sql` SELECT payload_json FROM orchestration_v2_projection_threads - WHERE deleted_at IS NULL + WHERE deleted_at IS NULL${threadId === undefined ? sql`` : sql` AND thread_id = ${threadId}`} AND json_extract(payload_json, '$.archivedAt') IS NULL AND json_array_length(payload_json, '$.pullRequests') > 0 ORDER BY updated_at ASC, thread_id ASC @@ -5583,13 +5585,14 @@ export const layerMemory: Layer.Layer = Layer.effect( left.id.localeCompare(right.id), ); }), - getThreadsWithPullRequests: () => + getThreadsWithPullRequests: (threadId) => Ref.get(replayState).pipe( Effect.map((state) => [...state.projections.values()] .map(({ thread }) => thread) .filter( (thread) => + (threadId === undefined || thread.id === threadId) && thread.deletedAt === null && thread.archivedAt === null && (thread.pullRequests ?? []).length > 0, diff --git a/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts b/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts index 7871b3ed6ee3..c09afd08b387 100644 --- a/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts +++ b/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts @@ -1,10 +1,15 @@ import * as Stream from "effect/Stream"; import { + EventId, + MessageId, ProjectId, ProviderInstanceId, PullRequestOperationError, + RunId, ThreadId, type OrchestrationV2Command as OrchestrationCommand, + type OrchestrationV2DomainEvent, + type OrchestrationV2Run, type OrchestrationProjectShell, type PullRequestRef, type PullRequestStack, @@ -171,6 +176,7 @@ const makeHarness = Effect.fn("makePullRequestSyncHarness")(function* (options: const linkCommands = yield* Ref.make>([]); const summaryCalls = yield* Ref.make>([]); const stackCalls = yield* Ref.make>([]); + const domainEvents = yield* Queue.unbounded(); const summary: PullRequestService.PullRequestService["Service"]["summary"] = ( input, @@ -211,12 +217,17 @@ const makeHarness = Effect.fn("makePullRequestSyncHarness")(function* (options: }), Layer.mock(ProjectionStore.ProjectionStoreV2)({ // Mirrors the store's filter: active threads that have at least one link. - getThreadsWithPullRequests: () => + getThreadsWithPullRequests: (threadId) => Queue.offer(snapshotReads, undefined).pipe( Effect.andThen(Ref.get(snapshots)), Effect.map((snapshot) => snapshot.threads - .filter((thread) => thread.archivedAt === null && thread.pullRequests.length > 0) + .filter( + (thread) => + (threadId === undefined || thread.id === threadId) && + thread.archivedAt === null && + thread.pullRequests.length > 0, + ) .map((thread) => ({ id: thread.id, projectId: thread.projectId, @@ -233,7 +244,7 @@ const makeHarness = Effect.fn("makePullRequestSyncHarness")(function* (options: Effect.andThen(Effect.die(new Error("pull request sync must not read the shell"))), ), dispatch, - streamDomainEvents: Stream.empty, + streamDomainEvents: Stream.fromQueue(domainEvents), }), Layer.succeed(ServerActivation.ServerActivation, Deferred.await(activation)), Layer.succeed(Crypto.Crypto, testCrypto), @@ -248,6 +259,7 @@ const makeHarness = Effect.fn("makePullRequestSyncHarness")(function* (options: linkCommands, summaryCalls, stackCalls, + domainEvents, layer: PullRequestSyncReactor.layer.pipe(Layer.provide(dependencies)), }; }); @@ -272,6 +284,38 @@ const sweepAgain = Effect.fn("sweepPullRequestSyncHarness")(function* ( yield* reactor.drain; }); +function runUpdated( + threadId: ThreadId, + status: OrchestrationV2Run["status"], +): OrchestrationV2DomainEvent { + const runId = RunId.make(`run:${threadId}:1`); + const at = DateTime.makeUnsafe(NOW); + return { + type: "run.updated", + id: EventId.make(`event:${threadId}:${status}`), + threadId, + runId, + occurredAt: at, + payload: { + id: runId, + threadId, + ordinal: 1, + providerInstanceId: ProviderInstanceId.make("codex"), + modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5" }, + providerThreadId: null, + userMessageId: MessageId.make(`message:${threadId}:1`), + rootNodeId: null, + activeAttemptId: null, + status, + requestedAt: at, + startedAt: at, + completedAt: at, + checkpointId: null, + contextHandoffId: null, + }, + }; +} + /** What the reactor would have persisted, so the next sweep sees its own writes. */ function applySync( snapshot: TestShellSnapshot, @@ -832,4 +876,54 @@ describe("PullRequestSyncReactor", () => { }), ), ); + + it.effect("reads the open links of a thread fresh when its run ends", () => + Effect.scoped( + Effect.gen(function* () { + yield* TestClock.setTime(Date.parse(NOW)); + const state = yield* Ref.make("open"); + const invalidated = yield* Ref.make>([]); + const fixture = yield* makeHarness({ + snapshot: makeSnapshot([ + makeThread("agent", { pullRequests: [makeLink(7, { state: "open" })] }), + makeThread("other", { pullRequests: [makeLink(9, { state: "open" })] }), + ]), + summary: (input) => + Ref.get(state).pipe( + Effect.map((current) => + makeSummary(input, current === "merged" ? { state: current, mergedAt: NOW } : {}), + ), + ), + invalidate: ({ reference }) => + Ref.update(invalidated, (numbers) => [...numbers, reference?.number ?? -1]), + }); + + yield* Effect.gen(function* () { + const reactor = yield* startAndSweep(fixture); + // The agent merges from a shell during its turn, inside the summary cache window. + yield* Ref.set(state, "merged"); + yield* Ref.set(fixture.summaryCalls, []); + + yield* Queue.offer(fixture.domainEvents, runUpdated(ThreadId.make("agent"), "completed")); + // The ended run's thread lookup, then the requested sweep. + yield* Queue.take(fixture.snapshotReads); + yield* Queue.take(fixture.snapshotReads); + yield* reactor.drain; + + assert.deepStrictEqual(yield* Ref.get(invalidated), [7]); + assert.deepStrictEqual( + (yield* Ref.get(fixture.summaryCalls)).map((call) => call.number), + [7], + ); + assert.deepStrictEqual( + (yield* Ref.get(fixture.syncCommands)).map((command) => [ + command.number, + command.snapshot.state, + ]), + [[7, "merged"]], + ); + }).pipe(Effect.provide(fixture.layer)); + }), + ), + ); }); diff --git a/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts b/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts index 670d401004b6..77f749b1d7d6 100644 --- a/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts +++ b/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts @@ -2,6 +2,7 @@ import { siblingPullRequestUrl } from "@t3tools/shared/changeRequestUrl"; import { CommandId, type PullRequestSummary, + type ThreadId, type ThreadPullRequestKey, type ThreadPullRequestLink, type ThreadPullRequestSnapshot, @@ -29,6 +30,7 @@ import * as PullRequestService from "../pullRequest/PullRequestService.ts"; import { forkParked } from "../serverActivation.ts"; import * as Orchestrator from "./Orchestrator.ts"; import * as ProjectionStore from "./ProjectionStore.ts"; +import { isTerminalRunStatus } from "./ThreadManagementService.ts"; const SLOW_SYNC_INTERVAL_MS = 15 * 60 * 1_000; @@ -330,22 +332,48 @@ export const make = Effect.gen(function* () { }).pipe(Effect.catchCause(logSkipped("pull request sync sweep failed", {}))), ); + const refreshOpenLinks = (threadId: ThreadId) => + projections.getThreadsWithPullRequests(threadId).pipe( + Effect.flatMap((threads) => + Effect.forEach( + threads.flatMap((thread) => + visibleThreadPullRequests(thread.pullRequests ?? []).filter( + (link) => link.snapshot?.state === "open", + ), + ), + requestSync, + { discard: true }, + ), + ), + Effect.catchCause(logSkipped("pull request refresh after run skipped", { threadId })), + ); + const start: PullRequestSyncReactor["Service"]["start"] = Effect.fn( "PullRequestSyncReactor.start", )(function* () { const events = engine.streamDomainEvents; yield* forkParked( - Stream.runForEach(events, (event) => - event.type === "thread.pull-request-synced" - ? Effect.forEach( + Stream.runForEach(events, (event) => { + switch (event.type) { + case "thread.pull-request-synced": + return Effect.forEach( visibleThreadPullRequests(event.payload.pullRequests ?? []).filter( (link) => link.snapshot === null, ), requestSync, { discard: true }, - ) - : Effect.void, - ).pipe(Effect.catchCause(logSkipped("pull request sync event stream failed", {}))), + ); + // An agent can merge or close its pull request from a shell (`gh pr merge`), which + // sends no merge notification. Read the thread's open links fresh when its run ends, + // so settlement does not wait for the next sweep and the cached summary. + case "run.updated": + return isTerminalRunStatus(event.payload.status) + ? refreshOpenLinks(event.threadId) + : Effect.void; + default: + return Effect.void; + } + }).pipe(Effect.catchCause(logSkipped("pull request sync event stream failed", {}))), ); yield* forkParked( Effect.gen(function* () { diff --git a/docs/internals/overview.md b/docs/internals/overview.md index 3d3218a7ca3d..fe22cc7e1b71 100644 --- a/docs/internals/overview.md +++ b/docs/internals/overview.md @@ -87,7 +87,9 @@ must reject that operation before changing the filesystem. Thread settlement is server-owned. The [settlement service](../../apps/server/src/orchestration-v2/ThreadSettlementService.ts) evaluates PR and inactivity settings without a connected client. Merge notifications invalidate cached PR state -and trigger a check. The guarded `thread.auto-settle` command rejects newer activity, explicit +and trigger a check. A merge outside T3, such as an agent running `gh pr merge`, sends no +notification, so the [PR sync reactor](../../apps/server/src/orchestration-v2/PullRequestSyncReactor.ts) +also re-reads a thread's open links when its run ends. The guarded `thread.auto-settle` command rejects newer activity, explicit settlement overrides, and live or blocked work. It records the activity timestamp for stable sorting and detaches idle provider sessions. Clients render the persisted result; they do not derive settlement from their own clocks or PR caches. From a430a720cf2ec7edbf71a62a990753e862bdee1d Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Fri, 2 Oct 2026 21:52:10 -0700 Subject: [PATCH 2/3] fix(server): refresh only after a run that ran a merge or close command Co-Authored-By: Claude Opus 5.5 (1M context) --- .../PullRequestSyncReactor.test.ts | 43 +++++++++++++++++-- .../PullRequestSyncReactor.ts | 20 +++++++-- docs/internals/overview.md | 2 +- 3 files changed, 58 insertions(+), 7 deletions(-) diff --git a/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts b/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts index c09afd08b387..6febdb9bbce3 100644 --- a/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts +++ b/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts @@ -7,6 +7,7 @@ import { PullRequestOperationError, RunId, ThreadId, + TurnItemId, type OrchestrationV2Command as OrchestrationCommand, type OrchestrationV2DomainEvent, type OrchestrationV2Run, @@ -316,6 +317,36 @@ function runUpdated( }; } +function commandRan(threadId: ThreadId, input: string): OrchestrationV2DomainEvent { + const runId = RunId.make(`run:${threadId}:1`); + const at = DateTime.makeUnsafe(NOW); + return { + type: "turn-item.updated", + id: EventId.make(`event:${threadId}:command`), + threadId, + runId, + occurredAt: at, + payload: { + id: TurnItemId.make(`item:${threadId}:command`), + threadId, + runId, + nodeId: null, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 1, + status: "completed", + title: null, + startedAt: at, + completedAt: at, + updatedAt: at, + type: "command_execution", + input, + }, + }; +} + /** What the reactor would have persisted, so the next sweep sees its own writes. */ function applySync( snapshot: TestShellSnapshot, @@ -877,7 +908,7 @@ describe("PullRequestSyncReactor", () => { ), ); - it.effect("reads the open links of a thread fresh when its run ends", () => + it.effect("reads open links fresh only when a run that ran a merge command ends", () => Effect.scoped( Effect.gen(function* () { yield* TestClock.setTime(Date.parse(NOW)); @@ -904,8 +935,14 @@ describe("PullRequestSyncReactor", () => { yield* Ref.set(state, "merged"); yield* Ref.set(fixture.summaryCalls, []); - yield* Queue.offer(fixture.domainEvents, runUpdated(ThreadId.make("agent"), "completed")); - // The ended run's thread lookup, then the requested sweep. + yield* Queue.offerAll(fixture.domainEvents, [ + // A run that only reads its pull request costs no host read when it ends. + commandRan(ThreadId.make("other"), "gh pr view 9"), + runUpdated(ThreadId.make("other"), "completed"), + commandRan(ThreadId.make("agent"), "gh pr merge 7 --squash 2>&1 | tail -3"), + runUpdated(ThreadId.make("agent"), "completed"), + ]); + // The agent thread's lookup, then the requested sweep. yield* Queue.take(fixture.snapshotReads); yield* Queue.take(fixture.snapshotReads); yield* reactor.drain; diff --git a/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts b/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts index 77f749b1d7d6..1b526d8f532e 100644 --- a/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts +++ b/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts @@ -33,6 +33,8 @@ import * as ProjectionStore from "./ProjectionStore.ts"; import { isTerminalRunStatus } from "./ThreadManagementService.ts"; const SLOW_SYNC_INTERVAL_MS = 15 * 60 * 1_000; +/** Shell commands that can merge or close a pull request without a merge notification. */ +const PULL_REQUEST_CLOSE_COMMAND = /\b(?:gh\s+pr|glab\s+mr)\s+(?:merge|close)\b/u; type SnapshotFields = Omit; @@ -332,6 +334,8 @@ export const make = Effect.gen(function* () { }).pipe(Effect.catchCause(logSkipped("pull request sync sweep failed", {}))), ); + // Threads whose current run ran a merge or close command, until that run ends. + const closeCommandThreads = new Set(); const refreshOpenLinks = (threadId: ThreadId) => projections.getThreadsWithPullRequests(threadId).pipe( Effect.flatMap((threads) => @@ -364,10 +368,20 @@ export const make = Effect.gen(function* () { { discard: true }, ); // An agent can merge or close its pull request from a shell (`gh pr merge`), which - // sends no merge notification. Read the thread's open links fresh when its run ends, - // so settlement does not wait for the next sweep and the cached summary. + // sends no merge notification. When a run that ran such a command ends, read the + // thread's open links fresh, so settlement does not wait for the next sweep and the + // cached summary. Other runs add no host reads. + case "turn-item.updated": + if ( + event.payload.type === "command_execution" && + PULL_REQUEST_CLOSE_COMMAND.test(event.payload.input) + ) { + closeCommandThreads.add(event.threadId); + } + return Effect.void; case "run.updated": - return isTerminalRunStatus(event.payload.status) + return isTerminalRunStatus(event.payload.status) && + closeCommandThreads.delete(event.threadId) ? refreshOpenLinks(event.threadId) : Effect.void; default: diff --git a/docs/internals/overview.md b/docs/internals/overview.md index fe22cc7e1b71..4a54e9c1b3c1 100644 --- a/docs/internals/overview.md +++ b/docs/internals/overview.md @@ -89,7 +89,7 @@ Thread settlement is server-owned. The and inactivity settings without a connected client. Merge notifications invalidate cached PR state and trigger a check. A merge outside T3, such as an agent running `gh pr merge`, sends no notification, so the [PR sync reactor](../../apps/server/src/orchestration-v2/PullRequestSyncReactor.ts) -also re-reads a thread's open links when its run ends. The guarded `thread.auto-settle` command rejects newer activity, explicit +re-reads a thread's open links when a run that ran a merge or close command ends. The guarded `thread.auto-settle` command rejects newer activity, explicit settlement overrides, and live or blocked work. It records the activity timestamp for stable sorting and detaches idle provider sessions. Clients render the persisted result; they do not derive settlement from their own clocks or PR caches. From aeacff9bfde4a9dd131c9c53082f8e65f8ae68ab Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Fri, 2 Oct 2026 21:52:22 -0700 Subject: [PATCH 3/3] docs: reflow settlement paragraph Co-Authored-By: Claude Opus 5.5 (1M context) --- docs/internals/overview.md | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/docs/internals/overview.md b/docs/internals/overview.md index 4a54e9c1b3c1..149bcf57dff0 100644 --- a/docs/internals/overview.md +++ b/docs/internals/overview.md @@ -89,8 +89,9 @@ Thread settlement is server-owned. The and inactivity settings without a connected client. Merge notifications invalidate cached PR state and trigger a check. A merge outside T3, such as an agent running `gh pr merge`, sends no notification, so the [PR sync reactor](../../apps/server/src/orchestration-v2/PullRequestSyncReactor.ts) -re-reads a thread's open links when a run that ran a merge or close command ends. The guarded `thread.auto-settle` command rejects newer activity, explicit -settlement overrides, and live or blocked work. It records the activity timestamp for stable +re-reads a thread's open links when a run that ran a merge or close command ends. The guarded +`thread.auto-settle` command rejects newer activity, explicit settlement overrides, and live or +blocked work. It records the activity timestamp for stable sorting and detaches idle provider sessions. Clients render the persisted result; they do not derive settlement from their own clocks or PR caches.