From 6108ef3d3d0ef2a598088a1bb64980580e935830 Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Sat, 3 Oct 2026 01:43:08 -0700 Subject: [PATCH] fix(server): runs no longer get stuck (#15048) Co-authored-by: Claude Opus 5.5 (1M context) --- .../src/orchestration-v2/EffectOutbox.ts | 24 +- apps/server/src/orchestration-v2/EventSink.ts | 91 +++-- .../FoundationPersistence.test.ts | 159 ++++++++ .../ProviderSessionManager.test.ts | 350 +++++++++++++++++- .../ProviderSessionManager.ts | 172 +++++++-- .../ProviderTurnStartService.test.ts | 59 ++- .../ProviderTurnStartService.ts | 4 +- .../RunExecutionService.test.ts | 161 +++++++- .../orchestration-v2/RunExecutionService.ts | 54 +-- 9 files changed, 959 insertions(+), 115 deletions(-) diff --git a/apps/server/src/orchestration-v2/EffectOutbox.ts b/apps/server/src/orchestration-v2/EffectOutbox.ts index 4841a576248022d9fd4d3e3d8ae555895da32d4c..9f63d8c568b670b75c0d352dbb4902d6df0722e4 100644 --- a/apps/server/src/orchestration-v2/EffectOutbox.ts +++ b/apps/server/src/orchestration-v2/EffectOutbox.ts @@ -278,9 +278,16 @@ export const layer: Layer.Layer = La available, Array.from({ length: Math.min(64, Math.max(0, Math.floor(count))) }, () => undefined), ).pipe(Effect.asVoid); + // Each thread runs its effects one at a time, in enqueue (rowid) order. An earlier + // effect waiting out a retry backoff still blocks later ones, so a turn + // cannot start while a failed rollback is about to restore files. A claim + // that skips restart continuations is not blocked by them either. // Title generation is correlated metadata work, so it has its own // per-thread lane and cannot delay provider lifecycle effects. - const claimableCandidatePredicate = (availableBefore?: string) => + const claimableCandidatePredicate = ( + availableBefore?: string, + excludeRestartContinuations = false, + ) => sql` ${ availableBefore === undefined @@ -292,7 +299,18 @@ export const layer: Layer.Layer = La SELECT 1 FROM orchestration_v2_effect_outbox AS active WHERE active.thread_id = candidate.thread_id - AND active.status = 'running' + AND ( + active.status = 'running' + OR ( + active.status = 'pending' + AND active.rowid < candidate.rowid + AND ${ + excludeRestartContinuations + ? sql`active.effect_type != 'provider-runtime.continue'` + : sql`1 = 1` + } + ) + ) AND ( ( candidate.effect_type = 'thread-title.generate' @@ -492,7 +510,7 @@ export const layer: Layer.Layer = La WHERE effect_id = ( SELECT candidate.effect_id FROM orchestration_v2_effect_outbox AS candidate - WHERE ${claimableCandidatePredicate(nowIso)} + WHERE ${claimableCandidatePredicate(nowIso, excludeRestartContinuations)} AND ${excludeRestartContinuations ? sql`candidate.effect_type != 'provider-runtime.continue'` : sql`1 = 1`} ORDER BY candidate.available_at ASC, candidate.created_at ASC, candidate.effect_id ASC LIMIT 1 diff --git a/apps/server/src/orchestration-v2/EventSink.ts b/apps/server/src/orchestration-v2/EventSink.ts index c61ce6ba8b7bf2dfd2aef651b978863f6c84485a..71b6f0ea994fee7be92af994f3d7cc774af0636a 100644 --- a/apps/server/src/orchestration-v2/EventSink.ts +++ b/apps/server/src/orchestration-v2/EventSink.ts @@ -17,6 +17,7 @@ import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as PubSub from "effect/PubSub"; +import * as Semaphore from "effect/Semaphore"; import * as Schema from "effect/Schema"; import * as Stream from "effect/Stream"; import * as SqlClient from "effect/unstable/sql/SqlClient"; @@ -224,6 +225,40 @@ const baseLayer: Layer.Layer< ); } }); + const publishStoredEvents = (events: ReadonlyArray) => + eventStore.publishCommitted(events).pipe(Effect.andThen(publishLiveEvents(events))); + + // Transactions commit one at a time, but each writer publishes after its + // commit. If a writer is descheduled in between, a later commit reaches + // subscribers first, and clients drop any event at or below the newest + // sequence they have applied. So a writer takes this lane as the last step + // of its transaction and holds it until it has published. Publishing never + // waits, so a writer that holds the transaction while it waits for the + // lane is not blocked for long. + const publishLane = yield* Semaphore.make(1); + const commitThenPublish = ( + transaction: Effect.Effect, + publish: (committed: A) => Effect.Effect, + ) => + Effect.suspend(() => { + let holdsLane = false; + const takeLane = publishLane.take(1).pipe( + Effect.andThen( + Effect.sync(() => { + holdsLane = true; + }), + ), + Effect.uninterruptible, + ); + return sql + .withTransaction(Effect.tap(transaction, () => takeLane)) + .pipe( + Effect.tap(publish), + Effect.ensuring( + Effect.suspend(() => (holdsLane ? publishLane.release(1) : Effect.void)), + ), + ); + }); // A user can answer after terminal normalization reads the pending request. // Recheck inside the write transaction so stale cleanup cannot erase answers. @@ -332,7 +367,7 @@ const baseLayer: Layer.Layer< "orchestration_v2.thread_id": input.events[0]?.threadId ?? null, }); - const storedEvents = yield* sql.withTransaction( + return yield* commitThenPublish( Effect.gen(function* () { const normalized = yield* normalizeEvents( input.guardPendingUserInputCancellations === true @@ -347,13 +382,14 @@ const baseLayer: Layer.Layer< yield* effectOutbox.enqueue(input.effects); return committed; }), + (storedEvents) => + Effect.gen(function* () { + if (input.effects.length > 0) { + yield* effectOutbox.notifyAvailable(input.effects.length); + } + yield* publishStoredEvents(storedEvents); + }), ); - if (input.effects.length > 0) { - yield* effectOutbox.notifyAvailable(input.effects.length); - } - yield* eventStore.publishCommitted(storedEvents); - yield* publishLiveEvents(storedEvents); - return storedEvents; }); const writeIfRunCurrentEffect = Effect.fn("orchestrationV2.EventSink.writeIfRunCurrent")( @@ -365,7 +401,7 @@ const baseLayer: Layer.Layer< "orchestration_v2.thread_id": input.threadId, }); - const result = yield* sql.withTransaction( + return yield* commitThenPublish( Effect.gen(function* () { const rows = yield* sql<{ readonly status: string; @@ -403,12 +439,8 @@ const baseLayer: Layer.Layer< yield* applyStoredEvents(storedEvents); return { committed: true as const, storedEvents }; }), + (result) => (result.committed ? publishStoredEvents(result.storedEvents) : Effect.void), ); - if (result.committed) { - yield* eventStore.publishCommitted(result.storedEvents); - yield* publishLiveEvents(result.storedEvents); - } - return result; }, ); @@ -424,7 +456,7 @@ const baseLayer: Layer.Layer< "orchestration_v2.expected_last_run_ordinal": input.expectedLastRunOrdinal, }); - const result = yield* sql.withTransaction( + return yield* commitThenPublish( Effect.gen(function* () { const rows = yield* sql<{ readonly active_attempt_id: string | null; @@ -464,12 +496,8 @@ const baseLayer: Layer.Layer< yield* applyStoredEvents(storedEvents); return { committed: true as const, storedEvents }; }), + (result) => (result.committed ? publishStoredEvents(result.storedEvents) : Effect.void), ); - if (result.committed) { - yield* eventStore.publishCommitted(result.storedEvents); - yield* publishLiveEvents(result.storedEvents); - } - return result; }); const existingCommandResult = (commandId: CommandId) => @@ -490,7 +518,7 @@ const baseLayer: Layer.Layer< const commitCommandEffect = Effect.fn("orchestrationV2.EventSink.commitCommand")(function* ( input: Parameters[0], ) { - const result = yield* sql.withTransaction( + const result = yield* commitThenPublish( Effect.gen(function* () { const reserved = yield* commandReceipts.insertIfAbsent({ commandId: input.commandId, @@ -538,15 +566,15 @@ const baseLayer: Layer.Layer< }); return { receipt, storedEvents, committed: true as const, cancelledEffectIds }; }), + (result) => + Effect.gen(function* () { + yield* effectOutbox.signalCancellations(result.cancelledEffectIds); + if (result.committed && input.effects.length > 0) { + yield* effectOutbox.notifyAvailable(input.effects.length); + } + if (result.committed) yield* publishStoredEvents(result.storedEvents); + }), ); - yield* effectOutbox.signalCancellations(result.cancelledEffectIds); - if (result.committed && input.effects.length > 0) { - yield* effectOutbox.notifyAvailable(input.effects.length); - } - if (result.committed) { - yield* eventStore.publishCommitted(result.storedEvents); - yield* publishLiveEvents(result.storedEvents); - } return { receipt: result.receipt, storedEvents: result.storedEvents, @@ -593,7 +621,7 @@ const baseLayer: Layer.Layer< const commitProjectCommandEffect = Effect.fn("orchestrationV2.EventSink.commitProjectCommand")( function* (input: Parameters[0]) { - const result = yield* sql.withTransaction( + const result = yield* commitThenPublish( Effect.gen(function* () { const reserved: CommandReceiptStore.ProjectCommandReceiptV2 = { commandId: input.commandId, @@ -613,10 +641,9 @@ const baseLayer: Layer.Layer< yield* commandReceipts.upsert(receipt); return { receipt, event }; }), + (result) => + result.event === undefined ? Effect.void : eventStore.publishCommitted([result.event]), ); - if (result.event !== undefined) { - yield* eventStore.publishCommitted([result.event]); - } return { receipt: result.receipt, committed: result.event !== undefined }; }, ); diff --git a/apps/server/src/orchestration-v2/FoundationPersistence.test.ts b/apps/server/src/orchestration-v2/FoundationPersistence.test.ts index 22b93178711c1ea3afecaf9a02685580a7af331b..dc7eacc59ba50ceca886d0a8702bb4f05795a931 100644 --- a/apps/server/src/orchestration-v2/FoundationPersistence.test.ts +++ b/apps/server/src/orchestration-v2/FoundationPersistence.test.ts @@ -51,7 +51,9 @@ import * as EventStore from "./EventStore.ts"; import * as IdAllocator from "./IdAllocator.ts"; import * as ProjectionMaintenance from "./ProjectionMaintenance.ts"; import * as ProjectionStore from "./ProjectionStore.ts"; +import * as ProjectStore from "./ProjectStore.ts"; import * as ProviderRuntimeRecovery from "./ProviderRuntimeRecoveryService.ts"; +import * as TurnItemPositionStore from "./TurnItemPositionStore.ts"; const isLiveStreamBufferError = Schema.is(LiveStreamBufferError); const encodeJson = Schema.encodeEffect(Schema.fromJsonString(Schema.Unknown)); @@ -2303,6 +2305,76 @@ it.layer(TestLayer)("orchestration V2 foundation persistence", (it) => { }).pipe(Effect.provide(Layer.fresh(effectOutboxProvided))), ); + it.effect("keeps later thread effects behind an earlier effect waiting to retry", () => + Effect.gen(function* () { + const outbox = yield* EffectOutbox.EffectOutboxV2; + const workerId = "retry-order-worker"; + const commandId = CommandId.make("command:foundation-retry-order"); + const threadId = ThreadId.make("thread:foundation-retry-order"); + yield* outbox.enqueue([ + { + id: "effect:foundation-retry-order:z-rollback", + commandId, + threadId, + request: { + type: "provider-thread.rollback", + providerThreadId: ProviderThreadId.make("provider-thread:foundation-retry-order"), + checkpointId: CheckpointId.make("checkpoint:foundation-retry-order"), + scopeId: CheckpointScopeId.make("scope:foundation-retry-order"), + }, + }, + ]); + const rollback = yield* outbox.claimNext({ workerId, leaseDurationMs: 30_000 }); + assert.isTrue(Option.isSome(rollback)); + if (Option.isNone(rollback)) return; + yield* outbox.retry({ + effectId: rollback.value.id, + workerId, + error: "rollback failed once", + delayMs: 60_000, + }); + + // A turn the user starts during the rollback's backoff must not run first, + // even when its timestamp ties and its id sorts first. + yield* outbox.enqueue([ + { + id: "effect:foundation-retry-order:a-start", + commandId: CommandId.make("command:foundation-retry-order:start"), + threadId, + request: { type: "provider-turn.start", runId: RunId.make("run:foundation-retry-order") }, + }, + { + id: "effect:foundation-retry-order:b-title", + commandId: CommandId.make("command:foundation-retry-order:title"), + threadId, + request: { type: "thread-title.generate", kind: { type: "regenerate" } }, + }, + ]); + const title = yield* outbox.claimNext({ workerId, leaseDurationMs: 30_000 }); + assert.equal(Option.getOrUndefined(title)?.id, "effect:foundation-retry-order:b-title"); + const blocked = yield* outbox.claimNext({ workerId, leaseDurationMs: 30_000 }); + assert.isTrue(Option.isNone(blocked)); + const nextClaimable = yield* outbox.nextClaimableAt; + assert.isTrue(Option.isSome(nextClaimable)); + if (Option.isSome(nextClaimable)) { + assert.equal( + DateTime.formatIso(nextClaimable.value), + (yield* outbox.get(rollback.value.id)).pipe(Option.getOrThrow).availableAt, + ); + } + + yield* outbox.cancelUnsettled({ + threadId, + effectTypes: ["provider-thread.rollback"], + reason: "Test cleanup.", + }); + const unblocked = yield* outbox.claimNext({ workerId, leaseDurationMs: 30_000 }); + assert.equal(Option.getOrUndefined(unblocked)?.id, "effect:foundation-retry-order:a-start"); + yield* outbox.succeed({ effectId: "effect:foundation-retry-order:a-start", workerId }); + yield* outbox.succeed({ effectId: "effect:foundation-retry-order:b-title", workerId }); + }).pipe(Effect.provide(Layer.fresh(effectOutboxProvided))), + ); + it.effect("executes a retry at its durable deadline instead of the liveness interval", () => Effect.gen(function* () { const outbox = yield* EffectOutbox.EffectOutboxV2; @@ -3222,3 +3294,90 @@ it.live("keeps claiming new work after repeated idle periods", () => }).pipe(Effect.provide(workerLayer), Effect.scoped); }).pipe(Effect.provide(TestLayer)), ); + +it.effect("publishes live events in commit order across concurrent writers", () => + Effect.gen(function* () { + const firstCommitted = yield* Deferred.make(); + const releaseFirst = yield* Deferred.make(); + // The first writer's post-commit wakeup stands in for any scheduler yield + // between its commit and its publish. + const pausingOutbox = Layer.effect( + EffectOutbox.EffectOutboxV2, + Effect.gen(function* () { + const delegate = yield* EffectOutbox.EffectOutboxV2; + return EffectOutbox.EffectOutboxV2.of({ + ...delegate, + notifyAvailable: (count) => + Deferred.succeed(firstCommitted, undefined).pipe( + Effect.andThen(Deferred.await(releaseFirst)), + Effect.andThen(delegate.notifyAvailable(count)), + ), + }); + }), + ).pipe(Layer.provide(effectOutboxProvided)); + const eventSinkLayer = EventSink.layerFromStores.pipe( + Layer.provide( + Layer.mergeAll( + storesProvided, + pausingOutbox, + commandReceiptStoreProvided, + ProjectStore.layer.pipe(Layer.provide(databaseLayer)), + TurnItemPositionStore.layer.pipe(Layer.provide(databaseLayer)), + ), + ), + ); + + yield* Effect.gen(function* () { + const eventSink = yield* EventSink.EventSinkV2; + const now = yield* DateTime.now; + const first = makeThread(ThreadId.make("thread:foundation-publish-order:first"), now); + const second = makeThread(ThreadId.make("thread:foundation-publish-order:second"), now); + const published = yield* eventSink + .stream({ afterSequence: yield* eventSink.latestSequence() }) + .pipe(Stream.take(2), Stream.runCollect, Effect.forkScoped({ startImmediately: true })); + + const firstWrite = yield* eventSink + .writeWithEffects({ + events: [ + threadCreatedEvent({ id: "event:foundation-publish-order:first", thread: first, now }), + ], + effects: [ + { + id: "effect:foundation-publish-order:first", + commandId: CommandId.make("command:foundation-publish-order:first"), + threadId: first.id, + request: { type: "terminal.cleanup" }, + }, + ], + }) + .pipe(Effect.forkScoped); + yield* Deferred.await(firstCommitted); + // The second writer commits after the first. It may run as far as it can + // before the first writer resumes. + const secondWrite = yield* eventSink + .write({ + events: [ + threadCreatedEvent({ + id: "event:foundation-publish-order:second", + thread: second, + now, + }), + ], + }) + .pipe( + Effect.provideService(Scheduler.MaxOpsBeforeYield, Number.POSITIVE_INFINITY), + Effect.forkScoped, + ); + yield* Effect.yieldNow; + yield* Deferred.succeed(releaseFirst, undefined); + yield* Fiber.join(firstWrite); + yield* Fiber.join(secondWrite); + + const sequences = Array.from(yield* Fiber.join(published), (stored) => stored.sequence); + assert.deepEqual( + sequences, + [...sequences].sort((left, right) => left - right), + ); + }).pipe(Effect.provide(eventSinkLayer)); + }).pipe(Effect.provide(databaseLayer)), +); diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts index a567a4ab4841085f1a7148e2d0bbff020399746d..696a40d8d17614245b125f9bb3e36bf32a801c72 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts @@ -20,6 +20,7 @@ import * as Cause from "effect/Cause"; import * as DateTime from "effect/DateTime"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; import * as Fiber from "effect/Fiber"; import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; @@ -80,6 +81,52 @@ const FailingReleaseEventSinkLayer = Layer.effect( }), ).pipe(Layer.provide(TestEventSinkLayer)); +interface FlakyReleaseWrites { + /** Which release writes fail right now. */ + readonly failing: Ref.Ref<"none" | "session" | "session-and-requests">; + /** Receives one item per failed write. */ + readonly failures: Queue.Queue; + /** Holds runtime request writes: completes `paused`, then waits for `resume`. */ + readonly pauseRequestWrites?: { + readonly paused: Deferred.Deferred; + readonly resume: Deferred.Deferred; + }; +} + +// Fails release writes with a defect, the way a failed SQL commit surfaces. +const makeFlakyReleaseEventSinkLayer = (flaky: FlakyReleaseWrites) => + Layer.effect( + EventSink.EventSinkV2, + Effect.gen(function* () { + const delegate = yield* EventSink.EventSinkV2; + return EventSink.EventSinkV2.of({ + ...delegate, + write: (input) => + Effect.gen(function* () { + const failing = yield* Ref.get(flaky.failing); + const fails = input.events.some( + (event) => + (failing !== "none" && + event.type === "provider-session.updated" && + (event.payload.status === "stopped" || event.payload.status === "error")) || + (failing === "session-and-requests" && event.type === "runtime-request.updated"), + ); + const pause = flaky.pauseRequestWrites; + if ( + pause !== undefined && + input.events.some((event) => event.type === "runtime-request.updated") + ) { + yield* Deferred.succeed(pause.paused, undefined); + yield* Deferred.await(pause.resume); + } + if (!fails) return yield* delegate.write(input); + yield* Queue.offer(flaky.failures, undefined); + return yield* Effect.die(new Error("simulated commit failure")); + }), + }); + }), + ).pipe(Layer.provide(TestEventSinkLayer)); + const CodexCapabilities: OrchestrationV2ProviderCapabilities = CodexProviderCapabilitiesV2; const ExclusiveCapabilities: OrchestrationV2ProviderCapabilities = { ...CodexCapabilities, @@ -359,15 +406,19 @@ function makeTestLayer(input: { readonly initialProviderItemIdentityVersion?: 2; }) => Effect.Effect; readonly failReleaseEventWrites?: boolean; + readonly flakyReleaseWrites?: FlakyReleaseWrites; readonly hasPendingBackgroundWork?: Effect.Effect; readonly hangSessionScopeClose?: boolean; readonly beforeUnload?: Effect.Effect; readonly serverSettingsLayer?: ReturnType; readonly projectServiceLayer?: Layer.Layer; }) { - const configuredEventSinkLayer = input.failReleaseEventWrites - ? FailingReleaseEventSinkLayer - : TestEventSinkLayer; + const configuredEventSinkLayer = + input.flakyReleaseWrites !== undefined + ? makeFlakyReleaseEventSinkLayer(input.flakyReleaseWrites) + : input.failReleaseEventWrites + ? FailingReleaseEventSinkLayer + : TestEventSinkLayer; const registryLayer = ProviderAdapterRegistry.makeSingleLayer( makeProviderAdapter(input.state, { failEventStream: input.failEventStream ?? false, @@ -2299,6 +2350,299 @@ it.effect("ProviderSessionManagerV2 marks pending runtime requests non-live on r yield* effect.pipe(Effect.provide(makeTestLayer({ state, idleTimeoutMs: 1000 }))); }), ); +it.effect("ProviderSessionManagerV2 retries release records that failed to persist", () => + Effect.gen(function* () { + const state = yield* Ref.make(emptyState); + const flaky: FlakyReleaseWrites = { + failing: yield* Ref.make<"none" | "session" | "session-and-requests">("session"), + failures: yield* Queue.unbounded(), + }; + const effect = Effect.gen(function* () { + const eventSink = yield* EventSink.EventSinkV2; + const idAllocator = yield* IdAllocator.IdAllocatorV2; + const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; + const projectionStore = yield* ProjectionStore.ProjectionStoreV2; + const now = yield* DateTime.now; + const threadId = ThreadId.make("thread-provider-session-manager-release-retry"); + const providerSessionId = yield* idAllocator.allocate.providerSession({ + providerInstanceId: modelSelection.instanceId, + threadId, + }); + const providerThread = makeProviderThread({ idAllocator, threadId, providerSessionId, now }); + yield* eventSink.write({ + events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })], + }); + const pendingRequest = yield* makePendingRuntimeRequestEvents({ + idAllocator, + threadId, + providerSessionId, + providerThread, + now, + }); + yield* eventSink.write({ events: pendingRequest.events }); + yield* manager.open({ threadId, providerSessionId, modelSelection, runtimePolicy }); + + assert.isTrue(Exit.isFailure(yield* Effect.exit(manager.close(providerSessionId)))); + yield* Queue.take(flaky.failures); + // The failed session write does not keep the approval answerable. + const afterClose = yield* projectionStore.getThreadProjection(threadId); + assert.equal(afterClose.runtimeRequests.at(-1)?.responseCapability.type, "not_resumable"); + assert.equal(afterClose.providerSessions.at(-1)?.status, "ready"); + + // The first retry fails as well, and the retries continue. + yield* TestClock.adjust("1 second"); + yield* Queue.take(flaky.failures); + yield* Ref.set(flaky.failing, "none"); + const stopped = yield* eventSink + .stream({ + threadId, + afterSequence: yield* eventSink.latestSequence({ threadId }), + eventType: "provider-session.updated", + }) + .pipe(Stream.runHead, Effect.forkScoped); + yield* TestClock.adjust("1 second"); + yield* Fiber.join(stopped); + + const afterRetry = yield* projectionStore.getThreadProjection(threadId); + assert.equal(afterRetry.providerSessions.at(-1)?.status, "stopped"); + }); + + yield* effect.pipe( + Effect.provide(makeTestLayer({ state, idleTimeoutMs: 60_000, flakyReleaseWrites: flaky })), + ); + }), +); + +it.effect("ProviderSessionManagerV2 release retries leave a replacement session alone", () => + Effect.gen(function* () { + const state = yield* Ref.make(emptyState); + const flaky: FlakyReleaseWrites = { + failing: yield* Ref.make<"none" | "session" | "session-and-requests">("none"), + failures: yield* Queue.unbounded(), + }; + const effect = Effect.gen(function* () { + const eventSink = yield* EventSink.EventSinkV2; + const idAllocator = yield* IdAllocator.IdAllocatorV2; + const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; + const projectionStore = yield* ProjectionStore.ProjectionStoreV2; + const threadId = ThreadId.make("thread-provider-session-manager-release-replacement"); + const providerSessionId = yield* idAllocator.allocate.providerSession({ + providerInstanceId: modelSelection.instanceId, + threadId, + }); + const writePendingRequest = Effect.gen(function* () { + const now = yield* DateTime.now; + const request = yield* makePendingRuntimeRequestEvents({ + idAllocator, + threadId, + providerSessionId, + providerThread: makeProviderThread({ idAllocator, threadId, providerSessionId, now }), + now, + }); + yield* eventSink.write({ events: request.events }); + return request.requestId; + }); + yield* eventSink.write({ + events: [ + yield* makeThreadCreatedEvent({ idAllocator, threadId, now: yield* DateTime.now }), + ], + }); + const oldRequestId = yield* writePendingRequest; + yield* manager.open({ threadId, providerSessionId, modelSelection, runtimePolicy }); + yield* Ref.set(flaky.failing, "session-and-requests"); + assert.isTrue(Exit.isFailure(yield* Effect.exit(manager.close(providerSessionId)))); + yield* Queue.take(flaky.failures); + yield* Queue.take(flaky.failures); + yield* Ref.set(flaky.failing, "none"); + + // A replacement opens with the same id before the retry runs. + yield* TestClock.adjust("500 millis"); + yield* manager.open({ threadId, providerSessionId, modelSelection, runtimePolicy }); + const newRequestId = yield* writePendingRequest; + const replacementStatus = (yield* projectionStore.getThreadProjection( + threadId, + )).providerSessions.at(-1)?.status; + const settled = yield* eventSink + .stream({ + threadId, + afterSequence: yield* eventSink.latestSequence({ threadId }), + eventType: "runtime-request.updated", + }) + .pipe(Stream.runHead, Effect.forkScoped); + yield* TestClock.adjust("500 millis"); + yield* Fiber.join(settled); + + const projection = yield* projectionStore.getThreadProjection(threadId); + const request = (id: typeof oldRequestId) => + projection.runtimeRequests.find((candidate) => candidate.id === id); + assert.equal(request(oldRequestId)?.responseCapability.type, "not_resumable"); + assert.equal(request(newRequestId)?.responseCapability.type, "live"); + assert.equal(projection.providerSessions.at(-1)?.status, replacementStatus); + }); + + yield* effect.pipe( + Effect.provide(makeTestLayer({ state, idleTimeoutMs: 60_000, flakyReleaseWrites: flaky })), + ); + }), +); + +it.effect("ProviderSessionManagerV2 keeps each failed release's cleanup", () => + Effect.gen(function* () { + const state = yield* Ref.make(emptyState); + const flaky: FlakyReleaseWrites = { + failing: yield* Ref.make<"none" | "session" | "session-and-requests">("none"), + failures: yield* Queue.unbounded(), + }; + const effect = Effect.gen(function* () { + const eventSink = yield* EventSink.EventSinkV2; + const idAllocator = yield* IdAllocator.IdAllocatorV2; + const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; + const projectionStore = yield* ProjectionStore.ProjectionStoreV2; + const now = yield* DateTime.now; + const firstThreadId = ThreadId.make("thread-provider-session-manager-release-each-a"); + const secondThreadId = ThreadId.make("thread-provider-session-manager-release-each-b"); + const providerSessionId = yield* idAllocator.allocate.providerSession({ + providerInstanceId: modelSelection.instanceId, + threadId: firstThreadId, + }); + yield* eventSink.write({ + events: [ + yield* makeThreadCreatedEvent({ idAllocator, threadId: firstThreadId, now }), + yield* makeThreadCreatedEvent({ idAllocator, threadId: secondThreadId, now }), + ], + }); + const secondThreadRequest = yield* makePendingRuntimeRequestEvents({ + idAllocator, + threadId: secondThreadId, + providerSessionId, + providerThread: makeProviderThread({ + idAllocator, + threadId: secondThreadId, + providerSessionId, + now, + }), + now, + }); + yield* eventSink.write({ events: secondThreadRequest.events }); + const failRelease = (failedWrites: number) => + Effect.gen(function* () { + yield* Ref.set(flaky.failing, "session-and-requests"); + assert.isTrue(Exit.isFailure(yield* Effect.exit(manager.close(providerSessionId)))); + yield* Effect.repeat(Queue.take(flaky.failures), { times: failedWrites - 1 }); + yield* Ref.set(flaky.failing, "none"); + }); + + // The first session serves both threads. Its replacement serves one. + yield* manager.open({ + threadId: firstThreadId, + providerSessionId, + modelSelection, + runtimePolicy, + }); + yield* manager.open({ + threadId: secondThreadId, + providerSessionId, + modelSelection, + runtimePolicy, + }); + // Both the session write and the second thread's request write fail. + yield* failRelease(2); + yield* TestClock.adjust("500 millis"); + yield* manager.open({ + threadId: firstThreadId, + providerSessionId, + modelSelection, + runtimePolicy, + }); + // Only the session write fails: the first thread has no requests. + yield* failRelease(1); + + const settled = yield* eventSink + .stream({ + threadId: secondThreadId, + afterSequence: yield* eventSink.latestSequence({ threadId: secondThreadId }), + eventType: "runtime-request.updated", + }) + .pipe(Stream.runHead, Effect.forkScoped); + yield* TestClock.adjust("500 millis"); + yield* Fiber.join(settled); + + const projection = yield* projectionStore.getThreadProjection(secondThreadId); + assert.equal(projection.runtimeRequests.at(-1)?.responseCapability.type, "not_resumable"); + }); + + yield* effect.pipe( + Effect.provide(makeTestLayer({ state, idleTimeoutMs: 60_000, flakyReleaseWrites: flaky })), + ); + }), +); + +it.effect("ProviderSessionManagerV2 settles a request the event pump persists during release", () => + Effect.gen(function* () { + const state = yield* Ref.make(emptyState); + const flaky: FlakyReleaseWrites = { + failing: yield* Ref.make<"none" | "session" | "session-and-requests">("none"), + failures: yield* Queue.unbounded(), + pauseRequestWrites: { + paused: yield* Deferred.make(), + resume: yield* Deferred.make(), + }, + }; + const pause = flaky.pauseRequestWrites!; + const effect = Effect.gen(function* () { + const eventSink = yield* EventSink.EventSinkV2; + const idAllocator = yield* IdAllocator.IdAllocatorV2; + const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; + const projectionStore = yield* ProjectionStore.ProjectionStoreV2; + const now = yield* DateTime.now; + const threadId = ThreadId.make("thread-provider-session-manager-release-pump"); + const providerSessionId = yield* idAllocator.allocate.providerSession({ + providerInstanceId: modelSelection.instanceId, + threadId, + }); + yield* eventSink.write({ + events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })], + }); + yield* manager.open({ threadId, providerSessionId, modelSelection, runtimePolicy }); + // The runtime creates the request a moment after the release starts. + const createdAt = DateTime.add(now, { seconds: 1 }); + const pendingRequest = yield* makePendingRuntimeRequestEvents({ + idAllocator, + threadId, + providerSessionId, + providerThread: makeProviderThread({ + idAllocator, + threadId, + providerSessionId, + now: createdAt, + }), + now: createdAt, + }); + const adapterEvents = (yield* Ref.get(state)).eventQueues.get(String(providerSessionId)); + assert.isDefined(adapterEvents); + yield* Queue.offerAll(adapterEvents!, pendingRequest.providerEvents); + // The event pump holds the request permit while it persists the request. + yield* Deferred.await(pause.paused); + const closed = yield* manager + .close(providerSessionId) + .pipe(Effect.forkScoped({ startImmediately: true })); + yield* TestClock.adjust("1 second"); + yield* Deferred.succeed(pause.resume, undefined); + yield* Fiber.join(closed); + + const projection = yield* projectionStore.getThreadProjection(threadId); + const request = projection.runtimeRequests.find( + (candidate) => candidate.id === pendingRequest.requestId, + ); + assert.equal(request?.responseCapability.type, "not_resumable"); + }); + + yield* effect.pipe( + Effect.provide(makeTestLayer({ state, idleTimeoutMs: 60_000, flakyReleaseWrites: flaky })), + ); + }), +); + it.effect("ProviderSessionManagerV2 terminalizes a pending input transcript item on release", () => Effect.gen(function* () { const state = yield* Ref.make(emptyState); diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.ts index 786eb5a6bf0ae71772bc05bea685fe98be3123a5..c1039c2a0207d9d81e22caf00a5a3d287e1a5524 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.ts @@ -17,11 +17,13 @@ import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; import * as Fiber from "effect/Fiber"; +import * as FiberSet from "effect/FiberSet"; import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; +import * as Schedule from "effect/Schedule"; import * as Schema from "effect/Schema"; import * as Scope from "effect/Scope"; import * as Semaphore from "effect/Semaphore"; @@ -361,6 +363,9 @@ export const layerWithOptions = ( ); const layerScope = yield* Effect.scope; const sessions = yield* Ref.make(new Map()); + // One retry per released entry, so a later release with the same id + // cannot drop cleanup for threads only the earlier session served. + const releaseRecordRetries = yield* FiberSet.make(); const nextSubscriberId = yield* Ref.make(0); const sessionOpen = yield* makeKeyedSerialExecutor(); // Orders a thread's attach against a detach unloading it on the same session. @@ -608,6 +613,8 @@ export const layerWithOptions = ( const writeReleasedRuntimeRequestEvents = (input: { readonly entry: LiveSessionEntry; readonly reason: ProviderSessionReleaseReason; + /** Requests created later belong to a replacement session with the same id. */ + readonly releasedAt: DateTime.Utc; }) => Effect.gen(function* () { const providerSessionId = input.entry.runtime.providerSessionId; @@ -629,7 +636,8 @@ export const layerWithOptions = ( (request) => request.status === "pending" && request.responseCapability.type === "live" && - request.responseCapability.providerSessionId === providerSessionId, + request.responseCapability.providerSessionId === providerSessionId && + DateTime.isLessThanOrEqualTo(request.createdAt, input.releasedAt), ); for (const request of releasedRequests) { @@ -708,6 +716,125 @@ export const layerWithOptions = ( } }); + // Records a released session as stopped and resolves the live runtime + // requests it left. Each write runs even if the other fails. Once a + // replacement session opens with the same id, it owns the session status, + // so only the requests are settled. + const writeReleaseRecords = (input: { + readonly entry: LiveSessionEntry; + readonly reason: ProviderSessionReleaseReason; + readonly detail?: string; + readonly releasedAt: DateTime.Utc; + readonly replaced: boolean; + }) => + Effect.all( + [ + input.replaced + ? Effect.succeed(Exit.void) + : Effect.exit(writeReleasedSessionEvents(input)), + Effect.exit( + writeReleasedRuntimeRequestEvents(input).pipe( + input.entry.requestEventPermit.withPermits(1), + ), + ), + ], + { concurrency: 1 }, + ).pipe(Effect.flatMap(Exit.asVoidAll)); + + // The session already left the live map, so a later release finds + // nothing to do. Without a retry the UI would keep a ready session and + // answerable approvals until a server restart. Each attempt holds the + // session's open lock, so it sees a replacement that opened meanwhile. + const retryReleaseRecords = ( + input: Omit[0], "replaced">, + ) => { + const providerSessionId = input.entry.runtime.providerSessionId; + const attempt = Effect.gen(function* () { + const exit = yield* Effect.exit( + sessionOpen.withLock( + providerSessionId, + Effect.gen(function* () { + const replaced = (yield* Ref.get(sessions)).has(sessionKey(providerSessionId)); + yield* writeReleaseRecords({ ...input, replaced }); + }), + ), + ); + if (Exit.isSuccess(exit) || Cause.hasInterruptsOnly(exit.cause)) return yield* exit; + yield* Effect.logWarning("orchestration-v2.provider-session-release-records-failed", { + providerSessionId, + cause: exit.cause, + }); + // A failed SQL commit is a defect, so every failure but interruption + // is retried. + return yield* Effect.fail(exit.cause); + }); + return attempt.pipe( + Effect.retry({ + schedule: Schedule.exponential("1 second").pipe( + Schedule.modifyDelay(({ duration }) => + Effect.succeed(Duration.min(duration, Duration.seconds(30))), + ), + ), + }), + Effect.delay("1 second"), + FiberSet.run(releaseRecordRetries), + ); + }; + + const logReleaseFailure = + (providerSessionId: ProviderSessionId) => + (release: Effect.Effect) => + release.pipe( + Effect.catchCause((cause) => + Effect.logWarning("orchestration-v2.provider-session-release-failed", { + providerSessionId, + cause, + }), + ), + ); + + // Removes the live entry and reads the request cleanup cutoff while + // holding the entry's request permit. A request the event pump is + // persisting for this runtime lands before the cutoff, and once the + // entry is gone the pump persists no more for it. A replacement's + // requests come after its own open. + const removeLiveEntry = (input: { + readonly providerSessionId: ProviderSessionId; + readonly onlyIfIdleGeneration?: number; + }): Effect.Effect, DateTime.Utc]> => + Effect.gen(function* () { + const key = sessionKey(input.providerSessionId); + const candidate = (yield* Ref.get(sessions)).get(key); + if (candidate === undefined) { + return [Option.none(), yield* DateTime.now] as const; + } + const removed = yield* Effect.zip( + Ref.modify(sessions, (current) => { + const existing = current.get(key); + if (existing !== candidate) { + return [existing === undefined ? "gone" : "changed", current] as const; + } + if ( + input.onlyIfIdleGeneration !== undefined && + (existing.busyCount > 0 || existing.idleGeneration !== input.onlyIfIdleGeneration) + ) { + return ["kept", current] as const; + } + const updated = new Map(current); + updated.delete(key); + return ["removed", updated] as const; + }), + DateTime.now, + ).pipe(candidate.requestEventPermit.withPermits(1)); + const [outcome, releasedAt] = removed; + // Another entry took this id while the permit was held; release it instead. + if (outcome === "changed") return yield* removeLiveEntry(input); + return [ + outcome === "removed" ? Option.some(candidate) : Option.none(), + releasedAt, + ] as const; + }); + const releaseEntry = (input: { readonly providerSessionId: ProviderSessionId; readonly reason: ProviderSessionReleaseReason; @@ -717,23 +844,8 @@ export const layerWithOptions = ( readonly gracefulSubscribers?: boolean; }) => Effect.acquireUseRelease( - Ref.modify(sessions, (current) => { - const key = sessionKey(input.providerSessionId); - const existing = current.get(key); - if (existing === undefined) { - return [Option.none(), current] as const; - } - if ( - input.onlyIfIdleGeneration !== undefined && - (existing.busyCount > 0 || existing.idleGeneration !== input.onlyIfIdleGeneration) - ) { - return [Option.none(), current] as const; - } - const updated = new Map(current); - updated.delete(key); - return [Option.some(existing), updated] as const; - }), - (entry) => + removeLiveEntry(input), + ([entry, releasedAt]) => Option.match(entry, { onNone: () => Effect.void, onSome: (entry) => @@ -794,21 +906,25 @@ export const layerWithOptions = ( Effect.forkDetach, ); } - yield* writeReleasedSessionEvents({ + const records = { entry, reason: input.reason, ...(input.detail === undefined ? {} : { detail: input.detail }), - }); - yield* writeReleasedRuntimeRequestEvents({ - entry, - reason: input.reason, - }).pipe(entry.requestEventPermit.withPermits(1)); + releasedAt, + }; + const recorded = yield* Effect.exit( + writeReleaseRecords({ ...records, replaced: false }), + ); + if (Exit.isFailure(recorded)) { + yield* retryReleaseRecords(records); + return yield* recorded; + } if (Option.isSome(closeExit) && Exit.isFailure(closeExit.value)) { return yield* Effect.failCause(closeExit.value.cause); } }), }), - (entry) => + ([entry]) => Option.match(entry, { onNone: () => Effect.void, onSome: (entry) => @@ -1495,7 +1611,7 @@ export const layerWithOptions = ( providerSessionId: entry.runtime.providerSessionId, reason: "manual_shutdown", gracefulSubscribers: true, - }).pipe(Effect.ignore); + }).pipe(logReleaseFailure(entry.runtime.providerSessionId)); return; } const cause = Exit.isFailure(exit) @@ -1516,7 +1632,7 @@ export const layerWithOptions = ( providerSessionId: entry.runtime.providerSessionId, reason: "runtime_error", detail: Cause.pretty(cause), - }).pipe(Effect.ignore); + }).pipe(logReleaseFailure(entry.runtime.providerSessionId)); }), ), Effect.forkIn(layerScope), @@ -1705,7 +1821,7 @@ export const layerWithOptions = ( providerSessionId: input.providerSessionId, reason: "runtime_error", detail: "Failed to persist the provider-session attachment.", - }).pipe(Effect.ignore), + }).pipe(logReleaseFailure(input.providerSessionId)), ), ); yield* startEventPump(entry); diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts index 746fbe235d99dcbd4ba5892e256b5c2d3212daca..fdb45f1931794250e96e5939bb4eeeab0eb03b55 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts @@ -30,6 +30,7 @@ import * as ProviderAuthService from "../provider/Services/ProviderAuthService.t import * as ContextHandoffService from "./ContextHandoffService.ts"; import * as EventSink from "./EventSink.ts"; import * as IdAllocator from "./IdAllocator.ts"; +import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts"; import * as ProjectionStore from "./ProjectionStore.ts"; import { ProviderAdapterEventStreamError } from "./ProviderAdapter.ts"; import * as ProviderSessionManager from "./ProviderSessionManager.ts"; @@ -168,6 +169,8 @@ function makeLocalCommandHarness(input: { readonly interruptOpen?: boolean; readonly interruptRunBeforeOpenFailure?: boolean; readonly writeFailure?: unknown; + /** Loads the thread and starts the run, then fails every later state read. */ + readonly failReadsAfterRunning?: boolean; }) { const now = DateTime.makeUnsafe("2026-09-04T12:00:00Z"); const threadId = ThreadId.make("thread-native-account-command"); @@ -411,9 +414,40 @@ function makeLocalCommandHarness(input: { ), ), ) - : Effect.die("A local command must not open a native session."), + : input.failReadsAfterRunning === true + ? Effect.succeed({ + driver: providerThread.driver, + providerSession: { + id: providerSessionId, + driver: providerThread.driver, + providerInstanceId: newInstanceId, + status: "ready", + cwd: "/tmp/native-account-command", + model: null, + capabilities: CodexProviderCapabilitiesV2, + createdAt: now, + updatedAt: now, + lastError: null, + }, + ensureThread: () => Effect.succeed(providerThread), + } as never) + : Effect.die("A local command must not open a native session."), + ); + const startRootRun = vi.fn< + (input: RunExecutionService.RunExecutionServiceV2StartRootRunInput) => Effect.Effect + >(() => + input.failReadsAfterRunning === true + ? Effect.void + : Effect.die("A local command must not start a native turn."), + ); + const failReadIfRunning = Effect.suspend(() => + input.failReadsAfterRunning === true && + projection.runs.find((candidate) => candidate.id === runId)?.status === "running" + ? Effect.fail( + new ProjectionStore.ProjectionStoreReadError({ threadId, cause: "database unavailable" }), + ) + : Effect.void, ); - const startRootRun = vi.fn(() => Effect.die("A local command must not start a native turn.")); const tryHandlePromptCommand = vi.fn(() => input.logoutFailure === undefined ? Effect.succeed(true) @@ -471,7 +505,7 @@ function makeLocalCommandHarness(input: { ), }), getRuntimeRecoveryProjection: () => - Effect.succeed({ + Effect.as(failReadIfRunning, { ...projection, hasConversation: projection.messages.some( (m) => @@ -703,6 +737,25 @@ effectIt.effect( }), ); +effectIt.effect("does not mistake a failed state read for a superseded run", () => + Effect.gen(function* () { + const harness = makeLocalCommandHarness({ text: "Continue", failReadsAfterRunning: true }); + + yield* harness.start; + + expect(harness.projection().runs.at(-1)?.status).toBe("running"); + const controls = harness.startRootRun.mock.calls[0]?.[0]; + expect(controls).toBeDefined(); + if (controls === undefined) return; + // "false" would skip the provider turn or the terminal write and leave the + // run active. A read failure must reach the caller instead. + const startCheck = yield* Effect.flip(controls.shouldStartProviderTurn!()); + const finalizeCheck = yield* Effect.flip(controls.shouldFinalizeRun!()); + expect(startCheck._tag).toBe("ProjectionStoreReadError"); + expect(finalizeCheck._tag).toBe("ProjectionStoreReadError"); + }), +); + effectIt.effect("does not overwrite a run interrupted while its thread loads", () => Effect.gen(function* () { const harness = makeLocalCommandHarness({ diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts index 781491ac7fd65ae1ea4752d6c5e5807cf9ce2b7b..ce2088d911d4c29f9d80af2f997c8bea01f01b0d 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts @@ -124,13 +124,14 @@ export const layer: Layer.Layer< }) => { // Guards and background routing need live execution state, not a fresh // allocation of every completed message and tool output in the thread. + // `false` means the run moved on or is gone. A failed read is an error, + // so the caller fails the start or the run instead of skipping it. const isCurrentAttemptInStatus = (expectedStatus: OrchestrationV2Run["status"]) => projectionStore.getRuntimeRecoveryProjection(input.threadId).pipe( Effect.map((current) => { const run = current.runs.find((candidate) => candidate.id === input.runId); return run?.activeAttemptId === input.attemptId && run.status === expectedStatus; }), - Effect.catchCause(() => Effect.succeed(false)), ); return { isCurrentAttemptInStatus, @@ -157,7 +158,6 @@ export const layer: Layer.Layer< (run.status === "starting" || run.status === "running") ); }), - Effect.catchCause(() => Effect.succeed(false)), ), hasUnpairedRunInterruptRequest: () => projectionStore diff --git a/apps/server/src/orchestration-v2/RunExecutionService.test.ts b/apps/server/src/orchestration-v2/RunExecutionService.test.ts index 370c3005834675ac6e02ad8eb6d938b26a5758a5..3d6dca6151ab460f47cebcf36a739431dfe9cb89 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.test.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.test.ts @@ -50,6 +50,7 @@ import { type ProviderAdapterV2Event, type ProviderAdapterV2SessionRuntime, } from "./ProviderAdapter.ts"; +import * as ProjectionStore from "./ProjectionStore.ts"; import * as ProviderEventIngestor from "./ProviderEventIngestor.ts"; import * as RunExecutionService from "./RunExecutionService.ts"; import * as RunFinalizationService from "./RunFinalizationService.ts"; @@ -590,6 +591,113 @@ it.effect("rechecks run ownership immediately before calling the provider", () = }).pipe(Effect.provide(RunExecutionTestLayer)), ); +it.effect("fails the run when its ownership check cannot be read before calling the provider", () => + Effect.gen(function* () { + const guardCalls = yield* Ref.make(0); + const providerStarts = yield* Ref.make(0); + const writes = yield* Ref.make>([]); + const threadId = ThreadId.make("thread:run-execution-start-guard-read"); + const runId = RunId.make("run:run-execution-start-guard-read"); + const attemptId = RunAttemptId.make("attempt:run-execution-start-guard-read"); + const providerInstanceId = ProviderInstanceId.make("codex"); + const testLayer = RunExecutionService.layer.pipe( + Layer.provide( + Layer.mergeAll( + Layer.mock(CheckpointService.CheckpointServiceV2)({ captureBaseline: () => Effect.void }), + Layer.mock(EventSink.EventSinkV2)({ + writeIfRunCurrent: (input) => + Ref.update(writes, (current) => [...current, ...input.events]).pipe( + Effect.as({ committed: true, storedEvents: [] }), + ), + }), + IdAllocator.layer, + Layer.mock(ProviderEventIngestor.ProviderEventIngestorV2)({ + ingestNormalized: () => Effect.succeed([]), + }), + ServerSettings.layerTest(), + ), + ), + ); + + yield* Effect.gen(function* () { + const runExecution = yield* RunExecutionService.RunExecutionServiceV2; + yield* runExecution.startRootRun({ + commandId: CommandId.make("command:run-execution-start-guard-read"), + appThread: { id: threadId } as OrchestrationV2AppThread, + providerSessionId: ProviderSessionId.make("session:run-execution-start-guard-read"), + session: { + events: Stream.never, + startTurn: () => Ref.update(providerStarts, (count) => count + 1), + } as unknown as ProviderAdapterV2SessionRuntime, + run: { id: runId, threadId, ordinal: 1, providerInstanceId } as OrchestrationV2Run, + rootNode: { + id: NodeId.make("node:run-execution-start-guard-read"), + } as OrchestrationV2ExecutionNode, + checkpointScope: { + id: CheckpointScopeId.make("checkpoint-scope:run-execution-start-guard-read"), + } as OrchestrationV2CheckpointScope, + providerThread: { + id: ProviderThreadId.make("provider-thread:run-execution-start-guard-read"), + driver, + } as OrchestrationV2ProviderThread, + attempt: { id: attemptId, providerTurnId: null } as OrchestrationV2RunAttempt, + attemptId, + providerTurnOrdinal: 1, + // The preparation check passes; the check right before the provider + // call cannot read the run. + shouldStartProviderTurn: () => + Ref.getAndUpdate(guardCalls, (calls) => calls + 1).pipe( + Effect.flatMap((calls) => + calls === 0 + ? Effect.succeed(true) + : Effect.fail( + new ProjectionStore.ProjectionStoreReadError({ + threadId, + cause: "database unavailable", + }), + ), + ), + ), + // The failure is settled by the guarded write, not by another read. + shouldFinalizeRun: () => + Effect.fail( + new ProjectionStore.ProjectionStoreReadError({ + threadId, + cause: "database unavailable", + }), + ), + message: { + messageId: MessageId.make("message:run-execution-start-guard-read"), + text: "Start while the store is down.", + attachments: [], + createdBy: "user", + creationSource: "web", + }, + modelSelection: { instanceId: providerInstanceId, model: "gpt-5.4" }, + runtimePolicy: { + runtimeMode: "full-access", + interactionMode: "default", + cwd: process.cwd(), + approvalPolicy: "never", + sandboxPolicy: { + type: "readOnly", + access: { type: "fullAccess" }, + networkAccess: false, + }, + }, + }); + }).pipe(Effect.provide(testLayer)); + + assert.equal(yield* Ref.get(guardCalls), 2); + assert.equal(yield* Ref.get(providerStarts), 0); + const runUpdate = (yield* Ref.get(writes)).find((event) => event.type === "run.updated"); + assert.equal( + runUpdate?.type === "run.updated" ? runUpdate.payload.status : undefined, + "failed", + ); + }), +); + it.effect( "dispatches only attachment-free compact commands through the native compaction path", () => @@ -3165,6 +3273,25 @@ it.effect.each(["completed", "interrupted", "cancelled", "failed"] as const)( }), ); +it.effect("records a finished run as failed when its ownership check cannot be read", () => + Effect.gen(function* () { + const { observed } = yield* captureRootRunTermination({ + key: "finalize-guard-read-failure", + shouldFinalizeRun: () => + Effect.fail( + new ProjectionStore.ProjectionStoreReadError({ + threadId: ThreadId.make("thread:finalize-guard-read-failure"), + cause: "database unavailable", + }), + ), + events: (ids) => Stream.make(rootTerminalEvent(ids, "completed")), + }); + // The fallback settles through the guarded write instead of the same + // failing read, so the run does not stay running. + assert.include(observed, "run:failed"); + }), +); + it.effect("does not refresh pull requests for auxiliary or stale provider terminals", () => Effect.gen(function* () { const { observed } = yield* captureRootRunTermination({ @@ -3254,7 +3381,7 @@ it.effect("keeps completed runs completed when pull request refresh fails", () = function captureRootRunTermination(input: { readonly key: string; - readonly shouldFinalizeRun: () => Effect.Effect; + readonly shouldFinalizeRun: () => Effect.Effect; readonly hasUnpairedRunInterruptRequest?: () => Effect.Effect; readonly seedOpenSubagent?: boolean; readonly events?: ( @@ -3278,6 +3405,17 @@ function captureRootRunTermination(input: { const ingestionDone = yield* Deferred.make(); const captureTurnItem = (payload: OrchestrationV2TurnItem) => Ref.update(writtenItems, (current) => [...current, payload]); + const captureFinalEvents = (events: ReadonlyArray) => + Effect.gen(function* () { + for (const event of events) { + if (event.type === "turn-item.updated") { + yield* captureTurnItem(event.payload); + } + if (event.type === "run.updated") { + yield* Ref.update(observed, (current) => [...current, `run:${event.payload.status}`]); + } + } + }); const testLayer = RunExecutionService.layer.pipe( Layer.provide( Layer.mergeAll( @@ -3292,22 +3430,11 @@ function captureRootRunTermination(input: { } return []; }), - writeWithEffects: (payload) => - Effect.gen(function* () { - for (const event of payload.events) { - if (event.type === "turn-item.updated") { - yield* captureTurnItem(event.payload); - } - if (event.type === "run.updated") { - yield* Ref.update(observed, (current) => [ - ...current, - `run:${event.payload.status}`, - ]); - } - } - return []; - }), - writeIfRunCurrent: () => Effect.succeed({ committed: true, storedEvents: [] }), + writeWithEffects: (payload) => captureFinalEvents(payload.events).pipe(Effect.as([])), + writeIfRunCurrent: (payload) => + captureFinalEvents(payload.events).pipe( + Effect.as({ committed: true, storedEvents: [] }), + ), }), IdAllocator.layer, Layer.mock(ProviderEventIngestor.ProviderEventIngestorV2)({ diff --git a/apps/server/src/orchestration-v2/RunExecutionService.ts b/apps/server/src/orchestration-v2/RunExecutionService.ts index 71211c126545913a2e32dd4632a82e51f57046dc..6aff5da24f1e101ebed640de7bf0d657003290eb 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.ts @@ -28,6 +28,7 @@ import * as Context from "effect/Context"; import * as Cause from "effect/Cause"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; import * as Ref from "effect/Ref"; @@ -47,6 +48,7 @@ import type { } from "./ProviderAdapter.ts"; import { ProviderAdapterTurnStartError } from "./ProviderAdapter.ts"; import * as ProviderEventIngestor from "./ProviderEventIngestor.ts"; +import type { ProjectionStoreV2Error } from "./ProjectionStore.ts"; import { makeProviderFailure, makeProviderFailureTurnItem } from "./ProviderFailure.ts"; import * as RunFinalizationService from "./RunFinalizationService.ts"; @@ -511,8 +513,8 @@ export interface RunExecutionServiceV2StartRootRunInput { >; readonly relatedThreadIds?: ReadonlyArray; readonly relatedProviderThreadIds?: ReadonlyArray; - readonly shouldStartProviderTurn?: () => Effect.Effect; - readonly shouldFinalizeRun?: () => Effect.Effect; + readonly shouldStartProviderTurn?: () => Effect.Effect; + readonly shouldFinalizeRun?: () => Effect.Effect; readonly hasUnpairedRunInterruptRequest?: () => Effect.Effect; readonly message: ProviderAdapterV2TurnMessage; readonly modelSelection: ModelSelection; @@ -557,7 +559,7 @@ export const layer: Layer.Layer< readonly checkpointScope: OrchestrationV2CheckpointScope; readonly providerThread: OrchestrationV2ProviderThread; readonly attempt: OrchestrationV2RunAttempt; - readonly shouldFinalizeRun?: () => Effect.Effect; + readonly shouldFinalizeRun?: () => Effect.Effect; readonly hasUnpairedRunInterruptRequest?: () => Effect.Effect; readonly openRunOwnedSubagents?: OpenRunOwnedSubagentProjection; readonly terminal: ProviderTerminalEvent; @@ -1286,15 +1288,12 @@ export const layer: Layer.Layer< checkpointScope: input.checkpointScope, providerThread, attempt: input.attempt, - ...(input.shouldFinalizeRun === undefined - ? {} - : { shouldFinalizeRun: input.shouldFinalizeRun }), - ...(input.hasUnpairedRunInterruptRequest === undefined - ? {} - : { - hasUnpairedRunInterruptRequest: - input.hasUnpairedRunInterruptRequest, - }), + // The failure may be the ownership + // read itself, so check in the write. + writeIfRunCurrent: { + activeAttemptId: input.attempt.id, + expectedStatus: "running", + }, openRunOwnedSubagents: openSubagents, terminal: makeFailedTerminalEvent( makeProviderFailure({ @@ -1328,10 +1327,13 @@ export const layer: Layer.Layer< Effect.forkDetach, ); - if ( - input.shouldStartProviderTurn !== undefined && - !(yield* input.shouldStartProviderTurn()) - ) { + // A failed read fails the start below, so the run is recorded as + // failed instead of staying active with no provider turn. + const shouldStart = + input.shouldStartProviderTurn === undefined + ? Exit.succeed(true) + : yield* Effect.exit(input.shouldStartProviderTurn()); + if (Exit.isSuccess(shouldStart) && !shouldStart.value) { yield* Fiber.interrupt(providerEventFiber); return; } @@ -1373,7 +1375,7 @@ export const layer: Layer.Layer< }), )) : input.session.startTurn(turnInput); - yield* startTurn.pipe( + yield* Effect.andThen(shouldStart, startTurn).pipe( Effect.catchCause((cause) => Effect.logError("orchestration V2 provider turn start failed", { runId: input.run.id, @@ -1392,20 +1394,18 @@ export const layer: Layer.Layer< checkpointScope: input.checkpointScope, providerThread, attempt: input.attempt, - ...(input.shouldFinalizeRun === undefined - ? {} - : { shouldFinalizeRun: input.shouldFinalizeRun }), - ...(input.hasUnpairedRunInterruptRequest === undefined - ? {} - : { - hasUnpairedRunInterruptRequest: - input.hasUnpairedRunInterruptRequest, - }), + // Checked in the write transaction, not by another + // read that can fail like the one before the start. + writeIfRunCurrent: { + activeAttemptId: input.attempt.id, + expectedStatus: "running", + }, openRunOwnedSubagents: openSubagents, terminal: makeFailedTerminalEvent( makeProviderFailure({ cause: Cause.squash(cause), - class: "provider_error", + // A failed ownership read is not the provider's fault. + class: Exit.isFailure(shouldStart) ? "unknown" : "provider_error", }), latestItemOrdinal + 1, ), -- 2.54.0