| author |  Patrik Votoček <patrik@votocek.cz> 2026-09-17 04:02:13+02:00 |
| committer |  GitHub <noreply@github.com> 2026-09-16 19:02:13-07:00 |
| log | 901db896691da5ab7b7428ef8a5da52883f831dd |
| tree | 81c47e02d9ba7eaf40b25a82f3b9fe554e9a9a2b |
| parent | d42a3bd2ae3e698141b26f6426aa28d8886e1754 |
| signature | Signed by PGP key B5690EEEBB952194 |
fix(server): keep VCS waits from blocking turn completion (#11970)
Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>
Co-authored-by: Julius Marminge <51714798+juliusmarminge@users.noreply.github.com>4 files changed, 416 insertions(+), 63 deletions(-)
apps
server
apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts+174-1
| ... | @@ -29,6 +29,7 @@ import * as Clock from "effect/Clock"; | ... | @@ -29,6 +29,7 @@ import * as Clock from "effect/Clock"; |
| 29 | import * as Deferred from "effect/Deferred"; | 29 | import * as Deferred from "effect/Deferred"; |
| 30 | import * as Effect from "effect/Effect"; | 30 | import * as Effect from "effect/Effect"; |
| 31 | import * as Exit from "effect/Exit"; | 31 | import * as Exit from "effect/Exit"; |
| 32 | import * as Fiber from "effect/Fiber"; | ||
| 32 | import * as Layer from "effect/Layer"; | 33 | import * as Layer from "effect/Layer"; |
| 33 | import * as ManagedRuntime from "effect/ManagedRuntime"; | 34 | import * as ManagedRuntime from "effect/ManagedRuntime"; |
| 34 | import * as Option from "effect/Option"; | 35 | import * as Option from "effect/Option"; |
| ... | @@ -42,6 +43,8 @@ import { afterEach, describe, expect, it } from "vite-plus/test"; | ... | @@ -42,6 +43,8 @@ import { afterEach, describe, expect, it } from "vite-plus/test"; |
| 42 | import { OrchestrationEventStoreLive } from "../../persistence/Layers/OrchestrationEventStore.ts"; | 43 | import { OrchestrationEventStoreLive } from "../../persistence/Layers/OrchestrationEventStore.ts"; |
| 43 | import { OrchestrationCommandReceiptRepositoryLive } from "../../persistence/Layers/OrchestrationCommandReceipts.ts"; | 44 | import { OrchestrationCommandReceiptRepositoryLive } from "../../persistence/Layers/OrchestrationCommandReceipts.ts"; |
| 44 | import { SqlitePersistenceMemory } from "../../persistence/Layers/Sqlite.ts"; | 45 | import { SqlitePersistenceMemory } from "../../persistence/Layers/Sqlite.ts"; |
| 46 | import { ProjectionTurnRepositoryLive } from "../../persistence/Layers/ProjectionTurns.ts"; | ||
| 47 | import { ProjectionTurnRepository } from "../../persistence/Services/ProjectionTurns.ts"; | ||
| 45 | import { | 48 | import { |
| 46 | ProviderService, | 49 | ProviderService, |
| 47 | type ProviderServiceShape, | 50 | type ProviderServiceShape, |
| ... | @@ -267,6 +270,7 @@ describe("ProviderRuntimeIngestion", () => { | ... | @@ -267,6 +270,7 @@ describe("ProviderRuntimeIngestion", () => { |
| 267 | serverSettings?: Partial<ServerSettings>; | 270 | serverSettings?: Partial<ServerSettings>; |
| 268 | threadTitle?: string; | 271 | threadTitle?: string; |
| 269 | workspaceSubdirectory?: string; | 272 | workspaceSubdirectory?: string; |
| 273 | isGitRepository?: CheckpointStore.CheckpointStore["Service"]["isGitRepository"]; | ||
| 270 | }) { | 274 | }) { |
| 271 | const repositoryRoot = makeTempDir("t3-provider-project-"); | 275 | const repositoryRoot = makeTempDir("t3-provider-project-"); |
| 272 | NodeChildProcess.execFileSync("git", ["init", "--initial-branch=main"], { | 276 | NodeChildProcess.execFileSync("git", ["init", "--initial-branch=main"], { |
| ... | @@ -327,7 +331,15 @@ describe("ProviderRuntimeIngestion", () => { | ... | @@ -327,7 +331,15 @@ describe("ProviderRuntimeIngestion", () => { |
| 327 | Layer.provideMerge(SqlitePersistenceMemory), | 331 | Layer.provideMerge(SqlitePersistenceMemory), |
| 328 | Layer.provideMerge(Layer.succeed(ProviderService, provider.service)), | 332 | Layer.provideMerge(Layer.succeed(ProviderService, provider.service)), |
| 329 | Layer.provideMerge(makeTestServerSettingsLayer(options?.serverSettings)), | 333 | Layer.provideMerge(makeTestServerSettingsLayer(options?.serverSettings)), |
| 330 | Layer.provideMerge(CheckpointStore.layer.pipe(Layer.provide(VcsDriverRegistry.layer))), | 334 | Layer.provideMerge( |
| 335 | Layer.effect( | ||
| 336 | CheckpointStore.CheckpointStore, | ||
| 337 | Effect.map(CheckpointStore.CheckpointStore, (store) => ({ | ||
| 338 | ...store, | ||
| 339 | isGitRepository: options?.isGitRepository ?? store.isGitRepository, | ||
| 340 | })), | ||
| 341 | ).pipe(Layer.provide(CheckpointStore.layer.pipe(Layer.provide(VcsDriverRegistry.layer)))), | ||
| 342 | ), | ||
| 331 | Layer.provideMerge(VcsProcess.layer), | 343 | Layer.provideMerge(VcsProcess.layer), |
| 332 | Layer.provideMerge(ServerConfig.layerTest(process.cwd(), process.cwd())), | 344 | Layer.provideMerge(ServerConfig.layerTest(process.cwd(), process.cwd())), |
| 333 | Layer.provideMerge(NodeServices.layer), | 345 | Layer.provideMerge(NodeServices.layer), |
| ... | @@ -405,6 +417,12 @@ describe("ProviderRuntimeIngestion", () => { | ... | @@ -405,6 +417,12 @@ describe("ProviderRuntimeIngestion", () => { |
| 405 | engine, | 417 | engine, |
| 406 | dispatch, | 418 | dispatch, |
| 407 | readModel: () => testRuntime.runPromise(snapshotQuery.getSnapshot()), | 419 | readModel: () => testRuntime.runPromise(snapshotQuery.getSnapshot()), |
| 420 | readTurn: (turnId: TurnId) => | ||
| 421 | testRuntime.runPromise( | ||
| 422 | Effect.flatMap(ProjectionTurnRepository, (turns) => | ||
| 423 | turns.getByTurnId({ threadId: asThreadId("thread-1"), turnId }), | ||
| 424 | ).pipe(Effect.map(Option.getOrUndefined), Effect.provide(ProjectionTurnRepositoryLive)), | ||
| 425 | ), | ||
| 408 | readThreadShell: () => | 426 | readThreadShell: () => |
| 409 | testRuntime.runPromise( | 427 | testRuntime.runPromise( |
| 410 | snapshotQuery | 428 | snapshotQuery |
| ... | @@ -4075,6 +4093,142 @@ describe("ProviderRuntimeIngestion", () => { | ... | @@ -4075,6 +4093,142 @@ describe("ProviderRuntimeIngestion", () => { |
| 4075 | }); | 4093 | }); |
| 4076 | }); | 4094 | }); |
| 4077 | 4095 | ||
| 4096 | effectIt.effect("settles the turn while repository detection for a diff is blocked", () => | ||
| 4097 | Effect.gen(function* () { | ||
| 4098 | const detectionStarted = yield* Deferred.make<void>(); | ||
| 4099 | const releaseDetection = yield* Deferred.make<boolean>(); | ||
| 4100 | const harness = yield* Effect.promise(() => | ||
| 4101 | createHarness({ | ||
| 4102 | isGitRepository: () => | ||
| 4103 | Deferred.succeed(detectionStarted, undefined).pipe( | ||
| 4104 | Effect.andThen(Deferred.await(releaseDetection)), | ||
| 4105 | ), | ||
| 4106 | }), | ||
| 4107 | ); | ||
| 4108 | yield* Effect.addFinalizer(() => Deferred.succeed(releaseDetection, true)); | ||
| 4109 | const base = { | ||
| 4110 | provider: ProviderDriverKind.make("codex"), | ||
| 4111 | threadId: asThreadId("thread-1"), | ||
| 4112 | turnId: asTurnId("blocked-diff-turn"), | ||
| 4113 | createdAt: "2026-01-01T00:00:00.000Z", | ||
| 4114 | }; | ||
| 4115 | yield* Effect.promise(() => | ||
| 4116 | harness.emitAndDrain([ | ||
| 4117 | { ...base, type: "turn.started", eventId: asEventId("evt-blocked-turn-start") }, | ||
| 4118 | ]), | ||
| 4119 | ); | ||
| 4120 | harness.emit({ | ||
| 4121 | ...base, | ||
| 4122 | type: "turn.diff.updated", | ||
| 4123 | eventId: asEventId("evt-blocked-diff"), | ||
| 4124 | payload: { unifiedDiff: "diff --git a/file.ts b/file.ts\n+new\n" }, | ||
| 4125 | }); | ||
| 4126 | yield* Deferred.await(detectionStarted); | ||
| 4127 | |||
| 4128 | const settled = yield* harness.engine.streamDomainEvents.pipe( | ||
| 4129 | Stream.filter( | ||
| 4130 | (event) => | ||
| 4131 | event.type === "thread.session-set" && | ||
| 4132 | event.payload.threadId === base.threadId && | ||
| 4133 | event.payload.session.status === "error", | ||
| 4134 | ), | ||
| 4135 | Stream.runHead, | ||
| 4136 | Effect.forkScoped({ startImmediately: true }), | ||
| 4137 | ); | ||
| 4138 | harness.emit({ | ||
| 4139 | ...base, | ||
| 4140 | type: "item.completed", | ||
| 4141 | eventId: asEventId("evt-blocked-final-reply"), | ||
| 4142 | itemId: asItemId("blocked-final-reply"), | ||
| 4143 | payload: { itemType: "assistant_message", status: "completed", detail: "Work finished." }, | ||
| 4144 | }); | ||
| 4145 | harness.emit({ | ||
| 4146 | ...base, | ||
| 4147 | type: "turn.completed", | ||
| 4148 | eventId: asEventId("evt-blocked-turn-completed"), | ||
| 4149 | payload: { state: "failed" }, | ||
| 4150 | }); | ||
| 4151 | // Resolves only if turn.completed is processed while detection is still blocked. | ||
| 4152 | yield* Fiber.join(settled); | ||
| 4153 | const blocked = yield* Effect.promise(harness.readModel); | ||
| 4154 | expect(blocked.threads[0]?.session).toMatchObject({ status: "error", activeTurnId: null }); | ||
| 4155 | expect(blocked.threads[0]?.messages).toEqual( | ||
| 4156 | expect.arrayContaining([expect.objectContaining({ text: "Work finished." })]), | ||
| 4157 | ); | ||
| 4158 | expect(blocked.threads[0]?.checkpoints).toEqual([]); | ||
| 4159 | |||
| 4160 | // A newer turn starts before detection returns. The late placeholder | ||
| 4161 | // must neither settle the failed turn as completed nor move the | ||
| 4162 | // latest-turn pointer back to it. | ||
| 4163 | const nextTurnId = asTurnId("next-turn"); | ||
| 4164 | const nextTurnStarted = yield* harness.engine.streamDomainEvents.pipe( | ||
| 4165 | Stream.filter( | ||
| 4166 | (event) => | ||
| 4167 | event.type === "thread.session-set" && | ||
| 4168 | event.payload.session.activeTurnId === nextTurnId, | ||
| 4169 | ), | ||
| 4170 | Stream.runHead, | ||
| 4171 | Effect.forkScoped({ startImmediately: true }), | ||
| 4172 | ); | ||
| 4173 | harness.emit({ | ||
| 4174 | ...base, | ||
| 4175 | type: "turn.started", | ||
| 4176 | turnId: nextTurnId, | ||
| 4177 | eventId: asEventId("evt-next-turn-start"), | ||
| 4178 | }); | ||
| 4179 | yield* Fiber.join(nextTurnStarted); | ||
| 4180 | yield* Deferred.succeed(releaseDetection, true); | ||
| 4181 | yield* Effect.promise(harness.drain); | ||
| 4182 | const released = yield* Effect.promise(harness.readModel); | ||
| 4183 | expect(released.threads[0]?.checkpoints).toEqual([]); | ||
| 4184 | expect(released.threads[0]?.latestTurn).toMatchObject({ | ||
| 4185 | turnId: nextTurnId, | ||
| 4186 | state: "running", | ||
| 4187 | }); | ||
| 4188 | expect(yield* Effect.promise(() => harness.readTurn(base.turnId))).toMatchObject({ | ||
| 4189 | state: "error", | ||
| 4190 | checkpointRef: null, | ||
| 4191 | }); | ||
| 4192 | }), | ||
| 4193 | ); | ||
| 4194 | |||
| 4195 | effectIt.effect("ignores a diff for a missing turn without moving the latest turn", () => | ||
| 4196 | Effect.gen(function* () { | ||
| 4197 | const harness = yield* Effect.promise(() => createHarness()); | ||
| 4198 | const base = { | ||
| 4199 | provider: ProviderDriverKind.make("codex"), | ||
| 4200 | threadId: asThreadId("thread-1"), | ||
| 4201 | createdAt: "2026-01-01T00:00:00.000Z", | ||
| 4202 | }; | ||
| 4203 | yield* Effect.promise(() => | ||
| 4204 | harness.emitAndDrain([ | ||
| 4205 | { | ||
| 4206 | ...base, | ||
| 4207 | type: "turn.started", | ||
| 4208 | eventId: asEventId("evt-existing-turn"), | ||
| 4209 | turnId: asTurnId("current-turn"), | ||
| 4210 | }, | ||
| 4211 | { | ||
| 4212 | ...base, | ||
| 4213 | type: "turn.diff.updated", | ||
| 4214 | eventId: asEventId("evt-missing-turn-diff"), | ||
| 4215 | turnId: asTurnId("missing-turn"), | ||
| 4216 | payload: { unifiedDiff: "diff --git a/file.ts b/file.ts\n+late\n" }, | ||
| 4217 | }, | ||
| 4218 | ]), | ||
| 4219 | ); | ||
| 4220 | const snapshot = yield* Effect.promise(harness.readModel); | ||
| 4221 | expect(snapshot.threads[0]?.checkpoints).toEqual([]); | ||
| 4222 | expect(snapshot.threads[0]?.latestTurn).toMatchObject({ | ||
| 4223 | turnId: "current-turn", | ||
| 4224 | state: "running", | ||
| 4225 | }); | ||
| 4226 | expect( | ||
| 4227 | yield* Effect.promise(() => harness.readTurn(asTurnId("missing-turn"))), | ||
| 4228 | ).toBeUndefined(); | ||
| 4229 | }), | ||
| 4230 | ); | ||
| 4231 | |||
| 4078 | effectIt.effect("tracks provider diff updates from a nested Git workspace", () => | 4232 | effectIt.effect("tracks provider diff updates from a nested Git workspace", () => |
| 4079 | Effect.gen(function* () { | 4233 | Effect.gen(function* () { |
| 4080 | const harness = yield* Effect.promise(() => | 4234 | const harness = yield* Effect.promise(() => |
| ... | @@ -4082,6 +4236,14 @@ describe("ProviderRuntimeIngestion", () => { | ... | @@ -4082,6 +4236,14 @@ describe("ProviderRuntimeIngestion", () => { |
| 4082 | ); | 4236 | ); |
| 4083 | yield* Effect.promise(() => | 4237 | yield* Effect.promise(() => |
| 4084 | harness.emitAndDrain([ | 4238 | harness.emitAndDrain([ |
| 4239 | { | ||
| 4240 | type: "turn.started", | ||
| 4241 | eventId: asEventId("evt-nested-turn-started"), | ||
| 4242 | provider: ProviderDriverKind.make("codex"), | ||
| 4243 | createdAt: "2026-01-01T00:00:00.000Z", | ||
| 4244 | threadId: asThreadId("thread-1"), | ||
| 4245 | turnId: asTurnId("nested-turn"), | ||
| 4246 | }, | ||
| 4085 | { | 4247 | { |
| 4086 | type: "turn.diff.updated", | 4248 | type: "turn.diff.updated", |
| 4087 | eventId: asEventId("evt-nested-diff"), | 4249 | eventId: asEventId("evt-nested-diff"), |
| ... | @@ -4106,6 +4268,17 @@ describe("ProviderRuntimeIngestion", () => { | ... | @@ -4106,6 +4268,17 @@ describe("ProviderRuntimeIngestion", () => { |
| 4106 | const harness = await createHarness(); | 4268 | const harness = await createHarness(); |
| 4107 | const now = "2026-01-01T00:00:00.000Z"; | 4269 | const now = "2026-01-01T00:00:00.000Z"; |
| 4108 | 4270 | ||
| 4271 | await harness.emitAndDrain([ | ||
| 4272 | { | ||
| 4273 | type: "turn.started", | ||
| 4274 | eventId: asEventId("evt-p1-turn-started"), | ||
| 4275 | provider: ProviderDriverKind.make("codex"), | ||
| 4276 | createdAt: now, | ||
| 4277 | threadId: asThreadId("thread-1"), | ||
| 4278 | turnId: asTurnId("turn-p1"), | ||
| 4279 | }, | ||
| 4280 | ]); | ||
| 4281 | |||
| 4109 | harness.emit({ | 4282 | harness.emit({ |
| 4110 | type: "thread.metadata.updated", | 4283 | type: "thread.metadata.updated", |
| 4111 | eventId: asEventId("evt-thread-metadata-updated"), | 4284 | eventId: asEventId("evt-thread-metadata-updated"), |
apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts+96-61
| ... | @@ -126,6 +126,8 @@ type TurnStartRequestedDomainEvent = Extract< | ... | @@ -126,6 +126,8 @@ type TurnStartRequestedDomainEvent = Extract< |
| 126 | { type: "thread.turn-start-requested" } | 126 | { type: "thread.turn-start-requested" } |
| 127 | >; | 127 | >; |
| 128 | 128 | ||
| 129 | type ProviderDiffEvent = Extract<ProviderRuntimeEvent, { type: "turn.diff.updated" }>; | ||
| 130 | |||
| 129 | type RuntimeIngestionInput = | 131 | type RuntimeIngestionInput = |
| 130 | | { | 132 | | { |
| 131 | source: "runtime"; | 133 | source: "runtime"; |
| ... | @@ -134,6 +136,11 @@ type RuntimeIngestionInput = | ... | @@ -134,6 +136,11 @@ type RuntimeIngestionInput = |
| 134 | | { | 136 | | { |
| 135 | source: "domain"; | 137 | source: "domain"; |
| 136 | event: TurnStartRequestedDomainEvent; | 138 | event: TurnStartRequestedDomainEvent; |
| 139 | } | ||
| 140 | | { | ||
| 141 | /** A diff whose workspace the diff worker confirmed is a Git repository. */ | ||
| 142 | source: "diff"; | ||
| 143 | event: ProviderDiffEvent; | ||
| 137 | }; | 144 | }; |
| 138 | 145 | ||
| 139 | function toTurnId(value: TurnId | string | undefined): TurnId | undefined { | 146 | function toTurnId(value: TurnId | string | undefined): TurnId | undefined { |
| ... | @@ -2436,48 +2443,6 @@ const make = Effect.gen(function* () { | ... | @@ -2436,48 +2443,6 @@ const make = Effect.gen(function* () { |
| 2436 | } | 2443 | } |
| 2437 | } | 2444 | } |
| 2438 | 2445 | ||
| 2439 | if (event.type === "turn.diff.updated") { | ||
| 2440 | const turnId = toTurnId(event.turnId); | ||
| 2441 | const checkpointContext = turnId | ||
| 2442 | ? yield* projectionSnapshotQuery | ||
| 2443 | .getThreadCheckpointContext(thread.id) | ||
| 2444 | .pipe(Effect.map(Option.getOrUndefined)) | ||
| 2445 | : undefined; | ||
| 2446 | const workspaceCwd = | ||
| 2447 | checkpointContext?.worktreePath ?? checkpointContext?.workspaceRoot ?? undefined; | ||
| 2448 | if ( | ||
| 2449 | turnId && | ||
| 2450 | checkpointContext && | ||
| 2451 | workspaceCwd && | ||
| 2452 | (yield* checkpointStore.isGitRepository(workspaceCwd)) | ||
| 2453 | ) { | ||
| 2454 | // Skip if a checkpoint already exists for this turn. A real | ||
| 2455 | // (non-placeholder) capture from CheckpointReactor should not | ||
| 2456 | // be clobbered, and dispatching a duplicate placeholder for the | ||
| 2457 | // same turnId would produce an unstable checkpointTurnCount. | ||
| 2458 | if (hasCheckpointForTurn(checkpointContext.checkpoints, turnId)) { | ||
| 2459 | // Already tracked; no-op. | ||
| 2460 | } else { | ||
| 2461 | const assistantMessageId = MessageId.make( | ||
| 2462 | `assistant:${event.itemId ?? event.turnId ?? event.eventId}`, | ||
| 2463 | ); | ||
| 2464 | yield* orchestrationEngine.dispatch({ | ||
| 2465 | type: "thread.turn.diff.complete", | ||
| 2466 | commandId: yield* providerCommandId(event, "thread-turn-diff-complete"), | ||
| 2467 | threadId: thread.id, | ||
| 2468 | turnId, | ||
| 2469 | completedAt: now, | ||
| 2470 | checkpointRef: CheckpointRef.make(`provider-diff:${event.eventId}`), | ||
| 2471 | status: "missing", | ||
| 2472 | files: [], | ||
| 2473 | assistantMessageId, | ||
| 2474 | checkpointTurnCount: maxCheckpointTurnCount(checkpointContext.checkpoints) + 1, | ||
| 2475 | createdAt: now, | ||
| 2476 | }); | ||
| 2477 | } | ||
| 2478 | } | ||
| 2479 | } | ||
| 2480 | |||
| 2481 | if (event.type === "task.started" || event.type === "task.progress") { | 2446 | if (event.type === "task.started" || event.type === "task.progress") { |
| 2482 | const description = event.payload.description?.trim(); | 2447 | const description = event.payload.description?.trim(); |
| 2483 | if (description) { | 2448 | if (description) { |
| ... | @@ -2626,31 +2591,100 @@ const make = Effect.gen(function* () { | ... | @@ -2626,31 +2591,100 @@ const make = Effect.gen(function* () { |
| 2626 | 2591 | ||
| 2627 | const processDomainEvent = (_event: TurnStartRequestedDomainEvent) => Effect.void; | 2592 | const processDomainEvent = (_event: TurnStartRequestedDomainEvent) => Effect.void; |
| 2628 | 2593 | ||
| 2629 | const processInput = (input: RuntimeIngestionInput) => | 2594 | // Records a mid-turn placeholder checkpoint for a provider diff. Runs on the |
| 2630 | input.source === "runtime" ? processRuntimeEvent(input.event) : processDomainEvent(input.event); | 2595 | // lifecycle worker, after repository detection, so the running-turn check |
| 2596 | // and the dispatch are ordered with the turn's terminal events: a diff that | ||
| 2597 | // resolved after turn.completed must not rewrite the settled turn's state or | ||
| 2598 | // move the latest-turn pointer back. | ||
| 2599 | const recordProviderDiff = Effect.fn("recordProviderDiff")(function* (event: ProviderDiffEvent) { | ||
| 2600 | const thread = yield* resolveThreadRuntimeContext(event.threadId); | ||
| 2601 | const turnId = toTurnId(event.turnId); | ||
| 2602 | if (!thread || !turnId) return; | ||
| 2603 | const turn = yield* projectionTurnRepository.getByTurnId({ threadId: thread.id, turnId }); | ||
| 2604 | if (Option.isNone(turn) || turn.value.state !== "running") return; | ||
| 2605 | const checkpointContext = yield* projectionSnapshotQuery | ||
| 2606 | .getThreadCheckpointContext(thread.id) | ||
| 2607 | .pipe(Effect.map(Option.getOrUndefined)); | ||
| 2608 | // Skip if a checkpoint already exists for this turn. A real | ||
| 2609 | // (non-placeholder) capture from CheckpointReactor should not | ||
| 2610 | // be clobbered, and dispatching a duplicate placeholder for the | ||
| 2611 | // same turnId would produce an unstable checkpointTurnCount. | ||
| 2612 | if (!checkpointContext || hasCheckpointForTurn(checkpointContext.checkpoints, turnId)) return; | ||
| 2613 | const now = event.createdAt; | ||
| 2614 | yield* orchestrationEngine.dispatch({ | ||
| 2615 | type: "thread.turn.diff.complete", | ||
| 2616 | commandId: yield* providerCommandId(event, "thread-turn-diff-complete"), | ||
| 2617 | threadId: thread.id, | ||
| 2618 | turnId, | ||
| 2619 | completedAt: now, | ||
| 2620 | checkpointRef: CheckpointRef.make(`provider-diff:${event.eventId}`), | ||
| 2621 | status: "missing", | ||
| 2622 | files: [], | ||
| 2623 | assistantMessageId: MessageId.make( | ||
| 2624 | `assistant:${event.itemId ?? event.turnId ?? event.eventId}`, | ||
| 2625 | ), | ||
| 2626 | checkpointTurnCount: maxCheckpointTurnCount(checkpointContext.checkpoints) + 1, | ||
| 2627 | createdAt: now, | ||
| 2628 | }); | ||
| 2629 | }); | ||
| 2630 | |||
| 2631 | const processInput = (input: RuntimeIngestionInput) => { | ||
| 2632 | switch (input.source) { | ||
| 2633 | case "runtime": | ||
| 2634 | return processRuntimeEvent(input.event); | ||
| 2635 | case "domain": | ||
| 2636 | return processDomainEvent(input.event); | ||
| 2637 | case "diff": | ||
| 2638 | return recordProviderDiff(input.event); | ||
| 2639 | } | ||
| 2640 | }; | ||
| 2631 | 2641 | ||
| 2632 | const processInputSafely = (input: RuntimeIngestionInput) => | 2642 | const logIngestionFailure = |
| 2633 | processInput(input).pipe( | 2643 | (source: string, event: { readonly eventId: string; readonly type: string }) => |
| 2634 | Effect.catchCause((cause) => { | 2644 | <E, R>(effect: Effect.Effect<void, E, R>) => |
| 2635 | if (Cause.hasInterruptsOnly(cause)) { | 2645 | effect.pipe( |
| 2636 | return Effect.failCause(cause); | 2646 | Effect.catchCause((cause) => { |
| 2637 | } | 2647 | if (Cause.hasInterruptsOnly(cause)) { |
| 2638 | return Effect.logWarning("provider runtime ingestion failed to process event", { | 2648 | return Effect.failCause(cause); |
| 2639 | source: input.source, | 2649 | } |
| 2640 | eventId: input.event.eventId, | 2650 | return Effect.logWarning("provider runtime ingestion failed to process event", { |
| 2641 | eventType: input.event.type, | 2651 | source, |
| 2642 | cause: Cause.pretty(cause), | 2652 | eventId: event.eventId, |
| 2643 | }); | 2653 | eventType: event.type, |
| 2644 | }), | 2654 | cause: Cause.pretty(cause), |
| 2645 | ); | 2655 | }); |
| 2656 | }), | ||
| 2657 | ); | ||
| 2658 | |||
| 2659 | const worker = yield* makeDrainableWorker((input: RuntimeIngestionInput) => | ||
| 2660 | processInput(input).pipe(logIngestionFailure(input.source, input.event)), | ||
| 2661 | ); | ||
| 2646 | 2662 | ||
| 2647 | const worker = yield* makeDrainableWorker(processInputSafely); | 2663 | // Repository detection for a diff goes through VCS subprocesses, which can |
| 2664 | // stall behind slow or hung git. It runs on its own worker so a stuck diff | ||
| 2665 | // never delays the lifecycle worker; confirmed diffs are handed back to it. | ||
| 2666 | const detectProviderDiffRepository = Effect.fn("detectProviderDiffRepository")(function* ( | ||
| 2667 | event: ProviderDiffEvent, | ||
| 2668 | ) { | ||
| 2669 | if (!toTurnId(event.turnId)) return; | ||
| 2670 | const checkpointContext = yield* projectionSnapshotQuery | ||
| 2671 | .getThreadCheckpointContext(event.threadId) | ||
| 2672 | .pipe(Effect.map(Option.getOrUndefined)); | ||
| 2673 | const workspaceCwd = checkpointContext?.worktreePath ?? checkpointContext?.workspaceRoot; | ||
| 2674 | if (!workspaceCwd || !(yield* checkpointStore.isGitRepository(workspaceCwd))) return; | ||
| 2675 | yield* worker.enqueue({ source: "diff", event }); | ||
| 2676 | }); | ||
| 2677 | const diffWorker = yield* makeDrainableWorker((event: ProviderDiffEvent) => | ||
| 2678 | detectProviderDiffRepository(event).pipe(logIngestionFailure("diff", event)), | ||
| 2679 | ); | ||
| 2648 | 2680 | ||
| 2649 | const start: ProviderRuntimeIngestionShape["start"] = () => | 2681 | const start: ProviderRuntimeIngestionShape["start"] = () => |
| 2650 | Effect.gen(function* () { | 2682 | Effect.gen(function* () { |
| 2651 | yield* forkParked( | 2683 | yield* forkParked( |
| 2652 | Stream.runForEach(providerService.streamEvents, (event) => | 2684 | Stream.runForEach(providerService.streamEvents, (event) => |
| 2653 | worker.enqueue({ source: "runtime", event }), | 2685 | event.type === "turn.diff.updated" |
| 2686 | ? diffWorker.enqueue(event) | ||
| 2687 | : worker.enqueue({ source: "runtime", event }), | ||
| 2654 | ), | 2688 | ), |
| 2655 | ); | 2689 | ); |
| 2656 | yield* forkParked( | 2690 | yield* forkParked( |
| ... | @@ -2665,7 +2699,8 @@ const make = Effect.gen(function* () { | ... | @@ -2665,7 +2699,8 @@ const make = Effect.gen(function* () { |
| 2665 | 2699 | ||
| 2666 | return { | 2700 | return { |
| 2667 | start, | 2701 | start, |
| 2668 | drain: worker.drain, | 2702 | // The diff worker feeds the lifecycle worker, so drain it first. |
| 2703 | drain: diffWorker.drain.pipe(Effect.andThen(worker.drain)), | ||
| 2669 | } satisfies ProviderRuntimeIngestionShape; | 2704 | } satisfies ProviderRuntimeIngestionShape; |
| 2670 | }); | 2705 | }); |
| 2671 | 2706 |
apps/server/src/orchestration/decider.ts+19-1
| ... | @@ -2075,11 +2075,29 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" | ... | @@ -2075,11 +2075,29 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" |
| 2075 | } | 2075 | } |
| 2076 | 2076 | ||
| 2077 | case "thread.turn.diff.complete": { | 2077 | case "thread.turn.diff.complete": { |
| 2078 | yield* requireThread({ | 2078 | const thread = yield* requireThread({ |
| 2079 | readModel, | 2079 | readModel, |
| 2080 | command, | 2080 | command, |
| 2081 | threadId: command.threadId, | 2081 | threadId: command.threadId, |
| 2082 | }); | 2082 | }); |
| 2083 | // A placeholder (status "missing") must never replace a checkpoint that | ||
| 2084 | // was already captured with a real git ref. Provider diff ingestion | ||
| 2085 | // checks this before dispatching, but CheckpointReactor can commit the | ||
| 2086 | // real capture in between; the decider runs under the engine's command | ||
| 2087 | // lock, so rejecting here closes that window. | ||
| 2088 | const existingCheckpoint = thread.checkpoints.find( | ||
| 2089 | (checkpoint) => checkpoint.turnId === command.turnId, | ||
| 2090 | ); | ||
| 2091 | if ( | ||
| 2092 | command.status === "missing" && | ||
| 2093 | existingCheckpoint !== undefined && | ||
| 2094 | existingCheckpoint.status !== "missing" | ||
| 2095 | ) { | ||
| 2096 | return yield* new OrchestrationCommandInvariantError({ | ||
| 2097 | commandType: command.type, | ||
| 2098 | detail: `turn ${command.turnId} already has a captured checkpoint`, | ||
| 2099 | }); | ||
| 2100 | } | ||
| 2083 | return { | 2101 | return { |
| 2084 | ...(yield* withEventBase({ | 2102 | ...(yield* withEventBase({ |
| 2085 | aggregateKind: "thread", | 2103 | aggregateKind: "thread", |
apps/server/src/orchestration/decider.turnDiffComplete.test.ts created+127
| ... | @@ -0,0 +1,127 @@ | ||
| 1 | import { | ||
| 2 | CheckpointRef, | ||
| 3 | CommandId, | ||
| 4 | MessageId, | ||
| 5 | ProjectId, | ||
| 6 | ProviderInstanceId, | ||
| 7 | ThreadId, | ||
| 8 | TurnId, | ||
| 9 | type OrchestrationCheckpointSummary, | ||
| 10 | type OrchestrationReadModel, | ||
| 11 | } from "@t3tools/contracts"; | ||
| 12 | import * as NodeServices from "@effect/platform-node/NodeServices"; | ||
| 13 | import { expect, it } from "@effect/vitest"; | ||
| 14 | import * as Effect from "effect/Effect"; | ||
| 15 | import * as Exit from "effect/Exit"; | ||
| 16 | |||
| 17 | import { decideOrchestrationCommand } from "./decider.ts"; | ||
| 18 | |||
| 19 | const NOW = "2026-01-01T00:00:00.000Z"; | ||
| 20 | const THREAD_ID = ThreadId.make("thread-1"); | ||
| 21 | const TURN_ID = TurnId.make("turn-1"); | ||
| 22 | |||
| 23 | function makeReadModel(checkpoints: ReadonlyArray<OrchestrationCheckpointSummary>) { | ||
| 24 | return { | ||
| 25 | snapshotSequence: 0, | ||
| 26 | projects: [], | ||
| 27 | threads: [ | ||
| 28 | { | ||
| 29 | id: THREAD_ID, | ||
| 30 | projectId: ProjectId.make("project-1"), | ||
| 31 | title: "Thread", | ||
| 32 | modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" }, | ||
| 33 | runtimeMode: "full-access", | ||
| 34 | interactionMode: "default", | ||
| 35 | branch: null, | ||
| 36 | worktreePath: null, | ||
| 37 | pullRequests: [], | ||
| 38 | latestTurn: null, | ||
| 39 | createdAt: NOW, | ||
| 40 | updatedAt: NOW, | ||
| 41 | archivedAt: null, | ||
| 42 | settledOverride: null, | ||
| 43 | settledAt: null, | ||
| 44 | snoozedUntil: null, | ||
| 45 | snoozedAt: null, | ||
| 46 | pinnedAt: null, | ||
| 47 | pinOrderKey: null, | ||
| 48 | deletedAt: null, | ||
| 49 | messages: [], | ||
| 50 | proposedPlans: [], | ||
| 51 | activities: [], | ||
| 52 | checkpoints, | ||
| 53 | session: null, | ||
| 54 | }, | ||
| 55 | ], | ||
| 56 | updatedAt: NOW, | ||
| 57 | } satisfies OrchestrationReadModel; | ||
| 58 | } | ||
| 59 | |||
| 60 | function makeCheckpoint(status: OrchestrationCheckpointSummary["status"]) { | ||
| 61 | return { | ||
| 62 | turnId: TURN_ID, | ||
| 63 | checkpointTurnCount: 1, | ||
| 64 | checkpointRef: CheckpointRef.make(`existing:${status}`), | ||
| 65 | status, | ||
| 66 | files: [], | ||
| 67 | assistantMessageId: null, | ||
| 68 | completedAt: NOW, | ||
| 69 | } satisfies OrchestrationCheckpointSummary; | ||
| 70 | } | ||
| 71 | |||
| 72 | function placeholderCommand() { | ||
| 73 | return { | ||
| 74 | type: "thread.turn.diff.complete", | ||
| 75 | commandId: CommandId.make("cmd-diff-placeholder"), | ||
| 76 | threadId: THREAD_ID, | ||
| 77 | turnId: TURN_ID, | ||
| 78 | completedAt: NOW, | ||
| 79 | checkpointRef: CheckpointRef.make("provider-diff:event-1"), | ||
| 80 | status: "missing", | ||
| 81 | files: [], | ||
| 82 | assistantMessageId: MessageId.make("assistant:turn-1"), | ||
| 83 | checkpointTurnCount: 2, | ||
| 84 | createdAt: NOW, | ||
| 85 | } as const; | ||
| 86 | } | ||
| 87 | |||
| 88 | it.layer(NodeServices.layer)("turn diff complete decider", (it) => { | ||
| 89 | it.effect("rejects a placeholder when the turn already has a captured checkpoint", () => | ||
| 90 | Effect.gen(function* () { | ||
| 91 | const exit = yield* Effect.exit( | ||
| 92 | decideOrchestrationCommand({ | ||
| 93 | command: placeholderCommand(), | ||
| 94 | readModel: makeReadModel([makeCheckpoint("ready")]), | ||
| 95 | }), | ||
| 96 | ); | ||
| 97 | expect(Exit.isFailure(exit)).toBe(true); | ||
| 98 | }), | ||
| 99 | ); | ||
| 100 | |||
| 101 | it.effect("accepts a placeholder when the turn only has a placeholder", () => | ||
| 102 | Effect.gen(function* () { | ||
| 103 | const event = yield* decideOrchestrationCommand({ | ||
| 104 | command: placeholderCommand(), | ||
| 105 | readModel: makeReadModel([makeCheckpoint("missing")]), | ||
| 106 | }); | ||
| 107 | const events = Array.isArray(event) ? event : [event]; | ||
| 108 | expect(events).toHaveLength(1); | ||
| 109 | expect(events[0]?.type).toBe("thread.turn-diff-completed"); | ||
| 110 | }), | ||
| 111 | ); | ||
| 112 | |||
| 113 | it.effect("lets a captured checkpoint replace an earlier placeholder", () => | ||
| 114 | Effect.gen(function* () { | ||
| 115 | const event = yield* decideOrchestrationCommand({ | ||
| 116 | command: { | ||
| 117 | ...placeholderCommand(), | ||
| 118 | checkpointRef: CheckpointRef.make("refs/t3/checkpoints/turn-1"), | ||
| 119 | status: "ready", | ||
| 120 | }, | ||
| 121 | readModel: makeReadModel([makeCheckpoint("missing")]), | ||
| 122 | }); | ||
| 123 | const events = Array.isArray(event) ? event : [event]; | ||
| 124 | expect(events[0]?.type).toBe("thread.turn-diff-completed"); | ||
| 125 | }), | ||
| 126 | ); | ||
| 127 | }); | ||