diff --git a/apps/server/src/orchestration-v2/ThreadSettlementService.test.ts b/apps/server/src/orchestration-v2/ThreadSettlementService.test.ts index 50ac83b814626..c16c8c160e68e 100644 --- a/apps/server/src/orchestration-v2/ThreadSettlementService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadSettlementService.test.ts @@ -8,6 +8,7 @@ import { ProviderInstanceId, RunId, ThreadId, + TurnItemId, type OrchestrationProjectShell, type OrchestrationV2AppThread, type OrchestrationV2Command, @@ -598,9 +599,12 @@ const makeHarness = Effect.fn("makeThreadSettlementHarness")(function* (options: return next; }); - const branchPullRequest: GitManager.GitManager["Service"]["branchPullRequest"] = (input) => + const branchPullRequest: GitManager.GitManager["Service"]["branchPullRequest"] = ( + input, + lookupOptions, + ) => Ref.update(branchCalls, (calls) => [...calls, input]).pipe( - Effect.andThen(options.branchPullRequest?.(input) ?? Effect.succeed(null)), + Effect.andThen(options.branchPullRequest?.(input, lookupOptions) ?? Effect.succeed(null)), ); const pullRequestSummary: PullRequestService.PullRequestService["Service"]["summary"] = ( input, @@ -1051,6 +1055,75 @@ describe("ThreadSettlementServiceV2 worker", () => { ), ); + it.effect("settles an unlinked branch thread its agent merged from a shell, past the cache", () => + Effect.scoped( + Effect.gen(function* () { + yield* TestClock.setTime(Date.parse(NOW)); + const settled = yield* Deferred.make(); + const thread = makeThread("shell-merged", { + branch: "saved-feature", + latestUserMessageAt: DateTime.makeUnsafe("2026-08-27T00:00:00.000Z"), + }); + const fixture = yield* makeHarness({ + snapshot: makeSnapshot([thread]), + // The host already merged it; only a fresh read sees that past the cached open lookup. + branchPullRequest: (_input, lookupOptions) => + Effect.succeed(makeBranchPullRequest(lookupOptions?.refresh ? "merged" : "open")), + onDispatch: () => Deferred.succeed(settled, undefined), + }); + const at = DateTime.makeUnsafe(NOW); + const runId = RunId.make("run:shell-merged"); + const runEnded: OrchestrationV2DomainEvent = { + type: "run.background-work-cancelled", + id: EventId.make("event:shell-merged:run-ended"), + threadId: thread.id, + occurredAt: at, + payload: { runId, restartCancelledBackgroundWork: [] }, + }; + + yield* Effect.gen(function* () { + const reactor = yield* ThreadSettlementService.ThreadSettlementServiceV2; + yield* startHarness(reactor, fixture.activation, fixture.snapshotReads); + yield* fixture.publishEvent(runEnded); + yield* Queue.take(fixture.snapshotReads); + yield* reactor.drain; + assert.deepStrictEqual(yield* Ref.get(fixture.commands), []); + + yield* fixture.publishEvent({ + type: "turn-item.updated", + id: EventId.make("event:shell-merged:command"), + threadId: thread.id, + runId, + occurredAt: at, + payload: { + id: TurnItemId.make("item:shell-merged:command"), + threadId: thread.id, + 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: "gh pr merge 42 --squash", + }, + }); + yield* Deferred.await(settled); + assert.deepStrictEqual( + (yield* Ref.get(fixture.commands)).map((command) => command.threadId), + [thread.id], + ); + }).pipe(Effect.provide(fixture.layer)); + }), + ), + ); + it.effect( "settles branch threads on a pull request merge without waiting for the next sweep", () => diff --git a/apps/server/src/orchestration-v2/ThreadSettlementService.ts b/apps/server/src/orchestration-v2/ThreadSettlementService.ts index a14601ae84054..35ed1d9721348 100644 --- a/apps/server/src/orchestration-v2/ThreadSettlementService.ts +++ b/apps/server/src/orchestration-v2/ThreadSettlementService.ts @@ -298,6 +298,9 @@ export const make = Effect.gen(function* () { const pullRequests = yield* PullRequestService.PullRequestService; const crypto = yield* Crypto.Crypto; const fileSystem = yield* FileSystem.FileSystem; + // A shell merge or close sends no event, and the branch lookup caches an open pull request for + // a minute, so these threads' next lookup reads the host fresh. + const shellClosedThreads = new Set(); const terminals = yield* TerminalManager.TerminalManager; const projectScripts = yield* ProjectSetupScriptRunner.ProjectSetupScriptRunner; // Repeated thread.settled retains settledAt; run the action only once per settlement. @@ -490,7 +493,8 @@ export const make = Effect.gen(function* () { if (cwd === undefined) { return yield* Effect.die(new Error("thread project not found")); } - return yield* git.branchPullRequest({ cwd, branch: thread.branch }); + const refresh = group.filter((member) => shellClosedThreads.delete(member.id)).length > 0; + return yield* git.branchPullRequest({ cwd, branch: thread.branch }, { refresh }); }); yield* Effect.forEach( @@ -615,10 +619,21 @@ export const make = Effect.gen(function* () { return backgroundWorkHoldsCompletion(event.payload.pendingBackgroundTasks ?? []) ? Effect.void : enqueueThread(event.threadId); - case "turn-item.updated": - return turnItemUpdateCanEndBackgroundWork(event.payload) + case "turn-item.updated": { + const item = event.payload; + if ( + item.type === "command_execution" && + !isOrchestrationV2WorkActive(item.status) && + sourceControlDefinitions.definitions.some( + (definition) => definition.changeRequestCloseCommand?.test(item.input) === true, + ) + ) { + shellClosedThreads.add(event.threadId); + } + return turnItemUpdateCanEndBackgroundWork(item) ? enqueueThread(event.threadId) : Effect.void; + } case "provider-session.updated": return event.payload.status !== "starting" && event.payload.status !== "running" ? enqueueThread(event.threadId)