authorgravatar for me@t3.ggTheo Browne <me@t3.gg> 2026-10-03 01:43:08-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-10-03 01:43:08-07:00
log6108ef3d3d0ef2a598088a1bb64980580e935830
treed18b977a832634312ba99f1cbce51671133195e0
parentfed41fa88bb27cb4325cb208d571393850bc63c2
signature Signed by PGP key B5690EEEBB952194

fix(server): runs no longer get stuck (#15048)

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

9 files changed, 959 insertions(+), 115 deletions(-)

apps/server/src/orchestration-v2/EffectOutbox.ts+21-3
......@@ -278,9 +278,16 @@ export const layer: Layer.Layer<EffectOutboxV2, never, SqlClient.SqlClient> = La
278278 available,
279279 Array.from({ length: Math.min(64, Math.max(0, Math.floor(count))) }, () => undefined),
280280 ).pipe(Effect.asVoid);
281 // Each thread runs its effects one at a time, in enqueue (rowid) order. An earlier
282 // effect waiting out a retry backoff still blocks later ones, so a turn
283 // cannot start while a failed rollback is about to restore files. A claim
284 // that skips restart continuations is not blocked by them either.
281285 // Title generation is correlated metadata work, so it has its own
282286 // per-thread lane and cannot delay provider lifecycle effects.
283 const claimableCandidatePredicate = (availableBefore?: string) =>
287 const claimableCandidatePredicate = (
288 availableBefore?: string,
289 excludeRestartContinuations = false,
290 ) =>
284291 sql`
285292 ${
286293 availableBefore === undefined
......@@ -292,7 +299,18 @@ export const layer: Layer.Layer<EffectOutboxV2, never, SqlClient.SqlClient> = La
292299 SELECT 1
293300 FROM orchestration_v2_effect_outbox AS active
294301 WHERE active.thread_id = candidate.thread_id
295 AND active.status = 'running'
302 AND (
303 active.status = 'running'
304 OR (
305 active.status = 'pending'
306 AND active.rowid < candidate.rowid
307 AND ${
308 excludeRestartContinuations
309 ? sql`active.effect_type != 'provider-runtime.continue'`
310 : sql`1 = 1`
311 }
312 )
313 )
296314 AND (
297315 (
298316 candidate.effect_type = 'thread-title.generate'
......@@ -492,7 +510,7 @@ export const layer: Layer.Layer<EffectOutboxV2, never, SqlClient.SqlClient> = La
492510 WHERE effect_id = (
493511 SELECT candidate.effect_id
494512 FROM orchestration_v2_effect_outbox AS candidate
495 WHERE ${claimableCandidatePredicate(nowIso)}
513 WHERE ${claimableCandidatePredicate(nowIso, excludeRestartContinuations)}
496514 AND ${excludeRestartContinuations ? sql`candidate.effect_type != 'provider-runtime.continue'` : sql`1 = 1`}
497515 ORDER BY candidate.available_at ASC, candidate.created_at ASC, candidate.effect_id ASC
498516 LIMIT 1
apps/server/src/orchestration-v2/EventSink.ts+59-32
......@@ -17,6 +17,7 @@ import * as Effect from "effect/Effect";
1717import * as Layer from "effect/Layer";
1818import * as Option from "effect/Option";
1919import * as PubSub from "effect/PubSub";
20import * as Semaphore from "effect/Semaphore";
2021import * as Schema from "effect/Schema";
2122import * as Stream from "effect/Stream";
2223import * as SqlClient from "effect/unstable/sql/SqlClient";
......@@ -224,6 +225,40 @@ const baseLayer: Layer.Layer<
224225 );
225226 }
226227 });
228 const publishStoredEvents = (events: ReadonlyArray<OrchestrationV2StoredEvent>) =>
229 eventStore.publishCommitted(events).pipe(Effect.andThen(publishLiveEvents(events)));
230
231 // Transactions commit one at a time, but each writer publishes after its
232 // commit. If a writer is descheduled in between, a later commit reaches
233 // subscribers first, and clients drop any event at or below the newest
234 // sequence they have applied. So a writer takes this lane as the last step
235 // of its transaction and holds it until it has published. Publishing never
236 // waits, so a writer that holds the transaction while it waits for the
237 // lane is not blocked for long.
238 const publishLane = yield* Semaphore.make(1);
239 const commitThenPublish = <A, E, R>(
240 transaction: Effect.Effect<A, E, R>,
241 publish: (committed: A) => Effect.Effect<void>,
242 ) =>
243 Effect.suspend(() => {
244 let holdsLane = false;
245 const takeLane = publishLane.take(1).pipe(
246 Effect.andThen(
247 Effect.sync(() => {
248 holdsLane = true;
249 }),
250 ),
251 Effect.uninterruptible,
252 );
253 return sql
254 .withTransaction(Effect.tap(transaction, () => takeLane))
255 .pipe(
256 Effect.tap(publish),
257 Effect.ensuring(
258 Effect.suspend(() => (holdsLane ? publishLane.release(1) : Effect.void)),
259 ),
260 );
261 });
227262
228263 // A user can answer after terminal normalization reads the pending request.
229264 // Recheck inside the write transaction so stale cleanup cannot erase answers.
......@@ -332,7 +367,7 @@ const baseLayer: Layer.Layer<
332367 "orchestration_v2.thread_id": input.events[0]?.threadId ?? null,
333368 });
334369
335 const storedEvents = yield* sql.withTransaction(
370 return yield* commitThenPublish(
336371 Effect.gen(function* () {
337372 const normalized = yield* normalizeEvents(
338373 input.guardPendingUserInputCancellations === true
......@@ -347,13 +382,14 @@ const baseLayer: Layer.Layer<
347382 yield* effectOutbox.enqueue(input.effects);
348383 return committed;
349384 }),
385 (storedEvents) =>
386 Effect.gen(function* () {
387 if (input.effects.length > 0) {
388 yield* effectOutbox.notifyAvailable(input.effects.length);
389 }
390 yield* publishStoredEvents(storedEvents);
391 }),
350392 );
351 if (input.effects.length > 0) {
352 yield* effectOutbox.notifyAvailable(input.effects.length);
353 }
354 yield* eventStore.publishCommitted(storedEvents);
355 yield* publishLiveEvents(storedEvents);
356 return storedEvents;
357393 });
358394
359395 const writeIfRunCurrentEffect = Effect.fn("orchestrationV2.EventSink.writeIfRunCurrent")(
......@@ -365,7 +401,7 @@ const baseLayer: Layer.Layer<
365401 "orchestration_v2.thread_id": input.threadId,
366402 });
367403
368 const result = yield* sql.withTransaction(
404 return yield* commitThenPublish(
369405 Effect.gen(function* () {
370406 const rows = yield* sql<{
371407 readonly status: string;
......@@ -403,12 +439,8 @@ const baseLayer: Layer.Layer<
403439 yield* applyStoredEvents(storedEvents);
404440 return { committed: true as const, storedEvents };
405441 }),
442 (result) => (result.committed ? publishStoredEvents(result.storedEvents) : Effect.void),
406443 );
407 if (result.committed) {
408 yield* eventStore.publishCommitted(result.storedEvents);
409 yield* publishLiveEvents(result.storedEvents);
410 }
411 return result;
412444 },
413445 );
414446
......@@ -424,7 +456,7 @@ const baseLayer: Layer.Layer<
424456 "orchestration_v2.expected_last_run_ordinal": input.expectedLastRunOrdinal,
425457 });
426458
427 const result = yield* sql.withTransaction(
459 return yield* commitThenPublish(
428460 Effect.gen(function* () {
429461 const rows = yield* sql<{
430462 readonly active_attempt_id: string | null;
......@@ -464,12 +496,8 @@ const baseLayer: Layer.Layer<
464496 yield* applyStoredEvents(storedEvents);
465497 return { committed: true as const, storedEvents };
466498 }),
499 (result) => (result.committed ? publishStoredEvents(result.storedEvents) : Effect.void),
467500 );
468 if (result.committed) {
469 yield* eventStore.publishCommitted(result.storedEvents);
470 yield* publishLiveEvents(result.storedEvents);
471 }
472 return result;
473501 });
474502
475503 const existingCommandResult = (commandId: CommandId) =>
......@@ -490,7 +518,7 @@ const baseLayer: Layer.Layer<
490518 const commitCommandEffect = Effect.fn("orchestrationV2.EventSink.commitCommand")(function* (
491519 input: Parameters<EventSinkV2Shape["commitCommand"]>[0],
492520 ) {
493 const result = yield* sql.withTransaction(
521 const result = yield* commitThenPublish(
494522 Effect.gen(function* () {
495523 const reserved = yield* commandReceipts.insertIfAbsent({
496524 commandId: input.commandId,
......@@ -538,15 +566,15 @@ const baseLayer: Layer.Layer<
538566 });
539567 return { receipt, storedEvents, committed: true as const, cancelledEffectIds };
540568 }),
569 (result) =>
570 Effect.gen(function* () {
571 yield* effectOutbox.signalCancellations(result.cancelledEffectIds);
572 if (result.committed && input.effects.length > 0) {
573 yield* effectOutbox.notifyAvailable(input.effects.length);
574 }
575 if (result.committed) yield* publishStoredEvents(result.storedEvents);
576 }),
541577 );
542 yield* effectOutbox.signalCancellations(result.cancelledEffectIds);
543 if (result.committed && input.effects.length > 0) {
544 yield* effectOutbox.notifyAvailable(input.effects.length);
545 }
546 if (result.committed) {
547 yield* eventStore.publishCommitted(result.storedEvents);
548 yield* publishLiveEvents(result.storedEvents);
549 }
550578 return {
551579 receipt: result.receipt,
552580 storedEvents: result.storedEvents,
......@@ -593,7 +621,7 @@ const baseLayer: Layer.Layer<
593621
594622 const commitProjectCommandEffect = Effect.fn("orchestrationV2.EventSink.commitProjectCommand")(
595623 function* (input: Parameters<EventSinkV2Shape["commitProjectCommand"]>[0]) {
596 const result = yield* sql.withTransaction(
624 const result = yield* commitThenPublish(
597625 Effect.gen(function* () {
598626 const reserved: CommandReceiptStore.ProjectCommandReceiptV2 = {
599627 commandId: input.commandId,
......@@ -613,10 +641,9 @@ const baseLayer: Layer.Layer<
613641 yield* commandReceipts.upsert(receipt);
614642 return { receipt, event };
615643 }),
644 (result) =>
645 result.event === undefined ? Effect.void : eventStore.publishCommitted([result.event]),
616646 );
617 if (result.event !== undefined) {
618 yield* eventStore.publishCommitted([result.event]);
619 }
620647 return { receipt: result.receipt, committed: result.event !== undefined };
621648 },
622649 );
apps/server/src/orchestration-v2/FoundationPersistence.test.ts+159
......@@ -51,7 +51,9 @@ import * as EventStore from "./EventStore.ts";
5151import * as IdAllocator from "./IdAllocator.ts";
5252import * as ProjectionMaintenance from "./ProjectionMaintenance.ts";
5353import * as ProjectionStore from "./ProjectionStore.ts";
54import * as ProjectStore from "./ProjectStore.ts";
5455import * as ProviderRuntimeRecovery from "./ProviderRuntimeRecoveryService.ts";
56import * as TurnItemPositionStore from "./TurnItemPositionStore.ts";
5557
5658const isLiveStreamBufferError = Schema.is(LiveStreamBufferError);
5759const encodeJson = Schema.encodeEffect(Schema.fromJsonString(Schema.Unknown));
......@@ -2303,6 +2305,76 @@ it.layer(TestLayer)("orchestration V2 foundation persistence", (it) => {
23032305 }).pipe(Effect.provide(Layer.fresh(effectOutboxProvided))),
23042306 );
23052307
2308 it.effect("keeps later thread effects behind an earlier effect waiting to retry", () =>
2309 Effect.gen(function* () {
2310 const outbox = yield* EffectOutbox.EffectOutboxV2;
2311 const workerId = "retry-order-worker";
2312 const commandId = CommandId.make("command:foundation-retry-order");
2313 const threadId = ThreadId.make("thread:foundation-retry-order");
2314 yield* outbox.enqueue([
2315 {
2316 id: "effect:foundation-retry-order:z-rollback",
2317 commandId,
2318 threadId,
2319 request: {
2320 type: "provider-thread.rollback",
2321 providerThreadId: ProviderThreadId.make("provider-thread:foundation-retry-order"),
2322 checkpointId: CheckpointId.make("checkpoint:foundation-retry-order"),
2323 scopeId: CheckpointScopeId.make("scope:foundation-retry-order"),
2324 },
2325 },
2326 ]);
2327 const rollback = yield* outbox.claimNext({ workerId, leaseDurationMs: 30_000 });
2328 assert.isTrue(Option.isSome(rollback));
2329 if (Option.isNone(rollback)) return;
2330 yield* outbox.retry({
2331 effectId: rollback.value.id,
2332 workerId,
2333 error: "rollback failed once",
2334 delayMs: 60_000,
2335 });
2336
2337 // A turn the user starts during the rollback's backoff must not run first,
2338 // even when its timestamp ties and its id sorts first.
2339 yield* outbox.enqueue([
2340 {
2341 id: "effect:foundation-retry-order:a-start",
2342 commandId: CommandId.make("command:foundation-retry-order:start"),
2343 threadId,
2344 request: { type: "provider-turn.start", runId: RunId.make("run:foundation-retry-order") },
2345 },
2346 {
2347 id: "effect:foundation-retry-order:b-title",
2348 commandId: CommandId.make("command:foundation-retry-order:title"),
2349 threadId,
2350 request: { type: "thread-title.generate", kind: { type: "regenerate" } },
2351 },
2352 ]);
2353 const title = yield* outbox.claimNext({ workerId, leaseDurationMs: 30_000 });
2354 assert.equal(Option.getOrUndefined(title)?.id, "effect:foundation-retry-order:b-title");
2355 const blocked = yield* outbox.claimNext({ workerId, leaseDurationMs: 30_000 });
2356 assert.isTrue(Option.isNone(blocked));
2357 const nextClaimable = yield* outbox.nextClaimableAt;
2358 assert.isTrue(Option.isSome(nextClaimable));
2359 if (Option.isSome(nextClaimable)) {
2360 assert.equal(
2361 DateTime.formatIso(nextClaimable.value),
2362 (yield* outbox.get(rollback.value.id)).pipe(Option.getOrThrow).availableAt,
2363 );
2364 }
2365
2366 yield* outbox.cancelUnsettled({
2367 threadId,
2368 effectTypes: ["provider-thread.rollback"],
2369 reason: "Test cleanup.",
2370 });
2371 const unblocked = yield* outbox.claimNext({ workerId, leaseDurationMs: 30_000 });
2372 assert.equal(Option.getOrUndefined(unblocked)?.id, "effect:foundation-retry-order:a-start");
2373 yield* outbox.succeed({ effectId: "effect:foundation-retry-order:a-start", workerId });
2374 yield* outbox.succeed({ effectId: "effect:foundation-retry-order:b-title", workerId });
2375 }).pipe(Effect.provide(Layer.fresh(effectOutboxProvided))),
2376 );
2377
23062378 it.effect("executes a retry at its durable deadline instead of the liveness interval", () =>
23072379 Effect.gen(function* () {
23082380 const outbox = yield* EffectOutbox.EffectOutboxV2;
......@@ -3222,3 +3294,90 @@ it.live("keeps claiming new work after repeated idle periods", () =>
32223294 }).pipe(Effect.provide(workerLayer), Effect.scoped);
32233295 }).pipe(Effect.provide(TestLayer)),
32243296);
3297
3298it.effect("publishes live events in commit order across concurrent writers", () =>
3299 Effect.gen(function* () {
3300 const firstCommitted = yield* Deferred.make<void>();
3301 const releaseFirst = yield* Deferred.make<void>();
3302 // The first writer's post-commit wakeup stands in for any scheduler yield
3303 // between its commit and its publish.
3304 const pausingOutbox = Layer.effect(
3305 EffectOutbox.EffectOutboxV2,
3306 Effect.gen(function* () {
3307 const delegate = yield* EffectOutbox.EffectOutboxV2;
3308 return EffectOutbox.EffectOutboxV2.of({
3309 ...delegate,
3310 notifyAvailable: (count) =>
3311 Deferred.succeed(firstCommitted, undefined).pipe(
3312 Effect.andThen(Deferred.await(releaseFirst)),
3313 Effect.andThen(delegate.notifyAvailable(count)),
3314 ),
3315 });
3316 }),
3317 ).pipe(Layer.provide(effectOutboxProvided));
3318 const eventSinkLayer = EventSink.layerFromStores.pipe(
3319 Layer.provide(
3320 Layer.mergeAll(
3321 storesProvided,
3322 pausingOutbox,
3323 commandReceiptStoreProvided,
3324 ProjectStore.layer.pipe(Layer.provide(databaseLayer)),
3325 TurnItemPositionStore.layer.pipe(Layer.provide(databaseLayer)),
3326 ),
3327 ),
3328 );
3329
3330 yield* Effect.gen(function* () {
3331 const eventSink = yield* EventSink.EventSinkV2;
3332 const now = yield* DateTime.now;
3333 const first = makeThread(ThreadId.make("thread:foundation-publish-order:first"), now);
3334 const second = makeThread(ThreadId.make("thread:foundation-publish-order:second"), now);
3335 const published = yield* eventSink
3336 .stream({ afterSequence: yield* eventSink.latestSequence() })
3337 .pipe(Stream.take(2), Stream.runCollect, Effect.forkScoped({ startImmediately: true }));
3338
3339 const firstWrite = yield* eventSink
3340 .writeWithEffects({
3341 events: [
3342 threadCreatedEvent({ id: "event:foundation-publish-order:first", thread: first, now }),
3343 ],
3344 effects: [
3345 {
3346 id: "effect:foundation-publish-order:first",
3347 commandId: CommandId.make("command:foundation-publish-order:first"),
3348 threadId: first.id,
3349 request: { type: "terminal.cleanup" },
3350 },
3351 ],
3352 })
3353 .pipe(Effect.forkScoped);
3354 yield* Deferred.await(firstCommitted);
3355 // The second writer commits after the first. It may run as far as it can
3356 // before the first writer resumes.
3357 const secondWrite = yield* eventSink
3358 .write({
3359 events: [
3360 threadCreatedEvent({
3361 id: "event:foundation-publish-order:second",
3362 thread: second,
3363 now,
3364 }),
3365 ],
3366 })
3367 .pipe(
3368 Effect.provideService(Scheduler.MaxOpsBeforeYield, Number.POSITIVE_INFINITY),
3369 Effect.forkScoped,
3370 );
3371 yield* Effect.yieldNow;
3372 yield* Deferred.succeed(releaseFirst, undefined);
3373 yield* Fiber.join(firstWrite);
3374 yield* Fiber.join(secondWrite);
3375
3376 const sequences = Array.from(yield* Fiber.join(published), (stored) => stored.sequence);
3377 assert.deepEqual(
3378 sequences,
3379 [...sequences].sort((left, right) => left - right),
3380 );
3381 }).pipe(Effect.provide(eventSinkLayer));
3382 }).pipe(Effect.provide(databaseLayer)),
3383);
apps/server/src/orchestration-v2/ProviderSessionManager.test.ts+347-3
......@@ -20,6 +20,7 @@ import * as Cause from "effect/Cause";
2020import * as DateTime from "effect/DateTime";
2121import * as Deferred from "effect/Deferred";
2222import * as Effect from "effect/Effect";
23import * as Exit from "effect/Exit";
2324import * as Fiber from "effect/Fiber";
2425import * as FileSystem from "effect/FileSystem";
2526import * as Layer from "effect/Layer";
......@@ -80,6 +81,52 @@ const FailingReleaseEventSinkLayer = Layer.effect(
8081 }),
8182).pipe(Layer.provide(TestEventSinkLayer));
8283
84interface FlakyReleaseWrites {
85 /** Which release writes fail right now. */
86 readonly failing: Ref.Ref<"none" | "session" | "session-and-requests">;
87 /** Receives one item per failed write. */
88 readonly failures: Queue.Queue<void>;
89 /** Holds runtime request writes: completes `paused`, then waits for `resume`. */
90 readonly pauseRequestWrites?: {
91 readonly paused: Deferred.Deferred<void>;
92 readonly resume: Deferred.Deferred<void>;
93 };
94}
95
96// Fails release writes with a defect, the way a failed SQL commit surfaces.
97const makeFlakyReleaseEventSinkLayer = (flaky: FlakyReleaseWrites) =>
98 Layer.effect(
99 EventSink.EventSinkV2,
100 Effect.gen(function* () {
101 const delegate = yield* EventSink.EventSinkV2;
102 return EventSink.EventSinkV2.of({
103 ...delegate,
104 write: (input) =>
105 Effect.gen(function* () {
106 const failing = yield* Ref.get(flaky.failing);
107 const fails = input.events.some(
108 (event) =>
109 (failing !== "none" &&
110 event.type === "provider-session.updated" &&
111 (event.payload.status === "stopped" || event.payload.status === "error")) ||
112 (failing === "session-and-requests" && event.type === "runtime-request.updated"),
113 );
114 const pause = flaky.pauseRequestWrites;
115 if (
116 pause !== undefined &&
117 input.events.some((event) => event.type === "runtime-request.updated")
118 ) {
119 yield* Deferred.succeed(pause.paused, undefined);
120 yield* Deferred.await(pause.resume);
121 }
122 if (!fails) return yield* delegate.write(input);
123 yield* Queue.offer(flaky.failures, undefined);
124 return yield* Effect.die(new Error("simulated commit failure"));
125 }),
126 });
127 }),
128 ).pipe(Layer.provide(TestEventSinkLayer));
129
83130const CodexCapabilities: OrchestrationV2ProviderCapabilities = CodexProviderCapabilitiesV2;
84131const ExclusiveCapabilities: OrchestrationV2ProviderCapabilities = {
85132 ...CodexCapabilities,
......@@ -359,15 +406,19 @@ function makeTestLayer(input: {
359406 readonly initialProviderItemIdentityVersion?: 2;
360407 }) => Effect.Effect<void>;
361408 readonly failReleaseEventWrites?: boolean;
409 readonly flakyReleaseWrites?: FlakyReleaseWrites;
362410 readonly hasPendingBackgroundWork?: Effect.Effect<boolean>;
363411 readonly hangSessionScopeClose?: boolean;
364412 readonly beforeUnload?: Effect.Effect<void>;
365413 readonly serverSettingsLayer?: ReturnType<typeof ServerSettings.layerTest>;
366414 readonly projectServiceLayer?: Layer.Layer<ProjectService.ProjectService>;
367415}) {
368 const configuredEventSinkLayer = input.failReleaseEventWrites
369 ? FailingReleaseEventSinkLayer
370 : TestEventSinkLayer;
416 const configuredEventSinkLayer =
417 input.flakyReleaseWrites !== undefined
418 ? makeFlakyReleaseEventSinkLayer(input.flakyReleaseWrites)
419 : input.failReleaseEventWrites
420 ? FailingReleaseEventSinkLayer
421 : TestEventSinkLayer;
371422 const registryLayer = ProviderAdapterRegistry.makeSingleLayer(
372423 makeProviderAdapter(input.state, {
373424 failEventStream: input.failEventStream ?? false,
......@@ -2299,6 +2350,299 @@ it.effect("ProviderSessionManagerV2 marks pending runtime requests non-live on r
22992350 yield* effect.pipe(Effect.provide(makeTestLayer({ state, idleTimeoutMs: 1000 })));
23002351 }),
23012352);
2353it.effect("ProviderSessionManagerV2 retries release records that failed to persist", () =>
2354 Effect.gen(function* () {
2355 const state = yield* Ref.make(emptyState);
2356 const flaky: FlakyReleaseWrites = {
2357 failing: yield* Ref.make<"none" | "session" | "session-and-requests">("session"),
2358 failures: yield* Queue.unbounded<void>(),
2359 };
2360 const effect = Effect.gen(function* () {
2361 const eventSink = yield* EventSink.EventSinkV2;
2362 const idAllocator = yield* IdAllocator.IdAllocatorV2;
2363 const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
2364 const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
2365 const now = yield* DateTime.now;
2366 const threadId = ThreadId.make("thread-provider-session-manager-release-retry");
2367 const providerSessionId = yield* idAllocator.allocate.providerSession({
2368 providerInstanceId: modelSelection.instanceId,
2369 threadId,
2370 });
2371 const providerThread = makeProviderThread({ idAllocator, threadId, providerSessionId, now });
2372 yield* eventSink.write({
2373 events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })],
2374 });
2375 const pendingRequest = yield* makePendingRuntimeRequestEvents({
2376 idAllocator,
2377 threadId,
2378 providerSessionId,
2379 providerThread,
2380 now,
2381 });
2382 yield* eventSink.write({ events: pendingRequest.events });
2383 yield* manager.open({ threadId, providerSessionId, modelSelection, runtimePolicy });
2384
2385 assert.isTrue(Exit.isFailure(yield* Effect.exit(manager.close(providerSessionId))));
2386 yield* Queue.take(flaky.failures);
2387 // The failed session write does not keep the approval answerable.
2388 const afterClose = yield* projectionStore.getThreadProjection(threadId);
2389 assert.equal(afterClose.runtimeRequests.at(-1)?.responseCapability.type, "not_resumable");
2390 assert.equal(afterClose.providerSessions.at(-1)?.status, "ready");
2391
2392 // The first retry fails as well, and the retries continue.
2393 yield* TestClock.adjust("1 second");
2394 yield* Queue.take(flaky.failures);
2395 yield* Ref.set(flaky.failing, "none");
2396 const stopped = yield* eventSink
2397 .stream({
2398 threadId,
2399 afterSequence: yield* eventSink.latestSequence({ threadId }),
2400 eventType: "provider-session.updated",
2401 })
2402 .pipe(Stream.runHead, Effect.forkScoped);
2403 yield* TestClock.adjust("1 second");
2404 yield* Fiber.join(stopped);
2405
2406 const afterRetry = yield* projectionStore.getThreadProjection(threadId);
2407 assert.equal(afterRetry.providerSessions.at(-1)?.status, "stopped");
2408 });
2409
2410 yield* effect.pipe(
2411 Effect.provide(makeTestLayer({ state, idleTimeoutMs: 60_000, flakyReleaseWrites: flaky })),
2412 );
2413 }),
2414);
2415
2416it.effect("ProviderSessionManagerV2 release retries leave a replacement session alone", () =>
2417 Effect.gen(function* () {
2418 const state = yield* Ref.make(emptyState);
2419 const flaky: FlakyReleaseWrites = {
2420 failing: yield* Ref.make<"none" | "session" | "session-and-requests">("none"),
2421 failures: yield* Queue.unbounded<void>(),
2422 };
2423 const effect = Effect.gen(function* () {
2424 const eventSink = yield* EventSink.EventSinkV2;
2425 const idAllocator = yield* IdAllocator.IdAllocatorV2;
2426 const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
2427 const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
2428 const threadId = ThreadId.make("thread-provider-session-manager-release-replacement");
2429 const providerSessionId = yield* idAllocator.allocate.providerSession({
2430 providerInstanceId: modelSelection.instanceId,
2431 threadId,
2432 });
2433 const writePendingRequest = Effect.gen(function* () {
2434 const now = yield* DateTime.now;
2435 const request = yield* makePendingRuntimeRequestEvents({
2436 idAllocator,
2437 threadId,
2438 providerSessionId,
2439 providerThread: makeProviderThread({ idAllocator, threadId, providerSessionId, now }),
2440 now,
2441 });
2442 yield* eventSink.write({ events: request.events });
2443 return request.requestId;
2444 });
2445 yield* eventSink.write({
2446 events: [
2447 yield* makeThreadCreatedEvent({ idAllocator, threadId, now: yield* DateTime.now }),
2448 ],
2449 });
2450 const oldRequestId = yield* writePendingRequest;
2451 yield* manager.open({ threadId, providerSessionId, modelSelection, runtimePolicy });
2452 yield* Ref.set(flaky.failing, "session-and-requests");
2453 assert.isTrue(Exit.isFailure(yield* Effect.exit(manager.close(providerSessionId))));
2454 yield* Queue.take(flaky.failures);
2455 yield* Queue.take(flaky.failures);
2456 yield* Ref.set(flaky.failing, "none");
2457
2458 // A replacement opens with the same id before the retry runs.
2459 yield* TestClock.adjust("500 millis");
2460 yield* manager.open({ threadId, providerSessionId, modelSelection, runtimePolicy });
2461 const newRequestId = yield* writePendingRequest;
2462 const replacementStatus = (yield* projectionStore.getThreadProjection(
2463 threadId,
2464 )).providerSessions.at(-1)?.status;
2465 const settled = yield* eventSink
2466 .stream({
2467 threadId,
2468 afterSequence: yield* eventSink.latestSequence({ threadId }),
2469 eventType: "runtime-request.updated",
2470 })
2471 .pipe(Stream.runHead, Effect.forkScoped);
2472 yield* TestClock.adjust("500 millis");
2473 yield* Fiber.join(settled);
2474
2475 const projection = yield* projectionStore.getThreadProjection(threadId);
2476 const request = (id: typeof oldRequestId) =>
2477 projection.runtimeRequests.find((candidate) => candidate.id === id);
2478 assert.equal(request(oldRequestId)?.responseCapability.type, "not_resumable");
2479 assert.equal(request(newRequestId)?.responseCapability.type, "live");
2480 assert.equal(projection.providerSessions.at(-1)?.status, replacementStatus);
2481 });
2482
2483 yield* effect.pipe(
2484 Effect.provide(makeTestLayer({ state, idleTimeoutMs: 60_000, flakyReleaseWrites: flaky })),
2485 );
2486 }),
2487);
2488
2489it.effect("ProviderSessionManagerV2 keeps each failed release's cleanup", () =>
2490 Effect.gen(function* () {
2491 const state = yield* Ref.make(emptyState);
2492 const flaky: FlakyReleaseWrites = {
2493 failing: yield* Ref.make<"none" | "session" | "session-and-requests">("none"),
2494 failures: yield* Queue.unbounded<void>(),
2495 };
2496 const effect = Effect.gen(function* () {
2497 const eventSink = yield* EventSink.EventSinkV2;
2498 const idAllocator = yield* IdAllocator.IdAllocatorV2;
2499 const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
2500 const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
2501 const now = yield* DateTime.now;
2502 const firstThreadId = ThreadId.make("thread-provider-session-manager-release-each-a");
2503 const secondThreadId = ThreadId.make("thread-provider-session-manager-release-each-b");
2504 const providerSessionId = yield* idAllocator.allocate.providerSession({
2505 providerInstanceId: modelSelection.instanceId,
2506 threadId: firstThreadId,
2507 });
2508 yield* eventSink.write({
2509 events: [
2510 yield* makeThreadCreatedEvent({ idAllocator, threadId: firstThreadId, now }),
2511 yield* makeThreadCreatedEvent({ idAllocator, threadId: secondThreadId, now }),
2512 ],
2513 });
2514 const secondThreadRequest = yield* makePendingRuntimeRequestEvents({
2515 idAllocator,
2516 threadId: secondThreadId,
2517 providerSessionId,
2518 providerThread: makeProviderThread({
2519 idAllocator,
2520 threadId: secondThreadId,
2521 providerSessionId,
2522 now,
2523 }),
2524 now,
2525 });
2526 yield* eventSink.write({ events: secondThreadRequest.events });
2527 const failRelease = (failedWrites: number) =>
2528 Effect.gen(function* () {
2529 yield* Ref.set(flaky.failing, "session-and-requests");
2530 assert.isTrue(Exit.isFailure(yield* Effect.exit(manager.close(providerSessionId))));
2531 yield* Effect.repeat(Queue.take(flaky.failures), { times: failedWrites - 1 });
2532 yield* Ref.set(flaky.failing, "none");
2533 });
2534
2535 // The first session serves both threads. Its replacement serves one.
2536 yield* manager.open({
2537 threadId: firstThreadId,
2538 providerSessionId,
2539 modelSelection,
2540 runtimePolicy,
2541 });
2542 yield* manager.open({
2543 threadId: secondThreadId,
2544 providerSessionId,
2545 modelSelection,
2546 runtimePolicy,
2547 });
2548 // Both the session write and the second thread's request write fail.
2549 yield* failRelease(2);
2550 yield* TestClock.adjust("500 millis");
2551 yield* manager.open({
2552 threadId: firstThreadId,
2553 providerSessionId,
2554 modelSelection,
2555 runtimePolicy,
2556 });
2557 // Only the session write fails: the first thread has no requests.
2558 yield* failRelease(1);
2559
2560 const settled = yield* eventSink
2561 .stream({
2562 threadId: secondThreadId,
2563 afterSequence: yield* eventSink.latestSequence({ threadId: secondThreadId }),
2564 eventType: "runtime-request.updated",
2565 })
2566 .pipe(Stream.runHead, Effect.forkScoped);
2567 yield* TestClock.adjust("500 millis");
2568 yield* Fiber.join(settled);
2569
2570 const projection = yield* projectionStore.getThreadProjection(secondThreadId);
2571 assert.equal(projection.runtimeRequests.at(-1)?.responseCapability.type, "not_resumable");
2572 });
2573
2574 yield* effect.pipe(
2575 Effect.provide(makeTestLayer({ state, idleTimeoutMs: 60_000, flakyReleaseWrites: flaky })),
2576 );
2577 }),
2578);
2579
2580it.effect("ProviderSessionManagerV2 settles a request the event pump persists during release", () =>
2581 Effect.gen(function* () {
2582 const state = yield* Ref.make(emptyState);
2583 const flaky: FlakyReleaseWrites = {
2584 failing: yield* Ref.make<"none" | "session" | "session-and-requests">("none"),
2585 failures: yield* Queue.unbounded<void>(),
2586 pauseRequestWrites: {
2587 paused: yield* Deferred.make<void>(),
2588 resume: yield* Deferred.make<void>(),
2589 },
2590 };
2591 const pause = flaky.pauseRequestWrites!;
2592 const effect = Effect.gen(function* () {
2593 const eventSink = yield* EventSink.EventSinkV2;
2594 const idAllocator = yield* IdAllocator.IdAllocatorV2;
2595 const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
2596 const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
2597 const now = yield* DateTime.now;
2598 const threadId = ThreadId.make("thread-provider-session-manager-release-pump");
2599 const providerSessionId = yield* idAllocator.allocate.providerSession({
2600 providerInstanceId: modelSelection.instanceId,
2601 threadId,
2602 });
2603 yield* eventSink.write({
2604 events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })],
2605 });
2606 yield* manager.open({ threadId, providerSessionId, modelSelection, runtimePolicy });
2607 // The runtime creates the request a moment after the release starts.
2608 const createdAt = DateTime.add(now, { seconds: 1 });
2609 const pendingRequest = yield* makePendingRuntimeRequestEvents({
2610 idAllocator,
2611 threadId,
2612 providerSessionId,
2613 providerThread: makeProviderThread({
2614 idAllocator,
2615 threadId,
2616 providerSessionId,
2617 now: createdAt,
2618 }),
2619 now: createdAt,
2620 });
2621 const adapterEvents = (yield* Ref.get(state)).eventQueues.get(String(providerSessionId));
2622 assert.isDefined(adapterEvents);
2623 yield* Queue.offerAll(adapterEvents!, pendingRequest.providerEvents);
2624 // The event pump holds the request permit while it persists the request.
2625 yield* Deferred.await(pause.paused);
2626 const closed = yield* manager
2627 .close(providerSessionId)
2628 .pipe(Effect.forkScoped({ startImmediately: true }));
2629 yield* TestClock.adjust("1 second");
2630 yield* Deferred.succeed(pause.resume, undefined);
2631 yield* Fiber.join(closed);
2632
2633 const projection = yield* projectionStore.getThreadProjection(threadId);
2634 const request = projection.runtimeRequests.find(
2635 (candidate) => candidate.id === pendingRequest.requestId,
2636 );
2637 assert.equal(request?.responseCapability.type, "not_resumable");
2638 });
2639
2640 yield* effect.pipe(
2641 Effect.provide(makeTestLayer({ state, idleTimeoutMs: 60_000, flakyReleaseWrites: flaky })),
2642 );
2643 }),
2644);
2645
23022646it.effect("ProviderSessionManagerV2 terminalizes a pending input transcript item on release", () =>
23032647 Effect.gen(function* () {
23042648 const state = yield* Ref.make(emptyState);
apps/server/src/orchestration-v2/ProviderSessionManager.ts+144-28
......@@ -17,11 +17,13 @@ import * as Duration from "effect/Duration";
1717import * as Effect from "effect/Effect";
1818import * as Exit from "effect/Exit";
1919import * as Fiber from "effect/Fiber";
20import * as FiberSet from "effect/FiberSet";
2021import * as FileSystem from "effect/FileSystem";
2122import * as Layer from "effect/Layer";
2223import * as Option from "effect/Option";
2324import * as Queue from "effect/Queue";
2425import * as Ref from "effect/Ref";
26import * as Schedule from "effect/Schedule";
2527import * as Schema from "effect/Schema";
2628import * as Scope from "effect/Scope";
2729import * as Semaphore from "effect/Semaphore";
......@@ -361,6 +363,9 @@ export const layerWithOptions = (
361363 );
362364 const layerScope = yield* Effect.scope;
363365 const sessions = yield* Ref.make(new Map<string, LiveSessionEntry>());
366 // One retry per released entry, so a later release with the same id
367 // cannot drop cleanup for threads only the earlier session served.
368 const releaseRecordRetries = yield* FiberSet.make();
364369 const nextSubscriberId = yield* Ref.make(0);
365370 const sessionOpen = yield* makeKeyedSerialExecutor<ProviderSessionId>();
366371 // Orders a thread's attach against a detach unloading it on the same session.
......@@ -608,6 +613,8 @@ export const layerWithOptions = (
608613 const writeReleasedRuntimeRequestEvents = (input: {
609614 readonly entry: LiveSessionEntry;
610615 readonly reason: ProviderSessionReleaseReason;
616 /** Requests created later belong to a replacement session with the same id. */
617 readonly releasedAt: DateTime.Utc;
611618 }) =>
612619 Effect.gen(function* () {
613620 const providerSessionId = input.entry.runtime.providerSessionId;
......@@ -629,7 +636,8 @@ export const layerWithOptions = (
629636 (request) =>
630637 request.status === "pending" &&
631638 request.responseCapability.type === "live" &&
632 request.responseCapability.providerSessionId === providerSessionId,
639 request.responseCapability.providerSessionId === providerSessionId &&
640 DateTime.isLessThanOrEqualTo(request.createdAt, input.releasedAt),
633641 );
634642
635643 for (const request of releasedRequests) {
......@@ -708,6 +716,125 @@ export const layerWithOptions = (
708716 }
709717 });
710718
719 // Records a released session as stopped and resolves the live runtime
720 // requests it left. Each write runs even if the other fails. Once a
721 // replacement session opens with the same id, it owns the session status,
722 // so only the requests are settled.
723 const writeReleaseRecords = (input: {
724 readonly entry: LiveSessionEntry;
725 readonly reason: ProviderSessionReleaseReason;
726 readonly detail?: string;
727 readonly releasedAt: DateTime.Utc;
728 readonly replaced: boolean;
729 }) =>
730 Effect.all(
731 [
732 input.replaced
733 ? Effect.succeed(Exit.void)
734 : Effect.exit(writeReleasedSessionEvents(input)),
735 Effect.exit(
736 writeReleasedRuntimeRequestEvents(input).pipe(
737 input.entry.requestEventPermit.withPermits(1),
738 ),
739 ),
740 ],
741 { concurrency: 1 },
742 ).pipe(Effect.flatMap(Exit.asVoidAll));
743
744 // The session already left the live map, so a later release finds
745 // nothing to do. Without a retry the UI would keep a ready session and
746 // answerable approvals until a server restart. Each attempt holds the
747 // session's open lock, so it sees a replacement that opened meanwhile.
748 const retryReleaseRecords = (
749 input: Omit<Parameters<typeof writeReleaseRecords>[0], "replaced">,
750 ) => {
751 const providerSessionId = input.entry.runtime.providerSessionId;
752 const attempt = Effect.gen(function* () {
753 const exit = yield* Effect.exit(
754 sessionOpen.withLock(
755 providerSessionId,
756 Effect.gen(function* () {
757 const replaced = (yield* Ref.get(sessions)).has(sessionKey(providerSessionId));
758 yield* writeReleaseRecords({ ...input, replaced });
759 }),
760 ),
761 );
762 if (Exit.isSuccess(exit) || Cause.hasInterruptsOnly(exit.cause)) return yield* exit;
763 yield* Effect.logWarning("orchestration-v2.provider-session-release-records-failed", {
764 providerSessionId,
765 cause: exit.cause,
766 });
767 // A failed SQL commit is a defect, so every failure but interruption
768 // is retried.
769 return yield* Effect.fail(exit.cause);
770 });
771 return attempt.pipe(
772 Effect.retry({
773 schedule: Schedule.exponential("1 second").pipe(
774 Schedule.modifyDelay(({ duration }) =>
775 Effect.succeed(Duration.min(duration, Duration.seconds(30))),
776 ),
777 ),
778 }),
779 Effect.delay("1 second"),
780 FiberSet.run(releaseRecordRetries),
781 );
782 };
783
784 const logReleaseFailure =
785 (providerSessionId: ProviderSessionId) =>
786 <E, R>(release: Effect.Effect<void, E, R>) =>
787 release.pipe(
788 Effect.catchCause((cause) =>
789 Effect.logWarning("orchestration-v2.provider-session-release-failed", {
790 providerSessionId,
791 cause,
792 }),
793 ),
794 );
795
796 // Removes the live entry and reads the request cleanup cutoff while
797 // holding the entry's request permit. A request the event pump is
798 // persisting for this runtime lands before the cutoff, and once the
799 // entry is gone the pump persists no more for it. A replacement's
800 // requests come after its own open.
801 const removeLiveEntry = (input: {
802 readonly providerSessionId: ProviderSessionId;
803 readonly onlyIfIdleGeneration?: number;
804 }): Effect.Effect<readonly [Option.Option<LiveSessionEntry>, DateTime.Utc]> =>
805 Effect.gen(function* () {
806 const key = sessionKey(input.providerSessionId);
807 const candidate = (yield* Ref.get(sessions)).get(key);
808 if (candidate === undefined) {
809 return [Option.none<LiveSessionEntry>(), yield* DateTime.now] as const;
810 }
811 const removed = yield* Effect.zip(
812 Ref.modify(sessions, (current) => {
813 const existing = current.get(key);
814 if (existing !== candidate) {
815 return [existing === undefined ? "gone" : "changed", current] as const;
816 }
817 if (
818 input.onlyIfIdleGeneration !== undefined &&
819 (existing.busyCount > 0 || existing.idleGeneration !== input.onlyIfIdleGeneration)
820 ) {
821 return ["kept", current] as const;
822 }
823 const updated = new Map(current);
824 updated.delete(key);
825 return ["removed", updated] as const;
826 }),
827 DateTime.now,
828 ).pipe(candidate.requestEventPermit.withPermits(1));
829 const [outcome, releasedAt] = removed;
830 // Another entry took this id while the permit was held; release it instead.
831 if (outcome === "changed") return yield* removeLiveEntry(input);
832 return [
833 outcome === "removed" ? Option.some(candidate) : Option.none<LiveSessionEntry>(),
834 releasedAt,
835 ] as const;
836 });
837
711838 const releaseEntry = (input: {
712839 readonly providerSessionId: ProviderSessionId;
713840 readonly reason: ProviderSessionReleaseReason;
......@@ -717,23 +844,8 @@ export const layerWithOptions = (
717844 readonly gracefulSubscribers?: boolean;
718845 }) =>
719846 Effect.acquireUseRelease(
720 Ref.modify(sessions, (current) => {
721 const key = sessionKey(input.providerSessionId);
722 const existing = current.get(key);
723 if (existing === undefined) {
724 return [Option.none<LiveSessionEntry>(), current] as const;
725 }
726 if (
727 input.onlyIfIdleGeneration !== undefined &&
728 (existing.busyCount > 0 || existing.idleGeneration !== input.onlyIfIdleGeneration)
729 ) {
730 return [Option.none<LiveSessionEntry>(), current] as const;
731 }
732 const updated = new Map(current);
733 updated.delete(key);
734 return [Option.some(existing), updated] as const;
735 }),
736 (entry) =>
847 removeLiveEntry(input),
848 ([entry, releasedAt]) =>
737849 Option.match(entry, {
738850 onNone: () => Effect.void,
739851 onSome: (entry) =>
......@@ -794,21 +906,25 @@ export const layerWithOptions = (
794906 Effect.forkDetach,
795907 );
796908 }
797 yield* writeReleasedSessionEvents({
909 const records = {
798910 entry,
799911 reason: input.reason,
800912 ...(input.detail === undefined ? {} : { detail: input.detail }),
801 });
802 yield* writeReleasedRuntimeRequestEvents({
803 entry,
804 reason: input.reason,
805 }).pipe(entry.requestEventPermit.withPermits(1));
913 releasedAt,
914 };
915 const recorded = yield* Effect.exit(
916 writeReleaseRecords({ ...records, replaced: false }),
917 );
918 if (Exit.isFailure(recorded)) {
919 yield* retryReleaseRecords(records);
920 return yield* recorded;
921 }
806922 if (Option.isSome(closeExit) && Exit.isFailure(closeExit.value)) {
807923 return yield* Effect.failCause(closeExit.value.cause);
808924 }
809925 }),
810926 }),
811 (entry) =>
927 ([entry]) =>
812928 Option.match(entry, {
813929 onNone: () => Effect.void,
814930 onSome: (entry) =>
......@@ -1495,7 +1611,7 @@ export const layerWithOptions = (
14951611 providerSessionId: entry.runtime.providerSessionId,
14961612 reason: "manual_shutdown",
14971613 gracefulSubscribers: true,
1498 }).pipe(Effect.ignore);
1614 }).pipe(logReleaseFailure(entry.runtime.providerSessionId));
14991615 return;
15001616 }
15011617 const cause = Exit.isFailure(exit)
......@@ -1516,7 +1632,7 @@ export const layerWithOptions = (
15161632 providerSessionId: entry.runtime.providerSessionId,
15171633 reason: "runtime_error",
15181634 detail: Cause.pretty(cause),
1519 }).pipe(Effect.ignore);
1635 }).pipe(logReleaseFailure(entry.runtime.providerSessionId));
15201636 }),
15211637 ),
15221638 Effect.forkIn(layerScope),
......@@ -1705,7 +1821,7 @@ export const layerWithOptions = (
17051821 providerSessionId: input.providerSessionId,
17061822 reason: "runtime_error",
17071823 detail: "Failed to persist the provider-session attachment.",
1708 }).pipe(Effect.ignore),
1824 }).pipe(logReleaseFailure(input.providerSessionId)),
17091825 ),
17101826 );
17111827 yield* startEventPump(entry);
apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts+56-3
......@@ -30,6 +30,7 @@ import * as ProviderAuthService from "../provider/Services/ProviderAuthService.t
3030import * as ContextHandoffService from "./ContextHandoffService.ts";
3131import * as EventSink from "./EventSink.ts";
3232import * as IdAllocator from "./IdAllocator.ts";
33import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts";
3334import * as ProjectionStore from "./ProjectionStore.ts";
3435import { ProviderAdapterEventStreamError } from "./ProviderAdapter.ts";
3536import * as ProviderSessionManager from "./ProviderSessionManager.ts";
......@@ -168,6 +169,8 @@ function makeLocalCommandHarness(input: {
168169 readonly interruptOpen?: boolean;
169170 readonly interruptRunBeforeOpenFailure?: boolean;
170171 readonly writeFailure?: unknown;
172 /** Loads the thread and starts the run, then fails every later state read. */
173 readonly failReadsAfterRunning?: boolean;
171174}) {
172175 const now = DateTime.makeUnsafe("2026-09-04T12:00:00Z");
173176 const threadId = ThreadId.make("thread-native-account-command");
......@@ -411,9 +414,40 @@ function makeLocalCommandHarness(input: {
411414 ),
412415 ),
413416 )
414 : Effect.die("A local command must not open a native session."),
417 : input.failReadsAfterRunning === true
418 ? Effect.succeed({
419 driver: providerThread.driver,
420 providerSession: {
421 id: providerSessionId,
422 driver: providerThread.driver,
423 providerInstanceId: newInstanceId,
424 status: "ready",
425 cwd: "/tmp/native-account-command",
426 model: null,
427 capabilities: CodexProviderCapabilitiesV2,
428 createdAt: now,
429 updatedAt: now,
430 lastError: null,
431 },
432 ensureThread: () => Effect.succeed(providerThread),
433 } as never)
434 : Effect.die("A local command must not open a native session."),
435 );
436 const startRootRun = vi.fn<
437 (input: RunExecutionService.RunExecutionServiceV2StartRootRunInput) => Effect.Effect<void>
438 >(() =>
439 input.failReadsAfterRunning === true
440 ? Effect.void
441 : Effect.die("A local command must not start a native turn."),
442 );
443 const failReadIfRunning = Effect.suspend(() =>
444 input.failReadsAfterRunning === true &&
445 projection.runs.find((candidate) => candidate.id === runId)?.status === "running"
446 ? Effect.fail(
447 new ProjectionStore.ProjectionStoreReadError({ threadId, cause: "database unavailable" }),
448 )
449 : Effect.void,
415450 );
416 const startRootRun = vi.fn(() => Effect.die("A local command must not start a native turn."));
417451 const tryHandlePromptCommand = vi.fn(() =>
418452 input.logoutFailure === undefined
419453 ? Effect.succeed(true)
......@@ -471,7 +505,7 @@ function makeLocalCommandHarness(input: {
471505 ),
472506 }),
473507 getRuntimeRecoveryProjection: () =>
474 Effect.succeed({
508 Effect.as(failReadIfRunning, {
475509 ...projection,
476510 hasConversation: projection.messages.some(
477511 (m) =>
......@@ -703,6 +737,25 @@ effectIt.effect(
703737 }),
704738);
705739
740effectIt.effect("does not mistake a failed state read for a superseded run", () =>
741 Effect.gen(function* () {
742 const harness = makeLocalCommandHarness({ text: "Continue", failReadsAfterRunning: true });
743
744 yield* harness.start;
745
746 expect(harness.projection().runs.at(-1)?.status).toBe("running");
747 const controls = harness.startRootRun.mock.calls[0]?.[0];
748 expect(controls).toBeDefined();
749 if (controls === undefined) return;
750 // "false" would skip the provider turn or the terminal write and leave the
751 // run active. A read failure must reach the caller instead.
752 const startCheck = yield* Effect.flip(controls.shouldStartProviderTurn!());
753 const finalizeCheck = yield* Effect.flip(controls.shouldFinalizeRun!());
754 expect(startCheck._tag).toBe("ProjectionStoreReadError");
755 expect(finalizeCheck._tag).toBe("ProjectionStoreReadError");
756 }),
757);
758
706759effectIt.effect("does not overwrite a run interrupted while its thread loads", () =>
707760 Effect.gen(function* () {
708761 const harness = makeLocalCommandHarness({
apps/server/src/orchestration-v2/ProviderTurnStartService.ts+2-2
......@@ -124,13 +124,14 @@ export const layer: Layer.Layer<
124124 }) => {
125125 // Guards and background routing need live execution state, not a fresh
126126 // allocation of every completed message and tool output in the thread.
127 // `false` means the run moved on or is gone. A failed read is an error,
128 // so the caller fails the start or the run instead of skipping it.
127129 const isCurrentAttemptInStatus = (expectedStatus: OrchestrationV2Run["status"]) =>
128130 projectionStore.getRuntimeRecoveryProjection(input.threadId).pipe(
129131 Effect.map((current) => {
130132 const run = current.runs.find((candidate) => candidate.id === input.runId);
131133 return run?.activeAttemptId === input.attemptId && run.status === expectedStatus;
132134 }),
133 Effect.catchCause(() => Effect.succeed(false)),
134135 );
135136 return {
136137 isCurrentAttemptInStatus,
......@@ -157,7 +158,6 @@ export const layer: Layer.Layer<
157158 (run.status === "starting" || run.status === "running")
158159 );
159160 }),
160 Effect.catchCause(() => Effect.succeed(false)),
161161 ),
162162 hasUnpairedRunInterruptRequest: () =>
163163 projectionStore
apps/server/src/orchestration-v2/RunExecutionService.test.ts+144-17
......@@ -50,6 +50,7 @@ import {
5050 type ProviderAdapterV2Event,
5151 type ProviderAdapterV2SessionRuntime,
5252} from "./ProviderAdapter.ts";
53import * as ProjectionStore from "./ProjectionStore.ts";
5354import * as ProviderEventIngestor from "./ProviderEventIngestor.ts";
5455import * as RunExecutionService from "./RunExecutionService.ts";
5556import * as RunFinalizationService from "./RunFinalizationService.ts";
......@@ -590,6 +591,113 @@ it.effect("rechecks run ownership immediately before calling the provider", () =
590591 }).pipe(Effect.provide(RunExecutionTestLayer)),
591592);
592593
594it.effect("fails the run when its ownership check cannot be read before calling the provider", () =>
595 Effect.gen(function* () {
596 const guardCalls = yield* Ref.make(0);
597 const providerStarts = yield* Ref.make(0);
598 const writes = yield* Ref.make<ReadonlyArray<OrchestrationV2DomainEvent>>([]);
599 const threadId = ThreadId.make("thread:run-execution-start-guard-read");
600 const runId = RunId.make("run:run-execution-start-guard-read");
601 const attemptId = RunAttemptId.make("attempt:run-execution-start-guard-read");
602 const providerInstanceId = ProviderInstanceId.make("codex");
603 const testLayer = RunExecutionService.layer.pipe(
604 Layer.provide(
605 Layer.mergeAll(
606 Layer.mock(CheckpointService.CheckpointServiceV2)({ captureBaseline: () => Effect.void }),
607 Layer.mock(EventSink.EventSinkV2)({
608 writeIfRunCurrent: (input) =>
609 Ref.update(writes, (current) => [...current, ...input.events]).pipe(
610 Effect.as({ committed: true, storedEvents: [] }),
611 ),
612 }),
613 IdAllocator.layer,
614 Layer.mock(ProviderEventIngestor.ProviderEventIngestorV2)({
615 ingestNormalized: () => Effect.succeed([]),
616 }),
617 ServerSettings.layerTest(),
618 ),
619 ),
620 );
621
622 yield* Effect.gen(function* () {
623 const runExecution = yield* RunExecutionService.RunExecutionServiceV2;
624 yield* runExecution.startRootRun({
625 commandId: CommandId.make("command:run-execution-start-guard-read"),
626 appThread: { id: threadId } as OrchestrationV2AppThread,
627 providerSessionId: ProviderSessionId.make("session:run-execution-start-guard-read"),
628 session: {
629 events: Stream.never,
630 startTurn: () => Ref.update(providerStarts, (count) => count + 1),
631 } as unknown as ProviderAdapterV2SessionRuntime,
632 run: { id: runId, threadId, ordinal: 1, providerInstanceId } as OrchestrationV2Run,
633 rootNode: {
634 id: NodeId.make("node:run-execution-start-guard-read"),
635 } as OrchestrationV2ExecutionNode,
636 checkpointScope: {
637 id: CheckpointScopeId.make("checkpoint-scope:run-execution-start-guard-read"),
638 } as OrchestrationV2CheckpointScope,
639 providerThread: {
640 id: ProviderThreadId.make("provider-thread:run-execution-start-guard-read"),
641 driver,
642 } as OrchestrationV2ProviderThread,
643 attempt: { id: attemptId, providerTurnId: null } as OrchestrationV2RunAttempt,
644 attemptId,
645 providerTurnOrdinal: 1,
646 // The preparation check passes; the check right before the provider
647 // call cannot read the run.
648 shouldStartProviderTurn: () =>
649 Ref.getAndUpdate(guardCalls, (calls) => calls + 1).pipe(
650 Effect.flatMap((calls) =>
651 calls === 0
652 ? Effect.succeed(true)
653 : Effect.fail(
654 new ProjectionStore.ProjectionStoreReadError({
655 threadId,
656 cause: "database unavailable",
657 }),
658 ),
659 ),
660 ),
661 // The failure is settled by the guarded write, not by another read.
662 shouldFinalizeRun: () =>
663 Effect.fail(
664 new ProjectionStore.ProjectionStoreReadError({
665 threadId,
666 cause: "database unavailable",
667 }),
668 ),
669 message: {
670 messageId: MessageId.make("message:run-execution-start-guard-read"),
671 text: "Start while the store is down.",
672 attachments: [],
673 createdBy: "user",
674 creationSource: "web",
675 },
676 modelSelection: { instanceId: providerInstanceId, model: "gpt-5.4" },
677 runtimePolicy: {
678 runtimeMode: "full-access",
679 interactionMode: "default",
680 cwd: process.cwd(),
681 approvalPolicy: "never",
682 sandboxPolicy: {
683 type: "readOnly",
684 access: { type: "fullAccess" },
685 networkAccess: false,
686 },
687 },
688 });
689 }).pipe(Effect.provide(testLayer));
690
691 assert.equal(yield* Ref.get(guardCalls), 2);
692 assert.equal(yield* Ref.get(providerStarts), 0);
693 const runUpdate = (yield* Ref.get(writes)).find((event) => event.type === "run.updated");
694 assert.equal(
695 runUpdate?.type === "run.updated" ? runUpdate.payload.status : undefined,
696 "failed",
697 );
698 }),
699);
700
593701it.effect(
594702 "dispatches only attachment-free compact commands through the native compaction path",
595703 () =>
......@@ -3165,6 +3273,25 @@ it.effect.each(["completed", "interrupted", "cancelled", "failed"] as const)(
31653273 }),
31663274);
31673275
3276it.effect("records a finished run as failed when its ownership check cannot be read", () =>
3277 Effect.gen(function* () {
3278 const { observed } = yield* captureRootRunTermination({
3279 key: "finalize-guard-read-failure",
3280 shouldFinalizeRun: () =>
3281 Effect.fail(
3282 new ProjectionStore.ProjectionStoreReadError({
3283 threadId: ThreadId.make("thread:finalize-guard-read-failure"),
3284 cause: "database unavailable",
3285 }),
3286 ),
3287 events: (ids) => Stream.make(rootTerminalEvent(ids, "completed")),
3288 });
3289 // The fallback settles through the guarded write instead of the same
3290 // failing read, so the run does not stay running.
3291 assert.include(observed, "run:failed");
3292 }),
3293);
3294
31683295it.effect("does not refresh pull requests for auxiliary or stale provider terminals", () =>
31693296 Effect.gen(function* () {
31703297 const { observed } = yield* captureRootRunTermination({
......@@ -3254,7 +3381,7 @@ it.effect("keeps completed runs completed when pull request refresh fails", () =
32543381
32553382function captureRootRunTermination(input: {
32563383 readonly key: string;
3257 readonly shouldFinalizeRun: () => Effect.Effect<boolean, never>;
3384 readonly shouldFinalizeRun: () => Effect.Effect<boolean, ProjectionStore.ProjectionStoreV2Error>;
32583385 readonly hasUnpairedRunInterruptRequest?: () => Effect.Effect<boolean, never>;
32593386 readonly seedOpenSubagent?: boolean;
32603387 readonly events?: (
......@@ -3278,6 +3405,17 @@ function captureRootRunTermination(input: {
32783405 const ingestionDone = yield* Deferred.make<void>();
32793406 const captureTurnItem = (payload: OrchestrationV2TurnItem) =>
32803407 Ref.update(writtenItems, (current) => [...current, payload]);
3408 const captureFinalEvents = (events: ReadonlyArray<OrchestrationV2DomainEvent>) =>
3409 Effect.gen(function* () {
3410 for (const event of events) {
3411 if (event.type === "turn-item.updated") {
3412 yield* captureTurnItem(event.payload);
3413 }
3414 if (event.type === "run.updated") {
3415 yield* Ref.update(observed, (current) => [...current, `run:${event.payload.status}`]);
3416 }
3417 }
3418 });
32813419 const testLayer = RunExecutionService.layer.pipe(
32823420 Layer.provide(
32833421 Layer.mergeAll(
......@@ -3292,22 +3430,11 @@ function captureRootRunTermination(input: {
32923430 }
32933431 return [];
32943432 }),
3295 writeWithEffects: (payload) =>
3296 Effect.gen(function* () {
3297 for (const event of payload.events) {
3298 if (event.type === "turn-item.updated") {
3299 yield* captureTurnItem(event.payload);
3300 }
3301 if (event.type === "run.updated") {
3302 yield* Ref.update(observed, (current) => [
3303 ...current,
3304 `run:${event.payload.status}`,
3305 ]);
3306 }
3307 }
3308 return [];
3309 }),
3310 writeIfRunCurrent: () => Effect.succeed({ committed: true, storedEvents: [] }),
3433 writeWithEffects: (payload) => captureFinalEvents(payload.events).pipe(Effect.as([])),
3434 writeIfRunCurrent: (payload) =>
3435 captureFinalEvents(payload.events).pipe(
3436 Effect.as({ committed: true, storedEvents: [] }),
3437 ),
33113438 }),
33123439 IdAllocator.layer,
33133440 Layer.mock(ProviderEventIngestor.ProviderEventIngestorV2)({
apps/server/src/orchestration-v2/RunExecutionService.ts+27-27
......@@ -28,6 +28,7 @@ import * as Context from "effect/Context";
2828import * as Cause from "effect/Cause";
2929import * as DateTime from "effect/DateTime";
3030import * as Effect from "effect/Effect";
31import * as Exit from "effect/Exit";
3132import * as Fiber from "effect/Fiber";
3233import * as Layer from "effect/Layer";
3334import * as Ref from "effect/Ref";
......@@ -47,6 +48,7 @@ import type {
4748} from "./ProviderAdapter.ts";
4849import { ProviderAdapterTurnStartError } from "./ProviderAdapter.ts";
4950import * as ProviderEventIngestor from "./ProviderEventIngestor.ts";
51import type { ProjectionStoreV2Error } from "./ProjectionStore.ts";
5052import { makeProviderFailure, makeProviderFailureTurnItem } from "./ProviderFailure.ts";
5153import * as RunFinalizationService from "./RunFinalizationService.ts";
5254
......@@ -511,8 +513,8 @@ export interface RunExecutionServiceV2StartRootRunInput {
511513 >;
512514 readonly relatedThreadIds?: ReadonlyArray<ThreadId>;
513515 readonly relatedProviderThreadIds?: ReadonlyArray<ProviderThreadId>;
514 readonly shouldStartProviderTurn?: () => Effect.Effect<boolean, never>;
515 readonly shouldFinalizeRun?: () => Effect.Effect<boolean, never>;
516 readonly shouldStartProviderTurn?: () => Effect.Effect<boolean, ProjectionStoreV2Error>;
517 readonly shouldFinalizeRun?: () => Effect.Effect<boolean, ProjectionStoreV2Error>;
516518 readonly hasUnpairedRunInterruptRequest?: () => Effect.Effect<boolean, never>;
517519 readonly message: ProviderAdapterV2TurnMessage;
518520 readonly modelSelection: ModelSelection;
......@@ -557,7 +559,7 @@ export const layer: Layer.Layer<
557559 readonly checkpointScope: OrchestrationV2CheckpointScope;
558560 readonly providerThread: OrchestrationV2ProviderThread;
559561 readonly attempt: OrchestrationV2RunAttempt;
560 readonly shouldFinalizeRun?: () => Effect.Effect<boolean, never>;
562 readonly shouldFinalizeRun?: () => Effect.Effect<boolean, ProjectionStoreV2Error>;
561563 readonly hasUnpairedRunInterruptRequest?: () => Effect.Effect<boolean, never>;
562564 readonly openRunOwnedSubagents?: OpenRunOwnedSubagentProjection;
563565 readonly terminal: ProviderTerminalEvent;
......@@ -1286,15 +1288,12 @@ export const layer: Layer.Layer<
12861288 checkpointScope: input.checkpointScope,
12871289 providerThread,
12881290 attempt: input.attempt,
1289 ...(input.shouldFinalizeRun === undefined
1290 ? {}
1291 : { shouldFinalizeRun: input.shouldFinalizeRun }),
1292 ...(input.hasUnpairedRunInterruptRequest === undefined
1293 ? {}
1294 : {
1295 hasUnpairedRunInterruptRequest:
1296 input.hasUnpairedRunInterruptRequest,
1297 }),
1291 // The failure may be the ownership
1292 // read itself, so check in the write.
1293 writeIfRunCurrent: {
1294 activeAttemptId: input.attempt.id,
1295 expectedStatus: "running",
1296 },
12981297 openRunOwnedSubagents: openSubagents,
12991298 terminal: makeFailedTerminalEvent(
13001299 makeProviderFailure({
......@@ -1328,10 +1327,13 @@ export const layer: Layer.Layer<
13281327 Effect.forkDetach,
13291328 );
13301329
1331 if (
1332 input.shouldStartProviderTurn !== undefined &&
1333 !(yield* input.shouldStartProviderTurn())
1334 ) {
1330 // A failed read fails the start below, so the run is recorded as
1331 // failed instead of staying active with no provider turn.
1332 const shouldStart =
1333 input.shouldStartProviderTurn === undefined
1334 ? Exit.succeed(true)
1335 : yield* Effect.exit(input.shouldStartProviderTurn());
1336 if (Exit.isSuccess(shouldStart) && !shouldStart.value) {
13351337 yield* Fiber.interrupt(providerEventFiber);
13361338 return;
13371339 }
......@@ -1373,7 +1375,7 @@ export const layer: Layer.Layer<
13731375 }),
13741376 ))
13751377 : input.session.startTurn(turnInput);
1376 yield* startTurn.pipe(
1378 yield* Effect.andThen(shouldStart, startTurn).pipe(
13771379 Effect.catchCause((cause) =>
13781380 Effect.logError("orchestration V2 provider turn start failed", {
13791381 runId: input.run.id,
......@@ -1392,20 +1394,18 @@ export const layer: Layer.Layer<
13921394 checkpointScope: input.checkpointScope,
13931395 providerThread,
13941396 attempt: input.attempt,
1395 ...(input.shouldFinalizeRun === undefined
1396 ? {}
1397 : { shouldFinalizeRun: input.shouldFinalizeRun }),
1398 ...(input.hasUnpairedRunInterruptRequest === undefined
1399 ? {}
1400 : {
1401 hasUnpairedRunInterruptRequest:
1402 input.hasUnpairedRunInterruptRequest,
1403 }),
1397 // Checked in the write transaction, not by another
1398 // read that can fail like the one before the start.
1399 writeIfRunCurrent: {
1400 activeAttemptId: input.attempt.id,
1401 expectedStatus: "running",
1402 },
14041403 openRunOwnedSubagents: openSubagents,
14051404 terminal: makeFailedTerminalEvent(
14061405 makeProviderFailure({
14071406 cause: Cause.squash(cause),
1408 class: "provider_error",
1407 // A failed ownership read is not the provider's fault.
1408 class: Exit.isFailure(shouldStart) ? "unknown" : "provider_error",
14091409 }),
14101410 latestItemOrdinal + 1,
14111411 ),