diff --git a/src/browser/features/Tools/WorkflowRunToolCall.timeout.stories.tsx b/src/browser/features/Tools/WorkflowRunToolCall.timeout.stories.tsx new file mode 100644 index 0000000000..8cc8f79868 --- /dev/null +++ b/src/browser/features/Tools/WorkflowRunToolCall.timeout.stories.tsx @@ -0,0 +1,267 @@ +import type { Meta, StoryObj } from "@storybook/react-vite"; + +import { WorkflowRunToolCall } from "@/browser/features/Tools/WorkflowRunToolCall"; +import { CHROMATIC_DISABLED, lightweightMeta } from "@/browser/stories/meta.js"; +import type { WorkflowRunRecord } from "@/common/types/workflow"; + +const meta = { + ...lightweightMeta, + title: "App/Chat/Tools/WorkflowRun/Timeouts", + component: WorkflowRunToolCall, + parameters: { + chromatic: CHROMATIC_DISABLED, + }, +} satisfies Meta; + +export default meta; + +type Story = StoryObj; + +const timeoutRecoveredRun: WorkflowRunRecord = { + id: "wfr_story_timeout_recovered", + workspaceId: "workspace-1", + workflow: { + name: "timeout-demo", + description: "Timeout demo", + scope: "project", + sourcePath: "./workflows/timeout-demo.js", + requestedScriptPath: "./workflows/timeout-demo.js", + canonicalScriptPath: "./workflows/timeout-demo.js", + sourceKind: "workspace-file", + sourceHash: "sha256:timeout-story", + executable: true, + }, + source: "export default function workflow() { return null; }", + sourceHash: "sha256:timeout-story", + args: { topic: "workflow timeout finalization" }, + status: "completed", + createdAt: "2026-05-29T00:00:00.000Z", + updatedAt: "2026-05-29T00:00:05.000Z", + events: [ + { sequence: 1, type: "status", at: "2026-05-29T00:00:00.000Z", status: "running" }, + { + sequence: 2, + type: "task", + at: "2026-05-29T00:00:01.000Z", + stepId: "slow-investigation", + taskId: "task_timeout", + status: "started", + title: "Slow investigation", + }, + { + sequence: 3, + type: "timeout", + at: "2026-05-29T00:00:03.000Z", + stepId: "slow-investigation", + taskId: "task_timeout", + phase: "soft", + details: { softMs: 2000, graceMs: 2000 }, + }, + { + sequence: 4, + type: "task", + at: "2026-05-29T00:00:03.000Z", + stepId: "slow-investigation", + taskId: "task_timeout", + status: "finalizing", + title: "Slow investigation", + }, + { + sequence: 5, + type: "timeout", + at: "2026-05-29T00:00:03.100Z", + stepId: "slow-investigation", + taskId: "task_timeout", + phase: "finalization_prompt_sent", + }, + { + sequence: 6, + type: "timeout", + at: "2026-05-29T00:00:04.000Z", + stepId: "slow-investigation", + taskId: "task_timeout", + phase: "recovered", + details: { graceMs: 2000 }, + }, + { + sequence: 7, + type: "task", + at: "2026-05-29T00:00:04.000Z", + stepId: "slow-investigation", + taskId: "task_timeout", + status: "completed", + title: "Slow investigation", + }, + { sequence: 8, type: "status", at: "2026-05-29T00:00:05.000Z", status: "completed" }, + ], + steps: [], +}; + +export const TimeoutRecovered: Story = { + args: { + args: { + script_path: "./workflows/timeout-demo.js", + args: { topic: "workflow timeout finalization" }, + run_in_background: false, + }, + status: "completed", + result: { + status: "completed", + runId: "wfr_story_timeout_recovered", + result: { reportMarkdown: "Recovered during timeout finalization." }, + run: timeoutRecoveredRun, + }, + }, +}; + +export const TimeoutFinalizing: Story = { + args: { + ...TimeoutRecovered.args, + result: { + status: "running", + runId: "wfr_story_timeout_finalizing", + result: null, + run: { + ...timeoutRecoveredRun, + id: "wfr_story_timeout_finalizing", + status: "running", + updatedAt: "2026-05-29T00:00:03.100Z", + events: timeoutRecoveredRun.events.slice(0, 5), + }, + }, + }, +}; + +export const TimeoutHard: Story = { + args: { + ...TimeoutRecovered.args, + result: { + status: "failed", + runId: "wfr_story_timeout_hard", + result: null, + run: { + ...timeoutRecoveredRun, + id: "wfr_story_timeout_hard", + status: "failed", + updatedAt: "2026-05-29T00:00:06.000Z", + events: [ + ...timeoutRecoveredRun.events.slice(0, 5), + { + sequence: 6, + type: "timeout", + at: "2026-05-29T00:00:05.000Z", + stepId: "slow-investigation", + taskId: "task_timeout", + phase: "hard", + details: { + error: + "Workflow agent step slow-investigation exceeded its soft timeout (2000ms) and did not produce a valid agent_report within the grace period (2000ms).", + }, + }, + { + sequence: 7, + type: "task", + at: "2026-05-29T00:00:05.000Z", + stepId: "slow-investigation", + taskId: "task_timeout", + status: "timed_out", + title: "Slow investigation", + }, + { + sequence: 8, + type: "error", + at: "2026-05-29T00:00:05.100Z", + message: + "Workflow agent step slow-investigation exceeded its soft timeout (2000ms) and did not produce a valid agent_report within the grace period (2000ms).", + }, + { sequence: 9, type: "status", at: "2026-05-29T00:00:06.000Z", status: "failed" }, + ], + }, + }, + }, +}; + +export const TimeoutParallel: Story = { + args: { + ...TimeoutRecovered.args, + result: { + status: "completed", + runId: "wfr_story_timeout_parallel", + result: { + reportMarkdown: "Parallel workflow completed after one lane recovered during grace.", + }, + run: { + ...timeoutRecoveredRun, + id: "wfr_story_timeout_parallel", + status: "completed", + updatedAt: "2026-05-29T00:00:06.000Z", + events: [ + { sequence: 1, type: "status", at: "2026-05-29T00:00:00.000Z", status: "running" }, + { + sequence: 2, + type: "task", + at: "2026-05-29T00:00:01.000Z", + stepId: "fast-lane", + taskId: "task_fast", + status: "completed", + title: "Fast lane", + }, + { + sequence: 3, + type: "task", + at: "2026-05-29T00:00:01.000Z", + stepId: "slow-lane", + taskId: "task_slow", + status: "started", + title: "Slow lane", + }, + { + sequence: 4, + type: "timeout", + at: "2026-05-29T00:00:03.000Z", + stepId: "slow-lane", + taskId: "task_slow", + phase: "soft", + details: { softMs: 2000, graceMs: 2000 }, + }, + { + sequence: 5, + type: "task", + at: "2026-05-29T00:00:03.000Z", + stepId: "slow-lane", + taskId: "task_slow", + status: "finalizing", + title: "Slow lane", + }, + { + sequence: 6, + type: "timeout", + at: "2026-05-29T00:00:03.100Z", + stepId: "slow-lane", + taskId: "task_slow", + phase: "finalization_prompt_sent", + }, + { + sequence: 7, + type: "timeout", + at: "2026-05-29T00:00:04.000Z", + stepId: "slow-lane", + taskId: "task_slow", + phase: "recovered", + details: { graceMs: 2000 }, + }, + { + sequence: 8, + type: "task", + at: "2026-05-29T00:00:04.000Z", + stepId: "slow-lane", + taskId: "task_slow", + status: "completed", + title: "Slow lane", + }, + { sequence: 9, type: "status", at: "2026-05-29T00:00:06.000Z", status: "completed" }, + ], + }, + }, + }, +}; diff --git a/src/browser/features/Tools/WorkflowRunToolCall.tsx b/src/browser/features/Tools/WorkflowRunToolCall.tsx index ba8721eb19..230e69b6cc 100644 --- a/src/browser/features/Tools/WorkflowRunToolCall.tsx +++ b/src/browser/features/Tools/WorkflowRunToolCall.tsx @@ -354,6 +354,18 @@ function getWorkflowEventLabel(event: WorkflowRunEvent): string { // Prefer the human-readable sub-agent title (matches the spawned // workspace title); fall back to stepId for legacy events without one. return `${event.title ?? event.stepId} / ${event.taskId} / ${event.status}`; + case "timeout": + switch (event.phase) { + case "soft": + return `${event.stepId} / ${event.taskId} / Soft timeout reached; requesting final report`; + case "finalization_prompt_sent": + return `${event.stepId} / ${event.taskId} / Final report requested`; + case "recovered": + return `${event.stepId} / ${event.taskId} / Recovered during grace period`; + case "hard": + return `${event.stepId} / ${event.taskId} / Hard timeout; child terminated`; + } + return `${event.stepId} / ${event.taskId} / timeout`; case "workflow": return `${event.stepId} / ${event.name} / ${event.runId} / ${event.status}`; case "patch": @@ -387,6 +399,8 @@ function getWorkflowEventDetail(event: WorkflowRunEvent): unknown { return event.details; case "patch": return event.details; + case "timeout": + return event.details; case "action": return event.details; case "task": @@ -421,6 +435,9 @@ function getEventTone(event: WorkflowRunEvent): "normal" | "success" | "warning" } return event.status === "failed" ? "warning" : "normal"; } + if (event.type === "timeout") { + return event.phase === "recovered" ? "success" : "warning"; + } if (event.type === "workflow") { return event.status === "completed" ? "success" diff --git a/src/browser/hooks/useCoderWorkspace.test.ts b/src/browser/hooks/useCoderWorkspace.test.ts index 8b9563d55d..38af1a53fd 100644 --- a/src/browser/hooks/useCoderWorkspace.test.ts +++ b/src/browser/hooks/useCoderWorkspace.test.ts @@ -1,6 +1,7 @@ -import { afterEach, beforeEach, describe, expect, mock, test } from "bun:test"; +import { afterAll, afterEach, beforeEach, describe, expect, mock, test } from "bun:test"; import { act, cleanup, renderHook } from "@testing-library/react"; import { GlobalWindow } from "happy-dom"; +import * as APIModule from "@/browser/contexts/API"; import { buildAutoSelectedTemplateConfig, useCoderWorkspace, @@ -9,6 +10,8 @@ import { import type { CoderInfo, CoderTemplate } from "@/common/orpc/schemas/coder"; import type { CoderWorkspaceConfig } from "@/common/types/runtime"; +const actualAPIModule = { ...APIModule }; + const makeTemplate = (name: string, org = "default-org"): CoderTemplate => ({ name, displayName: name, @@ -34,6 +37,7 @@ const apiMock = { }; void mock.module("@/browser/contexts/API", () => ({ + ...actualAPIModule, useAPI: () => ({ api: apiMock, status: "connected" as const, @@ -41,6 +45,10 @@ void mock.module("@/browser/contexts/API", () => ({ }), })); +afterAll(async () => { + await mock.module("@/browser/contexts/API", () => actualAPIModule); +}); + function deferred() { let resolve!: (v: T) => void; let reject!: (e: unknown) => void; diff --git a/src/common/orpc/schemas/workflow.ts b/src/common/orpc/schemas/workflow.ts index 75864e73d8..5512474fb1 100644 --- a/src/common/orpc/schemas/workflow.ts +++ b/src/common/orpc/schemas/workflow.ts @@ -124,6 +124,15 @@ export const WorkflowRunEventSchema = z.discriminatedUnion("type", [ // Optional so legacy persisted events without it still parse. title: z.string().min(1).optional(), }), + z.object({ + sequence: z.number().int().positive(), + type: z.literal("timeout"), + at: IsoDateTimeSchema, + stepId: z.string().min(1), + taskId: z.string().min(1), + phase: z.enum(["soft", "finalization_prompt_sent", "recovered", "hard"]), + details: JsonValueSchema.optional(), + }), z.object({ sequence: z.number().int().positive(), type: z.literal("workflow"), @@ -195,6 +204,18 @@ export const WorkflowEventSequenceSchema = z export const WorkflowStepStatusSchema = z.enum(["started", "completed", "failed", "interrupted"]); +export const WorkflowStepTimeoutMetadataSchema = z + .object({ + executionStartedAt: IsoDateTimeSchema.optional(), + softDeadlineAt: IsoDateTimeSchema.optional(), + hardDeadlineAt: IsoDateTimeSchema.optional(), + softTimedOutAt: IsoDateTimeSchema.optional(), + finalizationToken: z.string().min(1).optional(), + finalizationPromptSentAt: IsoDateTimeSchema.optional(), + hardTimedOutAt: IsoDateTimeSchema.optional(), + }) + .strict(); + export const WorkflowStepRecordSchema = z.object({ stepId: z.string().min(1), inputHash: z.string().min(1), @@ -203,6 +224,7 @@ export const WorkflowStepRecordSchema = z.object({ startedAt: IsoDateTimeSchema, completedAt: IsoDateTimeSchema.optional(), result: StructuredTaskOutputSchema.optional(), + timeout: WorkflowStepTimeoutMetadataSchema.optional(), error: z.string().min(1).optional(), }); diff --git a/src/common/schemas/project.ts b/src/common/schemas/project.ts index 7b8be50f74..716592874a 100644 --- a/src/common/schemas/project.ts +++ b/src/common/schemas/project.ts @@ -135,6 +135,10 @@ export const WorkspaceConfigSchema = z.object({ taskLaunchError: z.string().optional().meta({ description: "Startup failure recorded before an agent task could begin streaming.", }), + taskTimeoutFinalizationTokens: z.array(z.string().min(1)).optional().meta({ + description: + "Idempotency tokens for workflow timeout finalization prompts already sent to this task.", + }), taskRecoveryAttempts: z.number().int().nonnegative().optional().meta({ description: "Completion-tool recovery prompts sent to this agent task since it last completed successfully. Persisted (not in-memory) so crash/restart recovery loops stay bounded; cleared on a successful report, on plan-to-exec handoff, and on user-initiated resume.", diff --git a/src/node/builtinSkills/workflow-authoring.md b/src/node/builtinSkills/workflow-authoring.md index 42faee79fe..2c305520b3 100644 --- a/src/node/builtinSkills/workflow-authoring.md +++ b/src/node/builtinSkills/workflow-authoring.md @@ -216,6 +216,29 @@ const reviews = parallel( ); ``` +Timeouts are optional and explicit. Mux does not provide default workflow-agent timeout durations. When `timeout` is present, both `softMs` and `graceMs` are required positive integer millisecond values: + +```js +const report = agent("Investigate and report useful partial findings if time expires", { + id: "investigate", + schema: { + type: "object", + required: ["summary", "remainingWork"], + properties: { + summary: { type: "string" }, + remainingWork: { type: "array", items: { type: "string" } }, + }, + }, + timeout: { + softMs: 20 * 60_000, + graceMs: 2 * 60_000, + finalInstructions: "Prioritize completed findings and validation evidence.", + }, +}); +``` + +The soft budget starts when the child task begins running; queued/starting time does not count. If the soft timeout expires, Mux soft-interrupts the child turn, sends a synthetic prompt requiring `agent_report` (or `propose_plan` for Plan agents), and waits for the explicit grace period. A valid report during grace completes the step normally; otherwise Mux hard-times-out the child and fails the step. For schema-backed agents, design the schema so partial-but-useful results can still be represented. + ### `parallel(thunks, options?)` Runs independent workflow agent branches concurrently and returns results in input order. Each thunk should call `agent(...)` once. `options.maxParallel` may cap live child tasks. diff --git a/src/node/services/agentSkills/builtInSkillContent.generated.ts b/src/node/services/agentSkills/builtInSkillContent.generated.ts index 362a67d7fa..5b1cd38627 100644 --- a/src/node/services/agentSkills/builtInSkillContent.generated.ts +++ b/src/node/services/agentSkills/builtInSkillContent.generated.ts @@ -7728,6 +7728,29 @@ export const BUILTIN_SKILL_FILES: Record> = { ");", "```", "", + "Timeouts are optional and explicit. Mux does not provide default workflow-agent timeout durations. When `timeout` is present, both `softMs` and `graceMs` are required positive integer millisecond values:", + "", + "```js", + 'const report = agent("Investigate and report useful partial findings if time expires", {', + ' id: "investigate",', + " schema: {", + ' type: "object",', + ' required: ["summary", "remainingWork"],', + " properties: {", + ' summary: { type: "string" },', + ' remainingWork: { type: "array", items: { type: "string" } },', + " },", + " },", + " timeout: {", + " softMs: 20 * 60_000,", + " graceMs: 2 * 60_000,", + ' finalInstructions: "Prioritize completed findings and validation evidence.",', + " },", + "});", + "```", + "", + "The soft budget starts when the child task begins running; queued/starting time does not count. If the soft timeout expires, Mux soft-interrupts the child turn, sends a synthetic prompt requiring `agent_report` (or `propose_plan` for Plan agents), and waits for the explicit grace period. A valid report during grace completes the step normally; otherwise Mux hard-times-out the child and fails the step. For schema-backed agents, design the schema so partial-but-useful results can still be represented.", + "", "### `parallel(thunks, options?)`", "", "Runs independent workflow agent branches concurrently and returns results in input order. Each thunk should call `agent(...)` once. `options.maxParallel` may cap live child tasks.", diff --git a/src/node/services/taskService.test.ts b/src/node/services/taskService.test.ts index 9827a233a2..7d94fca9ec 100644 --- a/src/node/services/taskService.test.ts +++ b/src/node/services/taskService.test.ts @@ -6818,6 +6818,234 @@ describe("TaskService", () => { expect(sendMessage).not.toHaveBeenCalled(); }); + test("requestAgentFinalReportForTimeout records finalization token only after prompt send succeeds", async () => { + const config = await createTestConfig(rootDir); + + const projectPath = path.join(rootDir, "repo"); + const parentWorkspaceId = "parent-111"; + const childTaskId = "task-timeout-child"; + let sendSucceeds = false; + + await saveWorkspaces( + config, + projectPath, + [ + projectWorkspace(projectPath, "parent", parentWorkspaceId, { + aiSettings: { model: "openai:gpt-5.2", thinkingLevel: "medium" }, + }), + projectWorkspace(projectPath, "child-task", childTaskId, { + name: "agent_explore_child", + parentWorkspaceId, + agentType: "explore", + taskStatus: "running", + taskModelString: "openai:gpt-5.2", + taskThinkingLevel: "medium", + }), + ], + testTaskSettings() + ); + + let isStreaming = false; + let callOnAccepted = true; + let queuedOnAccepted: (() => Promise | void) | undefined; + const { aiService, stopStream } = createAIServiceMocks(config, { + isStreaming: mock(() => isStreaming), + }); + const { workspaceService, sendMessage } = createWorkspaceServiceMocks({ + sendMessage: mock(async (...args: unknown[]): Promise> => { + if (!sendSucceeds) { + return Err("send failed"); + } + const internal = args[3] as { onAccepted?: () => Promise | void } | undefined; + if (callOnAccepted) { + await internal?.onAccepted?.(); + } else { + queuedOnAccepted = internal?.onAccepted; + } + return Ok(undefined); + }), + }); + const { taskService } = createTaskServiceHarness(config, { aiService, workspaceService }); + const request = { + workflowRunId: "wfr_timeout", + stepId: "slow-step", + inputHash: "hash", + finalizationToken: "token-1", + }; + + const failedPromptResult = await taskService.requestAgentFinalReportForTimeout( + childTaskId, + request + ); + expect(failedPromptResult).toBe("not_active"); + let childWorkspace = config + .loadConfigOrDefault() + .projects.get(projectPath) + ?.workspaces.find((workspace) => workspace.id === childTaskId); + expect(childWorkspace?.taskTimeoutFinalizationTokens).toBeUndefined(); + + sendSucceeds = true; + const promptedResult = await taskService.requestAgentFinalReportForTimeout( + childTaskId, + request + ); + expect(promptedResult).toBe("prompted"); + childWorkspace = config + .loadConfigOrDefault() + .projects.get(projectPath) + ?.workspaces.find((workspace) => workspace.id === childTaskId); + expect(sendMessage).toHaveBeenLastCalledWith( + childTaskId, + expect.any(String), + expect.anything(), + expect.objectContaining({ startStreamInBackground: true }) + ); + expect(childWorkspace?.taskTimeoutFinalizationTokens).toEqual(["token-1"]); + isStreaming = true; + const alreadyPromptedResult = await taskService.requestAgentFinalReportForTimeout( + childTaskId, + request + ); + expect(alreadyPromptedResult).toBe("prompted"); + expect(stopStream).not.toHaveBeenCalled(); + expect(sendMessage).toHaveBeenCalledTimes(2); + isStreaming = false; + callOnAccepted = false; + const queuedPromptResult = await taskService.requestAgentFinalReportForTimeout(childTaskId, { + ...request, + finalizationToken: "token-queued", + }); + expect(queuedPromptResult).toBe("queued"); + childWorkspace = config + .loadConfigOrDefault() + .projects.get(projectPath) + ?.workspaces.find((workspace) => workspace.id === childTaskId); + expect(childWorkspace?.taskTimeoutFinalizationTokens).toEqual(["token-1"]); + expect(sendMessage).toHaveBeenCalledTimes(3); + await queuedOnAccepted?.(); + childWorkspace = config + .loadConfigOrDefault() + .projects.get(projectPath) + ?.workspaces.find((workspace) => workspace.id === childTaskId); + expect(childWorkspace?.taskTimeoutFinalizationTokens).toEqual(["token-1", "token-queued"]); + }); + + test("requestAgentFinalReportForTimeout requires propose_plan for timed-out plan agents", async () => { + const config = await createTestConfig(rootDir); + + const projectPath = path.join(rootDir, "repo"); + const parentWorkspaceId = "parent-111"; + const childTaskId = "task-timeout-plan-child"; + let sentMessage = ""; + let sentToolPolicy: Array<{ regex_match: string; action: string }> | undefined; + + await saveWorkspaces( + config, + projectPath, + [ + projectWorkspace(projectPath, "parent", parentWorkspaceId, { + aiSettings: { model: "openai:gpt-5.2", thinkingLevel: "medium" }, + }), + projectWorkspace(projectPath, "child-task", childTaskId, { + name: "agent_plan_child", + parentWorkspaceId, + agentId: "plan", + agentType: "plan", + taskStatus: "running", + taskModelString: "openai:gpt-5.2", + taskThinkingLevel: "medium", + }), + ], + testTaskSettings() + ); + + const { aiService } = createAIServiceMocks(config); + const { workspaceService } = createWorkspaceServiceMocks({ + sendMessage: mock( + async ( + _workspaceId: string, + message: string, + options: { toolPolicy?: Array<{ regex_match: string; action: string }> }, + internal?: { onAccepted?: () => Promise | void } + ): Promise> => { + sentMessage = message; + sentToolPolicy = options.toolPolicy; + await internal?.onAccepted?.(); + return Ok(undefined); + } + ), + }); + const { taskService } = createTaskServiceHarness(config, { aiService, workspaceService }); + + const result = await taskService.requestAgentFinalReportForTimeout(childTaskId, { + workflowRunId: "wfr_timeout", + stepId: "plan-step", + inputHash: "hash", + finalizationToken: "plan-token", + }); + expect(result).toBe("prompted"); + + expect(sentToolPolicy).toEqual([{ regex_match: "^propose_plan$", action: "require" }]); + expect(sentMessage).toContain("propose_plan"); + expect(sentMessage).not.toContain("agent_report"); + }); + + test("failAgentTaskForHardTimeout clears queued finalization before aborting the stream", async () => { + const config = await createTestConfig(rootDir); + + const projectPath = path.join(rootDir, "repo"); + const parentWorkspaceId = "parent-111"; + const childTaskId = "task-timeout-child"; + const operations: string[] = []; + + await saveWorkspaces( + config, + projectPath, + [ + projectWorkspace(projectPath, "parent", parentWorkspaceId, { + aiSettings: { model: "openai:gpt-5.2", thinkingLevel: "medium" }, + }), + projectWorkspace(projectPath, "child-task", childTaskId, { + name: "agent_explore_child", + parentWorkspaceId, + agentType: "explore", + taskStatus: "running", + taskModelString: "openai:gpt-5.2", + taskThinkingLevel: "medium", + }), + ], + testTaskSettings() + ); + + const { aiService, stopStream } = createAIServiceMocks(config, { + stopStream: mock((): Promise> => { + operations.push("stopStream"); + return Promise.resolve(Ok(undefined)); + }), + }); + const { workspaceService, clearQueue } = createWorkspaceServiceMocks({ + clearQueue: mock((): Result => { + operations.push("clearQueue"); + return Ok(undefined); + }), + }); + const { taskService } = createTaskServiceHarness(config, { aiService, workspaceService }); + + await taskService.failAgentTaskForHardTimeout(childTaskId, { + workflowRunId: "wfr_timeout", + stepId: "slow-step", + inputHash: "hash", + reason: "timed out", + }); + + expect(clearQueue).toHaveBeenCalledWith(childTaskId); + expect(stopStream).toHaveBeenCalledWith(childTaskId, { + abandonPartial: true, + abortReason: "system", + }); + expect(operations.slice(0, 2)).toEqual(["clearQueue", "stopStream"]); + }); + test("terminateDescendantAgentTask stops stream, removes workspace, and rejects waiters", async () => { const config = await createTestConfig(rootDir); diff --git a/src/node/services/taskService.ts b/src/node/services/taskService.ts index 63a0949441..7b8c7c4be2 100644 --- a/src/node/services/taskService.ts +++ b/src/node/services/taskService.ts @@ -134,6 +134,13 @@ import { export type TaskKind = "agent"; +export class AgentReportWaitTimeoutError extends Error { + constructor() { + super("Timed out waiting for agent_report"); + this.name = "AgentReportWaitTimeoutError"; + } +} + export type AgentTaskStatus = NonNullable; /** @@ -961,6 +968,26 @@ export class ForegroundWaitBackgroundedError extends Error { } } +function buildWorkflowTimeoutFinalizationPrompt( + finalInstructions: string | undefined, + completionToolName: "agent_report" | "propose_plan" +): string { + const reportNoun = completionToolName === "propose_plan" ? "plan" : "report"; + const base = + `Your workflow step time budget has expired. Stop starting new work and prepare a final ${reportNoun} now.\n\n` + + `In your ${reportNoun}:\n` + + "- summarize work completed;\n" + + "- list files changed or inspected;\n" + + "- include validation/test results already obtained;\n" + + "- call out uncertainty and remaining work;\n" + + `- do not run additional long-running tools unless absolutely necessary to write the ${reportNoun}.\n\n` + + getTaskCompletionInstruction({ completionToolName }); + if (finalInstructions == null) { + return base; + } + return `${base}\n\nAdditional workflow-specific finalization instructions:\n${finalInstructions}`; +} + export class TaskService { // Serialize stream-end processing per workspace to avoid races when // finalizing reported tasks and cleanup state transitions. @@ -4196,6 +4223,185 @@ export class TaskService { }); } + async requestAgentFinalReportForTimeout( + taskId: string, + options: { + workflowRunId: string; + stepId: string; + inputHash: string; + finalizationToken: string; + finalInstructions?: string; + } + ): Promise<"prompted" | "queued" | "already_reported" | "not_active"> { + assert(taskId.length > 0, "requestAgentFinalReportForTimeout: taskId must be non-empty"); + assert( + options.finalizationToken.length > 0, + "requestAgentFinalReportForTimeout: finalizationToken must be non-empty" + ); + + const reservation = await this.workspaceEventLocks.withLock(taskId, async () => { + const cfg = this.config.loadConfigOrDefault(); + const entry = findWorkspaceEntry(cfg, taskId); + if (!entry?.workspace.parentWorkspaceId) { + return { status: "not_active" as const }; + } + if (hasCompletedAgentReport(entry.workspace) || this.completedReportsByTaskId.has(taskId)) { + return { status: "already_reported" as const }; + } + if (entry.workspace.taskStatus === "interrupted" && !this.aiService.isStreaming(taskId)) { + return { status: "not_active" as const }; + } + + const tokens = entry.workspace.taskTimeoutFinalizationTokens ?? []; + const alreadyPrompted = tokens.includes(options.finalizationToken); + if (!alreadyPrompted) { + await this.editWorkspaceEntry( + taskId, + (workspace) => { + workspace.taskStatus = "awaiting_report"; + }, + { allowMissing: true } + ); + } + return { status: "reserved" as const, alreadyPrompted }; + }); + + if (reservation.status !== "reserved") { + return reservation.status; + } + if (reservation.alreadyPrompted) { + return "prompted"; + } + if (this.aiService.isStreaming(taskId)) { + await this.aiService.stopStream(taskId, { + soft: true, + abandonPartial: false, + abortReason: "system", + }); + } + + const freshConfig = this.config.loadConfigOrDefault(); + const freshEntry = findWorkspaceEntry(freshConfig, taskId); + if (!freshEntry?.workspace.parentWorkspaceId) { + return "not_active"; + } + if ( + hasCompletedAgentReport(freshEntry.workspace) || + this.completedReportsByTaskId.has(taskId) + ) { + return "already_reported"; + } + let finalizationAccepted = false; + const persistFinalizationToken = async (): Promise => { + await this.workspaceEventLocks.withLock(taskId, async () => { + const cfg = this.config.loadConfigOrDefault(); + const entry = findWorkspaceEntry(cfg, taskId); + if (!entry?.workspace.parentWorkspaceId) { + return; + } + if (hasCompletedAgentReport(entry.workspace) || this.completedReportsByTaskId.has(taskId)) { + return; + } + await this.editWorkspaceEntry( + taskId, + (workspace) => { + const existing = workspace.taskTimeoutFinalizationTokens ?? []; + workspace.taskTimeoutFinalizationTokens = Array.from( + new Set([...existing, options.finalizationToken]) + ); + workspace.taskStatus = "awaiting_report"; + }, + { allowMissing: true } + ); + }); + finalizationAccepted = true; + }; + const completionToolName = (await this.isPlanLikeTaskWorkspace(freshEntry)) + ? "propose_plan" + : "agent_report"; + const model = freshEntry.workspace.taskModelString ?? defaultModel; + const agentId = resolveTaskAgentIdForResume(freshEntry.workspace); + const sendResult = await this.workspaceService.sendMessage( + taskId, + buildWorkflowTimeoutFinalizationPrompt(options.finalInstructions, completionToolName), + { + model, + agentId, + thinkingLevel: freshEntry.workspace.taskThinkingLevel, + experiments: freshEntry.workspace.taskExperiments, + toolPolicy: [{ regex_match: `^${completionToolName}$`, action: "require" }], + }, + { + synthetic: true, + agentInitiated: true, + startStreamInBackground: true, + onAccepted: persistFinalizationToken, + onCanceled: (reason) => { + log.debug("Workflow timeout finalization prompt was canceled", { + taskId, + workflowRunId: options.workflowRunId, + stepId: options.stepId, + reason, + }); + }, + } + ); + if (!sendResult.success) { + log.error("Failed to prompt workflow task for timeout final report", { + taskId, + workflowRunId: options.workflowRunId, + stepId: options.stepId, + error: sendResult.error, + }); + return "not_active"; + } + + return finalizationAccepted ? "prompted" : "queued"; + } + + async failAgentTaskForHardTimeout( + taskId: string, + options: { workflowRunId: string; stepId: string; inputHash: string; reason: string } + ): Promise { + assert(taskId.length > 0, "failAgentTaskForHardTimeout: taskId must be non-empty"); + assert(options.reason.length > 0, "failAgentTaskForHardTimeout: reason must be non-empty"); + + await this.workspaceEventLocks.withLock(taskId, async () => { + const cfg = this.config.loadConfigOrDefault(); + const entry = findWorkspaceEntry(cfg, taskId); + if (!entry?.workspace.parentWorkspaceId) { + return; + } + if (hasCompletedAgentReport(entry.workspace) || this.completedReportsByTaskId.has(taskId)) { + return; + } + try { + const clearQueueResult = this.workspaceService.clearQueue(taskId); + if (!clearQueueResult.success) { + log.debug("failAgentTaskForHardTimeout: clearQueue failed", { + taskId, + error: clearQueueResult.error, + }); + } + } catch (error: unknown) { + log.debug("failAgentTaskForHardTimeout: clearQueue threw", { taskId, error }); + } + try { + await this.aiService.stopStream(taskId, { + abandonPartial: true, + abortReason: "system", + }); + } catch (error: unknown) { + log.debug("failAgentTaskForHardTimeout: stopStream threw", { taskId, error }); + } + await this.terminateAllDescendantAgentTasks(taskId, { workflowRunId: options.workflowRunId }); + await this.failAgentTaskTerminally(taskId, entry, { + errorType: "workflow_agent_timeout", + errorMessage: options.reason, + }); + }); + } + async waitForAgentReport( taskId: string, options?: { @@ -4203,6 +4409,7 @@ export class TaskService { abortSignal?: AbortSignal; requestingWorkspaceId?: string; backgroundOnMessageQueued?: boolean; + onExecutionStarted?: () => void | Promise; } ): Promise<{ reportMarkdown: string; @@ -4386,8 +4593,18 @@ export class TaskService { ? this.startForegroundAwait(requestingWorkspaceId) : null; + let executionStartNotified = false; + const notifyExecutionStarted = () => { + if (executionStartNotified) return; + executionStartNotified = true; + void Promise.resolve(options?.onExecutionStarted?.()).catch((error: unknown) => { + log.error("waitForAgentReport execution-start callback failed", { taskId, error }); + }); + }; + const startReportTimeout = () => { if (timeout) return; + notifyExecutionStarted(); timeout = setTimeout(() => { // Prefer a persisted terminal failure over a generic timeout so late // awaits surface the typed failure (e.g. model_refusal) even when the @@ -4395,7 +4612,7 @@ export class TaskService { void (async () => { const persistedFailure = await tryReadPersistedFailureError().catch(() => null); entry.cleanup(); - reject(persistedFailure ?? new Error("Timed out waiting for agent_report")); + reject(persistedFailure ?? new AgentReportWaitTimeoutError()); })(); }, timeoutMs); }; diff --git a/src/node/services/workflows/WorkflowRunStore.ts b/src/node/services/workflows/WorkflowRunStore.ts index 041d06987e..e4ed37fb7c 100644 --- a/src/node/services/workflows/WorkflowRunStore.ts +++ b/src/node/services/workflows/WorkflowRunStore.ts @@ -578,6 +578,31 @@ export class WorkflowRunStore { }); } + async recordStepTimeoutMetadata( + runId: string, + input: { + stepId: string; + inputHash: string; + taskId: string; + startedAt: string; + timeout: NonNullable; + }, + options: AppendWorkflowRunEventOptions = {} + ): Promise { + await this.appendStepRecord( + runId, + { + stepId: input.stepId, + inputHash: input.inputHash, + taskId: input.taskId, + startedAt: input.startedAt, + status: "started", + timeout: input.timeout, + }, + options + ); + } + async recordStepFailed( runId: string, input: { @@ -1056,15 +1081,42 @@ function mergeWorkflowStepRecords( nextRecord?: WorkflowStepRecord ): WorkflowStepRecord[] { const byKey = new Map(); + const mergeRecord = (record: WorkflowStepRecord): void => { + const key = getWorkflowStepKey(record); + const previous = byKey.get(key); + byKey.set(key, { + ...record, + timeout: mergeWorkflowStepTimeoutMetadata(previous, record), + }); + }; for (const record of records) { - byKey.set(getWorkflowStepKey(record), record); + mergeRecord(record); } if (nextRecord !== undefined) { - byKey.set(getWorkflowStepKey(nextRecord), nextRecord); + mergeRecord(nextRecord); } return Array.from(byKey.values()); } +function mergeWorkflowStepTimeoutMetadata( + previous: WorkflowStepRecord | undefined, + next: WorkflowStepRecord +): WorkflowStepRecord["timeout"] { + if (next.timeout != null) { + if (previous?.timeout != null && previous.taskId === next.taskId) { + return { ...previous.timeout, ...next.timeout }; + } + return next.timeout; + } + if (previous?.timeout == null) { + return undefined; + } + if (previous.taskId != null && next.taskId != null && previous.taskId !== next.taskId) { + return undefined; + } + return previous.timeout; +} + function hashSource(source: string): string { return `sha256:${crypto.createHash("sha256").update(source).digest("hex")}`; } diff --git a/src/node/services/workflows/WorkflowRunner.test.ts b/src/node/services/workflows/WorkflowRunner.test.ts index c498cc3965..f08593b008 100644 --- a/src/node/services/workflows/WorkflowRunner.test.ts +++ b/src/node/services/workflows/WorkflowRunner.test.ts @@ -737,6 +737,74 @@ describe("WorkflowRunner", () => { expect(interruptRun).toHaveBeenCalledTimes(1); }); + test("pipeline hard timeouts keep the timed_out task event without adding a failed task event", async () => { + using tmp = new DisposableTempDir("workflow-runner-pipeline-timeout-terminal-event"); + const store = new WorkflowRunStore({ + sessionDir: tmp.path, + staleLeaseMs: WORKFLOW_RUNNER_TEST_STALE_LEASE_MS, + }); + const runId = "wfr_pipeline_timeout_terminal_event"; + await store.createRun({ + id: runId, + workspaceId: "workspace-1", + workflow: definition, + source: `export default function workflow({ agent, pipeline }) { + return pipeline(["slow", "sibling"], (item) => agent("Stage " + item, { id: item, timeout: { softMs: 1000, graceMs: 2000 } })); +} +`, + args: {}, + now: "2026-05-29T00:00:00.000Z", + }); + const timeoutError = new Error("wait expired"); + timeoutError.name = "AgentReportWaitTimeoutError"; + const releaseSibling = createDeferred(); + const interruptRun = mock(async () => { + releaseSibling.resolve(); + }); + const runner = createRunner(store, { + async runAgent() { + throw new Error("pipeline should start agent tasks without using runAgent"); + }, + async createAgentTasks(specs, lifecycle) { + for (const [index, spec] of specs.entries()) { + await lifecycle?.onTaskCreated?.(index, `task_${spec.id}`); + } + return specs.map((spec) => ({ taskId: `task_${spec.id}`, status: "running" as const })); + }, + async waitForAgentTask(taskId, _spec, waitOptions) { + await waitOptions?.onExecutionStarted?.(); + if (taskId === "task_slow") { + throw timeoutError; + } + await releaseSibling.promise; + return { taskId, reportMarkdown: "sibling", structuredOutput: { label: "sibling" } }; + }, + async requestAgentFinalReportForTimeout() { + return "prompted"; + }, + async failAgentTaskForHardTimeout() { + // The workflow timeout path records the timed_out event before surfacing the failure. + }, + interruptRun, + }); + + await expect(runner.run(runId)).rejects.toThrow("exceeded its soft timeout"); + expect(interruptRun).toHaveBeenCalledTimes(1); + const run = await store.getRun(runId); + expect( + run.events.some( + (event) => + event.type === "task" && event.taskId === "task_slow" && event.status === "timed_out" + ) + ).toBe(true); + expect( + run.events.some( + (event) => + event.type === "task" && event.taskId === "task_slow" && event.status === "failed" + ) + ).toBe(false); + }); + test("workflow primitive requires a stable id before creating child runs", async () => { using tmp = new DisposableTempDir("workflow-runner-nested-workflow-missing-id"); const store = new WorkflowRunStore({ @@ -986,6 +1054,695 @@ describe("WorkflowRunner", () => { expect(runAgent).not.toHaveBeenCalled(); }); + test("normalizes explicit workflow agent timeout options before spawning", async () => { + using tmp = new DisposableTempDir("workflow-runner-agent-timeout-options"); + const store = new WorkflowRunStore({ + sessionDir: tmp.path, + staleLeaseMs: WORKFLOW_RUNNER_TEST_STALE_LEASE_MS, + }); + await store.createRun({ + id: "wfr_agent_timeout_options", + workspaceId: "workspace-1", + workflow: definition, + source: `export default function workflow({ agent }) { + return agent("Return structured output", { + id: "timeout-step", + schema: { type: "object", properties: { ok: { type: "boolean" } }, required: ["ok"] }, + timeout: { softMs: 1000, graceMs: 2000, finalInstructions: "Summarize only completed work." }, + }); +} +`, + args: {}, + now: "2026-05-29T00:00:00.000Z", + }); + const seenSpecs: WorkflowAgentSpec[] = []; + const runner = createRunner(store, { + async runAgent(spec) { + seenSpecs.push(spec); + return { taskId: "task_timeout", reportMarkdown: "ok", structuredOutput: { ok: true } }; + }, + }); + + await expect(runner.run("wfr_agent_timeout_options")).resolves.toEqual({ + reportMarkdown: JSON.stringify({ ok: true }), + }); + expect(seenSpecs[0]?.timeout).toEqual({ + softMs: 1000, + graceMs: 2000, + finalInstructions: "Summarize only completed work.", + }); + }); + + test("fails before spawning when workflow agent timeout is incomplete", async () => { + using tmp = new DisposableTempDir("workflow-runner-agent-timeout-invalid"); + const store = new WorkflowRunStore({ + sessionDir: tmp.path, + staleLeaseMs: WORKFLOW_RUNNER_TEST_STALE_LEASE_MS, + }); + await store.createRun({ + id: "wfr_agent_timeout_invalid", + workspaceId: "workspace-1", + workflow: definition, + source: `export default function workflow({ agent }) { + return agent("Invalid timeout", { id: "invalid-timeout", timeout: { softMs: 1000 } }); +} +`, + args: {}, + now: "2026-05-29T00:00:00.000Z", + }); + const runAgent = mock(async () => ({ + taskId: "task_1", + reportMarkdown: "summary", + structuredOutput: {}, + })); + const runner = createRunner(store, { runAgent }); + + await expect(runner.run("wfr_agent_timeout_invalid")).rejects.toThrow( + "agent timeout.graceMs must be a positive integer" + ); + expect(runAgent).not.toHaveBeenCalled(); + }); + + test("requests a final report when an agent soft timeout recovers during grace", async () => { + using tmp = new DisposableTempDir("workflow-runner-agent-timeout-recovered"); + const store = new WorkflowRunStore({ + sessionDir: tmp.path, + staleLeaseMs: WORKFLOW_RUNNER_TEST_STALE_LEASE_MS, + }); + await store.createRun({ + id: "wfr_agent_timeout_recovered", + workspaceId: "workspace-1", + workflow: definition, + source: `export default function workflow({ agent }) { + const result = agent("Slow work", { + id: "slow-step", + schema: { type: "object", properties: { ok: { type: "boolean" } }, required: ["ok"] }, + timeout: { softMs: 1000, graceMs: 2000, finalInstructions: "No extra tools." }, + }); + return { reportMarkdown: result.ok ? "recovered" : "bad" }; +} +`, + args: {}, + now: "2026-05-29T00:00:00.000Z", + }); + const timeoutError = new Error("soft wait expired"); + timeoutError.name = "AgentReportWaitTimeoutError"; + const waitTimeouts: Array = []; + const finalizationRequests: unknown[] = []; + const runner = createRunner(store, { + async runAgent() { + throw new Error("timeout steps should use createAgentTasks so the runner controls waits"); + }, + async createAgentTasks(specs, lifecycle) { + await lifecycle?.onTaskCreated?.(0, "task_slow"); + expect(specs).toHaveLength(1); + return [{ taskId: "task_slow", status: "running" }]; + }, + async waitForAgentTask(_taskId, _spec, waitOptions) { + await waitOptions?.onExecutionStarted?.(); + waitTimeouts.push(waitOptions?.timeoutMs); + if (waitTimeouts.length === 1) { + throw timeoutError; + } + return { taskId: "task_slow", reportMarkdown: "ok", structuredOutput: { ok: true } }; + }, + async requestAgentFinalReportForTimeout(taskId, request) { + finalizationRequests.push({ taskId, request }); + return "prompted"; + }, + async failAgentTaskForHardTimeout() { + throw new Error("hard timeout should not run when grace recovers"); + }, + }); + + await expect(runner.run("wfr_agent_timeout_recovered")).resolves.toEqual({ + reportMarkdown: "recovered", + }); + + expect(waitTimeouts).toEqual([1000, 2000]); + expect(finalizationRequests).toHaveLength(1); + const run = await store.getRun("wfr_agent_timeout_recovered"); + expect( + run.events.filter((event) => event.type === "timeout").map((event) => event.phase) + ).toEqual(["soft", "finalization_prompt_sent", "recovered"]); + }); + + test("extends persisted hard deadline when timeout finalization prompt is accepted", async () => { + using tmp = new DisposableTempDir("workflow-runner-agent-timeout-finalization-deadline"); + const store = new WorkflowRunStore({ + sessionDir: tmp.path, + staleLeaseMs: WORKFLOW_RUNNER_TEST_STALE_LEASE_MS, + }); + const runId = "wfr_agent_timeout_finalization_deadline"; + await store.createRun({ + id: runId, + workspaceId: "workspace-1", + workflow: definition, + source: `export default function workflow({ agent }) { + const result = agent("Slow work", { + id: "slow-step", + schema: { type: "object", properties: { ok: { type: "boolean" } }, required: ["ok"] }, + timeout: { softMs: 1000, graceMs: 2000 }, + }); + return { reportMarkdown: result.ok ? "recovered" : "bad" }; +} +`, + args: {}, + now: "2026-05-29T00:00:00.000Z", + }); + const timeoutError = new Error("soft wait expired"); + timeoutError.name = "AgentReportWaitTimeoutError"; + let nowMs = Date.parse("2026-05-29T00:00:01.000Z"); + let waitCount = 0; + const runner = new WorkflowRunner({ + runStore: store, + runtimeFactory: new QuickJSRuntimeFactory(), + runnerId: "runner-a", + clock: { + nowIso: () => new Date(nowMs).toISOString(), + nowMs: () => nowMs, + }, + taskAdapter: { + async runAgent() { + throw new Error("timeout steps should use createAgentTasks so the runner controls waits"); + }, + async createAgentTasks(_specs, lifecycle) { + await lifecycle?.onTaskCreated?.(0, "task_slow"); + return [{ taskId: "task_slow", status: "running" }]; + }, + async waitForAgentTask(_taskId, _spec, waitOptions) { + await waitOptions?.onExecutionStarted?.(); + waitCount += 1; + if (waitCount === 1) { + throw timeoutError; + } + return { taskId: "task_slow", reportMarkdown: "ok", structuredOutput: { ok: true } }; + }, + async requestAgentFinalReportForTimeout() { + nowMs = Date.parse("2026-05-29T00:00:06.000Z"); + return "prompted"; + }, + }, + }); + + await expect(runner.run(runId)).resolves.toEqual({ reportMarkdown: "recovered" }); + const run = await store.getRun(runId); + const timeout = run.steps.find((step) => step.stepId === "slow-step")?.timeout; + expect(timeout?.finalizationPromptSentAt).toBe("2026-05-29T00:00:06.000Z"); + expect(timeout?.hardDeadlineAt).toBe("2026-05-29T00:00:08.000Z"); + }); + + test("does not mark timeout finalization prompt sent before TaskService accepts it", async () => { + using tmp = new DisposableTempDir("workflow-runner-agent-timeout-finalization-queued"); + const store = new WorkflowRunStore({ + sessionDir: tmp.path, + staleLeaseMs: WORKFLOW_RUNNER_TEST_STALE_LEASE_MS, + }); + const runId = "wfr_agent_timeout_finalization_queued"; + await store.createRun({ + id: runId, + workspaceId: "workspace-1", + workflow: definition, + source: `export default function workflow({ agent }) { + const result = agent("Slow work", { + id: "slow-step", + schema: { type: "object", properties: { ok: { type: "boolean" } }, required: ["ok"] }, + timeout: { softMs: 1000, graceMs: 2000 }, + }); + return { reportMarkdown: result.ok ? "recovered" : "bad" }; +} +`, + args: {}, + now: "2026-05-29T00:00:00.000Z", + }); + const timeoutError = new Error("soft wait expired"); + timeoutError.name = "AgentReportWaitTimeoutError"; + let waitCount = 0; + const runner = createRunner(store, { + async runAgent() { + throw new Error("timeout steps should use createAgentTasks so the runner controls waits"); + }, + async createAgentTasks(_specs, lifecycle) { + await lifecycle?.onTaskCreated?.(0, "task_slow"); + return [{ taskId: "task_slow", status: "running" }]; + }, + async waitForAgentTask(_taskId, _spec, waitOptions) { + await waitOptions?.onExecutionStarted?.(); + waitCount += 1; + if (waitCount === 1) { + throw timeoutError; + } + return { taskId: "task_slow", reportMarkdown: "ok", structuredOutput: { ok: true } }; + }, + async requestAgentFinalReportForTimeout() { + return "queued"; + }, + }); + + await expect(runner.run(runId)).resolves.toEqual({ reportMarkdown: "recovered" }); + const run = await store.getRun(runId); + expect( + run.events.filter((event) => event.type === "timeout").map((event) => event.phase) + ).toEqual(["soft", "recovered"]); + expect( + run.steps.find((step) => step.stepId === "slow-step")?.timeout?.finalizationPromptSentAt + ).toBeUndefined(); + }); + + test("fails and hard-times-out an agent that does not report during grace", async () => { + using tmp = new DisposableTempDir("workflow-runner-agent-timeout-hard"); + const store = new WorkflowRunStore({ + sessionDir: tmp.path, + staleLeaseMs: WORKFLOW_RUNNER_TEST_STALE_LEASE_MS, + }); + await store.createRun({ + id: "wfr_agent_timeout_hard", + workspaceId: "workspace-1", + workflow: definition, + source: `export default function workflow({ agent }) { + return agent("Slow work", { id: "slow-step", timeout: { softMs: 1000, graceMs: 2000 } }); +} +`, + args: {}, + now: "2026-05-29T00:00:00.000Z", + }); + const timeoutError = new Error("wait expired"); + timeoutError.name = "AgentReportWaitTimeoutError"; + const hardTimeouts: unknown[] = []; + const runner = createRunner(store, { + async runAgent() { + throw new Error("timeout steps should use createAgentTasks so the runner controls waits"); + }, + async createAgentTasks(_specs, lifecycle) { + await lifecycle?.onTaskCreated?.(0, "task_slow"); + return [{ taskId: "task_slow", status: "running" }]; + }, + async waitForAgentTask(_taskId, _spec, waitOptions) { + await waitOptions?.onExecutionStarted?.(); + throw timeoutError; + }, + async requestAgentFinalReportForTimeout() { + return "prompted"; + }, + async failAgentTaskForHardTimeout(taskId, request) { + hardTimeouts.push({ taskId, request }); + }, + }); + + await expect(runner.run("wfr_agent_timeout_hard")).rejects.toThrow( + "Workflow agent step slow-step exceeded its soft timeout (1000ms) and did not produce a valid agent_report within the grace period (2000ms)." + ); + expect(hardTimeouts).toHaveLength(1); + const run = await store.getRun("wfr_agent_timeout_hard"); + expect( + run.events.filter((event) => event.type === "timeout").map((event) => event.phase) + ).toContain("hard"); + }); + + test("does not persist workflow agent timeout deadlines before the child starts running", async () => { + using tmp = new DisposableTempDir("workflow-runner-agent-timeout-queued-start"); + const store = new WorkflowRunStore({ + sessionDir: tmp.path, + staleLeaseMs: WORKFLOW_RUNNER_TEST_STALE_LEASE_MS, + }); + const runId = "wfr_agent_timeout_queued_start"; + await store.createRun({ + id: runId, + workspaceId: "workspace-1", + workflow: definition, + source: `export default function workflow({ agent }) { + const result = agent("Queued work", { + id: "queued-step", + schema: { type: "object", properties: { ok: { type: "boolean" } }, required: ["ok"] }, + timeout: { softMs: 1000, graceMs: 2000 }, + }); + return { reportMarkdown: result.ok ? "started" : "bad" }; +} +`, + args: {}, + now: "2026-05-29T00:00:00.000Z", + }); + const spec: WorkflowAgentSpec = { + id: "queued-step", + prompt: "Queued work", + outputSchema: { + type: "object", + properties: { ok: { type: "boolean" } }, + required: ["ok"], + }, + timeout: { softMs: 1000, graceMs: 2000 }, + }; + const inputHash = hashWorkflowStepInput(spec.id, spec); + let inspectedBeforeStart = false; + const runner = new WorkflowRunner({ + runStore: store, + runtimeFactory: new QuickJSRuntimeFactory(), + runnerId: "runner-a", + clock: { + nowIso: () => "2026-05-29T00:00:01.000Z", + nowMs: () => Date.parse("2026-05-29T00:00:01.000Z"), + }, + taskAdapter: { + async runAgent() { + throw new Error("timeout steps should use createAgentTasks so the runner controls waits"); + }, + async createAgentTasks(_specs, lifecycle) { + await lifecycle?.onTaskCreated?.(0, "task_queued"); + return [{ taskId: "task_queued", status: "queued" }]; + }, + async waitForAgentTask(_taskId, _spec, waitOptions) { + const stepBeforeStart = await store.getStep(runId, spec.id, inputHash); + expect(stepBeforeStart?.timeout).toBeUndefined(); + inspectedBeforeStart = true; + await waitOptions?.onExecutionStarted?.(); + return { taskId: "task_queued", reportMarkdown: "ok", structuredOutput: { ok: true } }; + }, + }, + }); + + await expect(runner.run(runId)).resolves.toEqual({ reportMarkdown: "started" }); + expect(inspectedBeforeStart).toBe(true); + const stepAfterStart = await store.getStep(runId, spec.id, inputHash); + expect(stepAfterStart?.timeout).toEqual({ + executionStartedAt: "2026-05-29T00:00:01.000Z", + softDeadlineAt: "2026-05-29T00:00:02.000Z", + }); + }); + + test("resets timeout metadata when validation retries start a fresh agent task", async () => { + using tmp = new DisposableTempDir("workflow-runner-agent-timeout-retry-reset"); + const store = new WorkflowRunStore({ + sessionDir: tmp.path, + staleLeaseMs: WORKFLOW_RUNNER_TEST_STALE_LEASE_MS, + }); + const runId = "wfr_agent_timeout_retry_reset"; + await store.createRun({ + id: runId, + workspaceId: "workspace-1", + workflow: definition, + source: `export default function workflow({ agent }) { + const result = agent("Slow work", { + id: "slow-step", + schema: { type: "object", properties: { ok: { type: "boolean" } }, required: ["ok"] }, + timeout: { softMs: 1000, graceMs: 2000 }, + }); + return { reportMarkdown: result.ok ? "recovered" : "bad" }; +} +`, + args: {}, + now: "2026-05-29T00:00:00.000Z", + }); + const timeoutError = new Error("soft wait expired"); + timeoutError.name = "AgentReportWaitTimeoutError"; + const taskIds = ["task_first", "task_retry"]; + const waitCalls: Array<{ taskId: string; timeoutMs: number | undefined }> = []; + const finalizationRequests: string[] = []; + const runner = createRunner(store, { + async runAgent() { + throw new Error("timeout steps should use createAgentTasks so the runner controls waits"); + }, + async createAgentTasks(_specs, lifecycle) { + const taskId = taskIds.shift(); + expect(taskId).toBeDefined(); + if (taskId == null) { + throw new Error("test expected another task id"); + } + await lifecycle?.onTaskCreated?.(0, taskId); + return [{ taskId, status: "running" }]; + }, + async waitForAgentTask(taskId, _spec, waitOptions) { + await waitOptions?.onExecutionStarted?.(); + waitCalls.push({ taskId, timeoutMs: waitOptions?.timeoutMs }); + if (taskId === "task_first" && waitCalls.length === 1) { + throw timeoutError; + } + if (taskId === "task_first") { + return { taskId, reportMarkdown: "invalid", structuredOutput: { ok: "not boolean" } }; + } + return { taskId, reportMarkdown: "ok", structuredOutput: { ok: true } }; + }, + async requestAgentFinalReportForTimeout(taskId) { + finalizationRequests.push(taskId); + return "prompted"; + }, + async failAgentTaskForHardTimeout() { + throw new Error("hard timeout should not run when retry gets a fresh budget"); + }, + }); + + await expect(runner.run(runId)).resolves.toEqual({ reportMarkdown: "recovered" }); + expect(waitCalls).toEqual([ + { taskId: "task_first", timeoutMs: 1000 }, + { taskId: "task_first", timeoutMs: 2000 }, + { taskId: "task_retry", timeoutMs: 1000 }, + ]); + expect(finalizationRequests).toEqual(["task_first"]); + }); + + test("uses a task-accepted report instead of recording hard timeout failure", async () => { + using tmp = new DisposableTempDir("workflow-runner-agent-timeout-hard-report-race"); + const store = new WorkflowRunStore({ + sessionDir: tmp.path, + staleLeaseMs: WORKFLOW_RUNNER_TEST_STALE_LEASE_MS, + }); + const runId = "wfr_agent_timeout_hard_report_race"; + await store.createRun({ + id: runId, + workspaceId: "workspace-1", + workflow: definition, + source: `export default function workflow({ agent }) { + const result = agent("Slow work", { + id: "slow-step", + schema: { type: "object", properties: { ok: { type: "boolean" } }, required: ["ok"] }, + timeout: { softMs: 1000, graceMs: 2000 }, + }); + return { reportMarkdown: result.ok ? "accepted" : "bad" }; +} +`, + args: {}, + now: "2026-05-29T00:00:00.000Z", + }); + const timeoutError = new Error("wait expired"); + timeoutError.name = "AgentReportWaitTimeoutError"; + let waitCount = 0; + const failHardTimeout = mock(async () => undefined); + const runner = createRunner(store, { + async runAgent() { + throw new Error("timeout steps should use createAgentTasks so the runner controls waits"); + }, + async createAgentTasks(_specs, lifecycle) { + await lifecycle?.onTaskCreated?.(0, "task_slow"); + return [{ taskId: "task_slow", status: "running" }]; + }, + async waitForAgentTask(_taskId, _spec, waitOptions) { + await waitOptions?.onExecutionStarted?.(); + waitCount += 1; + if (waitCount <= 2) { + throw timeoutError; + } + return { taskId: "task_slow", reportMarkdown: "ok", structuredOutput: { ok: true } }; + }, + async requestAgentFinalReportForTimeout() { + return "prompted"; + }, + failAgentTaskForHardTimeout: failHardTimeout, + }); + + await expect(runner.run(runId)).resolves.toEqual({ reportMarkdown: "accepted" }); + expect(failHardTimeout).not.toHaveBeenCalled(); + const run = await store.getRun(runId); + expect(run.steps.find((step) => step.stepId === "slow-step")?.status).toBe("completed"); + }); + + test("restarts timeout-configured resumed agent steps whose task workspace is gone", async () => { + using tmp = new DisposableTempDir("workflow-runner-agent-timeout-missing-task-restart"); + const store = new WorkflowRunStore({ + sessionDir: tmp.path, + staleLeaseMs: WORKFLOW_RUNNER_TEST_STALE_LEASE_MS, + }); + const runId = "wfr_agent_timeout_missing_task_restart"; + const source = `export default function workflow({ agent }) { + const result = agent("Slow work", { + id: "slow-step", + schema: { type: "object", properties: { ok: { type: "boolean" } }, required: ["ok"] }, + timeout: { softMs: 1000, graceMs: 2000 }, + }); + return { reportMarkdown: result.ok ? "restarted" : "bad" }; +} +`; + await store.createRun({ + id: runId, + workspaceId: "workspace-1", + workflow: definition, + source, + args: {}, + now: "2026-05-29T00:00:00.000Z", + }); + const spec: WorkflowAgentSpec = { + id: "slow-step", + prompt: "Slow work", + outputSchema: { + type: "object", + properties: { ok: { type: "boolean" } }, + required: ["ok"], + }, + timeout: { softMs: 1000, graceMs: 2000 }, + }; + const inputHash = hashWorkflowStepInput(spec.id, spec); + await store.recordStepStarted(runId, { + stepId: spec.id, + inputHash, + taskId: "task_missing", + startedAt: "2026-05-29T00:00:00.000Z", + }); + const waitCalls: string[] = []; + const runner = createRunner(store, { + async runAgent() { + throw new Error("timeout steps should use createAgentTasks so the runner controls waits"); + }, + async createAgentTasks(_specs, lifecycle) { + await lifecycle?.onTaskCreated?.(0, "task_restarted"); + return [{ taskId: "task_restarted", status: "running" }]; + }, + async waitForAgentTask(taskId, _spec, waitOptions) { + waitCalls.push(taskId); + if (taskId === "task_missing") { + throw new Error("Task not found"); + } + await waitOptions?.onExecutionStarted?.(); + return { taskId, reportMarkdown: "ok", structuredOutput: { ok: true } }; + }, + }); + + await expect(runner.run(runId)).resolves.toEqual({ reportMarkdown: "restarted" }); + expect(waitCalls).toEqual(["task_missing", "task_restarted"]); + }); + + test("records terminal task events for non-timeout failures from timeout waits", async () => { + using tmp = new DisposableTempDir("workflow-runner-agent-timeout-terminal-event"); + const store = new WorkflowRunStore({ + sessionDir: tmp.path, + staleLeaseMs: WORKFLOW_RUNNER_TEST_STALE_LEASE_MS, + }); + const runId = "wfr_agent_timeout_terminal_event"; + await store.createRun({ + id: runId, + workspaceId: "workspace-1", + workflow: definition, + source: `export default function workflow({ agent }) { + return agent("Slow work", { id: "slow-step", timeout: { softMs: 1000, graceMs: 2000 } }); +} +`, + args: {}, + now: "2026-05-29T00:00:00.000Z", + }); + const runner = createRunner(store, { + async runAgent() { + throw new Error("timeout steps should use createAgentTasks so the runner controls waits"); + }, + async createAgentTasks(_specs, lifecycle) { + await lifecycle?.onTaskCreated?.(0, "task_failed"); + return [{ taskId: "task_failed", status: "running" }]; + }, + async waitForAgentTask() { + throw new Error("model refused"); + }, + }); + + await expect(runner.run(runId)).rejects.toThrow("model refused"); + const run = await store.getRun(runId); + expect( + run.events.some( + (event) => + event.type === "task" && event.taskId === "task_failed" && event.status === "failed" + ) + ).toBe(true); + }); + + test("resumes a timeout-finalizing agent without sending a duplicate finalization prompt", async () => { + using tmp = new DisposableTempDir("workflow-runner-agent-timeout-resume-grace"); + const store = new WorkflowRunStore({ + sessionDir: tmp.path, + staleLeaseMs: WORKFLOW_RUNNER_TEST_STALE_LEASE_MS, + }); + const source = `export default function workflow({ agent }) { + const result = agent("Slow work", { + id: "slow-step", + schema: { type: "object", properties: { ok: { type: "boolean" } }, required: ["ok"] }, + timeout: { softMs: 1000, graceMs: 2000 }, + }); + return { reportMarkdown: result.ok ? "recovered" : "bad" }; +} +`; + await store.createRun({ + id: "wfr_agent_timeout_resume_grace", + workspaceId: "workspace-1", + workflow: definition, + source, + args: {}, + now: "2026-05-29T00:00:00.000Z", + }); + const spec: WorkflowAgentSpec = { + id: "slow-step", + prompt: "Slow work", + outputSchema: { + type: "object", + properties: { ok: { type: "boolean" } }, + required: ["ok"], + }, + timeout: { softMs: 1000, graceMs: 2000 }, + }; + const inputHash = hashWorkflowStepInput(spec.id, spec); + await store.recordStepStarted("wfr_agent_timeout_resume_grace", { + stepId: spec.id, + inputHash, + taskId: "task_slow", + startedAt: "2026-05-29T00:00:00.000Z", + }); + await store.recordStepTimeoutMetadata("wfr_agent_timeout_resume_grace", { + stepId: spec.id, + inputHash, + taskId: "task_slow", + startedAt: "2026-05-29T00:00:00.000Z", + timeout: { + executionStartedAt: "2026-05-29T00:00:00.000Z", + softDeadlineAt: "2026-05-29T00:00:01.000Z", + softTimedOutAt: "2026-05-29T00:00:01.000Z", + hardDeadlineAt: "2026-05-29T00:00:03.000Z", + finalizationToken: "existing-token", + finalizationPromptSentAt: "2026-05-29T00:00:01.100Z", + }, + }); + const waitTimeouts: Array = []; + const requestFinalReport = mock(async () => "prompted" as const); + const runner = new WorkflowRunner({ + runStore: store, + runtimeFactory: new QuickJSRuntimeFactory(), + runnerId: "runner-a", + clock: { + nowIso: () => "2026-05-29T00:00:01.500Z", + nowMs: () => Date.parse("2026-05-29T00:00:01.500Z"), + }, + taskAdapter: { + async runAgent() { + throw new Error("resume should wait on the started task"); + }, + async waitForAgentTask(_taskId, _spec, waitOptions) { + await waitOptions?.onExecutionStarted?.(); + waitTimeouts.push(waitOptions?.timeoutMs); + return { taskId: "task_slow", reportMarkdown: "ok", structuredOutput: { ok: true } }; + }, + requestAgentFinalReportForTimeout: requestFinalReport, + }, + }); + + await expect(runner.run("wfr_agent_timeout_resume_grace")).resolves.toEqual({ + reportMarkdown: "recovered", + }); + expect(waitTimeouts).toEqual([1500]); + expect(requestFinalReport).not.toHaveBeenCalled(); + }); + test("fails before spawning when workflow agent outputSchema is invalid", async () => { using tmp = new DisposableTempDir("workflow-runner-invalid-output-schema"); const store = new WorkflowRunStore({ diff --git a/src/node/services/workflows/WorkflowRunner.ts b/src/node/services/workflows/WorkflowRunner.ts index 5041b15e07..c19d8a77df 100644 --- a/src/node/services/workflows/WorkflowRunner.ts +++ b/src/node/services/workflows/WorkflowRunner.ts @@ -35,6 +35,12 @@ class WorkflowAgentOutputValidationError extends Error { } } +export interface WorkflowAgentTimeoutSpec { + softMs: number; + graceMs: number; + finalInstructions?: string; +} + export interface WorkflowAgentSpec { id: string; prompt: string; @@ -44,6 +50,7 @@ export interface WorkflowAgentSpec { thinkingLevel?: ParsedThinkingInput; isolation?: "fork" | "none"; outputSchema?: unknown; + timeout?: WorkflowAgentTimeoutSpec; /** Internal marker for new `agent(prompt, { id })` prose-only steps. */ markdownOnly?: boolean; /** @@ -59,6 +66,7 @@ export interface WorkflowAgentWaitOptions { abortSignal?: AbortSignal; timeoutMs?: number; backgroundOnMessageQueued?: boolean; + onExecutionStarted?: () => void | Promise; } export type WorkflowAgentResult = StructuredTaskOutput & { taskId: string }; @@ -166,6 +174,25 @@ export interface WorkflowTaskAdapter { spec: WorkflowAgentSpec, waitOptions?: WorkflowAgentWaitOptions ): Promise; + requestAgentFinalReportForTimeout?( + taskId: string, + options: { + workflowRunId: string; + stepId: string; + inputHash: string; + finalizationToken: string; + finalInstructions?: string; + } + ): Promise<"prompted" | "queued" | "already_reported" | "not_active">; + failAgentTaskForHardTimeout?( + taskId: string, + options: { + workflowRunId: string; + stepId: string; + inputHash: string; + reason: string; + } + ): Promise; applyPatch?( spec: WorkflowApplyPatchSpec, options?: { abortSignal?: AbortSignal } @@ -248,6 +275,28 @@ function parseParallelAgentsOptions(raw: unknown): { maxParallel?: number } { return parseWorkflowParallelOptions(raw, "parallel"); } +function isAgentReportWaitTimeoutError(error: unknown): boolean { + return error instanceof Error && error.name === "AgentReportWaitTimeoutError"; +} + +function buildWorkflowAgentTimeoutFinalizationToken( + runId: string, + step: { spec: WorkflowAgentSpec; inputHash: string; taskId: string }, + softTimedOutAt: string +): string { + return `workflow-agent-timeout:${runId}:${step.spec.id}:${step.inputHash}:${step.taskId}:${softTimedOutAt}`; +} + +function createWorkflowAgentHardTimeoutError(message: string): Error { + const error = new Error(message); + error.name = "WorkflowAgentHardTimeoutError"; + return error; +} + +function isWorkflowAgentHardTimeoutError(error: unknown): boolean { + return error instanceof Error && error.name === "WorkflowAgentHardTimeoutError"; +} + function shouldRestartUnrecoverableStartedTask(error: unknown): boolean { const message = getErrorMessage(error); return message === "Task not found" || message === "Task interrupted"; @@ -747,6 +796,15 @@ export class WorkflowRunner { await this.runStore.recordStepCompleted(runId, input, { expectedLeaseOwnerId: this.runnerId }); } + private async recordStepTimeoutMetadata( + runId: string, + input: Parameters[1] + ): Promise { + await this.runStore.recordStepTimeoutMetadata(runId, input, { + expectedLeaseOwnerId: this.runnerId, + }); + } + private async recordStepFailed( runId: string, input: Parameters[1] @@ -1233,11 +1291,18 @@ export class WorkflowRunner { } assert(state.taskId != null, `pipeline agent ${state.spec.id} has no taskId to wait for`); const taskId = state.taskId; - state.rawResultPromise = this.taskAdapter.waitForAgentTask( - taskId, - state.resultSpec, - options.waitOptions - ); + state.rawResultPromise = + state.spec.timeout == null + ? this.taskAdapter.waitForAgentTask(taskId, state.resultSpec, options.waitOptions) + : this.waitForAgentTaskWithGracefulTimeout(runId, sequence, { + spec: state.spec, + inputHash: state.inputHash, + startedAt: state.startedAt, + taskId, + resultSpec: state.resultSpec, + waitOptions: options.waitOptions, + leaseGuard: options.leaseGuard, + }); } let interruptPromise: Promise | undefined; @@ -1263,7 +1328,7 @@ export class WorkflowRunner { if ("error" in settled) { if (!isForegroundWaitBackgroundedError(settled.error)) { - if (settled.state.taskId != null) { + if (!isWorkflowAgentHardTimeoutError(settled.error) && settled.state.taskId != null) { options.leaseGuard.throwIfLost(); await this.recordTaskTerminalEventIfMissing(runId, sequence, { stepId: settled.state.spec.id, @@ -1709,6 +1774,310 @@ export class WorkflowRunner { }); } + private async waitForAgentTaskWithGracefulTimeout( + runId: string, + sequence: WorkflowEventSequence, + step: { + spec: WorkflowAgentSpec; + inputHash: string; + startedAt: string; + taskId: string; + resultSpec: WorkflowAgentSpec; + waitOptions?: WorkflowAgentWaitOptions; + leaseGuard: WorkflowRunnerLeaseGuard; + } + ): Promise { + const timeout = step.spec.timeout; + assert(timeout != null, "WorkflowRunner timeout wait requires timeout spec"); + assert( + this.taskAdapter.waitForAgentTask != null, + "WorkflowRunner timeout wait requires waitForAgentTask" + ); + const waitForAgentTask = this.taskAdapter.waitForAgentTask.bind(this.taskAdapter); + const existingStep = await this.runStore.getStep(runId, step.spec.id, step.inputHash); + let existingTimeout = existingStep?.timeout; + let executionStartedRecord: Promise | undefined; + const recordExecutionStarted = (): void => { + if (existingTimeout?.executionStartedAt != null && existingTimeout.softDeadlineAt != null) { + return; + } + if (executionStartedRecord != null) { + return; + } + const executionStartedAt = this.clock.nowIso(); + const softDeadlineAt = new Date(this.clock.nowMs() + timeout.softMs).toISOString(); + existingTimeout = { ...existingTimeout, executionStartedAt, softDeadlineAt }; + executionStartedRecord = this.recordStepTimeoutMetadata(runId, { + stepId: step.spec.id, + inputHash: step.inputHash, + taskId: step.taskId, + startedAt: step.startedAt, + timeout: { executionStartedAt, softDeadlineAt }, + }); + }; + const waitForExecutionStartRecord = async (): Promise => { + if (executionStartedRecord != null) { + await executionStartedRecord; + } + }; + const waitForReport = async (timeoutMs: number): Promise => { + assert(timeoutMs > 0, "WorkflowRunner timeout wait requires positive timeoutMs"); + try { + const result = await waitForAgentTask(step.taskId, step.resultSpec, { + ...step.waitOptions, + timeoutMs, + onExecutionStarted: () => { + recordExecutionStarted(); + return step.waitOptions?.onExecutionStarted?.(); + }, + }); + await waitForExecutionStartRecord(); + return result; + } catch (error) { + await waitForExecutionStartRecord(); + throw error; + } + }; + const remainingMsUntil = (deadlineIso: string | undefined, fallbackMs: number): number => { + if (deadlineIso == null) { + return fallbackMs; + } + const parsedDeadlineMs = Date.parse(deadlineIso); + if (!Number.isFinite(parsedDeadlineMs)) { + return fallbackMs; + } + return Math.max(1, parsedDeadlineMs - this.clock.nowMs()); + }; + const tryReadAcceptedReport = async (): Promise => { + const completedStep = await this.runStore.getCompletedStep( + runId, + step.spec.id, + step.inputHash + ); + if (completedStep?.result != null) { + return { ...completedStep.result, taskId: step.taskId }; + } + try { + return await waitForReport(1); + } catch { + return null; + } + }; + const hardTimeout = async (): Promise => { + const acceptedReportBeforeHardTimeout = await tryReadAcceptedReport(); + if (acceptedReportBeforeHardTimeout != null) { + return acceptedReportBeforeHardTimeout; + } + + const errorMessage = `Workflow agent step ${step.spec.id} exceeded its soft timeout (${timeout.softMs}ms) and did not produce a valid agent_report within the grace period (${timeout.graceMs}ms).`; + const hardTimedOutAt = this.clock.nowIso(); + await this.recordStepTimeoutMetadata(runId, { + stepId: step.spec.id, + inputHash: step.inputHash, + taskId: step.taskId, + startedAt: step.startedAt, + timeout: { hardTimedOutAt }, + }); + await this.appendEvent(runId, { + sequence: sequence.next(), + type: "timeout", + at: hardTimedOutAt, + stepId: step.spec.id, + taskId: step.taskId, + phase: "hard", + details: { error: errorMessage }, + }); + await this.recordTaskEventIfMissing(runId, sequence, { + stepId: step.spec.id, + taskId: step.taskId, + title: step.spec.title, + status: "timed_out", + }); + await this.taskAdapter.failAgentTaskForHardTimeout?.(step.taskId, { + workflowRunId: runId, + stepId: step.spec.id, + inputHash: step.inputHash, + reason: errorMessage, + }); + const acceptedReportAfterHardTimeout = await tryReadAcceptedReport(); + if (acceptedReportAfterHardTimeout != null) { + return acceptedReportAfterHardTimeout; + } + await this.recordStepFailed(runId, { + stepId: step.spec.id, + inputHash: step.inputHash, + taskId: step.taskId, + error: errorMessage, + startedAt: step.startedAt, + completedAt: hardTimedOutAt, + }); + throw createWorkflowAgentHardTimeoutError(errorMessage); + }; + const buildHardDeadlineFromNow = (): string => + new Date(this.clock.nowMs() + timeout.graceMs).toISOString(); + const recordFinalizationPromptAccepted = async ( + finalizationToken: string | undefined, + finalizationResult: "prompted" | "queued" | "already_reported" | "not_active" + ): Promise => { + const finalizationPromptSentAt = this.clock.nowIso(); + const hardDeadlineAt = buildHardDeadlineFromNow(); + step.leaseGuard.throwIfLost(); + await this.recordStepTimeoutMetadata(runId, { + stepId: step.spec.id, + inputHash: step.inputHash, + taskId: step.taskId, + startedAt: step.startedAt, + timeout: { + ...(finalizationToken != null ? { finalizationToken } : {}), + finalizationPromptSentAt, + hardDeadlineAt, + }, + }); + await this.appendEvent(runId, { + sequence: sequence.next(), + type: "timeout", + at: finalizationPromptSentAt, + stepId: step.spec.id, + taskId: step.taskId, + phase: "finalization_prompt_sent", + details: { result: finalizationResult }, + }); + return hardDeadlineAt; + }; + const waitDuringGrace = async (graceTimeoutMs: number): Promise => { + try { + const result = await waitForReport(graceTimeoutMs); + await this.appendEvent(runId, { + sequence: sequence.next(), + type: "timeout", + at: this.clock.nowIso(), + stepId: step.spec.id, + taskId: step.taskId, + phase: "recovered", + details: { graceMs: timeout.graceMs }, + }); + return result; + } catch (error) { + if (!isAgentReportWaitTimeoutError(error)) { + throw error; + } + } + return await hardTimeout(); + }; + + if (existingTimeout?.softTimedOutAt != null) { + assert( + this.taskAdapter.requestAgentFinalReportForTimeout != null, + "WorkflowRunner timeout wait requires requestAgentFinalReportForTimeout" + ); + const finalizationToken = + existingTimeout.finalizationToken ?? + buildWorkflowAgentTimeoutFinalizationToken(runId, step, existingTimeout.softTimedOutAt); + if (existingTimeout.finalizationPromptSentAt == null) { + const finalizationResult = await this.taskAdapter.requestAgentFinalReportForTimeout( + step.taskId, + { + workflowRunId: runId, + stepId: step.spec.id, + inputHash: step.inputHash, + finalizationToken, + finalInstructions: timeout.finalInstructions, + } + ); + if (finalizationResult === "already_reported") { + return await waitForReport( + remainingMsUntil(existingTimeout.hardDeadlineAt, timeout.graceMs) + ); + } + if (finalizationResult === "prompted") { + const hardDeadlineAt = await recordFinalizationPromptAccepted( + finalizationToken, + finalizationResult + ); + existingTimeout = { ...existingTimeout, hardDeadlineAt }; + } + } + return await waitDuringGrace( + remainingMsUntil(existingTimeout.hardDeadlineAt, timeout.graceMs) + ); + } + + try { + return await waitForReport(remainingMsUntil(existingTimeout?.softDeadlineAt, timeout.softMs)); + } catch (error) { + if (!isAgentReportWaitTimeoutError(error)) { + throw error; + } + } + + const completedAfterSoftTimeout = await this.runStore.getCompletedStep( + runId, + step.spec.id, + step.inputHash + ); + if (completedAfterSoftTimeout?.result != null) { + return { ...completedAfterSoftTimeout.result, taskId: step.taskId }; + } + + assert( + this.taskAdapter.requestAgentFinalReportForTimeout != null, + "WorkflowRunner timeout wait requires requestAgentFinalReportForTimeout" + ); + const softTimedOutAt = this.clock.nowIso(); + const finalizationToken = buildWorkflowAgentTimeoutFinalizationToken( + runId, + step, + softTimedOutAt + ); + const hardDeadlineAt = buildHardDeadlineFromNow(); + await this.recordStepTimeoutMetadata(runId, { + stepId: step.spec.id, + inputHash: step.inputHash, + taskId: step.taskId, + startedAt: step.startedAt, + timeout: { + softTimedOutAt, + hardDeadlineAt, + finalizationToken, + }, + }); + await this.appendEvent(runId, { + sequence: sequence.next(), + type: "timeout", + at: softTimedOutAt, + stepId: step.spec.id, + taskId: step.taskId, + phase: "soft", + details: { softMs: timeout.softMs, graceMs: timeout.graceMs }, + }); + await this.recordTaskEventIfMissing(runId, sequence, { + stepId: step.spec.id, + taskId: step.taskId, + title: step.spec.title, + status: "finalizing", + }); + + const finalizationResult = await this.taskAdapter.requestAgentFinalReportForTimeout( + step.taskId, + { + workflowRunId: runId, + stepId: step.spec.id, + inputHash: step.inputHash, + finalizationToken, + finalInstructions: timeout.finalInstructions, + } + ); + if (finalizationResult === "already_reported") { + return await waitForReport(timeout.graceMs); + } + const acceptedHardDeadlineAt = + finalizationResult === "prompted" + ? await recordFinalizationPromptAccepted(finalizationToken, finalizationResult) + : hardDeadlineAt; + + return await waitDuringGrace(remainingMsUntil(acceptedHardDeadlineAt, timeout.graceMs)); + } + private async runOrResumeAgentStep( runId: string, sequence: WorkflowEventSequence, @@ -1723,6 +2092,90 @@ export class WorkflowRunner { } ): Promise { step.leaseGuard.throwIfLost(); + if (step.spec.timeout != null && this.taskAdapter.waitForAgentTask != null) { + const resultSpec = normalizeWorkflowAgentSpecForExecution(step.spec, { + allowMissingOutputSchema: step.allowMissingOutputSchema, + }); + let taskId = step.taskId; + if (taskId == null) { + assert( + this.taskAdapter.createAgentTasks != null, + "agent timeout requires workflow task adapter support for nonblocking agent starts" + ); + const createdTasks = await this.taskAdapter.createAgentTasks([resultSpec], { + onTaskCreated: async (index, createdTaskId) => { + assert(index === 0, "WorkflowRunner timeout agent start lifecycle index mismatch"); + taskId = createdTaskId; + step.leaseGuard.throwIfLost(); + await this.recordStepStarted(runId, { + stepId: step.spec.id, + inputHash: step.inputHash, + taskId: createdTaskId, + startedAt: step.startedAt, + }); + await this.recordTaskStartedEventIfMissing(runId, sequence, { + stepId: step.spec.id, + taskId: createdTaskId, + title: step.spec.title, + }); + }, + }); + assert(createdTasks.length === 1, "timeout agent start returned the wrong number of tasks"); + const createdTask = createdTasks[0]; + assert(createdTask != null, "timeout agent start must return a task"); + if (taskId == null) { + taskId = createdTask.taskId; + await this.recordStepStarted(runId, { + stepId: step.spec.id, + inputHash: step.inputHash, + taskId, + startedAt: step.startedAt, + }); + await this.recordTaskStartedEventIfMissing(runId, sequence, { + stepId: step.spec.id, + taskId, + title: step.spec.title, + }); + } + } else { + await this.recordTaskStartedEventIfMissing(runId, sequence, { + stepId: step.spec.id, + taskId, + title: step.spec.title, + }); + } + try { + const rawResult = await this.waitForAgentTaskWithGracefulTimeout(runId, sequence, { + spec: step.spec, + inputHash: step.inputHash, + startedAt: step.startedAt, + taskId, + resultSpec, + waitOptions: step.waitOptions, + leaseGuard: step.leaseGuard, + }); + return { rawResult, resultSpec }; + } catch (error) { + if (!isForegroundWaitBackgroundedError(error) && !isWorkflowAgentHardTimeoutError(error)) { + step.leaseGuard.throwIfLost(); + await this.recordTaskTerminalEventIfMissing(runId, sequence, { + stepId: step.spec.id, + taskId, + title: step.spec.title, + status: getTaskTerminalStatusForError(error, step.waitOptions?.abortSignal), + }); + } + if (step.taskId == null || !shouldRestartUnrecoverableStartedTask(error)) { + throw error; + } + return await this.runOrResumeAgentStep(runId, sequence, { + ...step, + startedAt: this.clock.nowIso(), + taskId: undefined, + }); + } + } + if (step.taskId != null && this.taskAdapter.waitForAgentTask != null) { await this.recordTaskStartedEventIfMissing(runId, sequence, { stepId: step.spec.id, @@ -2342,6 +2795,49 @@ function parseWorkflowAgentThinkingLevel(rawValue: unknown): ParsedThinkingInput return parsed; } +const WORKFLOW_AGENT_TIMEOUT_MIN_MS = 1_000; +const WORKFLOW_AGENT_SOFT_TIMEOUT_MAX_MS = 24 * 60 * 60 * 1000; +const WORKFLOW_AGENT_GRACE_TIMEOUT_MAX_MS = 60 * 60 * 1000; +const WORKFLOW_AGENT_FINAL_INSTRUCTIONS_MAX_LENGTH = 4_000; + +function parseWorkflowAgentTimeoutSpec(rawValue: unknown): WorkflowAgentTimeoutSpec | undefined { + if (rawValue === undefined) { + return undefined; + } + assert( + rawValue != null && typeof rawValue === "object" && !Array.isArray(rawValue), + "agent timeout must be an object" + ); + const timeout = rawValue as Record; + const softMs = timeout.softMs; + assert( + typeof softMs === "number" && + Number.isInteger(softMs) && + softMs >= WORKFLOW_AGENT_TIMEOUT_MIN_MS && + softMs <= WORKFLOW_AGENT_SOFT_TIMEOUT_MAX_MS, + "agent timeout.softMs must be a positive integer between 1000ms and 24h" + ); + const graceMs = timeout.graceMs; + assert( + typeof graceMs === "number" && + Number.isInteger(graceMs) && + graceMs >= WORKFLOW_AGENT_TIMEOUT_MIN_MS && + graceMs <= WORKFLOW_AGENT_GRACE_TIMEOUT_MAX_MS, + "agent timeout.graceMs must be a positive integer between 1000ms and 1h" + ); + const parsed: WorkflowAgentTimeoutSpec = { softMs, graceMs }; + if (timeout.finalInstructions !== undefined) { + assert( + typeof timeout.finalInstructions === "string" && + timeout.finalInstructions.trim().length > 0 && + timeout.finalInstructions.length <= WORKFLOW_AGENT_FINAL_INSTRUCTIONS_MAX_LENGTH, + "agent timeout.finalInstructions must be a non-empty string under 4000 characters" + ); + parsed.finalInstructions = timeout.finalInstructions; + } + return parsed; +} + function parseWorkflowAgentSpec( rawSpec: unknown, options: { allowMissingOutputSchema: boolean } @@ -2378,6 +2874,10 @@ function parseWorkflowAgentSpec( ); parsed.isolation = spec.isolation; } + const timeout = parseWorkflowAgentTimeoutSpec(spec.timeout); + if (timeout !== undefined) { + parsed.timeout = timeout; + } if (spec.markdownOnly !== undefined) { assert(spec.markdownOnly === true, "agent markdownOnly must be true when provided"); parsed.markdownOnly = true; diff --git a/src/node/services/workflows/WorkflowTaskServiceAdapter.ts b/src/node/services/workflows/WorkflowTaskServiceAdapter.ts index 1f14bdf344..294fbd6d9b 100644 --- a/src/node/services/workflows/WorkflowTaskServiceAdapter.ts +++ b/src/node/services/workflows/WorkflowTaskServiceAdapter.ts @@ -92,6 +92,25 @@ interface WorkflowTaskServiceLike { structuredOutput?: unknown; planFilePath?: string; }>; + requestAgentFinalReportForTimeout?( + taskId: string, + options: { + workflowRunId: string; + stepId: string; + inputHash: string; + finalizationToken: string; + finalInstructions?: string; + } + ): Promise<"prompted" | "queued" | "already_reported" | "not_active">; + failAgentTaskForHardTimeout?( + taskId: string, + options: { + workflowRunId: string; + stepId: string; + inputHash: string; + reason: string; + } + ): Promise; terminateAllDescendantAgentTasks?( workspaceId: string, options?: { workflowRunId?: string } @@ -460,6 +479,34 @@ export class WorkflowTaskServiceAdapter implements WorkflowTaskAdapter { return { ...experiments, subagentFileReports: false }; } + async requestAgentFinalReportForTimeout( + taskId: string, + options: { + workflowRunId: string; + stepId: string; + inputHash: string; + finalizationToken: string; + finalInstructions?: string; + } + ): Promise<"prompted" | "queued" | "already_reported" | "not_active"> { + assert( + this.taskService.requestAgentFinalReportForTimeout != null, + "WorkflowTaskServiceAdapter requires TaskService timeout finalization support" + ); + return await this.taskService.requestAgentFinalReportForTimeout(taskId, options); + } + + async failAgentTaskForHardTimeout( + taskId: string, + options: { workflowRunId: string; stepId: string; inputHash: string; reason: string } + ): Promise { + assert( + this.taskService.failAgentTaskForHardTimeout != null, + "WorkflowTaskServiceAdapter requires TaskService hard timeout support" + ); + await this.taskService.failAgentTaskForHardTimeout(taskId, options); + } + async waitForAgentTask( taskId: string, _spec: WorkflowAgentSpec, @@ -468,6 +515,9 @@ export class WorkflowTaskServiceAdapter implements WorkflowTaskAdapter { const report = await this.taskService.waitForAgentReport(taskId, { ...(waitOptions?.abortSignal != null ? { abortSignal: waitOptions.abortSignal } : {}), ...(waitOptions?.timeoutMs != null ? { timeoutMs: waitOptions.timeoutMs } : {}), + ...(waitOptions?.onExecutionStarted != null + ? { onExecutionStarted: waitOptions.onExecutionStarted } + : {}), requestingWorkspaceId: this.parentWorkspaceId, backgroundOnMessageQueued: waitOptions?.backgroundOnMessageQueued ?? true, }); diff --git a/tests/ui/storybook/budget.test.ts b/tests/ui/storybook/budget.test.ts index 645504a48b..d83a11118a 100644 --- a/tests/ui/storybook/budget.test.ts +++ b/tests/ui/storybook/budget.test.ts @@ -7,9 +7,9 @@ const COLOCATED_STORY_DIRS = ["src/browser/components", "src/browser/features"]; const MAX_SNAPSHOT_ENABLED_FILES = 70; // Keep a buffer under Chromatic's 300 snapshot limit. This reflects the current // retained snapshot set while still blocking accidental growth. -// 262 keeps the retained snapshot set under Chromatic's limit while leaving -// headroom for the pinned-mobile WorkflowListNarrow story guarding narrow rows. -const MAX_ESTIMATED_SNAPSHOTS = 262; +// 271 keeps the retained snapshot set under Chromatic's limit while leaving +// headroom for disabled/dogfood-only stories that must not add Chromatic snapshots. +const MAX_ESTIMATED_SNAPSHOTS = 271; const STORY_EXPORT_PATTERN = /^export const \w+/gm; const SMOKE_MODE_PATTERN = /modes:\s*CHROMATIC_SMOKE_MODES/g; const INLINE_MODE_OBJECT_PATTERN = /modes:\s*{/g;