From 7fb9da4986af8883063579d2660ae88479dd0dac Mon Sep 17 00:00:00 2001 From: shivam <91240327+shivamhwp@users.noreply.github.com> Date: Fri, 6 Nov 2026 01:16:20 +0000 Subject: [PATCH 1/2] fix(server): a usage-limit auto-resume survives a server restart #29838 made the auto-resume an agent message. A restart before the provider starts it cancels the run, and the limit worker then rebuilds the same already-receipted resume command, so the thread never resumes. Runs now record usageLimitContinuationOfRunId. When recovery cancels a never-started run while auto-resume is armed, it gives limitRecovery a fresh request id so the worker sends the resume again, and it marks the cancelled resume's prompt as queued input so the shared timeline rule hides it. The thread resumes on its own with one resume message. Co-Authored-By: Claude Opus 5.5 --- .../src/orchestration-v2/Orchestrator.ts | 9 +++ .../ProviderRuntimeRecoveryService.ts | 59 ++++++++++++++++++ .../src/orchestration-v2/runtimeLayer.test.ts | 62 ++++++++++++++++++- packages/contracts/src/orchestrationV2.ts | 1 + 4 files changed, 130 insertions(+), 1 deletion(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index bcec0a13cf20a..5d3c4391b1fe8 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -6279,6 +6279,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ...(command.restartContinuationOfRunId === undefined ? {} : { restartContinuationOfRunId: command.restartContinuationOfRunId }), + ...(command.usageLimitContinuationOfRunId === undefined + ? {} + : { usageLimitContinuationOfRunId: command.usageLimitContinuationOfRunId }), ...(continuedTaskId === undefined ? {} : { delegatedTaskId: continuedTaskId }), ...(freshContext ? { freshContext: true } : {}), }; @@ -6659,6 +6662,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ...(command.restartContinuationOfRunId === undefined ? {} : { restartContinuationOfRunId: command.restartContinuationOfRunId }), + ...(command.usageLimitContinuationOfRunId === undefined + ? {} + : { usageLimitContinuationOfRunId: command.usageLimitContinuationOfRunId }), ...(dispatchMode.type === "defer_start" && dispatchMode.workspaceStrategy !== undefined ? { workspacePreparation: dispatchMode.workspaceStrategy } : {}), @@ -7445,6 +7451,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ...(command.restartContinuationOfRunId === undefined ? {} : { restartContinuationOfRunId: command.restartContinuationOfRunId }), + ...(command.usageLimitContinuationOfRunId === undefined + ? {} + : { usageLimitContinuationOfRunId: command.usageLimitContinuationOfRunId }), ...(continuedTaskId === undefined ? {} : { delegatedTaskId: continuedTaskId }), ...wakeWorkStartedAt(projection.runs, command), }; diff --git a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts index 222747c413982..59304673203d2 100644 --- a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts +++ b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts @@ -442,6 +442,64 @@ export const make = Effect.gen(function* () { }, }); } + const limitRecovery = projection.thread.limitRecovery; + // Cancelling a never-started run leaves the thread limited by the same run, so a resume sent + // before the restart spent the worker's request id. A fresh id lets the worker send it again. + if (limitRecovery?.autoResume === true && runs.some((run) => run.startedAt === null)) { + events.push({ + id: yield* allocateEventId(), + type: "thread.metadata-updated", + threadId: projection.thread.id, + occurredAt: now, + payload: { + ...projection.thread, + limitRecovery: { ...limitRecovery, requestId: commandId }, + updatedAt: now, + }, + }); + } + // The worker sends the auto-resume again, so the unsent copy leaves the transcript the way + // cancelled queued input does. + const unsentResumeRunIds = runs.flatMap((run) => + run.startedAt === null && run.usageLimitContinuationOfRunId !== undefined ? [run.id] : [], + ); + const unsentResumePrompts = + unsentResumeRunIds.length === 0 + ? [] + : (yield* projections + .getThreadRecords(projection.thread.id, ["turnItems"], { + turnItemTypes: ["user_message"], + turnItemRunIds: unsentResumeRunIds, + }) + .pipe( + Effect.mapError( + (cause) => + new ProviderRuntimeRecoveryError({ + operation: "read-projections", + threadId: projection.thread.id, + cause, + }), + ), + )).turnItems; + const unsentResumePromptIds = new Set(unsentResumePrompts.map((item) => item.id)); + for (const item of unsentResumePrompts) { + if (item.type !== "user_message") continue; + const active = + item.status === "pending" || item.status === "running" || item.status === "waiting"; + events.push({ + id: yield* allocateEventId(), + type: "turn-item.updated", + threadId: projection.thread.id, + ...(item.runId === null ? {} : { runId: item.runId }), + occurredAt: now, + payload: { + ...item, + ...(active ? { status: "cancelled", completedAt: now } : {}), + inputIntent: "queued_turn", + updatedAt: now, + }, + }); + } for (const question of questions) { events.push({ id: yield* allocateEventId(), @@ -552,6 +610,7 @@ export const make = Effect.gen(function* () { for (const item of projection.turnItems.filter( (candidate) => candidate.runId === run.id && + !unsentResumePromptIds.has(candidate.id) && (candidate.nodeId === null || !messageRequestNodeIds.has(candidate.nodeId)) && !isAppOwnedDelegationItem(candidate) && // A link card waits on the user, not the run, and answers by message. diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index b90f491dfa94e..c8019f483023f 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -1,6 +1,7 @@ import { PullRequestProviderError } from "@t3tools/source-control-core/server/PullRequestProvider"; import { SourceControlProviderKind } from "@t3tools/contracts"; import { limitRecoveryCommand } from "./UsageLimitRecoveryWorker.ts"; +import { createOrchestrationV2TurnItemVisibility } from "@t3tools/shared/orchestrationV2Timeline"; import * as McpProviderSessions from "@t3tools/provider-core/server/McpProviderSessions"; import * as SourceControlProviderRegistry from "../sourceControl/SourceControlProviderRegistry.ts"; import * as NodeCrypto from "@effect/platform-node/NodeCrypto"; @@ -8136,6 +8137,7 @@ it.layer(layerTest)("usage-limit recovery", (it) => { it.effect.each([ "resume", + "resume-started", "queued-resume", "rearm", "snooze-race", @@ -8614,17 +8616,74 @@ it.layer(layerTest)("usage-limit recovery", (it) => { yield* orchestrator.dispatch(resume); } const after = yield* orchestrator.getThreadProjection(threadId); - if (scenario === "resume") { + if (scenario === "resume" || scenario === "resume-started") { assert.deepInclude(after.messages.at(-1), { text: "Continue where you left off.", createdBy: "agent", creationSource: "server", }); + const continuation = after.runs.at(-1)!; + assert.equal(continuation.status, "starting"); + if (scenario === "resume-started") { + const startedAt = yield* DateTime.now; + yield* events.write({ + events: [ + { + id: EventId.make("recovery:continuation-started:resume-started"), + type: "run.updated", + threadId, + runId: continuation.id, + occurredAt: startedAt, + payload: { ...continuation, status: "running", startedAt }, + }, + ], + }); + } + yield* (yield* ProviderRuntimeRecoveryService.ProviderRuntimeRecoveryService).reconcile( + "startup", + ); + const restartedShell = (yield* orchestrator.getShellSnapshot()).threads.find( + (thread) => thread.id === threadId, + )!; + const retry = limitRecoveryCommand( + restartedShell, + true, + DateTime.toEpochMillis(yield* DateTime.now), + ); + if (scenario === "resume-started") { + assert.isNull(retry); + assert.equal( + restartedShell.limitRecovery?.requestId, + after.thread.limitRecovery?.requestId, + ); + } else { + assert(retry?.type === "message.dispatch"); + assert.notEqual(retry.commandId, resume!.commandId); + yield* orchestrator.dispatch(retry); + yield* orchestrator.dispatch(retry); + const resumed = yield* orchestrator.getThreadProjection(threadId); + assert.deepEqual( + resumed.runs.map((row) => row.status), + ["failed", "cancelled", "starting"], + ); + const visible = createOrchestrationV2TurnItemVisibility({ + runs: resumed.runs, + attempts: resumed.attempts, + items: resumed.turnItems, + }); + assert.deepEqual( + resumed.turnItems.flatMap((item) => + item.type === "user_message" && visible(item) ? [item.runId] : [], + ), + [run.id, resumed.runs[2]!.id], + ); + } } assert.lengthOf( after.runs, before.runs.length + (scenario === "resume" || + scenario === "resume-started" || stopBeforeArm || scenario === "queued-resume" || scenario === "independent-patches" || @@ -8636,6 +8695,7 @@ it.layer(layerTest)("usage-limit recovery", (it) => { after.messages, before.messages.length + (scenario === "resume" || + scenario === "resume-started" || stopBeforeArm || scenario === "queued-resume" || scenario === "independent-patches" || diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index 461ec5e4b3f3e..5c430ef75581a 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -692,6 +692,7 @@ export const OrchestrationV2Run = Schema.Struct({ checkpointId: Schema.NullOr(CheckpointId), contextHandoffId: Schema.NullOr(ContextHandoffId), restartContinuationOfRunId: Schema.optional(RunId), + usageLimitContinuationOfRunId: Schema.optional(RunId), delegatedTaskId: Schema.optional(NodeId), workStartedAt: Schema.optional(Schema.DateTimeUtc), restartCancelledBackgroundWork: Schema.optional( From 5ef11ff9ebdc56243ad69ef724547db05709e118 Mon Sep 17 00:00:00 2001 From: shivam <91240327+shivamhwp@users.noreply.github.com> Date: Sat, 7 Nov 2026 00:20:21 +0000 Subject: [PATCH 2/2] fix(server): only re-send a resume restart recovery can hide Restart recovery now re-arms the usage-limit auto-resume only when the cancelled run carries the resume tag. An untagged resume from an older server stays cancelled, as on main, instead of showing twice, and other cancelled runs no longer trigger an extra resume. Co-Authored-By: Claude Opus 5.5 --- .../ProviderRuntimeRecoveryService.ts | 12 ++-- .../src/orchestration-v2/runtimeLayer.test.ts | 69 +++++++++++++------ 2 files changed, 53 insertions(+), 28 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts index 59304673203d2..89e43e5b1ede4 100644 --- a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts +++ b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts @@ -442,10 +442,13 @@ export const make = Effect.gen(function* () { }, }); } + const unsentResumeRunIds = runs.flatMap((run) => + run.startedAt === null && run.usageLimitContinuationOfRunId !== undefined ? [run.id] : [], + ); const limitRecovery = projection.thread.limitRecovery; - // Cancelling a never-started run leaves the thread limited by the same run, so a resume sent - // before the restart spent the worker's request id. A fresh id lets the worker send it again. - if (limitRecovery?.autoResume === true && runs.some((run) => run.startedAt === null)) { + // The unsent resume spent the worker's request id, so a fresh one lets it send again. An + // untagged resume from an older server stays cancelled: its copy couldn't be hidden. + if (limitRecovery?.autoResume === true && unsentResumeRunIds.length > 0) { events.push({ id: yield* allocateEventId(), type: "thread.metadata-updated", @@ -460,9 +463,6 @@ export const make = Effect.gen(function* () { } // The worker sends the auto-resume again, so the unsent copy leaves the transcript the way // cancelled queued input does. - const unsentResumeRunIds = runs.flatMap((run) => - run.startedAt === null && run.usageLimitContinuationOfRunId !== undefined ? [run.id] : [], - ); const unsentResumePrompts = unsentResumeRunIds.length === 0 ? [] diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index c8019f483023f..170ceee73a331 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -8138,6 +8138,7 @@ it.layer(layerTest)("usage-limit recovery", (it) => { it.effect.each([ "resume", "resume-started", + "resume-untagged", "queued-resume", "rearm", "snooze-race", @@ -8616,7 +8617,11 @@ it.layer(layerTest)("usage-limit recovery", (it) => { yield* orchestrator.dispatch(resume); } const after = yield* orchestrator.getThreadProjection(threadId); - if (scenario === "resume" || scenario === "resume-started") { + if ( + scenario === "resume" || + scenario === "resume-started" || + scenario === "resume-untagged" + ) { assert.deepInclude(after.messages.at(-1), { text: "Continue where you left off.", createdBy: "agent", @@ -8639,6 +8644,22 @@ it.layer(layerTest)("usage-limit recovery", (it) => { ], }); } + if (scenario === "resume-untagged") { + // A server older than the run tag stored the resume run without it. + const { usageLimitContinuationOfRunId: _, ...untagged } = continuation; + yield* events.write({ + events: [ + { + id: EventId.make("recovery:continuation-untagged"), + type: "run.updated", + threadId, + runId: continuation.id, + occurredAt: yield* DateTime.now, + payload: untagged, + }, + ], + }); + } yield* (yield* ProviderRuntimeRecoveryService.ProviderRuntimeRecoveryService).reconcile( "startup", ); @@ -8650,40 +8671,43 @@ it.layer(layerTest)("usage-limit recovery", (it) => { true, DateTime.toEpochMillis(yield* DateTime.now), ); - if (scenario === "resume-started") { - assert.isNull(retry); + if (scenario === "resume-started") assert.isNull(retry); + else assert(retry?.type === "message.dispatch"); + if (scenario === "resume") { + assert.notEqual(retry!.commandId, resume!.commandId); + } else { assert.equal( restartedShell.limitRecovery?.requestId, after.thread.limitRecovery?.requestId, ); - } else { - assert(retry?.type === "message.dispatch"); - assert.notEqual(retry.commandId, resume!.commandId); + } + if (retry !== null) { yield* orchestrator.dispatch(retry); yield* orchestrator.dispatch(retry); - const resumed = yield* orchestrator.getThreadProjection(threadId); - assert.deepEqual( - resumed.runs.map((row) => row.status), - ["failed", "cancelled", "starting"], - ); - const visible = createOrchestrationV2TurnItemVisibility({ - runs: resumed.runs, - attempts: resumed.attempts, - items: resumed.turnItems, - }); - assert.deepEqual( - resumed.turnItems.flatMap((item) => - item.type === "user_message" && visible(item) ? [item.runId] : [], - ), - [run.id, resumed.runs[2]!.id], - ); } + const resumed = yield* orchestrator.getThreadProjection(threadId); + const visible = createOrchestrationV2TurnItemVisibility({ + runs: resumed.runs, + attempts: resumed.attempts, + items: resumed.turnItems, + }); + assert.deepEqual( + resumed.turnItems.flatMap((item) => + item.type === "user_message" && visible(item) ? [item.runId] : [], + ), + [run.id, resumed.runs.at(-1)!.id], + ); + assert.deepEqual( + resumed.runs.map((row) => row.status), + scenario === "resume" ? ["failed", "cancelled", "starting"] : ["failed", "cancelled"], + ); } assert.lengthOf( after.runs, before.runs.length + (scenario === "resume" || scenario === "resume-started" || + scenario === "resume-untagged" || stopBeforeArm || scenario === "queued-resume" || scenario === "independent-patches" || @@ -8696,6 +8720,7 @@ it.layer(layerTest)("usage-limit recovery", (it) => { before.messages.length + (scenario === "resume" || scenario === "resume-started" || + scenario === "resume-untagged" || stopBeforeArm || scenario === "queued-resume" || scenario === "independent-patches" ||