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..89e43e5b1ede4 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 unsentResumeRunIds = runs.flatMap((run) => + run.startedAt === null && run.usageLimitContinuationOfRunId !== undefined ? [run.id] : [], + ); + const limitRecovery = projection.thread.limitRecovery; + // 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", + 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 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..170ceee73a331 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,8 @@ it.layer(layerTest)("usage-limit recovery", (it) => { it.effect.each([ "resume", + "resume-started", + "resume-untagged", "queued-resume", "rearm", "snooze-race", @@ -8614,17 +8617,97 @@ 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" || + scenario === "resume-untagged" + ) { 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 }, + }, + ], + }); + } + 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", + ); + 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); + 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, + ); + } + if (retry !== null) { + yield* orchestrator.dispatch(retry); + yield* orchestrator.dispatch(retry); + } + 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" || @@ -8636,6 +8719,8 @@ it.layer(layerTest)("usage-limit recovery", (it) => { after.messages, before.messages.length + (scenario === "resume" || + scenario === "resume-started" || + scenario === "resume-untagged" || 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(