diff --git a/apps/server/src/orchestration-v2/EffectOutbox.ts b/apps/server/src/orchestration-v2/EffectOutbox.ts index 0a250bb2b865c..e0814f1781084 100644 --- a/apps/server/src/orchestration-v2/EffectOutbox.ts +++ b/apps/server/src/orchestration-v2/EffectOutbox.ts @@ -35,6 +35,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"), @@ -705,6 +706,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 @@ -713,6 +715,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 7ef6b8df286b4..94444eb8e5c69 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"; @@ -777,3 +780,60 @@ it.effect.each(["confirmed", "unconfirmed", "failed"] as const)( ); }), ); + +it.effect.each([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, + lineage: { parentThreadId: null, relationshipToParent: 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 16bacc22ec080..41b2d234f26b3 100644 --- a/apps/server/src/orchestration-v2/EffectWorker.ts +++ b/apps/server/src/orchestration-v2/EffectWorker.ts @@ -160,7 +160,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 841e4b3dfb080..3f00941c8daa3 100644 --- a/apps/server/src/orchestration-v2/FoundationPersistence.test.ts +++ b/apps/server/src/orchestration-v2/FoundationPersistence.test.ts @@ -2752,6 +2752,34 @@ it.layer(layerTest)("orchestration V2 foundation persistence", (it) => { }).pipe(Effect.provide(layerIsolatedOutbox)), ); + it.effect.each(["queued", "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; @@ -3402,6 +3430,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/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/ProviderRuntimeRecoveryService.ts b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts index 46b739bec5732..979c5a0fc3ae7 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"; @@ -839,13 +840,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, + }, }, ] : []; @@ -1020,7 +1047,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..d333aad6626bd 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,75 @@ const continuationDispatches = ( return dispatches; }); +it.effect.each([ + [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 +1245,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 242e366d4b605..7454c2154461b 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/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", diff --git a/docs/internals/server-updates.md b/docs/internals/server-updates.md index 10df2ef464f8b..5e9937c03f5a0 100644 --- a/docs/internals/server-updates.md +++ b/docs/internals/server-updates.md @@ -130,8 +130,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 5c543dcd0bccf..369e675d7f993 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)