From 5e826ef0b3a824951d5afd99ef9861602c0bdf67 Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Fri, 30 Oct 2026 21:07:28 -0400 Subject: [PATCH 1/2] feat(web): resume a run that failed with a provider error Signed-off-by: Yordis Prieto --- .../src/orchestration-v2/Orchestrator.ts | 9 ++---- .../src/orchestration-v2/runtimeLayer.test.ts | 32 ++++++++++++------- apps/web/src/components/ChatView.tsx | 19 +++-------- .../shared/src/orchestrationV2ThreadError.ts | 10 ++++++ 4 files changed, 39 insertions(+), 31 deletions(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 3a3190e8f46a3..59f5812534418 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -7,6 +7,7 @@ import type { import { latestExecutedRun, latestRootProviderFailure, + manuallyResumableRun, runRanAfter, usageLimitBlockedRun, } from "@t3tools/shared/orchestrationV2ThreadError"; @@ -5240,14 +5241,10 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ) { let projection = yield* getProjectionWithPendingEvents(command.threadId, events); if (command.manualContinuationOfRunId !== undefined) { - const source = projection.runs.find((run) => run.id === command.manualContinuationOfRunId); - const limited = latestRootProviderFailure(source ?? null, projection.turnItems); if ( command.dispatchMode.type !== "start_immediately" || - source === undefined || - (source.status !== "interrupted" && - !(source.status === "failed" && limited?.class === "usage_limit")) || - latestExecutedRun(projection.runs)?.id !== source.id || + manuallyResumableRun(projection.runs, projection.turnItems)?.id !== + command.manualContinuationOfRunId || projection.thread.archivedAt !== null || projection.thread.deletedAt !== null || projection.runtimeRequests.some((request) => request.status === "pending") diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index 0fae8c3455268..6bfb8dded692f 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -7841,7 +7841,7 @@ it.layer(layerSharedApplicationDataPlaneTest)("shared application data plane", ( }); it.layer(layerTest)("usage-limit recovery", (it) => { - it.effect.each(["interrupted", "usage_limit"] as const)( + it.effect.each(["interrupted", "usage_limit", "provider_error"] as const)( "manually resumes an %s run ahead of its queued message only once", (reason) => Effect.gen(function* () { @@ -7922,8 +7922,8 @@ it.layer(layerTest)("usage-limit recovery", (it) => { ], }); let scheduledResume: ReturnType = null; - if (reason === "usage_limit") { - const resetAt = DateTime.formatIso(DateTime.add(now, { minutes: 1 })); + const resetAt = DateTime.formatIso(DateTime.add(now, { minutes: 1 })); + if (reason !== "interrupted") { yield* events.write({ events: [ { @@ -7943,21 +7943,31 @@ it.layer(layerTest)("usage-limit recovery", (it) => { parentItemId: null, ordinal: 2, status: "failed", - title: "Usage limit reached", + title: "Provider error", startedAt: now, completedAt: now, updatedAt: now, - failure: { - class: "usage_limit", - message: "Plan limit reached.", - code: "usageLimitExceeded", - retryable: null, - resetAt, - }, + failure: + reason === "usage_limit" + ? { + class: "usage_limit", + message: "Plan limit reached.", + code: "usageLimitExceeded", + retryable: null, + resetAt, + } + : { + class: "provider_error", + message: "Usage balance exhausted.", + code: null, + retryable: null, + }, }, }, ], }); + } + if (reason === "usage_limit") { const shell = (yield* orchestrator.getShellSnapshot()).threads.find( (thread) => thread.id === threadId, )!; diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index 63ff8b8463eee..da104d8da76ba 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -42,10 +42,7 @@ import { RightPanelSurfaceContent } from "./chat/RightPanelSurfaceContent"; import { usePendingThreadRequests } from "./chat/usePendingThreadRequests"; import { isScratchProject } from "@t3tools/client-runtime/state/projects"; import { visibleThreadPullRequests } from "@t3tools/shared/threadPullRequests"; -import { - latestExecutedRun, - latestRootProviderFailure, -} from "@t3tools/shared/orchestrationV2ThreadError"; +import { manuallyResumableRun } from "@t3tools/shared/orchestrationV2ThreadError"; import type { UsageLimitSourceSnapshots } from "@t3tools/contracts"; import { feedbackBannerItem } from "./chat/ComposerFeedback"; import { useUsageLimitsPanel } from "./chat/useUsageLimitsPanel"; @@ -814,16 +811,10 @@ export default function ChatView(props: ChatViewProps) { isServerThread && serverProjection?.runs.some((run) => run.status === "queued" && run.queueHeld === true) === true; - const resumableRunId = (() => { - if (!isServerThread || serverProjection === null) return null; - const run = latestExecutedRun(serverProjection.runs); - if (run?.status === "interrupted") return run.id; - return run?.status === "failed" && - serverRuntime?.lastErrorClass === "usage_limit" && - latestRootProviderFailure(run, serverProjection.turnItems)?.class === "usage_limit" - ? run.id - : null; - })(); + const resumableRunId = + !isServerThread || serverProjection === null + ? null + : (manuallyResumableRun(serverProjection.runs, serverProjection.turnItems)?.id ?? null); const parentSubagentThreadId = activeThread?.lineage.relationshipToParent === "subagent" ? activeThread.lineage.parentThreadId diff --git a/packages/shared/src/orchestrationV2ThreadError.ts b/packages/shared/src/orchestrationV2ThreadError.ts index 5b9940729e2e1..288994b71afee 100644 --- a/packages/shared/src/orchestrationV2ThreadError.ts +++ b/packages/shared/src/orchestrationV2ThreadError.ts @@ -64,6 +64,16 @@ export function latestExecutedRun( return latest; } +/** A failed run can resume on any model, so a provider-side failure is not a dead end. */ +export function manuallyResumableRun( + runs: ReadonlyArray, + turnItems: ReadonlyArray, +): OrchestrationV2Run | null { + const run = latestExecutedRun(runs); + if (run?.status === "interrupted") return run; + return latestRootProviderFailure(run, turnItems) === null ? null : run; +} + /** Held queues can execute out of ordinal order; unfinished runs count as latest. */ export function runRanAfter( run: Pick, From 4e1f3b70a1993f7ba7ffe662deeddca7002902c8 Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Fri, 30 Oct 2026 21:22:13 -0400 Subject: [PATCH 2/2] fix(server): keep workspace preparation failures on their own retry path Signed-off-by: Yordis Prieto --- .../src/orchestration-v2/runtimeLayer.test.ts | 318 +++++++++--------- .../shared/src/orchestrationV2ThreadError.ts | 19 +- 2 files changed, 179 insertions(+), 158 deletions(-) diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index 6bfb8dded692f..91bd558cc7ef6 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -32,6 +32,7 @@ import { RunId, ScheduledTaskId, ThreadId, + ORCHESTRATION_V2_WORKSPACE_PREPARATION_FAILURE_CODE, } from "@t3tools/contracts"; import * as Effect from "effect/Effect"; import * as DateTime from "effect/DateTime"; @@ -7841,172 +7842,185 @@ it.layer(layerSharedApplicationDataPlaneTest)("shared application data plane", ( }); it.layer(layerTest)("usage-limit recovery", (it) => { - it.effect.each(["interrupted", "usage_limit", "provider_error"] as const)( - "manually resumes an %s run ahead of its queued message only once", - (reason) => - Effect.gen(function* () { - const orchestrator = yield* Orchestrator.OrchestratorV2; - const events = yield* EventSink.EventSinkV2; - const threadId = ThreadId.make(`manual-resume:${reason}`); - const projectId = ProjectId.make(`manual-resume:project:${reason}`); - const now = yield* DateTime.now; - const createdAt = DateTime.formatIso(now); - yield* seedProject({ - projectId, - title: "Resume project", - workspaceRoot: process.cwd(), - defaultModelSelection: modelSelection, - createdAt, - }); - yield* orchestrator.dispatch({ - type: "thread.create", - commandId: CommandId.make(`manual-resume:create:${reason}`), - threadId, - projectId, - title: "Interrupted thread", - modelSelection, - runtimeMode: "full-access", - interactionMode: "default", - branch: null, - worktreePath: null, - createdBy: "user", - creationSource: "web", - }); - yield* orchestrator.dispatch({ - type: "message.dispatch", - commandId: CommandId.make(`manual-resume:start:${reason}`), - threadId, - messageId: MessageId.make(`manual-resume:start:${reason}`), - text: "Start work.", - attachments: [], - dispatchMode: { type: "defer_start" }, - createdBy: "user", - creationSource: "web", - }); - yield* orchestrator.dispatch({ - type: "message.dispatch", - commandId: CommandId.make(`manual-resume:queue:${reason}`), - threadId, - messageId: MessageId.make(`manual-resume:queue:${reason}`), - text: "Follow up.", - attachments: [], - dispatchMode: { type: "queue_after_active" }, - createdBy: "user", - creationSource: "web", - }); - const [source, queued] = (yield* orchestrator.getThreadProjection(threadId)).runs as [ - OrchestrationV2Run, - OrchestrationV2Run, - ]; - // Hold the queue like user Stop so the terminal worker cannot start it before explicit resume. + it.effect.each([ + "interrupted", + "usage_limit", + "provider_error", + "workspace_preparation", + ] as const)("guards a manual resume of an %s run ahead of its queued message", (reason) => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const events = yield* EventSink.EventSinkV2; + const threadId = ThreadId.make(`manual-resume:${reason}`); + const projectId = ProjectId.make(`manual-resume:project:${reason}`); + const now = yield* DateTime.now; + const createdAt = DateTime.formatIso(now); + yield* seedProject({ + projectId, + title: "Resume project", + workspaceRoot: process.cwd(), + defaultModelSelection: modelSelection, + createdAt, + }); + yield* orchestrator.dispatch({ + type: "thread.create", + commandId: CommandId.make(`manual-resume:create:${reason}`), + threadId, + projectId, + title: "Interrupted thread", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + createdBy: "user", + creationSource: "web", + }); + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make(`manual-resume:start:${reason}`), + threadId, + messageId: MessageId.make(`manual-resume:start:${reason}`), + text: "Start work.", + attachments: [], + dispatchMode: { type: "defer_start" }, + createdBy: "user", + creationSource: "web", + }); + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make(`manual-resume:queue:${reason}`), + threadId, + messageId: MessageId.make(`manual-resume:queue:${reason}`), + text: "Follow up.", + attachments: [], + dispatchMode: { type: "queue_after_active" }, + createdBy: "user", + creationSource: "web", + }); + const [source, queued] = (yield* orchestrator.getThreadProjection(threadId)).runs as [ + OrchestrationV2Run, + OrchestrationV2Run, + ]; + // Hold the queue like user Stop so the terminal worker cannot start it before explicit resume. + yield* events.write({ + events: [ + { + id: EventId.make(`manual-resume:hold:${reason}`), + type: "run.updated", + threadId, + occurredAt: now, + payload: { ...queued, queueHeld: true }, + }, + { + id: EventId.make(`manual-resume:stop:${reason}`), + type: "run.updated", + threadId, + occurredAt: now, + payload: { + ...source, + status: reason === "interrupted" ? "interrupted" : "failed", + completedAt: now, + }, + }, + ], + }); + let scheduledResume: ReturnType = null; + const resetAt = DateTime.formatIso(DateTime.add(now, { minutes: 1 })); + if (reason !== "interrupted") { yield* events.write({ events: [ { - id: EventId.make(`manual-resume:hold:${reason}`), - type: "run.updated", - threadId, - occurredAt: now, - payload: { ...queued, queueHeld: true }, - }, - { - id: EventId.make(`manual-resume:stop:${reason}`), - type: "run.updated", + id: EventId.make(`manual-resume:error:${reason}`), + type: "turn-item.updated", threadId, occurredAt: now, payload: { - ...source, - status: reason === "interrupted" ? "interrupted" : "failed", + id: TurnItemId.make(`manual-resume:error:${reason}`), + type: "error", + threadId, + runId: source.id, + nodeId: source.rootNodeId, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 2, + status: "failed", + title: "Provider error", + startedAt: now, completedAt: now, + updatedAt: now, + failure: + reason === "usage_limit" + ? { + class: "usage_limit", + message: "Plan limit reached.", + code: "usageLimitExceeded", + retryable: null, + resetAt, + } + : { + class: "provider_error", + message: "Usage balance exhausted.", + code: + reason === "workspace_preparation" + ? ORCHESTRATION_V2_WORKSPACE_PREPARATION_FAILURE_CODE + : null, + retryable: null, + }, }, }, ], }); - let scheduledResume: ReturnType = null; - const resetAt = DateTime.formatIso(DateTime.add(now, { minutes: 1 })); - if (reason !== "interrupted") { - yield* events.write({ - events: [ - { - id: EventId.make(`manual-resume:error:${reason}`), - type: "turn-item.updated", - threadId, - occurredAt: now, - payload: { - id: TurnItemId.make(`manual-resume:error:${reason}`), - type: "error", - threadId, - runId: source.id, - nodeId: source.rootNodeId, - providerThreadId: null, - providerTurnId: null, - nativeItemRef: null, - parentItemId: null, - ordinal: 2, - status: "failed", - title: "Provider error", - startedAt: now, - completedAt: now, - updatedAt: now, - failure: - reason === "usage_limit" - ? { - class: "usage_limit", - message: "Plan limit reached.", - code: "usageLimitExceeded", - retryable: null, - resetAt, - } - : { - class: "provider_error", - message: "Usage balance exhausted.", - code: null, - retryable: null, - }, - }, - }, - ], - }); - } - if (reason === "usage_limit") { - const shell = (yield* orchestrator.getShellSnapshot()).threads.find( - (thread) => thread.id === threadId, - )!; - yield* orchestrator.dispatch( - limitRecoveryCommand(shell, true, DateTime.toEpochMillis(now))!, - ); - const armed = (yield* orchestrator.getShellSnapshot()).threads.find( - (thread) => thread.id === threadId, - )!; - scheduledResume = limitRecoveryCommand(armed, true, Date.parse(resetAt)); - assert.isNotNull(scheduledResume); - } - const resume = (suffix: string) => ({ - type: "message.dispatch" as const, - commandId: CommandId.make(`manual-resume:${suffix}:${reason}`), - threadId, - messageId: MessageId.make(`manual-resume:${suffix}:${reason}`), - manualContinuationOfRunId: source.id, - text: "Continue where you left off.", - attachments: [], - dispatchMode: { type: "start_immediately" as const }, - createdBy: "user" as const, - creationSource: "web" as const, - }); - yield* orchestrator.dispatch(resume("first")); - const after = yield* orchestrator.getThreadProjection(threadId); - assert.lengthOf(after.runs, 3); - assert.equal(after.runs[1]?.status, "queued"); - assert.equal(after.runs[2]?.status, "starting"); + } + if (reason === "usage_limit") { + const shell = (yield* orchestrator.getShellSnapshot()).threads.find( + (thread) => thread.id === threadId, + )!; + yield* orchestrator.dispatch( + limitRecoveryCommand(shell, true, DateTime.toEpochMillis(now))!, + ); + const armed = (yield* orchestrator.getShellSnapshot()).threads.find( + (thread) => thread.id === threadId, + )!; + scheduledResume = limitRecoveryCommand(armed, true, Date.parse(resetAt)); + assert.isNotNull(scheduledResume); + } + const resume = (suffix: string) => ({ + type: "message.dispatch" as const, + commandId: CommandId.make(`manual-resume:${suffix}:${reason}`), + threadId, + messageId: MessageId.make(`manual-resume:${suffix}:${reason}`), + manualContinuationOfRunId: source.id, + text: "Continue where you left off.", + attachments: [], + dispatchMode: { type: "start_immediately" as const }, + createdBy: "user" as const, + creationSource: "web" as const, + }); + if (reason === "workspace_preparation") { assert.equal( - (yield* orchestrator.dispatch(resume("second")).pipe(Effect.exit))._tag, + (yield* orchestrator.dispatch(resume("first")).pipe(Effect.exit))._tag, "Failure", ); - if (scheduledResume !== null) { - yield* TestClock.adjust("1 minute"); - yield* orchestrator.dispatch(scheduledResume); - } - assert.lengthOf((yield* orchestrator.getThreadProjection(threadId)).runs, 3); - }), + return; + } + yield* orchestrator.dispatch(resume("first")); + const after = yield* orchestrator.getThreadProjection(threadId); + assert.lengthOf(after.runs, 3); + assert.equal(after.runs[1]?.status, "queued"); + assert.equal(after.runs[2]?.status, "starting"); + assert.equal( + (yield* orchestrator.dispatch(resume("second")).pipe(Effect.exit))._tag, + "Failure", + ); + if (scheduledResume !== null) { + yield* TestClock.adjust("1 minute"); + yield* orchestrator.dispatch(scheduledResume); + } + assert.lengthOf((yield* orchestrator.getThreadProjection(threadId)).runs, 3); + }), ); it.effect.each([ diff --git a/packages/shared/src/orchestrationV2ThreadError.ts b/packages/shared/src/orchestrationV2ThreadError.ts index 288994b71afee..1a3f4d077580d 100644 --- a/packages/shared/src/orchestrationV2ThreadError.ts +++ b/packages/shared/src/orchestrationV2ThreadError.ts @@ -1,7 +1,8 @@ -import type { - OrchestrationV2ProviderFailure, - OrchestrationV2Run, - OrchestrationV2TurnItem, +import { + ORCHESTRATION_V2_WORKSPACE_PREPARATION_FAILURE_CODE, + type OrchestrationV2ProviderFailure, + type OrchestrationV2Run, + type OrchestrationV2TurnItem, } from "@t3tools/contracts"; import * as DateTime from "effect/DateTime"; @@ -64,14 +65,20 @@ export function latestExecutedRun( return latest; } -/** A failed run can resume on any model, so a provider-side failure is not a dead end. */ +/** + * A failed run can resume on any model, so a provider-side failure is not a dead end. Workspace + * failures retry through prepared-run.retry, or the agent would run outside its worktree. + */ export function manuallyResumableRun( runs: ReadonlyArray, turnItems: ReadonlyArray, ): OrchestrationV2Run | null { const run = latestExecutedRun(runs); if (run?.status === "interrupted") return run; - return latestRootProviderFailure(run, turnItems) === null ? null : run; + const failure = latestRootProviderFailure(run, turnItems); + return failure === null || failure.code === ORCHESTRATION_V2_WORKSPACE_PREPARATION_FAILURE_CODE + ? null + : run; } /** Held queues can execute out of ordinal order; unfinished runs count as latest. */