From 6d528a01d4e89878d1f7d9ed69c8ad56038bda96 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Mon, 26 Oct 2026 19:10:56 +0000 Subject: [PATCH 1/4] fix(server): preserve the crash recovery budget across restarts Co-authored-by: Tin Zeljar <35663757+Zheruel@users.noreply.github.com> --- .../src/orchestration-v2/EffectOutbox.ts | 12 +++ .../src/orchestration-v2/EffectWorker.test.ts | 59 +++++++++++++ .../src/orchestration-v2/EffectWorker.ts | 8 +- .../FoundationPersistence.test.ts | 29 +++++++ .../ProviderRuntimeRecoveryService.ts | 35 +++++++- .../RestartContinuation.test.ts | 87 ++++++++++++++++++- .../orchestration-v2/RestartContinuation.ts | 12 ++- docs/internals/server-updates.md | 7 +- docs/user/updating.md | 6 +- 9 files changed, 243 insertions(+), 12 deletions(-) diff --git a/apps/server/src/orchestration-v2/EffectOutbox.ts b/apps/server/src/orchestration-v2/EffectOutbox.ts index 92e1e01ad43f2..501f049d458a1 100644 --- a/apps/server/src/orchestration-v2/EffectOutbox.ts +++ b/apps/server/src/orchestration-v2/EffectOutbox.ts @@ -33,6 +33,7 @@ export const OrchestrationEffectRequestV2 = Schema.Union([ Schema.Struct({ type: Schema.Literal("provider-runtime.continue"), sourceRunId: RunId, + processLossCount: Schema.optional(Schema.Int.check(Schema.isGreaterThanOrEqualTo(0))), }), Schema.Struct({ type: Schema.Literal("provider-session.detach"), @@ -669,6 +670,7 @@ export const layer: Layer.Layer = La const cutoff = DateTime.formatIso( DateTime.subtractDuration(yield* DateTime.now, SETTLED_EFFECT_RETENTION), ); + // Live continuations retain their preceding intent's cumulative crash budget. // Separate batch statements release the write lock between batches. const deleteBatch = sql<{ readonly effect_id: string }>` DELETE FROM orchestration_v2_effect_outbox @@ -677,6 +679,16 @@ export const layer: Layer.Layer = La FROM orchestration_v2_effect_outbox WHERE status IN ('succeeded', 'cancelled') AND completed_at < ${cutoff} + AND NOT ( + effect_type = 'provider-runtime.continue' + AND EXISTS ( + SELECT 1 FROM orchestration_v2_projection_runs AS recovery_run + WHERE recovery_run.thread_id = orchestration_v2_effect_outbox.thread_id + AND recovery_run.status IN ('queued', 'preparing', 'starting', 'running', 'waiting') + AND json_extract(recovery_run.payload_json, '$.restartContinuationOfRunId') = + json_extract(orchestration_v2_effect_outbox.payload_json, '$.sourceRunId') + ) + ) LIMIT ${PRUNE_BATCH_SIZE} ) RETURNING effect_id diff --git a/apps/server/src/orchestration-v2/EffectWorker.test.ts b/apps/server/src/orchestration-v2/EffectWorker.test.ts index 2329f2aa20cf0..5a1bdf9998037 100644 --- a/apps/server/src/orchestration-v2/EffectWorker.test.ts +++ b/apps/server/src/orchestration-v2/EffectWorker.test.ts @@ -4,6 +4,8 @@ import { CheckpointScopeId, CommandId, NodeId, + ProjectId, + ProviderInstanceId, ProviderSessionId, ProviderThreadId, ProviderTurnId, @@ -12,6 +14,7 @@ import { RuntimeRequestId, ThreadId, VcsProcessExitError, + type OrchestrationV2ThreadProjection, } from "@t3tools/contracts"; import * as Cause from "effect/Cause"; import * as DateTime from "effect/DateTime"; @@ -635,3 +638,59 @@ it.effect.each(["confirmed", "unconfirmed", "failed"] as const)( ); }), ); + +it.effect.each([0, 3, 4])( + "enforces process-loss count %s after 22 minutes through the executor", + (processLossCount) => + Effect.gen(function* () { + const now = yield* DateTime.now; + const instanceId = ProviderInstanceId.make("codex"); + const events = yield* Ref.make>([]); + const layer = layerExecutorFor({ + events, + continueAfterRestart: true, + threads: { + getThreadRecords: () => + Effect.succeed({ + thread: { + id: threadId, + projectId: ProjectId.make("project:executor-recovery"), + providerInstanceId: instanceId, + archivedAt: null, + deletedAt: null, + }, + runs: [ + { + id: runId, + providerInstanceId: instanceId, + status: "cancelled", + requestedAt: DateTime.subtract(now, { minutes: 22 }), + completedAt: now, + ordinal: 1, + }, + ], + messages: [], + providerTurns: [], + runtimeRequests: [], + turnItems: [], + attempts: [], + } as unknown as OrchestrationV2ThreadProjection), + dispatch: () => + Ref.update(events, (entries) => [...entries, "dispatch"]).pipe(Effect.as({} as never)), + recoverDelegatedTask: () => + Ref.update(events, (entries) => [...entries, "recover-child"]), + }, + }); + yield* Effect.gen(function* () { + const executor = yield* EffectWorker.OrchestrationEffectExecutorV2; + yield* executor.execute({ + ...restartEffect(now, { type: "detach" }), + request: { type: "provider-runtime.continue", sourceRunId: runId, processLossCount }, + }); + }).pipe(Effect.provide(layer)); + assert.deepEqual( + yield* Ref.get(events), + processLossCount > 3 ? ["recover-child"] : ["dispatch", "recover-child"], + ); + }), +); diff --git a/apps/server/src/orchestration-v2/EffectWorker.ts b/apps/server/src/orchestration-v2/EffectWorker.ts index c5b5ea9b383a9..bbc9fbfab58d0 100644 --- a/apps/server/src/orchestration-v2/EffectWorker.ts +++ b/apps/server/src/orchestration-v2/EffectWorker.ts @@ -100,7 +100,13 @@ export const layerExecutor: Layer.Layer< switch (effect.request.type) { case "provider-runtime.continue": { const sourceRunId = effect.request.sourceRunId; - return continueRestartedRun({ threadId: effect.threadId, sourceRunId }).pipe( + return continueRestartedRun({ + threadId: effect.threadId, + sourceRunId, + ...(effect.request.processLossCount === undefined + ? {} + : { processLossCount: effect.request.processLossCount }), + }).pipe( Effect.provideService(ThreadManagementService.ThreadManagementService, threads), Effect.provideService(ServerSettings.ServerSettingsService, settings), Effect.tapError(() => diff --git a/apps/server/src/orchestration-v2/FoundationPersistence.test.ts b/apps/server/src/orchestration-v2/FoundationPersistence.test.ts index e1b2e7610663c..b30540727cb64 100644 --- a/apps/server/src/orchestration-v2/FoundationPersistence.test.ts +++ b/apps/server/src/orchestration-v2/FoundationPersistence.test.ts @@ -2510,6 +2510,34 @@ it.layer(layerTest)("orchestration V2 foundation persistence", (it) => { }).pipe(Effect.provide(layerIsolatedOutbox)), ); + it.effect.each(["queued", "preparing", "starting", "running", "waiting"])( + "keeps a %s continuation's crash budget past outbox retention", + (status) => + Effect.gen(function* () { + const outbox = yield* EffectOutbox.EffectOutboxV2; + const sql = yield* SqlClient.SqlClient; + const now = yield* DateTime.now; + const old = DateTime.formatIso(DateTime.subtractDuration(now, Duration.days(8))); + const timestamp = DateTime.formatIso(now); + yield* sql` + INSERT INTO orchestration_v2_effect_outbox + (effect_id, command_id, thread_id, effect_type, payload_json, status, available_at, created_at, updated_at, completed_at) + VALUES ('effect:retained-recovery', 'command:retained-recovery', 'thread:retained-recovery', + 'provider-runtime.continue', '{"type":"provider-runtime.continue","sourceRunId":"run:retained-source","processLossCount":3}', + 'succeeded', ${old}, ${old}, ${old}, ${old}) + `; + yield* sql` + INSERT INTO orchestration_v2_projection_runs + (run_id, thread_id, ordinal, provider, status, requested_at, payload_json) + VALUES ('run:retained-recovery', 'thread:retained-recovery', 1, 'codex', ${status}, ${timestamp}, + '{"restartContinuationOfRunId":"run:retained-source"}') + `; + assert.equal(yield* outbox.pruneSettled, 0); + yield* sql`UPDATE orchestration_v2_projection_runs SET status = 'completed' WHERE run_id = 'run:retained-recovery'`; + assert.equal(yield* outbox.pruneSettled, 1); + }).pipe(Effect.provide(layerIsolatedOutbox)), + ); + it.effect("wakes claimers when cancellation unblocks same-thread work", () => Effect.gen(function* () { const outbox = yield* EffectOutbox.EffectOutboxV2; @@ -2966,6 +2994,7 @@ it.layer(layerTest)("orchestration V2 foundation persistence", (it) => { assert.deepEqual(effects[0]?.request, { type: "provider-runtime.continue", sourceRunId: runId, + processLossCount: 0, }); yield* outbox.cancelUnsettled({ threadId, diff --git a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts index c374db2abfb97..8f06cc0c8120e 100644 --- a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts +++ b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts @@ -16,6 +16,7 @@ import * as Context from "effect/Context"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; import * as Schema from "effect/Schema"; import * as EffectOutbox from "./EffectOutbox.ts"; @@ -748,13 +749,39 @@ export const make = Effect.gen(function* () { now, ) : undefined; + let processLossCount = 1; + // Normal upgrades must not exhaust an unfinished turn's crash allowance. + if ( + continuationRun?.restartContinuationOfRunId !== undefined && + continuationRun.status !== "completed" && + continuationRun.status !== "waiting" + ) { + const previous = yield* outbox + .get(`effect:restart-continuation:${continuationRun.restartContinuationOfRunId}`) + .pipe( + Effect.mapError( + (cause) => + new ProviderRuntimeRecoveryError({ + operation: "reconcile", + threadId: projection.thread.id, + cause, + }), + ), + ); + if (Option.isSome(previous) && previous.value.request.type === "provider-runtime.continue") + processLossCount += previous.value.request.processLossCount ?? 0; + } const effects: Array = continuationRun ? [ { id: `effect:restart-continuation:${continuationRun.id}`, commandId, threadId: projection.thread.id, - request: { type: "provider-runtime.continue", sourceRunId: continuationRun.id }, + request: { + type: "provider-runtime.continue", + sourceRunId: continuationRun.id, + processLossCount, + }, }, ] : []; @@ -909,7 +936,11 @@ export const make = Effect.gen(function* () { id: `effect:restart-continuation:${run.id}`, commandId, threadId, - request: { type: "provider-runtime.continue", sourceRunId: run.id }, + request: { + type: "provider-runtime.continue", + sourceRunId: run.id, + processLossCount: 0, + }, }, ], }); diff --git a/apps/server/src/orchestration-v2/RestartContinuation.test.ts b/apps/server/src/orchestration-v2/RestartContinuation.test.ts index d2f21a772ac99..ab12e5392069b 100644 --- a/apps/server/src/orchestration-v2/RestartContinuation.test.ts +++ b/apps/server/src/orchestration-v2/RestartContinuation.test.ts @@ -20,6 +20,7 @@ import * as Effect from "effect/Effect"; import * as Deferred from "effect/Deferred"; import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; import * as ServerSettings from "../serverSettings.ts"; import { restartContinuationRun, continueRestartedRun } from "./RestartContinuation.ts"; import { continuationStartRetryDelayMs } from "./RestartBackgroundNote.ts"; @@ -293,6 +294,7 @@ it.effect.each([ assert.deepEqual(committed!.effects[0]?.request, { type: "provider-runtime.continue", sourceRunId: runId, + processLossCount: 1, }); }), ); @@ -425,7 +427,7 @@ it.effect.each(["running", "waiting"])( ]); assert.deepEqual( committed?.effects.map((effect) => effect.request), - [{ type: "provider-runtime.continue", sourceRunId: runId }], + [{ type: "provider-runtime.continue", sourceRunId: runId, processLossCount: 1 }], ); }), ); @@ -563,6 +565,7 @@ it.effect("prepares no continuation for background work another provider thread assert.deepEqual(writes[0]!.effects[0]?.request, { type: "provider-runtime.continue", sourceRunId: runId, + processLossCount: 0, }); }), ); @@ -668,7 +671,15 @@ it.effect.each([ const recorded = [...writes.flatMap((write) => write.effects), ...commits[0]!.effects]; assert.deepEqual( recorded.map((effect) => effect.request), - wakes ? [{ type: "provider-runtime.continue", sourceRunId: runId }] : [], + wakes + ? [ + { + type: "provider-runtime.continue", + sourceRunId: runId, + processLossCount: trigger === "shutdown" ? 0 : 1, + }, + ] + : [], ); assert.lengthOf(commits, 1); const events = commits[0]!.events; @@ -868,6 +879,76 @@ const continuationDispatches = ( return dispatches; }); +it.effect.each([ + [0, "running"], + [3, "running"], + [3, "completed"], + [3, "waiting"], +] as const)("persists recovery after count %s and %s work", ([previousCount, status]) => + Effect.gen(function* () { + const base = makeProjection(); + const source = RunId.make("run:previous-loss"); + const projection = { + ...base, + runs: [ + { + ...base.runs[0]!, + status, + restartContinuationOfRunId: source, + requestedAt: DateTime.subtract(now, { minutes: 22 }), + }, + ], + providerTurns: [ + { ...base.providerTurns[0]!, status: status === "running" ? "running" : "completed" }, + ], + turnItems: status === "running" ? [] : [backgroundSubagent.turnItem], + } as unknown as OrchestrationV2ThreadProjection; + const commits: Parameters[0][] = []; + const recovery = yield* ProviderRuntimeRecovery.make.pipe( + Effect.provide( + Layer.mergeAll( + ServerSettings.layerTest({ continueThreadsAfterServerUpdate: true }), + Layer.mock(ProjectionStore.ProjectionStoreV2)({ + getRecoveryThreadIds: () => Effect.succeed([threadId]), + getRuntimeRecoveryProjection: () => Effect.succeed(projection), + }), + Layer.mock(EventSink.EventSinkV2)({ + commitCommand: (input) => { + commits.push(input); + return Effect.succeed({ committed: true, cancelledEffectCount: 0 } as never); + }, + }), + IdAllocator.layer, + Layer.mock(EffectWorker.OrchestrationEffectWorkerV2)({}), + Layer.mock(EffectOutbox.EffectOutboxV2)({ + get: (id) => { + return Effect.succeed( + id === `effect:restart-continuation:${source}` + ? Option.some({ + request: { + type: "provider-runtime.continue", + sourceRunId: source, + processLossCount: previousCount, + }, + } as EffectOutbox.OrchestrationEffectV2) + : Option.none(), + ); + }, + listByCommandId: () => Effect.succeed([]), + reconcileAfterProcessLoss: Effect.succeed({ requeued: 0, cancelled: 0 }), + }), + ), + ), + ); + yield* recovery.reconcile("startup"); + assert.deepEqual(commits[0]?.effects[0]?.request, { + type: "provider-runtime.continue", + sourceRunId: runId, + processLossCount: status === "running" ? previousCount + 1 : 1, + }); + }), +); + const continuationTexts = (projection: OrchestrationV2ThreadProjection) => continuationDispatches(projection).pipe( Effect.map((dispatches) => dispatches.map((command) => command.text)), @@ -1165,7 +1246,7 @@ it.effect("prepares later threads' continuations when one thread fails", () => yield* recovery.prepareForShutdown; assert.deepEqual( writes.map((write) => write.effects[0]?.request), - [{ type: "provider-runtime.continue", sourceRunId: runId }], + [{ type: "provider-runtime.continue", sourceRunId: runId, processLossCount: 0 }], ); }), ); diff --git a/apps/server/src/orchestration-v2/RestartContinuation.ts b/apps/server/src/orchestration-v2/RestartContinuation.ts index 20d110d335aa2..e5d0b8ae4f6d1 100644 --- a/apps/server/src/orchestration-v2/RestartContinuation.ts +++ b/apps/server/src/orchestration-v2/RestartContinuation.ts @@ -183,7 +183,11 @@ export function restartContinuationRun( } export const continueRestartedRun = Effect.fn("RestartContinuation.continueRestartedRun")( - function* (input: { readonly threadId: ThreadId; readonly sourceRunId: RunId }) { + function* (input: { + readonly threadId: ThreadId; + readonly sourceRunId: RunId; + readonly processLossCount?: number; + }) { const settings = yield* ServerSettings.ServerSettingsService; const enabled = yield* settings.getSettings.pipe(Effect.orElseSucceed(() => null)); if (!enabled) return; @@ -239,7 +243,11 @@ export const continueRestartedRun = Effect.fn("RestartContinuation.continueResta if (DateTime.toEpochMillis(run.requestedAt) < loopWindowStart) break; chainLength += 1; } - if (chainLength >= MAX_CONTINUATION_CHAIN) { + if ( + input.processLossCount === undefined + ? chainLength >= MAX_CONTINUATION_CHAIN + : input.processLossCount > MAX_CONTINUATION_CHAIN + ) { yield* Effect.logWarning("Not continuing a run restarted too many times in a row", { threadId: input.threadId, sourceRunId: input.sourceRunId, diff --git a/docs/internals/server-updates.md b/docs/internals/server-updates.md index 1536884578d4b..0dbca531908e6 100644 --- a/docs/internals/server-updates.md +++ b/docs/internals/server-updates.md @@ -91,8 +91,11 @@ and maintenance turns such as `/compact` before dispatching. Intent and delivery refuse a thread that is archived, settled, or still snoozed, a cut turn whose provider turn already ended, and a run waiting on the user: an approval, question, or secret that was pending, or that recovery closed on the cut turn. Message-capable questions stay answerable, so -they do not block. Delivery also stops after three consecutive continuations in one -`restartContinuationOfRunId` chain, so a turn that crashes the server cannot loop. Stable +they do not block. New continuation intents persist a cumulative process-loss count; +delivery stops before the fourth crash recovery of an unfinished turn, regardless +of elapsed time. Planned shutdowns and completed turns reset it. A live continuation +retains its preceding intent past outbox retention so cleanup cannot erase the count. +Legacy intents without a count keep the five-minute chain guard. Stable command and message IDs prevent duplicate submissions after an outbox retry. Codex resumes without adding provider prompt text unless the turn lost background work; other adapters receive the continuation message through their normal turn path. diff --git a/docs/user/updating.md b/docs/user/updating.md index 6f8b35945cf6b..7d51ca42c4078 100644 --- a/docs/user/updating.md +++ b/docs/user/updating.md @@ -77,8 +77,10 @@ command or monitor the agent said it was waiting on is woken and told that this stopped. Other background shells and monitors, such as a dev server and its watchdog, do not wake it; the thread learns about them with your next message. A thread is not resumed while it waits for your approval, answer, or secret, or -while it is archived, snoozed, or settled. After three automatic resumes in a row, -the thread waits for your next message. +while it is archived, snoozed, or settled. After three consecutive crash recoveries +of an unfinished turn, the thread waits for your next message, even when crashes +are more than five minutes apart. A completed turn or a planned shutdown resets +that limit. Updates from the previous orchestration system preserve conversation transcripts but cannot carry every kind of runtime history forward. Read [Threads from older T3 Code versions](./thread-migration.md) From 21c01ef34b45ffe307c248a7f8e32fbd95409cc4 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Fri, 6 Nov 2026 04:10:47 +0000 Subject: [PATCH 2/4] test(server): give the executor recovery fixture a thread lineage --- apps/server/src/orchestration-v2/EffectWorker.test.ts | 1 + 1 file changed, 1 insertion(+) diff --git a/apps/server/src/orchestration-v2/EffectWorker.test.ts b/apps/server/src/orchestration-v2/EffectWorker.test.ts index 5a1bdf9998037..11662d5d0a4e1 100644 --- a/apps/server/src/orchestration-v2/EffectWorker.test.ts +++ b/apps/server/src/orchestration-v2/EffectWorker.test.ts @@ -658,6 +658,7 @@ it.effect.each([0, 3, 4])( providerInstanceId: instanceId, archivedAt: null, deletedAt: null, + lineage: { parentThreadId: null, relationshipToParent: null }, }, runs: [ { From e112e6f18269594d30311ec6af71587b9cfcf5ed Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 14 Nov 2026 21:50:55 +0000 Subject: [PATCH 3/4] test(server): trim crash recovery budget tests to their boundaries Co-Authored-By: Claude Opus 5.5 --- apps/server/src/orchestration-v2/EffectWorker.test.ts | 2 +- apps/server/src/orchestration-v2/FoundationPersistence.test.ts | 2 +- apps/server/src/orchestration-v2/RestartContinuation.test.ts | 1 - 3 files changed, 2 insertions(+), 3 deletions(-) diff --git a/apps/server/src/orchestration-v2/EffectWorker.test.ts b/apps/server/src/orchestration-v2/EffectWorker.test.ts index f5bb61a920d4f..94444eb8e5c69 100644 --- a/apps/server/src/orchestration-v2/EffectWorker.test.ts +++ b/apps/server/src/orchestration-v2/EffectWorker.test.ts @@ -781,7 +781,7 @@ it.effect.each(["confirmed", "unconfirmed", "failed"] as const)( }), ); -it.effect.each([0, 3, 4])( +it.effect.each([3, 4])( "enforces process-loss count %s after 22 minutes through the executor", (processLossCount) => Effect.gen(function* () { diff --git a/apps/server/src/orchestration-v2/FoundationPersistence.test.ts b/apps/server/src/orchestration-v2/FoundationPersistence.test.ts index d437be009c2f5..3f00941c8daa3 100644 --- a/apps/server/src/orchestration-v2/FoundationPersistence.test.ts +++ b/apps/server/src/orchestration-v2/FoundationPersistence.test.ts @@ -2752,7 +2752,7 @@ it.layer(layerTest)("orchestration V2 foundation persistence", (it) => { }).pipe(Effect.provide(layerIsolatedOutbox)), ); - it.effect.each(["queued", "preparing", "starting", "running", "waiting"])( + it.effect.each(["queued", "waiting"])( "keeps a %s continuation's crash budget past outbox retention", (status) => Effect.gen(function* () { diff --git a/apps/server/src/orchestration-v2/RestartContinuation.test.ts b/apps/server/src/orchestration-v2/RestartContinuation.test.ts index ab12e5392069b..d333aad6626bd 100644 --- a/apps/server/src/orchestration-v2/RestartContinuation.test.ts +++ b/apps/server/src/orchestration-v2/RestartContinuation.test.ts @@ -880,7 +880,6 @@ const continuationDispatches = ( }); it.effect.each([ - [0, "running"], [3, "running"], [3, "completed"], [3, "waiting"], From 2432ccb54ed5380a506e348c8f040d15bbe54212 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 14 Nov 2026 23:09:51 +0000 Subject: [PATCH 4/4] fix(server): continuation start retries keep the crash recovery count A retried continuation start wrote its intent without processLossCount, so the next crash recovery restarted the count at 1 and a host that crashes under load could resume forever. The retry now copies the count from the intent of the run it continues. Co-Authored-By: Claude Opus 5.5 --- .../src/orchestration-v2/Orchestrator.ts | 17 +++++++++++++- .../src/orchestration-v2/runtimeLayer.test.ts | 23 +++++++++++++++++++ 2 files changed, 39 insertions(+), 1 deletion(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 226b7d703d0b1..5948d3fbf9a0d 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -3250,6 +3250,17 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio const delayMs = continuationStartRetryDelayMs(run, projection); if (delayMs === undefined) return; const commandId = CommandId.make(`command:restart-continuation-retry:${run.id}`); + // A start retry is not a crash: carry the crash budget forward unchanged. + const previous = + run.restartContinuationOfRunId === undefined + ? undefined + : Option.getOrUndefined( + yield* effectOutbox.get( + `effect:restart-continuation:${run.restartContinuationOfRunId}`, + ), + )?.request; + const processLossCount = + previous?.type === "provider-runtime.continue" ? previous.processLossCount : undefined; yield* writeSystemEvents( [], [ @@ -3257,7 +3268,11 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio id: `effect:restart-continuation:${run.id}`, commandId, threadId, - request: { type: "provider-runtime.continue", sourceRunId: run.id }, + request: { + type: "provider-runtime.continue", + sourceRunId: run.id, + ...(processLossCount === undefined ? {} : { processLossCount }), + }, availableAt: DateTime.add(yield* DateTime.now, { milliseconds: delayMs }), }, ], diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index 96add5d1fc48a..c3c9919988ab4 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -9254,6 +9254,21 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { }); const cut = (yield* runFor("cut"))!; yield* write("cut:cancelled", { ...cut, status: "cancelled" }); + yield* eventSink.writeWithEffects({ + events: [], + effects: [ + { + id: `effect:restart-continuation:${cut.id}`, + commandId: CommandId.make(`${threadId}:crash-recovery`), + threadId, + request: { + type: "provider-runtime.continue", + sourceRunId: cut.id, + processLossCount: 2, + }, + }, + ], + }); yield* orchestrator.dispatch({ type: "message.dispatch", createdBy: "agent", @@ -9337,6 +9352,14 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { ); yield* worker.drain(); assert.deepEqual(yield* retriesOf, [], "the retry waits out its backoff"); + const retryIntent = yield* (yield* EffectOutbox.EffectOutboxV2).get( + `effect:restart-continuation:${continuation.id}`, + ); + assert.deepEqual(Option.getOrUndefined(retryIntent)?.request, { + type: "provider-runtime.continue", + sourceRunId: continuation.id, + processLossCount: 2, + }); if (stopped) { yield* orchestrator.dispatch({ type: "thread.stop",