feat: execute durable retry and fallback policy (#1017)

* feat: execute durable retry and fallback policy

* fix: declare recovery control permissions

* fix: make recovery fail closed and durable

* test: preserve workflow recovery revision
This commit is contained in:
Brad Groux 2026-07-24 20:37:08 -05:00 committed by GitHub
parent 74647256bc
commit f7aea9a4d4
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
38 changed files with 2897 additions and 132 deletions

View file

@ -7,6 +7,21 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
## [Unreleased]
### Added
- Added one durable `run-recovery/v1` state machine for production task
attempts and workflow steps. Only classified transient transport, provider
availability, rate-limit, timeout, and verification failures retry; policy,
configuration, cancellation, destructive partial-side-effect, and unknown
failures fail closed or require operator review. Recovery records preserve
causal parents, exponential jittered backoff, selected routes, launch
manifest digests, and cumulative budgets; compatible fallbacks are probed
through runtime capability and sandbox gates before launch. Pending recovery
survives restart, duplicate terminal callbacks cannot create multiple
branches, and exact task recovery can be cancelled through REST, CLI, or MCP
controls. Workflow recovery exposes the same exact-parent cancellation
through REST (#861).
## [6.0.2] - 2026-07-24
Veritas Kanban 6.0.2 is a desktop recovery and supportability hotfix. It keeps

View file

@ -92,7 +92,7 @@ When the board is working, use [Setup Paths](docs/SETUP-PATHS.md) to choose the
- [Setup Paths](docs/SETUP-PATHS.md) — start here for board-only, CLI, MCP, OpenClaw, and self-hosted paths without mixing optional layers into first-run setup.
- [Getting Started Guide](docs/GETTING-STARTED.md) — zero ➝ agent-ready in 5 minutes, plus sanity checks and prompt registry tips.
- [MCP Server Guide](docs/mcp/README.md) — optional MCP setup, 41 tools, architecture, tool catalog, security model, and read/write smoke checks.
- [MCP Server Guide](docs/mcp/README.md) — optional MCP setup, 42 tools, architecture, tool catalog, security model, and read/write smoke checks.
- [Agent Guide and `AGENTS.md` Template](docs/AGENTS-TEMPLATE.md) — shared managed-run protocol, external self-reporting template, and unmanaged MCP setup.
- [Agent Providers](docs/AGENT-PROVIDERS.md) — evidence-backed Buzz, Grok Build, Codex, Claude Code, Copilot CLI, Hermes, OpenClaw, and optional model profiles.
- [v6 Agent Runtime Control Plane](docs/architecture/V6-AGENT-RUNTIME-CONTROL-PLANE.md) — authority, adapter, lifecycle, approval, tool, credential, Buzz, and certification boundaries.
@ -214,7 +214,7 @@ Tasks are markdown files. Settings are JSON. Workflows are YAML. No database, no
### 🔌 Optional Integration Surfaces
- **MCP Server** — 41 tools across 9 categories via Model Context Protocol
- **MCP Server** — 42 tools across 9 categories via Model Context Protocol
- **CLI**`vk begin <id>` / `vk done <id> "summary"` replaces 6 API calls with 2 commands
- **REST API** — Full lifecycle management. If it can make HTTP calls, it can drive the board.
@ -353,7 +353,7 @@ Tasks are markdown files. Settings are JSON. Workflows are YAML. No database, no
#### Integration
- **CLI**`vk` command for terminal workflows
- **MCP Server** — 41 tools across 9 categories via Model Context Protocol
- **MCP Server** — 42 tools across 9 categories via Model Context Protocol
- **Codex MCP setup** — documented `codex mcp add veritas-kanban` setup for local and API-key-backed deployments
- **Notifications** — Teams integration for task updates
@ -391,7 +391,7 @@ Veritas Kanban is neither. It's the **visual command center for agentic work**
| **YAML workflow pipelines** | ✅ Loops, gates, parallel | ⚠️ Code-defined only | ❌ |
| **Real-time agent dashboard** | ✅ Status, model attribution | ❌ | ❌ |
| **Agent communication** | ✅ Squad Chat with lifecycle events | ⚠️ Internal only | ❌ |
| **MCP server** | ✅ 41 tools | ❌ | ❌ |
| **MCP server** | ✅ 42 tools | ❌ | ❌ |
| **CLI** | ✅ Full lifecycle | ❌ | ⚠️ Limited |
| **Git worktrees + code review** | ✅ Built-in | ❌ | ❌ |
| **Task persistence** | ✅ Markdown files | ❌ In-memory | ✅ Database |
@ -753,7 +753,7 @@ vk agents:pending
## 🔗 MCP Server
Optional. The MCP server exposes 41 tools across 9 categories (tasks, agents, automation, notifications, summaries, sprints, comments, projects, and run-scoped tool control) via [Model Context Protocol](https://modelcontextprotocol.io/). Skip this for board-only use.
Optional. The MCP server exposes 42 tools across 9 categories (tasks, agents, automation, notifications, summaries, sprints, comments, projects, and run-scoped tool control) via [Model Context Protocol](https://modelcontextprotocol.io/). Skip this for board-only use.
**→ [Full MCP documentation](docs/mcp/README.md)** — architecture, quickstart, tool catalog with examples, security model, read/write smoke checks, and troubleshooting.

View file

@ -119,6 +119,22 @@ describe('vk agent runtime capability controls', () => {
});
});
it('binds recovery cancellation to the exact persisted parent attempt', async () => {
const program = new Command();
program.exitOverride();
registerAgentCommands(program);
await program.parseAsync(
['agent:cancel-recovery', 'task_1', '--attempt', 'attempt_parent', '--json'],
{ from: 'user' }
);
expect(mockApi).toHaveBeenCalledWith('/api/agents/task_1/recovery/cancel', {
method: 'POST',
body: JSON.stringify({ attemptId: 'attempt_parent' }),
});
});
it('starts a native history fork from an explicit source attempt and turn', async () => {
const program = new Command();
program.exitOverride();

View file

@ -11,6 +11,7 @@ import type {
AgentProfileValidationResult,
ConversationLifecycleRecord,
ConversationLifecycleResult,
RunRecoveryRecord,
RunLaunchManifestPreview,
} from '@veritas-kanban/shared';
@ -419,6 +420,59 @@ export function registerAgentCommands(program: Command): void {
}
});
program
.command('agent:recovery <id>')
.description('Show the latest durable retry or fallback decision for a task')
.option('--json', 'Output as JSON')
.action(async (id, options) => {
try {
const taskId = await resolveTaskId(id);
const result = await api<{ recovery: RunRecoveryRecord | null }>(
`/api/agents/${taskId}/recovery`
);
if (options.json) {
console.log(JSON.stringify(result, null, 2));
} else if (!result.recovery) {
console.log(chalk.dim('No recovery decision is recorded for this task'));
} else {
console.log(chalk.yellow(`Recovery: ${result.recovery.state}`));
console.log(` Action: ${result.recovery.action}`);
console.log(` Attempt: ${result.recovery.parentRunId}`);
console.log(` Sequence: ${result.recovery.sequence}`);
console.log(` Reason: ${result.recovery.reason}`);
}
} catch (err) {
console.error(chalk.red(`Error: ${(err as Error).message}`));
process.exit(1);
}
});
program
.command('agent:cancel-recovery <id>')
.description('Cancel the exact pending retry or fallback for a task')
.requiredOption('--attempt <attemptId>', 'Parent attempt that owns the pending recovery')
.option('--json', 'Output as JSON')
.action(async (id, options: { attempt: string; json?: boolean }) => {
try {
const taskId = await resolveTaskId(id);
const result = await api<{ cancelled: boolean; recovery: RunRecoveryRecord }>(
`/api/agents/${taskId}/recovery/cancel`,
{
method: 'POST',
body: JSON.stringify({ attemptId: options.attempt }),
}
);
if (options.json) {
console.log(JSON.stringify(result, null, 2));
} else {
console.log(chalk.yellow('✓ Automatic recovery cancelled'));
}
} catch (err) {
console.error(chalk.red(`Error: ${(err as Error).message}`));
process.exit(1);
}
});
registerConversationTurnCommand(
program,
'resume',

View file

@ -907,6 +907,34 @@ Recommended starting point:
3. Enable `ollama-cloud` only when the workflow is allowed to leave local execution.
4. Use explicit routing rules for local LLM profiles instead of making them global defaults on teams with mixed operating systems.
### Automatic retry and fallback
`agentRouting.maxRetries` and `agentRouting.fallbackOnFailure` are captured in
each new run launch manifest. Task attempts and workflow steps use the same
durable `run-recovery/v1` decision record:
- Only transient transport, provider unavailable, rate-limit, timeout, and
verification failures are automatically retryable.
- Invalid configuration and policy blocks require operator action. Explicit
cancellation stops recovery. A failed or partial run with destructive side
effects requires approval before another launch.
- Retry backoff is exponential, capped at 30 seconds, and jittered by 20
percent. The attempt chain retains its root and parent IDs, route, source and
launched manifest digests, and cumulative budget.
- A fallback is launched only after retry exhaustion and only after its runtime
capabilities and sandbox policy pass the normal launch preflight. An
incompatible fallback produces an actionable exhausted handoff.
- Scheduled task and workflow recovery is reconciled after server restart.
Revision claims and terminal idempotency prevent duplicate callbacks from
starting multiple branches.
Inspect a task decision with `vk agent:recovery <task> --json`. Cancel the
exact pending parent with
`vk agent:cancel-recovery <task> --attempt <attempt-id>`. MCP clients use
`cancel_agent_recovery` with the same task and parent attempt. REST clients use
`GET /api/agents/:taskId/recovery` and
`POST /api/agents/:taskId/recovery/cancel`.
---
## Hermes Agent (v2026.7.7.2)

View file

@ -2335,6 +2335,38 @@ Approval-required tools are omitted from native provider configuration and use
the mediated tool-call API below. Prompt text is never accepted as equivalent
enforcement.
### Automatic Run Recovery
```
GET /api/agents/:taskId/recovery
POST /api/agents/:taskId/recovery/cancel
POST /api/workflows/runs/:runId/recovery/cancel
```
The GET endpoint returns the latest durable `run-recovery/v1` decision or
`null`. Cancellation requires the exact parent attempt:
```json
{ "attemptId": "attempt_parent" }
```
Only a `scheduled` or `launching` recovery can be cancelled. A stale attempt
ID, already launched child, or terminal recovery returns `409 Conflict`.
Recovery records expose the normalized failure class, action, state, root and
parent IDs, sequence, backoff and not-before time, selected route, manifest
digests, cumulative budget, and any operator handoff.
Workflow cancellation requires the exact step and causal parent:
```json
{
"stepId": "implement",
"parentRunId": "run_123:implement:0"
}
```
The caller must have execute permission on the workflow.
### Conversation Lifecycle
```

View file

@ -450,6 +450,8 @@ Manage AI agents on code tasks.
| `vk start <id>` | Start an agent; optionally require runtime capabilities |
| `vk launch-preview <id>` | Preview effective launch inputs, blockers, and drift |
| `vk stop <id>` | Stop a run only when its persisted manifest supports stop |
| `vk agent:recovery <id>` | Inspect the latest retry or fallback decision |
| `vk agent:cancel-recovery <id> --attempt <id>` | Cancel the exact pending recovery parent |
| `vk agent:resume <id> --source-attempt <id> -m <text>` | Resume the exact persisted provider conversation |
| `vk agent:follow-up <id> --source-attempt <id> -m <text>` | Start a provider-native follow-up turn |
| `vk agent:fork <id> --source-attempt <id> -m <text>` | Fork provider history without mutating its source |
@ -510,6 +512,18 @@ the stop request, so a replacement run fails a delayed stop closed. The CLI
preserves the server's reason when `run.stop` is unavailable or the active and
persisted manifest digests do not match.
Automatic recovery is separate from stopping an active provider. Use
`vk agent:recovery TASK-001 --json` to inspect its classification, backoff,
route, causal parent, manifest evidence, and cumulative budget. A cancellation
must include the exact parent attempt:
```bash
vk agent:cancel-recovery TASK-001 --attempt attempt_parent --json
```
The server rejects stale parent IDs, recoveries that already launched, and
recoveries that are already terminal.
---
### Prompt Commands

View file

@ -958,6 +958,10 @@ Execute a single agent prompt with configurable retries.
- Template rendering with `{{variable}}` and `{{nested.path}}` substitution
- Acceptance criteria validation (substring, regex, JSON path)
- Retry routing: retry same step, retry different step, escalate
- Production retry/fallback state machine: only explicitly transient failure
classes retry; each decision persists causal parents, jittered backoff,
route and manifest evidence, and cumulative budget. Fallback agents must pass
runtime capability and sandbox preflight before launch.
#### 2. Loop Steps
@ -1074,8 +1078,15 @@ All three types are backward-compatible — substring matching was the original
Every workflow run persists its state to disk, enabling:
- **Server restart recovery** — Runs can resume from last checkpoint
- **Retry with exponential backoff** — Configurable `retry_delay_ms` prevents rapid retry loops
- **Server restart recovery** — Scheduled retries and fallbacks are restored
from their durable step records
- **Retry with exponential backoff**`retry_delay_ms` supplies the base delay;
recovery applies bounded exponential backoff with jitter
- **Fail-closed fallback** — Explicit `agent:<id>` escalation and compatible
workspace fallback routes run only after retry exhaustion and runtime/sandbox
preflight
- **Operator cancellation** — Exact pending workflow recovery can be cancelled
before another provider launch
- **Progress file tracking** — Shared `progress.md` per run for context passing:
- Each step appends its output with timestamp
- Templates can access `{{progress}}` for previous step context
@ -1866,7 +1877,7 @@ vk done <id> "Added OAuth2 with Google and GitHub providers"
## MCP Server
Model Context Protocol server for AI assistant integration (Claude Desktop, OpenClaw, Cursor, Codex, etc.). 41 tools across task management, agent orchestration, automation, notifications, summaries, sprint management, comments, projects, and run-scoped tool control.
Model Context Protocol server for AI assistant integration (Claude Desktop, OpenClaw, Cursor, Codex, etc.). 42 tools across task management, agent orchestration, automation, notifications, summaries, sprint management, comments, projects, and run-scoped tool control.
### Tools
@ -1880,6 +1891,7 @@ Model Context Protocol server for AI assistant integration (Claude Desktop, Open
| `delete_task` | Permanently delete a task |
| `start_agent` | Start an AI agent on a code task |
| `stop_agent` | Stop a running agent |
| `cancel_agent_recovery` | Cancel an exact pending retry or fallback |
| `list_pending_automation` | List automation tasks awaiting execution |
| `list_running_automation` | List currently running automation tasks |
| `start_automation` | Start an automation task via sub-agent |

View file

@ -1205,7 +1205,19 @@ If the server crashes mid-workflow, runs can be recovered:
curl -X POST http://localhost:3001/api/workflow-runs/run_XYZ/resume
```
> **📝 Note**: Automatic recovery is planned for a future release.
Automatic transient-failure recovery is persisted on the step as
`runRetry`. Scheduled retry and fallback timers are restored after server
restart. Cancel an exact pending workflow recovery with:
```bash
curl -X POST http://localhost:3001/api/workflows/runs/run_XYZ/recovery/cancel \
-H 'Content-Type: application/json' \
-d '{"stepId":"implement","parentRunId":"run_XYZ:implement:0"}'
```
Only explicitly transient failures retry automatically. Policy and
configuration failures do not retry, while fallback agents must pass the same
runtime capability and sandbox preflight as a normal workflow launch.
### Performance Issues

View file

@ -1,6 +1,6 @@
# MCP Server — Veritas Kanban
> **41 tools · 9 categories · stdio transport · zero external dependencies**
> **42 tools · 9 categories · stdio transport · zero external dependencies**
The Veritas Kanban MCP server lets any [Model Context Protocol](https://modelcontextprotocol.io/) client — Claude Desktop, OpenClaw, Cursor, Cline, Codex, or your own tooling — manage tasks, sprints, projects, comments, agents, automation, notifications, and summaries through a single stdio process.
@ -24,7 +24,7 @@ MCP is optional. The board, REST API, and CLI do not require MCP or OpenClaw. Us
- [Configuration Reference](#configuration-reference)
- [Tool Catalog](#tool-catalog)
- [Task Management (6 tools)](#task-management-6-tools)
- [Agent Control (2 tools)](#agent-control-2-tools)
- [Agent Control (4 tools)](#agent-control-4-tools)
- [Automation (4 tools)](#automation-4-tools)
- [Notifications (3 tools)](#notifications-3-tools)
- [Summaries (2 tools)](#summaries-2-tools)
@ -45,7 +45,7 @@ MCP is optional. The board, REST API, and CLI do not require MCP or OpenClaw. Us
Use the MCP server when:
- Your AI assistant (Claude Desktop, Cursor, etc.) needs **structured tool access** to VK — not raw HTTP calls.
- You want **one process** that exposes all 36 VK operations with typed inputs and validated outputs.
- You want **one process** that exposes all 42 VK operations with typed inputs and validated outputs.
- You're building **agent orchestration** and need task/sprint/automation lifecycle management over MCP.
Don't use it when:
@ -71,7 +71,7 @@ Don't use it when:
│ ┌────────────┐ ┌────────────┐ ┌───────────┐ │
│ │ Tool │ │ Resource │ │ Transport │ │
│ │ Registry │ │ Provider │ │ (stdio) │ │
│ │ (41 tools) │ │ (kanban:// │ │ │ │
│ │ (42 tools) │ │ (kanban:// │ │ │ │
│ │ │ │ URIs) │ │ │ │
│ └──────┬─────┘ └──────┬─────┘ └───────────┘ │
│ │ │ │
@ -356,12 +356,13 @@ Task write tools return concise confirmations. Use `get_task`, `list_tasks`, or
---
### Agent Control (3 tools)
### Agent Control (4 tools)
| Tool | Description | Required Inputs | Key Options |
| ---------------------------- | ----------------------------------------- | --------------------------- | --------------------------------------------------------------------- |
| `start_agent` | Start a coding agent on a task | `id` | `agent`; `requiredRuntimeCapabilities`; `commitPolicy` |
| `stop_agent` | Stop a running agent | `id` | Resolves status and binds the stop to that exact attempt and manifest |
| `cancel_agent_recovery` | Cancel a pending retry or fallback | `id`, `attemptId` | Requires the exact persisted parent attempt |
| `control_agent_conversation` | Invoke a supported conversation lifecycle | `id`, `attemptId`, `action` | `message`, `forkTurnId`, `commitPolicy` |
> **Constraints:** Only works on tasks with `type: "code"` that already have a git worktree attached.
@ -887,4 +888,4 @@ The `findTask` utility matches the last N characters of a task ID (minimum 6). I
---
_Last updated: 2026-07-24 · VK v6.0.2 · 41 tools / 9 categories_
_Last updated: 2026-07-24 · VK v6.0.2 · 42 tools / 9 categories_

View file

@ -651,6 +651,22 @@
"source": "cli/src/commands/agents.ts",
"denialReason": "Stopping an agent requires task read and task write access."
},
{
"id": "cli:agents:agent:recovery",
"kind": "cli",
"classification": "authenticated-read",
"permissions": ["task:read", "agent:read"],
"source": "cli/src/commands/agents.ts",
"denialReason": "Reading a task recovery decision requires task and agent read access."
},
{
"id": "cli:agents:agent:cancel-recovery",
"kind": "cli",
"classification": "agent-scoped",
"permissions": ["task:read", "task:write"],
"source": "cli/src/commands/agents.ts",
"denialReason": "Cancelling an exact pending recovery requires task read and task write access."
},
{
"id": "cli:agents:agent:resume",
"kind": "cli",
@ -1371,6 +1387,14 @@
"source": "mcp/src/tools/agents.ts",
"denialReason": "Stopping an agent requires task read and task write access."
},
{
"id": "mcp:cancel_agent_recovery",
"kind": "mcp",
"classification": "agent-scoped",
"permissions": ["task:read", "task:write"],
"source": "mcp/src/tools/agents.ts",
"denialReason": "Cancelling an exact pending recovery requires task read and task write access."
},
{
"id": "mcp:list_tool_servers",
"kind": "mcp",

View file

@ -118,6 +118,21 @@ describe('MCP agent runtime capability controls', () => {
});
});
it('publishes and forwards exact recovery cancellation provenance', async () => {
const cancel = agentTools.find((tool) => tool.name === 'cancel_agent_recovery');
expect(cancel?.inputSchema.required).toEqual(['id', 'attemptId']);
await handleAgentTool('cancel_agent_recovery', {
id: 'task_1',
attemptId: 'attempt_parent',
});
expect(mockApi).toHaveBeenCalledWith('/api/agents/task_1/recovery/cancel', {
method: 'POST',
body: JSON.stringify({ attemptId: 'attempt_parent' }),
});
});
it('forwards a native history fork with its source turn boundary', async () => {
await handleAgentTool('control_agent_conversation', {
id: 'task_1',

View file

@ -22,6 +22,11 @@ const TaskIdSchema = z.object({
id: z.string().min(1),
});
const CancelAgentRecoverySchema = z.object({
id: z.string().min(1),
attemptId: z.string().trim().min(1).max(120),
});
const ConversationActionSchema = z.enum([
'resume',
'follow-up',
@ -132,6 +137,24 @@ export const agentTools = [
required: ['id'],
},
},
{
name: 'cancel_agent_recovery',
description: 'Cancel the exact pending automatic retry or fallback for a task',
inputSchema: {
type: 'object',
properties: {
id: {
type: 'string',
description: 'Task ID or partial ID',
},
attemptId: {
type: 'string',
description: 'Exact parent attempt that owns the pending recovery',
},
},
required: ['id', 'attemptId'],
},
},
{
name: 'control_agent_conversation',
description:
@ -249,6 +272,24 @@ export async function handleAgentTool(name: string, args: any): Promise<any> {
};
}
case 'cancel_agent_recovery': {
const { id, attemptId } = CancelAgentRecoverySchema.parse(args);
const task = await findTask(id);
if (!task) {
return {
content: [{ type: 'text', text: `Task not found: ${id}` }],
isError: true,
};
}
const result = await api(`/api/agents/${task.id}/recovery/cancel`, {
method: 'POST',
body: JSON.stringify({ attemptId }),
});
return {
content: [{ type: 'text', text: JSON.stringify(result, null, 2) }],
};
}
case 'control_agent_conversation': {
const { id, action, sourceAttemptId, attemptId, message, forkTurnId } =
ConversationControlSchema.parse(args);

View file

@ -760,6 +760,21 @@ describe('AgentRoutingService', () => {
expect(result).toBeNull();
});
it('ignores a matched fallback that resolves to the failed agent', async () => {
const config = structuredClone(BASE_CONFIG);
const routing = requireRouting(config);
const [firstRule] = routing.rules;
if (!firstRule) throw new Error('Expected first routing rule in test fixture');
routing.rules = [firstRule];
firstRule.fallback = firstRule.agent;
routing.defaultAgent = firstRule.agent;
mockGetConfig.mockResolvedValue(config);
const result = await service.getFallback({ type: 'code', priority: 'high' }, firstRule.agent);
expect(result).toBeNull();
});
it('returns default agent when it differs from failed', async () => {
const result = await service.getFallback(
{ type: 'docs', priority: 'low' },

View file

@ -2197,7 +2197,10 @@ describe('ClawdbotAgentService Codex providers', () => {
expect(task.attempt?.completionResult?.idempotencyKey).toMatch(/^sha256:[a-f0-9]{64}$/);
expect(
mockUpdateTask.mock.calls.filter(
([, update]) => update.attempt?.id === active.attemptId && update.attempt?.ended
([, update]) =>
update.attempt?.id === active.attemptId &&
update.attempt?.ended &&
!update.attempt?.runRetry
)
).toHaveLength(1);
expect(
@ -2252,7 +2255,10 @@ describe('ClawdbotAgentService Codex providers', () => {
});
expect(
mockUpdateTask.mock.calls.filter(
([, update]) => update.attempt?.id === active.attemptId && update.attempt?.ended
([, update]) =>
update.attempt?.id === active.attemptId &&
update.attempt?.ended &&
!update.attempt?.runRetry
)
).toHaveLength(1);
expect(revokeRun).toHaveBeenCalledWith({
@ -2489,7 +2495,10 @@ describe('ClawdbotAgentService Codex providers', () => {
).toHaveLength(1);
expect(
mockUpdateTask.mock.calls.filter(
([, update]) => update.attempt?.id === active.attemptId && update.attempt?.ended
([, update]) =>
update.attempt?.id === active.attemptId &&
update.attempt?.ended &&
!update.attempt?.runRetry
)
).toHaveLength(1);
expect(task.attempt?.completionResult).toMatchObject({
@ -2522,11 +2531,19 @@ describe('ClawdbotAgentService Codex providers', () => {
).toHaveLength(1);
expect(
mockUpdateTask.mock.calls.filter(
([, update]) => update.attempt?.id === active.attemptId && update.attempt?.ended
([, update]) =>
update.attempt?.id === active.attemptId &&
update.attempt?.ended &&
!update.attempt?.runRetry
)
).toHaveLength(2);
const completionDigests = mockUpdateTask.mock.calls
.filter(([, update]) => update.attempt?.id === active.attemptId && update.attempt?.ended)
.filter(
([, update]) =>
update.attempt?.id === active.attemptId &&
update.attempt?.ended &&
!update.attempt?.runRetry
)
.map(([, update]) => update.attempt?.completionResult?.digest);
expect(new Set(completionDigests)).toEqual(new Set([task.attempt?.completionResult?.digest]));
await expect(service.getAgentStatus(task.id)).resolves.toBeNull();
@ -2551,7 +2568,10 @@ describe('ClawdbotAgentService Codex providers', () => {
expect(
mockUpdateTask.mock.calls.filter(
([, update]) => update.attempt?.id === active.attemptId && update.attempt?.ended
([, update]) =>
update.attempt?.id === active.attemptId &&
update.attempt?.ended &&
!update.attempt?.runRetry
)
).toHaveLength(1);
expect(mockTelemetryEmit).toHaveBeenCalledWith(
@ -2630,11 +2650,18 @@ describe('ClawdbotAgentService Codex providers', () => {
await service.stopAgent(task.id, active.attemptId);
const completionWrites = mockUpdateTask.mock.calls.filter(
([, update]) => update.attempt?.id === active.attemptId && update.attempt?.ended
([, update]) =>
update.attempt?.id === active.attemptId &&
update.attempt?.ended &&
!update.attempt?.runRetry
);
expect(completionWrites.map(([, update]) => update.expectedRevision)).toEqual([7, 8]);
expect(task.priority).toBe('high');
expect(task.revision).toBe(9);
expect(task.revision).toBe(10);
expect(task.attempt?.runRetry).toMatchObject({
state: 'cancelled',
action: 'cancelled',
});
});
it('rejects a completion retry when immutable attempt bindings change under the same ID', async () => {

View file

@ -379,7 +379,7 @@ describe('#780 — Bounded retry_step cycles', () => {
type: 'agent',
agent: 'a1',
name: 'Step',
on_fail: { retry: 2 },
on_fail: { retry: 2, retry_delay_ms: 100 },
},
])
);

View file

@ -0,0 +1,167 @@
import { describe, expect, it } from 'vitest';
import { ZERO_AGENT_BUDGET_USAGE } from '@veritas-kanban/shared';
import { RunRecoveryPolicyService } from '../services/run-recovery-policy-service.js';
describe('RunRecoveryPolicyService', () => {
it.each([
['rate-limit', { status: 'failed', error: 'HTTP 429 rate limited' }],
['timeout', { status: 'failed', error: 'Provider timed out after 30s' }],
['provider-unavailable', { status: 'failed', error: 'Provider unavailable' }],
['transient-transport', { status: 'failed', error: 'ECONNRESET from gateway' }],
['invalid-request', { status: 'failed', error: 'Invalid configuration' }],
['task-failure', { status: 'failed', error: 'Implementation did not work' }],
['verification-failure', { status: 'partial', error: 'Required verification failed' }],
['policy-block', { status: 'failed', error: 'Sandbox policy denied launch' }],
[
'cancellation',
{
status: 'interrupted',
terminalSource: 'operator-interruption',
error: 'Stopped by user',
},
],
[
'partial-side-effect',
{
status: 'failed',
error: 'ECONNRESET after write',
sideEffects: [
{
kind: 'external-write',
description: 'Created remote record',
target: 'record-1',
authorized: true,
verified: true,
},
],
},
],
['unknown', { status: 'success', summary: 'Unexpected classification input' }],
] as const)('classifies %s failures', (expected, evidence) => {
const policy = new RunRecoveryPolicyService(() => 0.5);
expect(policy.classify(evidence)).toMatchObject({ classification: expected });
});
it('retries only explicitly retryable classes within the configured bound', () => {
const policy = new RunRecoveryPolicyService(() => 0.5);
const failure = policy.classify({ status: 'failed', error: 'ECONNRESET from gateway' });
const decision = policy.decide(failure, baseDecision({ maxRetries: 2 }));
expect(decision).toMatchObject({
action: 'retry',
state: 'scheduled',
sequence: 1,
backoffMs: 1_000,
selectedAgent: 'codex',
});
});
it('falls back after retry exhaustion when the candidate passed policy checks', () => {
const policy = new RunRecoveryPolicyService(() => 0.5);
const failure = policy.classify({ status: 'failed', error: 'Provider unavailable' });
const decision = policy.decide(
failure,
baseDecision({
previousSequence: 1,
maxRetries: 1,
fallbackOnFailure: true,
fallbackAgent: 'claude-code',
fallbackEligible: true,
})
);
expect(decision).toMatchObject({
action: 'fallback',
state: 'scheduled',
sequence: 2,
fallbackUsed: true,
selectedAgent: 'claude-code',
});
});
it('fails closed when a fallback is capability or sandbox incompatible', () => {
const policy = new RunRecoveryPolicyService(() => 0.5);
const failure = policy.classify({ status: 'failed', error: 'Provider unavailable' });
const decision = policy.decide(
failure,
baseDecision({
maxRetries: 0,
fallbackOnFailure: true,
fallbackAgent: 'claude-code',
fallbackEligible: false,
fallbackReason: 'sandbox preset cannot be enforced',
})
);
expect(decision).toMatchObject({
action: 'terminal',
state: 'exhausted',
handoff: {
nextActions: expect.arrayContaining([
'Choose a fallback that satisfies runtime capabilities and sandbox policy.',
]),
},
});
});
it.each([
['policy-block', { status: 'failed', error: 'Permission denied by sandbox policy' }],
['invalid-request', { status: 'failed', error: 'Invalid configuration' }],
[
'partial-side-effect',
{
status: 'failed',
error: 'Transient provider failure after commit',
sideEffects: [
{
kind: 'git-commit',
description: 'Created commit',
target: 'abc123',
authorized: false,
verified: true,
},
],
},
],
] as const)('requires approval for unsafe %s recovery', (_expected, evidence) => {
const policy = new RunRecoveryPolicyService(() => 0.5);
const decision = policy.decide(policy.classify(evidence), baseDecision());
expect(decision).toMatchObject({ action: 'approval', state: 'approval-required' });
});
it('never retries explicit cancellation', () => {
const policy = new RunRecoveryPolicyService(() => 0.5);
const failure = policy.classify({
status: 'interrupted',
terminalSource: 'operator-interruption',
});
expect(policy.decide(failure, baseDecision())).toMatchObject({
action: 'cancelled',
state: 'cancelled',
backoffMs: 0,
});
});
it('keeps exponential jitter inside the documented bounds', () => {
expect(new RunRecoveryPolicyService(() => 0).backoffMs(2)).toBe(1_600);
expect(new RunRecoveryPolicyService(() => 1).backoffMs(2)).toBe(2_400);
expect(new RunRecoveryPolicyService(() => 1).backoffMs(32)).toBe(30_000);
});
});
function baseDecision(overrides: Record<string, unknown> = {}) {
return {
rootRunId: 'attempt-root',
parentRunId: 'attempt-parent',
selectedAgent: 'codex',
routingDecision: 'Matched code rule',
sourceManifestDigest: `sha256:${'a'.repeat(64)}`,
requiredRuntimeCapabilities: ['run.start'],
cumulativeBudget: { ...ZERO_AGENT_BUDGET_USAGE },
previousSequence: 0,
fallbackUsed: false,
maxRetries: 1,
fallbackOnFailure: false,
...overrides,
};
}

View file

@ -4,13 +4,14 @@
* attempts after a server restart and move legacy runs into an actionable blocked state.
*/
import { describe, it, expect, beforeEach, vi } from 'vitest';
import type { Task, TaskAttempt } from '@veritas-kanban/shared';
import type { RunRecoveryRecord, Task, TaskAttempt } from '@veritas-kanban/shared';
import { providerRuntimeManifestFixture } from '../fixtures/provider-runtime-manifest.js';
import {
TaskEnvelopeService,
type CompletionEvidenceSource,
} from '../../services/task-envelope-service.js';
import { ProviderCompletionService } from '../../services/provider-completion-service.js';
import { RunRecoveryPolicyService } from '../../services/run-recovery-policy-service.js';
// ─── Mocks ────────────────────────────────────────────────────────────────
@ -397,4 +398,386 @@ describe('ClawdbotAgentService.reconcileRunningAttempts (issue #781)', () => {
})
);
});
it('plans retry policy for a failed supervisor completion discovered after restart', async () => {
const completedAt = '2026-07-23T18:30:00.000Z';
const evidence: CompletionEvidenceSource = {
captureLaunchBaseline: async (_worktreePath, capturedAt) => ({
capturedAt,
headSha: 'a'.repeat(40),
dirty: false,
files: [],
}),
captureCompletionEvidence: async ({ taskEnvelope, capturedAt }) => ({
capturedAt,
headSha: taskEnvelope.workspace.baseline.headSha,
changedFiles: [],
commits: [],
artifacts: [],
verification: [],
sideEffects: [],
}),
};
const envelopes = new TaskEnvelopeService(evidence);
const completions = new ProviderCompletionService(evidence, () => completedAt);
const providerRuntimeManifest = providerRuntimeManifestFixture({
provider: 'codex-cli',
adapter: 'codex-cli',
});
let currentTask = {
...makeTask('task-supervisor-failure', 'running'),
revision: 1,
verificationSteps: [],
git: {
repo: 'BradGroux/veritas-kanban',
branch: 'feat/recovery',
baseBranch: 'main',
worktreePath: '/tmp/veritas-supervisor-recovery',
},
} as Task;
const runningAttempt = currentTask.attempt;
if (!runningAttempt) throw new Error('Expected a running attempt fixture');
const taskEnvelope = await envelopes.build({
task: currentTask,
attemptId: runningAttempt.id,
createdAt: '2026-07-23T17:00:00.000Z',
worktreePath: currentTask.git?.worktreePath ?? '/tmp/veritas-supervisor-recovery',
providerRuntimeManifest,
commitPolicy: 'allowed',
});
currentTask.attempt = {
...runningAttempt,
provider: 'codex-cli',
providerRuntimeManifest,
taskEnvelope,
};
currentTask.attempts = [currentTask.attempt];
const completionResult = await completions.complete({
task: currentTask,
taskEnvelope,
claim: {
terminalSource: 'process',
status: 'failed',
summary: 'ECONNRESET after restart',
error: 'ECONNRESET after restart',
},
});
mockGetTask.mockImplementation(async () => currentTask);
mockUpdateTask.mockImplementation(async (_id, update) => {
if (
update.expectedRevision !== undefined &&
update.expectedRevision !== (currentTask.revision ?? 1)
) {
throw Object.assign(new Error('stale revision'), { statusCode: 409 });
}
currentTask = {
...currentTask,
...update,
revision: (currentTask.revision ?? 1) + 1,
} as Task;
return currentTask;
});
const append = vi.fn(async (input: { taskId: string; attemptId: string; kind: string }) => ({
event: {
taskId: input.taskId,
attemptId: input.attemptId,
kind: input.kind,
sequence: 1,
},
}));
service = new ClawdbotAgentService(
undefined,
undefined,
envelopes,
undefined,
completions,
undefined,
undefined,
{ append } as never,
undefined,
undefined,
undefined,
undefined,
undefined,
new RunRecoveryPolicyService(() => 0.5)
);
const schedule = vi
.spyOn(service as never, 'scheduleTaskRecovery')
.mockImplementation(() => undefined);
await (
service as unknown as {
persistSupervisorCompletion(
task: Task,
attempt: TaskAttempt,
result: typeof completionResult
): Promise<void>;
}
).persistSupervisorCompletion(currentTask, currentTask.attempt, completionResult);
expect(currentTask.attempt).toMatchObject({
id: runningAttempt.id,
status: 'failed',
runRetry: {
parentRunId: runningAttempt.id,
state: 'scheduled',
action: 'retry',
failure: { classification: 'transient-transport' },
},
});
expect(schedule).toHaveBeenCalledWith(
currentTask.id,
runningAttempt.id,
expect.objectContaining({ state: 'scheduled', action: 'retry' })
);
});
it('persists one retry branch for duplicate failure planning and cancels the exact parent', async () => {
let currentTask = {
...makeTask('task-retry-once', 'failed'),
revision: 1,
} as Task;
const failedAttempt = currentTask.attempt;
if (!failedAttempt) throw new Error('Expected a failed attempt fixture');
failedAttempt.runLaunchManifest = {
digest: `sha256:${'a'.repeat(64)}`,
routing: {
requestedAgent: 'openclaw',
selectedAgent: 'openclaw',
selectedHost: 'local-process',
reason: 'Test routing decision.',
fallbackAgent: null,
fallbackAllowed: false,
fallbackOnFailure: false,
maxRetries: 3,
},
providerRequirements: { required: [], capabilities: [] },
} as TaskAttempt['runLaunchManifest'];
currentTask.attempts = [failedAttempt];
mockGetTask.mockImplementation(async () => currentTask);
mockUpdateTask.mockImplementation(async (_id, update) => {
if (
update.expectedRevision !== undefined &&
update.expectedRevision !== (currentTask.revision ?? 1)
) {
throw Object.assign(new Error('stale revision'), { statusCode: 409 });
}
currentTask = {
...currentTask,
...update,
revision: (currentTask.revision ?? 1) + 1,
} as Task;
return currentTask;
});
const append = vi.fn(async (input: { taskId: string; attemptId: string; kind: string }) => ({
event: {
taskId: input.taskId,
attemptId: input.attemptId,
kind: input.kind,
sequence: 1,
},
}));
service = new ClawdbotAgentService(
undefined,
undefined,
undefined,
undefined,
undefined,
undefined,
undefined,
{ append } as never,
undefined,
undefined,
undefined,
undefined,
undefined,
new RunRecoveryPolicyService(() => 0.5)
);
const testable = service as unknown as {
planTaskRecovery(
taskId: string,
attempt: TaskAttempt,
failure: RunRecoveryRecord['failure']
): Promise<RunRecoveryRecord | null>;
};
const failure: RunRecoveryRecord['failure'] = {
classification: 'transient-transport',
summary: 'ECONNRESET',
retryable: true,
approvalRequired: false,
destructiveSideEffects: false,
};
const [first, duplicate] = await Promise.all([
testable.planTaskRecovery(currentTask.id, failedAttempt, failure),
testable.planTaskRecovery(currentTask.id, failedAttempt, failure),
]);
expect(first?.state).toBe('scheduled');
expect(duplicate?.state).toBe('scheduled');
expect(mockUpdateTask).toHaveBeenCalledTimes(2);
expect(append).toHaveBeenCalledTimes(1);
const scheduled = currentTask.attempt?.runRetry;
expect(scheduled).toMatchObject({
parentRunId: failedAttempt.id,
sequence: 1,
action: 'retry',
state: 'scheduled',
});
const cancelled = await service.cancelTaskRecovery(
currentTask.id,
failedAttempt.id,
'test-operator'
);
expect(cancelled).toMatchObject({
state: 'cancelled',
action: 'cancelled',
cancelledBy: 'test-operator',
});
expect(currentTask.status).toBe('in-progress');
await expect(
service.cancelTaskRecovery(currentTask.id, 'attempt_wrong', 'test-operator')
).rejects.toMatchObject({ statusCode: 409 });
});
it('launches the exact scheduled task recovery with its causal record', async () => {
const recovery: RunRecoveryRecord = {
schemaVersion: 'run-recovery/v1',
rootRunId: 'attempt_root',
parentRunId: 'attempt_parent',
sequence: 1,
fallbackUsed: false,
state: 'scheduled',
action: 'retry',
failure: {
classification: 'transient-transport',
summary: 'ECONNRESET',
retryable: true,
approvalRequired: false,
destructiveSideEffects: false,
},
reason: 'Retry 1 of 1 after transient-transport.',
backoffMs: 100,
scheduledAt: '2026-07-24T00:00:00.000Z',
notBefore: '2026-07-24T00:00:00.100Z',
selectedAgent: 'openclaw',
routingDecision: 'Matched code rule.',
requiredRuntimeCapabilities: [],
cumulativeBudget: {
tokens: 0,
cost: 0,
runtimeSeconds: 0,
idleRuntimeSeconds: 0,
retries: 0,
fanOut: 1,
},
};
let currentTask = {
...makeTask('task-launch-recovery', 'failed'),
revision: 1,
} as Task;
const parentAttempt = currentTask.attempt;
if (!parentAttempt) throw new Error('Expected a failed attempt fixture');
parentAttempt.id = recovery.parentRunId;
parentAttempt.runRetry = recovery;
currentTask.attempts = [parentAttempt];
mockGetTask.mockImplementation(async () => currentTask);
mockUpdateTask.mockImplementation(async (_id, update) => {
currentTask = {
...currentTask,
...update,
revision: (currentTask.revision ?? 1) + 1,
} as Task;
return currentTask;
});
const appendRunEvent = vi
.spyOn(service as never, 'appendRunEvent')
.mockResolvedValue({} as never);
const startAgent = vi.spyOn(service, 'startAgent').mockResolvedValue({
taskId: currentTask.id,
attemptId: 'attempt_child',
agent: 'openclaw',
runLaunchManifest: { digest: `sha256:${'b'.repeat(64)}` },
} as never);
await (
service as unknown as {
launchScheduledTaskRecovery(taskId: string, attemptId: string): Promise<void>;
}
).launchScheduledTaskRecovery(currentTask.id, parentAttempt.id);
expect(mockUpdateTask).toHaveBeenCalledWith(
currentTask.id,
expect.objectContaining({
expectedRevision: 1,
attempt: expect.objectContaining({
id: parentAttempt.id,
runRetry: expect.objectContaining({ state: 'launching', sequence: 1 }),
}),
})
);
expect(startAgent).toHaveBeenCalledWith(
currentTask.id,
'openclaw',
expect.objectContaining({
parentAttemptId: parentAttempt.id,
recovery: expect.objectContaining({
parentRunId: parentAttempt.id,
state: 'launching',
}),
})
);
expect(appendRunEvent).toHaveBeenCalledWith(
currentTask.id,
'attempt_child',
'recovery.launched',
expect.objectContaining({ parentAttemptId: parentAttempt.id }),
expect.any(Object)
);
});
it('keeps a scheduled task recovery armed when cancellation persistence fails', async () => {
const task = {
...makeTask('task-cancel-race', 'failed'),
revision: 1,
} as Task;
const attempt = task.attempt;
if (!attempt) throw new Error('Expected a failed attempt fixture');
const recovery = new RunRecoveryPolicyService(() => 0.5).decide(
{
classification: 'transient-transport',
summary: 'ECONNRESET',
retryable: true,
approvalRequired: false,
destructiveSideEffects: false,
},
{
rootRunId: attempt.id,
parentRunId: attempt.id,
selectedAgent: attempt.agent,
routingDecision: 'Test route.',
maxRetries: 1,
fallbackOnFailure: false,
now: new Date(Date.now() + 60_000),
}
);
attempt.runRetry = recovery;
mockGetTask.mockResolvedValue(task);
mockUpdateTask.mockRejectedValue(new Error('concurrent task mutation'));
const testable = service as unknown as {
scheduleTaskRecovery(taskId: string, attemptId: string, recovery: RunRecoveryRecord): void;
clearScheduledRecovery(taskId: string, attemptId: string): void;
};
testable.scheduleTaskRecovery(task.id, attempt.id, recovery);
const clearTimeoutSpy = vi.spyOn(globalThis, 'clearTimeout');
await expect(service.cancelTaskRecovery(task.id, attempt.id, 'test-operator')).rejects.toThrow(
'concurrent task mutation'
);
expect(clearTimeoutSpy).not.toHaveBeenCalled();
testable.clearScheduledRecovery(task.id, attempt.id);
clearTimeoutSpy.mockRestore();
});
});

View file

@ -138,6 +138,7 @@ describe('SQLite workflow repositories', () => {
await internals.snapshotWorkflow(running.id, definition);
const completed = run({
revision: running.revision,
status: 'completed',
completedAt: '2026-03-01T00:05:00.000Z',
steps: [

View file

@ -2,7 +2,11 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
import fs from 'node:fs/promises';
import os from 'node:os';
import path from 'node:path';
import type { WorkflowDefinition } from '../../types/workflow.js';
import {
ZERO_AGENT_BUDGET_USAGE,
type RunRecoveryRecord,
} from '@veritas-kanban/shared';
import type { WorkflowDefinition, WorkflowRun } from '../../types/workflow.js';
import { WorkflowService } from '../../services/workflow-service.js';
import {
createTestSqliteDatabase,
@ -53,7 +57,7 @@ function workflow(): WorkflowDefinition {
type: 'agent',
agent: 'agent-1',
input: 'retry once',
on_fail: { retry: 1 },
on_fail: { retry: 1, retry_delay_ms: 100 },
},
{
id: 'approval',
@ -153,4 +157,102 @@ describe('SQLite workflow run execution', () => {
expect(mockBroadcastWorkflowStatus).toHaveBeenCalled();
await expect(fs.access(runsDir)).rejects.toThrow();
});
it('allows only one service instance to claim a persisted workflow recovery', async () => {
const { WorkflowRunService } = await import('../../services/workflow-run-service.js');
const definition = workflow();
await workflowService.saveWorkflow(definition);
const runsDir = path.join(testRoot, 'storage', 'workflow-runs');
const first = new WorkflowRunService({
runsDir,
storageType: 'sqlite',
sqliteDatabase: fixture.database,
workflowService,
});
const second = new WorkflowRunService({
runsDir,
storageType: 'sqlite',
sqliteDatabase: fixture.database,
workflowService,
});
const recovery: RunRecoveryRecord = {
schemaVersion: 'run-recovery/v1',
rootRunId: 'run_root',
parentRunId: 'run_parent',
sequence: 1,
fallbackUsed: false,
state: 'scheduled',
action: 'retry',
failure: {
classification: 'transient-transport',
summary: 'ECONNRESET',
retryable: true,
approvalRequired: false,
destructiveSideEffects: false,
},
reason: 'Retry after transient transport failure.',
backoffMs: 100,
scheduledAt: '2026-07-24T00:00:00.000Z',
notBefore: '2026-07-24T00:00:00.100Z',
selectedAgent: 'agent-1',
routingDecision: 'Workflow retry policy.',
requiredRuntimeCapabilities: ['run.start'],
cumulativeBudget: { ...ZERO_AGENT_BUDGET_USAGE },
};
const run: WorkflowRun = {
id: 'run_1784941000000_race01',
workflowId: definition.id,
workflowVersion: definition.version,
status: 'pending',
currentStep: 'retryable',
context: {},
startedAt: '2026-07-24T00:00:00.000Z',
steps: [
{
stepId: 'retryable',
status: 'failed',
agent: 'agent-1',
retries: 1,
runRetry: recovery,
},
],
};
await (
first as unknown as { saveRun(run: WorkflowRun): Promise<void> }
).saveRun(run);
const executeFirst = vi
.spyOn(first as never, 'executeRun')
.mockResolvedValue(undefined as never);
const executeSecond = vi
.spyOn(second as never, 'executeRun')
.mockResolvedValue(undefined as never);
await Promise.all([
(
first as unknown as {
resumeScheduledWorkflowRecovery(runId: string, stepId: string): Promise<void>;
}
).resumeScheduledWorkflowRecovery(run.id, 'retryable'),
(
second as unknown as {
resumeScheduledWorkflowRecovery(runId: string, stepId: string): Promise<void>;
}
).resumeScheduledWorkflowRecovery(run.id, 'retryable'),
]);
expect(executeFirst.mock.calls.length + executeSecond.mock.calls.length).toBe(1);
expect(await first.getRun(run.id)).toMatchObject({
revision: 2,
status: 'running',
steps: [
expect.objectContaining({
runRetry: expect.objectContaining({
state: 'launched',
parentRunId: recovery.parentRunId,
sequence: recovery.sequence,
}),
}),
],
});
});
});

View file

@ -6,9 +6,11 @@ import path from 'path';
const mockLoadWorkflow = vi.fn();
const mockListWorkflowsMetadata = vi.fn();
const mockExecuteStep = vi.fn();
const mockValidateFallbackAgent = vi.fn();
const mockBroadcastWorkflowStatus = vi.fn();
const mockGetTask = vi.fn();
const mockCheckWorkflowPermission = vi.fn();
const mockGetFallback = vi.fn();
vi.mock('../services/workflow-service.js', () => ({
getWorkflowService: () => ({
@ -23,6 +25,7 @@ vi.mock('../services/workflow-step-executor.js', async (importOriginal) => {
HumanGateBlockError: actual.HumanGateBlockError,
WorkflowStepExecutor: class {
executeStep = mockExecuteStep;
validateFallbackAgent = mockValidateFallbackAgent;
},
};
});
@ -39,6 +42,10 @@ vi.mock('../middleware/workflow-auth.js', () => ({
checkWorkflowPermission: mockCheckWorkflowPermission,
}));
vi.mock('../services/agent-routing-service.js', () => ({
getAgentRoutingService: () => ({ getFallback: mockGetFallback }),
}));
function makeWorkflow(overrides: Record<string, any> = {}) {
return {
id: 'wf-1',
@ -72,6 +79,8 @@ describe('WorkflowRunService', () => {
output: { done: step.id },
outputPath: `/tmp/${step.id}.json`,
}));
mockValidateFallbackAgent.mockResolvedValue({});
mockGetFallback.mockResolvedValue(null);
const mod = await import('../services/workflow-run-service.js');
service = new mod.WorkflowRunService(tmpDir);
});
@ -201,12 +210,16 @@ describe('WorkflowRunService', () => {
it('handles retry, retry_step, skip, block, and workflow failure', async () => {
const delaySpy = vi.spyOn(globalThis, 'setTimeout').mockImplementation(((fn: any) => {
fn();
return 0 as any;
queueMicrotask(fn);
return { unref: vi.fn() } as any;
}) as any);
mockLoadWorkflow.mockResolvedValue(
makeWorkflow({
agents: [
{ id: 'agent-1', name: 'Agent 1' },
{ id: 'TARS', name: 'TARS' },
],
steps: [
{ id: 'prep', type: 'agent', agent: 'agent-1', prompt: 'prep' },
{
@ -240,12 +253,16 @@ describe('WorkflowRunService', () => {
],
})
);
mockGetFallback.mockResolvedValue({
agent: 'TARS',
reason: 'Workspace fallback should not override retry_step.',
});
const counts: Record<string, number> = {};
mockExecuteStep.mockImplementation(async (step: any) => {
counts[step.id] = (counts[step.id] || 0) + 1;
if (step.id === 'retryable' && counts[step.id] === 1) throw new Error('fail once');
if (step.id === 'reroute' && counts[step.id] === 1) throw new Error('reroute me');
if (step.id === 'retryable' && counts[step.id] === 1) throw new Error('ECONNRESET fail once');
if (step.id === 'reroute' && counts[step.id] === 1) throw new Error('ECONNRESET reroute me');
if (step.id === 'skippable') throw new Error('skip me');
if (step.id === 'blocking') throw new Error('block me');
return { output: { done: step.id }, outputPath: `/tmp/${step.id}.json` };
@ -263,6 +280,151 @@ describe('WorkflowRunService', () => {
delaySpy.mockRestore();
});
it('routes an exhausted transient step to a validated fallback agent', async () => {
mockLoadWorkflow.mockResolvedValue(
makeWorkflow({
agents: [
{ id: 'agent-1', name: 'Agent 1' },
{ id: 'TARS', name: 'TARS' },
],
steps: [
{
id: 'step-1',
type: 'agent',
agent: 'agent-1',
prompt: 'x',
on_fail: {
retry: 0,
retry_delay_ms: 1,
escalate_to: 'agent:TARS',
},
},
],
})
);
mockExecuteStep
.mockRejectedValueOnce(new Error('ECONNRESET'))
.mockResolvedValueOnce({ output: { ok: true }, outputPath: '/tmp/fallback.json' });
const run = await service.startRun('wf-1');
await vi.waitFor(
async () => {
const saved = await service.getRun(run.id);
expect(saved.status).toBe('completed');
expect(saved.steps[0]).toMatchObject({
agent: 'TARS',
retries: 1,
runRetry: {
action: 'fallback',
state: 'launched',
fallbackUsed: true,
selectedAgent: 'TARS',
},
});
},
{ timeout: 2_000 }
);
expect(mockValidateFallbackAgent).toHaveBeenCalledWith(
expect.objectContaining({ id: 'step-1' }),
expect.objectContaining({ id: run.id }),
'TARS'
);
});
it('cancels the exact pending workflow retry before its timer can launch', async () => {
mockLoadWorkflow.mockResolvedValue(
makeWorkflow({
steps: [
{
id: 'step-1',
type: 'agent',
agent: 'agent-1',
prompt: 'x',
on_fail: { retry: 2, retry_delay_ms: 10_000 },
},
],
})
);
mockExecuteStep.mockRejectedValueOnce(new Error('ETIMEDOUT'));
const run = await service.startRun('wf-1');
let parentRunId = '';
await vi.waitFor(async () => {
const saved = await service.getRun(run.id);
expect(saved.status).toBe('pending');
expect(saved.steps[0].runRetry?.state).toBe('scheduled');
parentRunId = saved.steps[0].runRetry.parentRunId;
});
const cancelled = await service.cancelPendingRecovery(
run.id,
'step-1',
parentRunId,
'test-operator'
);
expect(cancelled).toMatchObject({
status: 'blocked',
steps: [
expect.objectContaining({
runRetry: expect.objectContaining({
state: 'cancelled',
action: 'cancelled',
cancelledBy: 'test-operator',
}),
}),
],
});
expect(mockExecuteStep).toHaveBeenCalledTimes(1);
});
it('blocks a restarted workflow when a launched recovery cannot be proven terminal', async () => {
mockLoadWorkflow.mockResolvedValue(
makeWorkflow({
steps: [
{
id: 'step-1',
type: 'agent',
agent: 'agent-1',
prompt: 'x',
on_fail: { retry: 1, retry_delay_ms: 10_000 },
},
],
})
);
mockExecuteStep.mockRejectedValueOnce(new Error('ECONNRESET'));
const run = await service.startRun('wf-1');
const persisted = await vi.waitFor(async () => {
const saved = await service.getRun(run.id);
expect(saved.steps[0].runRetry?.state).toBe('scheduled');
return saved;
});
service.clearScheduledWorkflowRecovery(run.id, 'step-1');
persisted.status = 'running';
persisted.steps[0].status = 'running';
persisted.steps[0].runRetry = {
...persisted.steps[0].runRetry,
state: 'launched',
};
await service.saveRun(persisted);
await service.reconcilePendingRecoveries();
expect(await service.getRun(run.id)).toMatchObject({
status: 'blocked',
error: 'Workflow recovery requires operator reconciliation after restart.',
steps: [
expect.objectContaining({
runRetry: expect.objectContaining({
state: 'approval-required',
action: 'approval',
}),
}),
],
});
expect(mockExecuteStep).toHaveBeenCalledTimes(1);
});
it('resumes blocked runs and validates invalid resume requests', async () => {
mockLoadWorkflow.mockResolvedValue(
makeWorkflow({
@ -443,7 +605,7 @@ describe('WorkflowRunService', () => {
]);
});
it('rejects invalid ids, missing workflows, invalid metadata reads, and unimplemented agent escalation', async () => {
it('rejects invalid ids, missing workflows, invalid metadata reads, and incompatible agent escalation', async () => {
await expect(service.getRun('../bad')).rejects.toThrow(/illegal path characters/);
await expect(service.getRun('run_invalid')).rejects.toThrow(/format is invalid/);
@ -458,14 +620,17 @@ describe('WorkflowRunService', () => {
type: 'agent',
agent: 'agent-1',
prompt: 'x',
on_fail: { escalate_to: 'agent:TARS' },
on_fail: { retry: 0, escalate_to: 'agent:TARS' },
},
],
})
);
mockExecuteStep.mockRejectedValueOnce(new Error('boom'));
mockExecuteStep.mockRejectedValueOnce(new Error('ECONNRESET'));
mockValidateFallbackAgent.mockRejectedValueOnce(
new Error('Workflow fallback agent TARS is not defined in the workflow')
);
const run = await service.startRun('wf-1');
await vi.waitFor(async () => expect((await service.getRun(run.id)).status).toBe('failed'));
expect((await service.getRun(run.id)).error).toMatch(/Agent escalation not yet implemented/);
expect((await service.getRun(run.id)).error).toMatch(/Fallback TARS was rejected/);
});
});

View file

@ -35,6 +35,7 @@ vi.mock('../services/governance-trace-service.js', () => ({
import { WorkflowStepExecutor } from '../services/workflow-step-executor.js';
import type { WorkflowRun, WorkflowStep } from '../types/workflow.js';
import type { RunRecoveryRecord } from '@veritas-kanban/shared';
import { providerRuntimeManifestFixture } from './fixtures/provider-runtime-manifest.js';
const runtimeManifestResolver = vi.fn();
@ -179,6 +180,73 @@ describe('WorkflowStepExecutor Codex integration', () => {
);
});
it('persists exact capability and manifest evidence for a launched recovery', async () => {
const persistRun = vi.fn().mockResolvedValue(undefined);
const executor = new WorkflowStepExecutor(tmpDir, { runtimeManifestResolver, persistRun });
const manifest = codexRuntimeManifest();
const recovery: RunRecoveryRecord = {
schemaVersion: 'run-recovery/v1',
rootRunId: 'run_root',
parentRunId: 'run_parent',
sequence: 1,
fallbackUsed: false,
state: 'launched',
action: 'retry',
failure: {
classification: 'transient-transport',
summary: 'ECONNRESET',
retryable: true,
approvalRequired: false,
destructiveSideEffects: false,
},
reason: 'Retry after transient transport failure.',
backoffMs: 100,
selectedAgent: 'codex',
routingDecision: 'Workflow retry policy.',
requiredRuntimeCapabilities: [],
cumulativeBudget: {
tokens: 0,
cost: 0,
runtimeSeconds: 0,
idleRuntimeSeconds: 0,
retries: 0,
fanOut: 1,
},
};
const run = {
id: 'run_1234567890_recover',
workflowId: 'wf-codex',
workflowVersion: 1,
status: 'running',
context: {},
startedAt: new Date().toISOString(),
steps: [{ stepId: 'recover', status: 'running', retries: 1, runRetry: recovery }],
} as WorkflowRun;
const step = { id: 'recover', type: 'agent', agent: 'codex' } as WorkflowStep;
await (
executor as unknown as {
recordRuntimeManifest(
run: WorkflowRun,
step: WorkflowStep,
manifest: typeof manifest,
required: string[]
): Promise<void>;
}
).recordRuntimeManifest(run, step, manifest, ['run.start', 'artifact.write']);
expect(run.steps[0]).toMatchObject({
providerRuntimeManifest: { digest: manifest.digest },
requiredRuntimeCapabilities: ['artifact.write', 'run.start'],
runRetry: {
state: 'launched',
launchedManifestDigest: manifest.digest,
requiredRuntimeCapabilities: ['artifact.write', 'run.start'],
},
});
expect(persistRun).toHaveBeenCalledWith(run);
});
it('rejects providers that have no workflow execution adapter before probing or launch', async () => {
const executor = new WorkflowStepExecutor(tmpDir, { runtimeManifestResolver });
const step: WorkflowStep = {

View file

@ -345,6 +345,38 @@ router.post(
})
);
// GET /api/agents/:taskId/recovery - Read the latest durable recovery decision.
router.get(
'/:taskId/recovery',
asyncHandler(async (req, res) => {
const recovery = await clawdbotAgentService.getTaskRecovery(req.params.taskId as string);
res.json({ recovery });
})
);
// POST /api/agents/:taskId/recovery/cancel - Cancel the exact pending retry/fallback.
router.post(
'/:taskId/recovery/cancel',
requireLocalAgentCapability,
asyncHandler(async (req, res) => {
let attemptId: string;
try {
({ attemptId } = runControlSchema.parse(req.body));
} catch (error) {
if (error instanceof z.ZodError) {
throw new ValidationError('Validation failed', error.issues);
}
throw error;
}
const recovery = await clawdbotAgentService.cancelTaskRecovery(
req.params.taskId as string,
attemptId,
requestActor(req)
);
res.json({ cancelled: true, recovery });
})
);
// POST /api/agents/:taskId/message - Send an attributed operator message to a running agent
router.post(
'/:taskId/message',

View file

@ -50,6 +50,13 @@ const resumeRunSchema = z.object({
context: z.record(z.string(), z.unknown()).optional(),
});
const cancelRecoverySchema = z
.object({
stepId: z.string().trim().min(1).max(160),
parentRunId: z.string().trim().min(1).max(320),
})
.strict();
const authoringContextSchema = z
.object({
taskId: z.string().optional(),
@ -618,6 +625,30 @@ router.post(
})
);
/**
* POST /api/workflows/runs/:id/recovery/cancel Cancel an exact pending step recovery.
*/
router.post(
'/runs/:id/recovery/cancel',
asyncHandler(async (req: AuthenticatedRequest, res) => {
const runId = getStringParam(req.params.id);
const userId = getUserId(req);
const run = await workflowRunService.getRun(runId);
if (!run) {
throw new NotFoundError(`Workflow run ${runId} not found`);
}
await assertWorkflowPermission(run.workflowId, userId, 'execute');
const { stepId, parentRunId } = cancelRecoverySchema.parse(req.body || {});
const cancelled = await workflowRunService.cancelPendingRecovery(
runId,
stepId,
parentRunId,
userId
);
res.json(cancelled);
})
);
/**
* POST /api/workflow-runs/:runId/steps/:stepId/approve Approve a gate step
* Phase 4: Gate approval endpoint fixed for human-gate blocking (#778)

View file

@ -91,6 +91,8 @@ export const RunLaunchManifestSchema = z
reason: safeTextSchema,
fallbackAgent: identifierSchema.nullable(),
fallbackAllowed: z.boolean(),
fallbackOnFailure: z.boolean().optional(),
maxRetries: z.number().int().min(0).max(3).optional(),
})
.strict(),
profile: z

View file

@ -37,6 +37,7 @@ import { initBroadcast, nextWebSocketEventSequence } from './services/broadcast-
import { runStartupMigrations } from './services/migration-service.js';
import { getPolicyService } from './services/policy-service.js';
import { getCredentialBrokerService } from './services/credential-broker-service.js';
import { getWorkflowRunService } from './services/workflow-run-service.js';
import { createBackup, runIntegrityChecks } from './services/integrity-service.js';
import { errorHandler, AppError } from './middleware/error-handler.js';
import { requestIdMiddleware } from './middleware/request-id.js';
@ -593,9 +594,11 @@ async function initializeServices(): Promise<void> {
// a previous server crash/restart (issue #781).
try {
await agentService.reconcileRunningAttempts();
await agentService.reconcilePendingRecoveries();
await getWorkflowRunService().reconcilePendingRecoveries();
} catch (reconcileErr) {
// Non-fatal: log and continue — the server can still serve requests.
log.warn({ err: reconcileErr }, 'Startup: agent attempt reconciliation failed');
log.warn({ err: reconcileErr }, 'Startup: agent run reconciliation failed');
}
await reconcileCredentialLeases('startup');
credentialReconciliationInterval ??= setInterval(

View file

@ -48,6 +48,11 @@ interface RoutingTraceContext {
requiredRuntimeCapabilities?: ProviderRuntimeCapabilityId[];
}
interface FallbackRoutingContext {
preferredFallback?: AgentType;
requiredRuntimeCapabilities?: ProviderRuntimeCapabilityId[];
}
interface RuntimeManifestRegistryReader {
list(): RegisteredAgent[];
}
@ -429,7 +434,8 @@ export class AgentRoutingService {
*/
async getFallback(
task: Pick<Task, 'type' | 'priority' | 'project' | 'subtasks'>,
failedAgent: AgentType
failedAgent: AgentType,
context: FallbackRoutingContext = {}
): Promise<RoutingResult | null> {
const config = await this.configService.getConfig();
const routing: AgentRoutingConfig = config.agentRouting || DEFAULT_ROUTING_CONFIG;
@ -437,15 +443,43 @@ export class AgentRoutingService {
if (!routing.fallbackOnFailure) {
return null;
}
const requiredRuntimeCapabilities = uniqueRuntimeCapabilities(
context.requiredRuntimeCapabilities ?? []
);
if (context.preferredFallback && context.preferredFallback !== failedAgent) {
const preferredAvailability = await this.getAgentAvailability(
config.agents,
context.preferredFallback,
requiredRuntimeCapabilities
);
if (!preferredAvailability.available) {
log.warn(
`Preferred fallback agent "${context.preferredFallback}" is unavailable: ${preferredAvailability.reason}`
);
return null;
}
return {
agent: context.preferredFallback,
model: preferredAvailability.agentConfig?.model,
reason: `Fallback: ${failedAgent} failed → ${context.preferredFallback} (captured launch route)`,
runtimeSelection: preferredAvailability.runtimeSelection,
};
}
// Find the rule that originally matched (to get its fallback)
for (const rule of routing.rules) {
if (!rule.enabled) continue;
if (rule.agent !== failedAgent) continue;
if (!rule.fallback) continue;
if (rule.fallback === failedAgent) continue;
if (!this.matchesRule(task, rule.match)) continue;
const fallbackAvailability = await this.getAgentAvailability(config.agents, rule.fallback);
const fallbackAvailability = await this.getAgentAvailability(
config.agents,
rule.fallback,
requiredRuntimeCapabilities
);
if (!fallbackAvailability.available) {
log.warn(
`Fallback agent "${rule.fallback}" for rule "${rule.name}" is unavailable: ${fallbackAvailability.reason}`
@ -456,20 +490,27 @@ export class AgentRoutingService {
log.info(`Falling back from ${failedAgent}${rule.fallback} (rule: ${rule.name})`);
return {
agent: rule.fallback,
model: fallbackAvailability.agentConfig?.model,
rule: rule.id,
reason: `Fallback: ${failedAgent} failed → ${rule.fallback} (rule: ${rule.name})`,
runtimeSelection: fallbackAvailability.runtimeSelection,
};
}
// No specific fallback found — try default if it's different from failed
const defaultAgent = routing.defaultAgent || config.defaultAgent;
if (defaultAgent !== failedAgent) {
const defaultAvailability = await this.getAgentAvailability(config.agents, defaultAgent);
const defaultAvailability = await this.getAgentAvailability(
config.agents,
defaultAgent,
requiredRuntimeCapabilities
);
if (defaultAvailability.available) {
return {
agent: defaultAgent,
model: routing.defaultModel,
reason: `Fallback: ${failedAgent} failed → default agent (${defaultAgent})`,
runtimeSelection: defaultAvailability.runtimeSelection,
};
}
}

View file

@ -53,8 +53,10 @@ import type { ThreadEvent } from '@openai/codex-sdk';
import {
evaluateTaskReadiness,
CONVERSATION_LIFECYCLE_SCHEMA_VERSION,
DEFAULT_ROUTING_CONFIG,
EXECUTABLE_AGENT_PROVIDERS,
RUN_LAUNCH_MANIFEST_SCHEMA_VERSION,
ZERO_AGENT_BUDGET_USAGE,
} from '@veritas-kanban/shared';
import type {
Task,
@ -76,6 +78,7 @@ import type {
AgentBudgetUsage,
AgentBudgetDecision,
AgentBudgetEvaluation,
RunRecoveryRecord,
AgentProfileLaunchMetadata,
AgentProfileResolvedLaunch,
ExecutableAgentProvider,
@ -119,7 +122,7 @@ import type {
RunApprovalRiskClass,
} from '@veritas-kanban/shared';
import { createLogger } from '../lib/logger.js';
import { ConflictError } from '../middleware/error-handler.js';
import { ConflictError, NotFoundError } from '../middleware/error-handler.js';
import type { AgentBudgetThresholdEvent } from '@veritas-kanban/shared';
import { getAgentProfilePackageService } from './agent-profile-package-service.js';
import {
@ -225,6 +228,7 @@ import {
type RunToolBridgeService,
} from './run-tool-bridge-service.js';
import { getToolPolicyService } from './tool-policy-service.js';
import { RunRecoveryPolicyService } from './run-recovery-policy-service.js';
const log = createLogger('clawdbot-agent-service');
const TRACE_SECRET_PATTERNS: Array<[RegExp, string]> = [
@ -290,6 +294,7 @@ export interface AgentStatus {
runLaunchManifest: RunLaunchManifest;
runLaunchParentAttemptId?: string;
runLaunchManifestDrift?: RunLaunchManifestDriftResult;
runRetry?: RunRecoveryRecord;
conversation: ConversationLifecycleRecord;
controls: ProviderRuntimeControlSet;
}
@ -309,6 +314,8 @@ export interface AgentStartOptions {
commitPolicy?: TaskCommitPolicy;
parentAttemptId?: string;
conversation?: ConversationLaunchRequest;
/** Internal durable recovery context. API callers cannot supply this field. */
recovery?: RunRecoveryRecord;
}
export interface AgentMessageOptions {
@ -349,6 +356,7 @@ interface PendingAgent {
provider: ExecutableAgentProvider;
model?: string;
budget?: AgentBudgetState;
recoveryBudgetBase?: AgentBudgetUsage;
budgetStopped?: boolean;
agentProfile?: AgentProfileLaunchMetadata;
providerRuntimeManifest: ProviderRuntimeManifest;
@ -358,6 +366,7 @@ interface PendingAgent {
runLaunchManifestTraceId: string;
runLaunchParentAttemptId?: string;
runLaunchManifestDrift?: RunLaunchManifestDriftResult;
runRetry?: RunRecoveryRecord;
conversation: ConversationLifecycleRecord;
supervisorId?: string;
recoveredControl?: boolean;
@ -415,6 +424,10 @@ const startingAgents = new Set<string>();
const finalizingAgents = new Map<PendingAgent, Promise<void>>();
const budgetEvaluations = new Map<PendingAgent, Promise<void>>();
const recoveredProcessMonitors = new Map<string, NodeJS.Timeout>();
const scheduledRecoveries = new Map<
string,
{ attemptId: string; timer: ReturnType<typeof setTimeout> }
>();
const COMPLETION_PERSISTENCE_ATTEMPTS = 3;
const NOOP_CREDENTIAL_LEASE_LIFECYCLE: CredentialLeaseLifecycle = {
async revokeRun() {
@ -464,6 +477,7 @@ export class ClawdbotAgentService {
private conversationLifecycle: ConversationLifecycleService;
private toolControlPlane: ToolControlPlaneService;
private runToolBridge: RunToolBridgeService;
private runRecoveryPolicy: RunRecoveryPolicyService;
private logsDir: string;
constructor(
@ -479,7 +493,8 @@ export class ClawdbotAgentService {
runSupervisor: RunSupervisorService = getRunSupervisorService(),
conversationLifecycle = new ConversationLifecycleService(),
toolControlPlane: ToolControlPlaneService = getToolControlPlaneService(),
runToolBridge: RunToolBridgeService = getRunToolBridgeService()
runToolBridge: RunToolBridgeService = getRunToolBridgeService(),
runRecoveryPolicy = new RunRecoveryPolicyService()
) {
this.configService = new ConfigService();
this.taskService = new TaskService();
@ -502,6 +517,7 @@ export class ClawdbotAgentService {
this.conversationLifecycle = conversationLifecycle;
this.toolControlPlane = toolControlPlane;
this.runToolBridge = runToolBridge;
this.runRecoveryPolicy = runRecoveryPolicy;
this.logsDir = getLogsDir();
this.ensureLogsDir();
}
@ -731,6 +747,442 @@ export class ClawdbotAgentService {
}
}
/**
* Restore durable retry/fallback timers after process restart.
*
* A record left in `launching` has no child attempt, otherwise the child
* would be the task's current attempt. Re-queueing that exact record is safe
* because the task revision and parent attempt ID are claimed again before
* launch.
*/
async reconcilePendingRecoveries(): Promise<void> {
let tasks: Task[];
try {
tasks = await this.taskService.listTasks();
} catch (error) {
log.warn({ err: error }, '[ClawdbotAgent] reconcilePendingRecoveries: failed to list tasks');
return;
}
let scheduledCount = 0;
for (const task of tasks) {
const attempt = task.attempt;
const recovery = attempt?.runRetry;
if (!attempt || !recovery) continue;
if (attempt.status === 'running' || !['scheduled', 'launching'].includes(recovery.state)) {
continue;
}
try {
let record = recovery;
if (record.state === 'launching') {
record = {
...record,
state: 'scheduled',
notBefore: new Date().toISOString(),
reason: `${record.reason} Re-queued after server restart before child launch.`,
};
const recoveredAttempt = { ...attempt, runRetry: record };
const updated = await this.taskService.updateTask(task.id, {
expectedRevision: normalizedTaskRevision(task),
attempt: recoveredAttempt,
attempts: upsertAttemptHistory(task.attempts, recoveredAttempt),
});
if (!updated) continue;
await this.appendRunEvent(
task.id,
attempt.id,
'recovery.reconciled',
{
action: record.action,
sequence: record.sequence,
state: record.state,
notBefore: record.notBefore,
},
{
provider: 'system',
adapter: 'run-recovery',
agent: record.selectedAgent,
dedupeKey: `recovery.reconciled:${record.sequence}`,
}
);
}
this.scheduleTaskRecovery(task.id, attempt.id, record);
scheduledCount += 1;
} catch (error) {
log.warn(
{ err: error, taskId: task.id, attemptId: attempt.id },
'[ClawdbotAgent] Failed to reconcile pending recovery'
);
}
}
if (scheduledCount > 0) {
log.info(
{ scheduledCount },
'[ClawdbotAgent] Durable retry/fallback reconciliation complete'
);
}
}
async getTaskRecovery(taskId: string): Promise<RunRecoveryRecord | null> {
const task = await this.taskService.getTask(taskId);
if (!task) throw new NotFoundError(`Task "${taskId}" not found`);
if (task.attempt?.runRetry) return task.attempt.runRetry;
return (
[...(task.attempts ?? [])].reverse().find((attempt) => attempt.runRetry)?.runRetry ?? null
);
}
async cancelTaskRecovery(
taskId: string,
expectedAttemptId: string,
actor = 'operator'
): Promise<RunRecoveryRecord> {
const task = await this.taskService.getTask(taskId);
if (!task) throw new NotFoundError(`Task "${taskId}" not found`);
const attempt = task.attempt;
const recovery = attempt?.runRetry;
if (!attempt || attempt.id !== expectedAttemptId || !recovery) {
throw new ConflictError('Recovery cancellation does not match the active attempt', {
activeAttemptId: attempt?.id,
requestedAttemptId: expectedAttemptId,
});
}
if (!['scheduled', 'launching'].includes(recovery.state)) {
throw new ConflictError('Recovery is not pending cancellation', {
attemptId: attempt.id,
recoveryState: recovery.state,
});
}
const cancelled: RunRecoveryRecord = {
...recovery,
state: 'cancelled',
action: 'cancelled',
reason: 'Automatic recovery was cancelled by an operator.',
backoffMs: 0,
cancelledAt: new Date().toISOString(),
cancelledBy: actor.trim() || 'operator',
handoff: {
summary: 'Automatic recovery was cancelled.',
nextActions: ['Launch a new attempt explicitly if the objective should continue.'],
},
};
const cancelledAttempt = { ...attempt, runRetry: cancelled };
const updated = await this.taskService.updateTask(taskId, {
expectedRevision: normalizedTaskRevision(task),
attempt: cancelledAttempt,
attempts: upsertAttemptHistory(task.attempts, cancelledAttempt),
});
if (!updated) throw new Error(`Task "${taskId}" disappeared during recovery cancellation`);
this.clearScheduledRecovery(taskId, expectedAttemptId);
await this.appendRunEvent(
taskId,
attempt.id,
'recovery.cancelled',
{
action: recovery.action,
sequence: recovery.sequence,
actor: cancelled.cancelledBy,
},
{
provider: 'operator',
adapter: 'run-recovery',
agent: recovery.selectedAgent,
dedupeKey: `recovery.cancelled:${recovery.sequence}`,
}
);
return cancelled;
}
private async planTaskRecovery(
taskId: string,
failedAttempt: TaskAttempt,
failure: RunRecoveryRecord['failure']
): Promise<RunRecoveryRecord | null> {
const task = await this.taskService.getTask(taskId);
if (!task || task.attempt?.id !== failedAttempt.id) return null;
const currentAttempt = task.attempt;
const currentRecovery = currentAttempt.runRetry;
if (
currentRecovery &&
['scheduled', 'approval-required', 'exhausted', 'cancelled'].includes(currentRecovery.state)
) {
return currentRecovery;
}
const redactedFailure = {
...failure,
summary: this.redactTraceText(failure.summary),
};
const launchManifest = currentAttempt.runLaunchManifest;
const routing = launchManifest?.routing;
const maxRetries = routing?.maxRetries ?? DEFAULT_ROUTING_CONFIG.maxRetries;
const fallbackOnFailure = routing?.fallbackOnFailure ?? routing?.fallbackAllowed ?? false;
const requiredRuntimeCapabilities = [
...(launchManifest?.providerRequirements.required ?? []),
] as ProviderRuntimeCapabilityId[];
const previousSequence = currentRecovery?.sequence ?? 0;
const fallbackUsed = currentRecovery?.fallbackUsed ?? false;
const cumulativeBudget =
currentAttempt.budget?.usage ??
currentRecovery?.cumulativeBudget ??
({ ...ZERO_AGENT_BUDGET_USAGE } satisfies AgentBudgetUsage);
const preferredFallback = currentRecovery?.fallbackAgent ?? routing?.fallbackAgent ?? undefined;
let fallbackAgent: AgentType | undefined = preferredFallback;
let fallbackEligible: boolean | undefined;
let fallbackReason: string | undefined;
if (failure.retryable && previousSequence >= maxRetries && fallbackOnFailure && !fallbackUsed) {
const fallback = await getAgentRoutingService().getFallback(task, currentAttempt.agent, {
...(preferredFallback ? { preferredFallback } : {}),
requiredRuntimeCapabilities,
});
fallbackAgent = fallback?.agent ?? preferredFallback;
fallbackEligible = Boolean(fallback);
fallbackReason =
fallback?.reason ??
(fallbackAgent
? `Fallback ${fallbackAgent} is unavailable or lacks required runtime capabilities.`
: 'No compatible fallback route is configured.');
}
const decisionInput = {
rootRunId: currentRecovery?.rootRunId ?? currentAttempt.id,
parentRunId: currentAttempt.id,
selectedAgent: currentAttempt.agent,
routingDecision:
currentRecovery?.routingDecision ??
routing?.reason ??
'Legacy run without captured routing evidence.',
...(launchManifest?.digest ? { sourceManifestDigest: launchManifest.digest } : {}),
requiredRuntimeCapabilities,
cumulativeBudget,
previousSequence,
fallbackUsed,
maxRetries,
fallbackOnFailure,
...(fallbackAgent ? { fallbackAgent } : {}),
...(fallbackEligible !== undefined ? { fallbackEligible } : {}),
...(fallbackReason ? { fallbackReason } : {}),
};
let decision = this.runRecoveryPolicy.decide(redactedFailure, decisionInput);
if (decision.action === 'fallback' && fallbackAgent) {
try {
const preview = await this.previewAgentLaunch(
taskId,
fallbackAgent,
this.recoveryLaunchOptions(currentAttempt, decision)
);
this.runLaunchManifests.assertEnforceable(preview.manifest);
} catch (error) {
decision = this.runRecoveryPolicy.decide(redactedFailure, {
...decisionInput,
fallbackEligible: false,
fallbackReason: this.redactTraceText(
error instanceof Error ? error.message : String(error)
),
});
}
}
const recoveredAttempt = { ...currentAttempt, runRetry: decision };
try {
const updated = await this.taskService.updateTask(taskId, {
expectedRevision: normalizedTaskRevision(task),
...(decision.state === 'approval-required' ? { status: 'blocked' as const } : {}),
attempt: recoveredAttempt,
attempts: upsertAttemptHistory(task.attempts, recoveredAttempt),
});
if (!updated) return null;
} catch (error) {
const latest = await this.taskService.getTask(taskId);
if (
latest?.attempt?.id === currentAttempt.id &&
latest.attempt.runRetry?.state === decision.state &&
latest.attempt.runRetry.sequence === decision.sequence
) {
return latest.attempt.runRetry;
}
throw error;
}
await this.appendRunEvent(
taskId,
currentAttempt.id,
`recovery.${decision.state}`,
{
action: decision.action,
state: decision.state,
sequence: decision.sequence,
failureClass: decision.failure.classification,
reason: decision.reason,
backoffMs: decision.backoffMs,
notBefore: decision.notBefore,
selectedAgent: decision.selectedAgent,
fallbackAgent: decision.fallbackAgent,
cumulativeBudget: decision.cumulativeBudget,
handoff: decision.handoff,
},
{
provider: 'system',
adapter: 'run-recovery',
agent: decision.selectedAgent,
dedupeKey: `recovery.${decision.state}:${decision.sequence}`,
}
);
if (decision.state === 'scheduled') {
this.scheduleTaskRecovery(taskId, currentAttempt.id, decision);
}
return decision;
}
private recoveryLaunchOptions(
parentAttempt: TaskAttempt,
recovery: RunRecoveryRecord
): AgentStartOptions {
const retryingSameAgent = recovery.action === 'retry';
return {
...(retryingSameAgent && parentAttempt.agentProfile?.id
? { profileId: parentAttempt.agentProfile.id }
: {}),
...(parentAttempt.runLaunchManifest?.sandbox.presetId
? { sandboxPresetId: parentAttempt.runLaunchManifest.sandbox.presetId }
: {}),
...(parentAttempt.runLaunchManifest?.budget
? { budget: parentAttempt.runLaunchManifest.budget }
: {}),
...(parentAttempt.runLaunchManifest?.providerRequirements.required.length
? {
requiredRuntimeCapabilities: [
...parentAttempt.runLaunchManifest.providerRequirements.required,
] as ProviderRuntimeCapabilityId[],
}
: {}),
...(parentAttempt.taskEnvelope?.commitPolicy
? { commitPolicy: parentAttempt.taskEnvelope.commitPolicy }
: {}),
parentAttemptId: parentAttempt.id,
recovery,
};
}
private scheduleTaskRecovery(
taskId: string,
attemptId: string,
recovery: RunRecoveryRecord
): void {
if (recovery.state !== 'scheduled') return;
this.clearScheduledRecovery(taskId);
const notBefore = recovery.notBefore ? Date.parse(recovery.notBefore) : Date.now();
const delay = Math.max(0, Math.min(2_147_483_647, notBefore - Date.now()));
const timer = setTimeout(() => {
const scheduled = scheduledRecoveries.get(taskId);
if (!scheduled || scheduled.attemptId !== attemptId) return;
scheduledRecoveries.delete(taskId);
void this.launchScheduledTaskRecovery(taskId, attemptId).catch((error) => {
log.error(
{ err: error, taskId, attemptId },
'[ClawdbotAgent] Scheduled recovery launch failed'
);
});
}, delay);
timer.unref?.();
scheduledRecoveries.set(taskId, { attemptId, timer });
}
private clearScheduledRecovery(taskId: string, expectedAttemptId?: string): void {
const scheduled = scheduledRecoveries.get(taskId);
if (!scheduled || (expectedAttemptId && scheduled.attemptId !== expectedAttemptId)) return;
clearTimeout(scheduled.timer);
scheduledRecoveries.delete(taskId);
}
private async launchScheduledTaskRecovery(taskId: string, attemptId: string): Promise<void> {
const task = await this.taskService.getTask(taskId);
const parentAttempt = task?.attempt;
const recovery = parentAttempt?.runRetry;
if (
!task ||
!parentAttempt ||
parentAttempt.id !== attemptId ||
parentAttempt.status === 'running' ||
recovery?.state !== 'scheduled'
) {
return;
}
if (recovery.notBefore && Date.parse(recovery.notBefore) > Date.now()) {
this.scheduleTaskRecovery(taskId, attemptId, recovery);
return;
}
const launching: RunRecoveryRecord = { ...recovery, state: 'launching' };
const claimedAttempt = { ...parentAttempt, runRetry: launching };
const claimed = await this.taskService.updateTask(taskId, {
expectedRevision: normalizedTaskRevision(task),
attempt: claimedAttempt,
attempts: upsertAttemptHistory(task.attempts, claimedAttempt),
});
if (!claimed) return;
await this.appendRunEvent(
taskId,
attemptId,
'recovery.launching',
{
action: launching.action,
sequence: launching.sequence,
selectedAgent: launching.selectedAgent,
},
{
provider: 'system',
adapter: 'run-recovery',
agent: launching.selectedAgent,
dedupeKey: `recovery.launching:${launching.sequence}`,
}
);
try {
const child = await this.startAgent(
taskId,
launching.selectedAgent,
this.recoveryLaunchOptions(claimedAttempt, launching)
);
await this.appendRunEvent(
taskId,
child.attemptId,
'recovery.launched',
{
action: launching.action,
sequence: launching.sequence,
parentAttemptId: attemptId,
launchedAttemptId: child.attemptId,
selectedAgent: child.agent,
sourceManifestDigest: launching.sourceManifestDigest,
launchedManifestDigest: child.runLaunchManifest.digest,
cumulativeBudget: launching.cumulativeBudget,
},
{
provider: 'system',
adapter: 'run-recovery',
agent: child.agent,
dedupeKey: `recovery.launched:${launching.sequence}`,
}
);
} catch (error) {
const latest = await this.taskService.getTask(taskId);
if (latest?.attempt?.id === attemptId) {
await this.planTaskRecovery(
taskId,
latest.attempt,
this.runRecoveryPolicy.classifyError(error)
);
}
throw error;
}
}
private async restoreRecoveredRun(
task: Task,
attempt: TaskAttempt,
@ -770,6 +1222,7 @@ export class ClawdbotAgentService {
provider,
model: attempt.model,
budget: supervisor.budget ?? attempt.budget,
recoveryBudgetBase: attempt.runRetry?.cumulativeBudget,
agentProfile: attempt.agentProfile,
providerRuntimeManifest: attempt.providerRuntimeManifest,
harnessSupport: attempt.harnessSupport,
@ -779,6 +1232,7 @@ export class ClawdbotAgentService {
attempt.runLaunchManifestTraceId ?? `run-supervisor:${supervisor.id}`,
runLaunchParentAttemptId: attempt.runLaunchParentAttemptId,
runLaunchManifestDrift: attempt.runLaunchManifestDrift,
runRetry: attempt.runRetry,
conversation: recoveredConversation,
supervisorId: supervisor.id,
recoveredControl: true,
@ -911,6 +1365,7 @@ export class ClawdbotAgentService {
}
const config = await this.configService.getConfig();
const routingPolicy = config.agentRouting ?? DEFAULT_ROUTING_CONFIG;
const profileLaunch = options.profileId
? await getAgentProfilePackageService().resolveLaunch(options.profileId)
: undefined;
@ -961,16 +1416,19 @@ export class ClawdbotAgentService {
agentBudget: budgetSources.agentBudget,
runBudget: budgetSources.runBudget ?? profileLaunch?.budget,
});
const budgetEvaluation = budgetService.evaluate(
budgetPolicy,
{ fanOut: 1 },
{
taskId,
agentId: agent,
actionType: 'agent.launch-preview',
project: task.project,
}
);
const recoveryBudgetUsage = options.recovery
? {
...options.recovery.cumulativeBudget,
retries: Math.max(options.recovery.cumulativeBudget.retries, options.recovery.sequence),
fanOut: Math.max(1, options.recovery.cumulativeBudget.fanOut),
}
: { fanOut: 1 };
const budgetEvaluation = budgetService.evaluate(budgetPolicy, recoveryBudgetUsage, {
taskId,
agentId: agent,
actionType: 'agent.launch-preview',
project: task.project,
});
if (this.isBlockingBudgetDecision(budgetEvaluation.decision)) {
throw new ConflictError('Agent run budget requires operator action before launch', {
decision: budgetEvaluation.decision,
@ -1055,6 +1513,8 @@ export class ClawdbotAgentService {
requestedAgent,
routingReason,
routingFallback,
routingFallbackOnFailure: routingPolicy.fallbackOnFailure,
routingMaxRetries: routingPolicy.maxRetries,
agent,
launchAgentConfig,
provider,
@ -1138,6 +1598,7 @@ export class ClawdbotAgentService {
// Get agent config — use routing engine when agent is "auto" or not specified
const config = await this.configService.getConfig();
const routingPolicy = config.agentRouting ?? DEFAULT_ROUTING_CONFIG;
const profileLaunch = options.profileId
? await getAgentProfilePackageService().resolveLaunch(options.profileId)
: undefined;
@ -1195,16 +1656,19 @@ export class ClawdbotAgentService {
agentBudget: budgetSources.agentBudget,
runBudget: budgetSources.runBudget ?? profileLaunch?.budget,
});
const budgetEvaluation = budgetService.evaluate(
budgetPolicy,
{ fanOut: 1 },
{
taskId,
agentId: agent,
actionType: 'agent.start',
project: task.project,
}
);
const recoveryBudgetUsage = options.recovery
? {
...options.recovery.cumulativeBudget,
retries: Math.max(options.recovery.cumulativeBudget.retries, options.recovery.sequence),
fanOut: Math.max(1, options.recovery.cumulativeBudget.fanOut),
}
: { fanOut: 1 };
const budgetEvaluation = budgetService.evaluate(budgetPolicy, recoveryBudgetUsage, {
taskId,
agentId: agent,
actionType: 'agent.start',
project: task.project,
});
const budgetTraceIds: string[] = [];
if (budgetEvaluation.trace) {
const trace = await getGovernanceTraceService().record(budgetEvaluation.trace);
@ -1345,6 +1809,8 @@ export class ClawdbotAgentService {
requestedAgent,
routingReason,
routingFallback,
routingFallbackOnFailure: routingPolicy.fallbackOnFailure,
routingMaxRetries: routingPolicy.maxRetries,
agent,
launchAgentConfig,
provider,
@ -1368,6 +1834,16 @@ export class ClawdbotAgentService {
const runLaunchManifestDrift = parentAttempt?.runLaunchManifest
? diffRunLaunchManifests(runLaunchManifest, parentAttempt.runLaunchManifest)
: undefined;
const runRetry = options.recovery
? {
...options.recovery,
state: 'launched' as const,
launchedAt: startedAt,
launchedRunId: attemptId,
launchedManifestDigest: runLaunchManifest.digest,
selectedAgent: agent,
}
: undefined;
const runLaunchTrace = await getGovernanceTraceService().record({
kind: 'policy',
outcome: runLaunchManifest.enforcement.enforceable ? 'allowed' : 'blocked',
@ -1433,6 +1909,7 @@ export class ClawdbotAgentService {
runLaunchManifestTraceId: runLaunchTrace.id,
runLaunchParentAttemptId: parentAttempt?.id,
runLaunchManifestDrift,
runRetry,
conversation,
budget: budgetPolicy
? {
@ -1445,6 +1922,7 @@ export class ClawdbotAgentService {
overrideReason: options.overrideReason,
}
: undefined,
recoveryBudgetBase: options.recovery?.cumulativeBudget,
});
// Initialize log file (ensure it stays within logs dir)
@ -1476,6 +1954,7 @@ export class ClawdbotAgentService {
runLaunchManifestTraceId: runLaunchTrace.id,
runLaunchParentAttemptId: parentAttempt?.id,
runLaunchManifestDrift,
runRetry,
conversation,
};
@ -1494,6 +1973,24 @@ export class ClawdbotAgentService {
}
task = claimedTask;
}
if (options.recovery) {
const claimedRecovery = task.attempt?.runRetry;
if (
task.attempt?.id !== options.parentAttemptId ||
claimedRecovery?.state !== 'launching' ||
claimedRecovery.sequence !== options.recovery.sequence ||
claimedRecovery.parentRunId !== options.recovery.parentRunId
) {
pendingAgents.delete(taskId);
throw new ConflictError('Recovery launch no longer matches the claimed parent attempt', {
taskId,
activeAttemptId: task.attempt?.id,
parentAttemptId: options.parentAttemptId,
recoveryState: claimedRecovery?.state,
recoverySequence: claimedRecovery?.sequence,
});
}
}
try {
const pending = pendingAgents.get(taskId);
if (!pending || pending.attemptId !== attemptId) {
@ -1514,6 +2011,7 @@ export class ClawdbotAgentService {
})
: undefined;
await this.taskService.updateTask(taskId, {
...(options.recovery ? { expectedRevision: normalizedTaskRevision(task) } : {}),
status: 'in-progress',
attempt,
attempts: task.attempt ? upsertAttemptHistory(task.attempts, task.attempt) : task.attempts,
@ -1727,6 +2225,42 @@ export class ClawdbotAgentService {
stackTrace: startError.stack,
harnessSupport: this.harnessTelemetry(harnessSupport, 'launch-failed'),
});
const launchCleanupEffects: Array<[string, () => void | Promise<void>]> = [
[
'release worktree ownership',
async () => {
if (usesManagedWorktree) {
await this.worktrees.releaseOwnership(taskId, attemptId);
}
},
],
[
'revoke run credential leases',
() =>
this.revokeRunCredentialLeases(taskId, attemptId, 'failed', runLaunchManifest.digest),
],
['close run tool sessions', () => this.toolControlPlane.closeRun(taskId, attemptId)],
];
for (const [effect, cleanup] of launchCleanupEffects) {
try {
await cleanup();
} catch (cleanupError) {
log.error(
{ err: cleanupError, taskId, attemptId, effect },
'Failed to clean up a failed recovery-capable launch'
);
}
}
await this.planTaskRecovery(
taskId,
failedAttempt,
this.runRecoveryPolicy.classifyError(startError)
).catch((recoveryError) => {
log.error(
{ err: recoveryError, taskId, attemptId },
'Failed to persist recovery policy after provider launch failure'
);
});
throw new Error(`Failed to start agent via ${adapter.label}: ${startError.message}`, {
cause: error,
});
@ -1746,6 +2280,7 @@ export class ClawdbotAgentService {
runLaunchManifest,
runLaunchParentAttemptId: parentAttempt?.id,
runLaunchManifestDrift,
runRetry,
conversation,
controls: providerRuntimeControls(providerRuntimeManifest),
};
@ -2135,6 +2670,13 @@ export class ClawdbotAgentService {
);
});
}
if (completionResult.status !== 'success') {
await this.planTaskRecovery(
task.id,
completedAttempt,
this.runRecoveryPolicy.classifyCompletion(completionResult)
);
}
}
private async persistRestartedProviderCompletion(
@ -2481,6 +3023,7 @@ export class ClawdbotAgentService {
runLaunchManifestTraceId: pending.runLaunchManifestTraceId,
runLaunchParentAttemptId: pending.runLaunchParentAttemptId,
runLaunchManifestDrift: pending.runLaunchManifestDrift,
runRetry: pending.runRetry,
conversation: pending.conversation,
completionResult,
};
@ -2663,6 +3206,19 @@ export class ClawdbotAgentService {
}
}
if (!successful) {
await this.planTaskRecovery(
taskId,
preparedCompletion.completedAttempt,
this.runRecoveryPolicy.classifyCompletion(completionResult)
).catch((recoveryError) => {
log.error(
{ err: recoveryError, taskId, attemptId },
'[ClawdbotAgent] Failed to persist automatic recovery decision'
);
});
}
log.info(`[ClawdbotAgent] Task ${taskId} completed with status: ${status}`);
}
@ -3222,6 +3778,19 @@ export class ClawdbotAgentService {
}
const budgetService = getAgentBudgetService();
const usage = budgetService.mergeUsage(pending.budget.usage, delta);
if (pending.recoveryBudgetBase) {
if (delta.runtimeSeconds !== undefined) {
usage.runtimeSeconds =
pending.recoveryBudgetBase.runtimeSeconds + Math.max(0, delta.runtimeSeconds);
}
if (delta.idleRuntimeSeconds !== undefined) {
usage.idleRuntimeSeconds =
pending.recoveryBudgetBase.idleRuntimeSeconds + Math.max(0, delta.idleRuntimeSeconds);
}
if (pending.runRetry) {
usage.retries = Math.max(usage.retries, pending.runRetry.sequence);
}
}
const nextEvaluation = budgetService.evaluate(pending.budget.policy, usage, {
taskId,
agentId: pending.agent,
@ -7110,6 +7679,7 @@ export class ClawdbotAgentService {
runLaunchManifest: pending.runLaunchManifest,
runLaunchParentAttemptId: pending.runLaunchParentAttemptId,
runLaunchManifestDrift: pending.runLaunchManifestDrift,
runRetry: pending.runRetry,
conversation: pending.conversation,
controls: providerRuntimeControls(pending.providerRuntimeManifest),
};
@ -7325,6 +7895,8 @@ export class ClawdbotAgentService {
requestedAgent: AgentType;
routingReason: string;
routingFallback?: AgentType;
routingFallbackOnFailure: boolean;
routingMaxRetries: number;
agent: AgentType;
launchAgentConfig?: AgentConfig;
provider: ExecutableAgentProvider;
@ -7951,7 +8523,9 @@ export class ClawdbotAgentService {
selectedHost: input.provider === 'openclaw' ? 'openclaw-gateway' : 'local-process',
reason: input.routingReason,
fallbackAgent: input.routingFallback ?? null,
fallbackAllowed: Boolean(input.routingFallback),
fallbackAllowed: Boolean(input.routingFallback && input.routingFallbackOnFailure),
fallbackOnFailure: input.routingFallbackOnFailure,
maxRetries: input.routingMaxRetries,
},
...(profile
? {

View file

@ -0,0 +1,331 @@
import {
RUN_RECOVERY_SCHEMA_VERSION,
ZERO_AGENT_BUDGET_USAGE,
type AgentBudgetUsage,
type CompletionResult,
type RunFailureClassification,
type RunRecoveryRecord,
type TaskCompletionBlocker,
type TaskCompletionSideEffect,
type TaskCompletionStatus,
type TaskTerminalSource,
} from '@veritas-kanban/shared';
const DEFAULT_BACKOFF_MS = 1_000;
const MAX_BACKOFF_MS = 30_000;
const JITTER_RATIO = 0.2;
const RATE_LIMIT_PATTERN =
/\b(?:429|rate[ -]?limit(?:ed|ing)?|too many requests|quota (?:exceeded|exhausted))\b/i;
const TIMEOUT_PATTERN = /\b(?:timed? ?out|timeout|deadline exceeded|etimedout|ehostunreach)\b/i;
const PROVIDER_UNAVAILABLE_PATTERN =
/\b(?:provider unavailable|service unavailable|temporarily unavailable|unhealthy|not found on path|command not found|enoent|connection refused|econnrefused|gateway unavailable)\b/i;
const TRANSIENT_TRANSPORT_PATTERN =
/\b(?:transient|econnreset|epipe|socket hang up|connection (?:closed|reset|lost)|network error|fetch failed|http 5\d\d|bad gateway|gateway timeout|dns|eai_again)\b/i;
const INVALID_REQUEST_PATTERN =
/\b(?:invalid (?:request|configuration|config|argument|option|schema)|validation failed|unsupported (?:flag|option|provider)|unknown (?:flag|option)|malformed|zod)\b/i;
const POLICY_BLOCK_PATTERN =
/\b(?:policy|sandbox|permission|approval|unauthori[sz]ed|forbidden|denied|budget)\b/i;
const VERIFICATION_PATTERN =
/\b(?:verification|acceptance criteri|test(?:s|ing)? failed|required (?:output|commit|evidence).*(?:missing|failed))\b/i;
export interface RunFailureEvidence {
status: TaskCompletionStatus;
terminalSource?: TaskTerminalSource;
summary?: string | null;
error?: string | null;
blockers?: TaskCompletionBlocker[];
sideEffects?: TaskCompletionSideEffect[];
}
export interface RunRecoveryDecisionInput {
rootRunId: string;
parentRunId: string;
selectedAgent: string;
routingDecision: string;
sourceManifestDigest?: string;
requiredRuntimeCapabilities?: string[];
cumulativeBudget?: AgentBudgetUsage;
previousSequence?: number;
fallbackUsed?: boolean;
maxRetries: number;
fallbackOnFailure: boolean;
fallbackAgent?: string;
fallbackEligible?: boolean;
fallbackReason?: string;
baseBackoffMs?: number;
now?: Date;
}
export class RunRecoveryPolicyService {
constructor(private readonly random: () => number = Math.random) {}
classifyCompletion(result: CompletionResult): RunFailureClassification {
return this.classify({
status: result.status,
terminalSource: result.terminalSource,
summary: result.summary,
error: result.error,
blockers: result.blockers,
sideEffects: result.sideEffects,
});
}
classifyError(error: unknown): RunFailureClassification {
const summary = error instanceof Error ? error.message : String(error || 'Unknown error');
return this.classify({ status: 'failed', summary, error: summary });
}
classify(evidence: RunFailureEvidence): RunFailureClassification {
const blockers = evidence.blockers ?? [];
const sideEffects = evidence.sideEffects ?? [];
const summary = boundedSummary(
evidence.error || evidence.summary || `Run reported ${evidence.status}.`
);
const text = [summary, ...blockers.flatMap((blocker) => [blocker.code, blocker.detail])].join(
' '
);
const destructiveSideEffects = sideEffects.some((effect) =>
[
'filesystem-write',
'process-execute',
'network-egress',
'git-commit',
'external-write',
'task-mutate',
].includes(effect.kind)
);
if (evidence.status === 'interrupted' || evidence.terminalSource === 'operator-interruption') {
return classification('cancellation', summary, false, false, false);
}
if (destructiveSideEffects) {
return classification('partial-side-effect', summary, false, true, true);
}
if (
evidence.status === 'blocked' ||
POLICY_BLOCK_PATTERN.test(text) ||
blockers.some((blocker) => !blocker.retryable || POLICY_BLOCK_PATTERN.test(blocker.code))
) {
return classification('policy-block', summary, false, true, false);
}
if (INVALID_REQUEST_PATTERN.test(text)) {
return classification('invalid-request', summary, false, true, false);
}
if (RATE_LIMIT_PATTERN.test(text)) {
return classification('rate-limit', summary, true, false, false);
}
if (TIMEOUT_PATTERN.test(text)) {
return classification('timeout', summary, true, false, false);
}
if (PROVIDER_UNAVAILABLE_PATTERN.test(text)) {
return classification('provider-unavailable', summary, true, false, false);
}
if (TRANSIENT_TRANSPORT_PATTERN.test(text)) {
return classification('transient-transport', summary, true, false, false);
}
if (
evidence.status === 'partial' ||
VERIFICATION_PATTERN.test(text) ||
(blockers.length > 0 && blockers.every((blocker) => blocker.retryable))
) {
return classification('verification-failure', summary, true, false, false);
}
if (evidence.status === 'failed') {
return classification('task-failure', summary, false, false, false);
}
return classification('unknown', summary, false, false, false);
}
decide(failure: RunFailureClassification, input: RunRecoveryDecisionInput): RunRecoveryRecord {
const previousSequence = Math.max(0, Math.trunc(input.previousSequence ?? 0));
const maxRetries = clampInteger(input.maxRetries, 0, 3);
const fallbackUsed = input.fallbackUsed === true;
const now = input.now ?? new Date();
const cumulativeBudget = input.cumulativeBudget ?? { ...ZERO_AGENT_BUDGET_USAGE };
const base = {
schemaVersion: RUN_RECOVERY_SCHEMA_VERSION,
rootRunId: input.rootRunId,
parentRunId: input.parentRunId,
sequence: previousSequence,
fallbackUsed,
failure,
selectedAgent: input.selectedAgent,
...(input.fallbackAgent ? { fallbackAgent: input.fallbackAgent } : {}),
routingDecision: input.routingDecision,
sourceManifestDigest: input.sourceManifestDigest,
requiredRuntimeCapabilities: [...new Set(input.requiredRuntimeCapabilities ?? [])].sort(),
cumulativeBudget: { ...cumulativeBudget },
} satisfies Omit<
RunRecoveryRecord,
'state' | 'action' | 'reason' | 'backoffMs' | 'scheduledAt' | 'notBefore' | 'handoff'
>;
if (failure.classification === 'cancellation') {
return {
...base,
state: 'cancelled',
action: 'cancelled',
reason: 'The operator or provider explicitly cancelled the run.',
backoffMs: 0,
cancelledAt: now.toISOString(),
handoff: {
summary: 'Automatic recovery was cancelled.',
nextActions: ['Start a new run explicitly if the objective should continue.'],
},
};
}
if (failure.approvalRequired) {
return {
...base,
state: 'approval-required',
action: 'approval',
reason: `Automatic recovery is unsafe for ${failure.classification}.`,
backoffMs: 0,
handoff: {
summary: `Recovery requires operator review after ${failure.classification}.`,
nextActions: [
'Inspect the terminal evidence and side effects.',
'Resolve the policy or configuration blocker.',
'Launch a new attempt explicitly after approval.',
],
},
};
}
if (!failure.retryable) {
return {
...base,
state: 'exhausted',
action: 'terminal',
reason: `Failure class ${failure.classification} is not explicitly retryable.`,
backoffMs: 0,
handoff: {
summary: `Automatic recovery stopped after ${failure.classification}.`,
nextActions: [
'Inspect the attempt log and normalized completion evidence.',
'Correct the task or provider inputs before launching another attempt.',
],
},
};
}
if (previousSequence < maxRetries) {
return this.scheduledRecord(
base,
'retry',
input.selectedAgent,
`Retry ${previousSequence + 1} of ${maxRetries} after ${failure.classification}.`,
now,
input.baseBackoffMs
);
}
if (
input.fallbackOnFailure &&
!fallbackUsed &&
input.fallbackAgent &&
input.fallbackEligible !== false
) {
return this.scheduledRecord(
{
...base,
fallbackUsed: true,
},
'fallback',
input.fallbackAgent,
input.fallbackReason ||
`Retry budget exhausted; route from ${input.selectedAgent} to ${input.fallbackAgent}.`,
now,
input.baseBackoffMs,
input.fallbackAgent
);
}
const fallbackDetail =
input.fallbackOnFailure && input.fallbackAgent && input.fallbackEligible === false
? ` Fallback ${input.fallbackAgent} was rejected: ${input.fallbackReason || 'incompatible runtime policy'}.`
: '';
return {
...base,
state: 'exhausted',
action: 'terminal',
reason: `Retry budget of ${maxRetries} exhausted.${fallbackDetail}`,
backoffMs: 0,
...(input.fallbackAgent ? { fallbackAgent: input.fallbackAgent } : {}),
handoff: {
summary: 'Automatic retry and fallback policy was exhausted.',
nextActions: [
'Review the failure classification and prior attempt chain.',
fallbackDetail
? 'Choose a fallback that satisfies runtime capabilities and sandbox policy.'
: 'Correct the provider or task failure before relaunching.',
],
},
};
}
private scheduledRecord(
base: Omit<
RunRecoveryRecord,
'state' | 'action' | 'reason' | 'backoffMs' | 'scheduledAt' | 'notBefore' | 'handoff'
>,
action: 'retry' | 'fallback',
selectedAgent: string,
reason: string,
now: Date,
requestedBaseBackoffMs?: number,
fallbackAgent?: string
): RunRecoveryRecord {
const sequence = base.sequence + 1;
const backoffMs = this.backoffMs(sequence, requestedBaseBackoffMs);
return {
...base,
sequence,
state: 'scheduled',
action,
reason,
backoffMs,
scheduledAt: now.toISOString(),
notBefore: new Date(now.getTime() + backoffMs).toISOString(),
selectedAgent,
...(fallbackAgent ? { fallbackAgent } : {}),
};
}
backoffMs(sequence: number, requestedBaseBackoffMs = DEFAULT_BACKOFF_MS): number {
const boundedSequence = clampInteger(sequence, 1, 32);
const base = clampInteger(requestedBaseBackoffMs, 100, MAX_BACKOFF_MS);
const exponential = Math.min(MAX_BACKOFF_MS, base * 2 ** (boundedSequence - 1));
const random = Math.max(0, Math.min(1, this.random()));
const jitterMultiplier = 1 - JITTER_RATIO + random * JITTER_RATIO * 2;
return Math.max(100, Math.min(MAX_BACKOFF_MS, Math.round(exponential * jitterMultiplier)));
}
}
function classification(
classificationValue: RunFailureClassification['classification'],
summary: string,
retryable: boolean,
approvalRequired: boolean,
destructiveSideEffects: boolean
): RunFailureClassification {
return {
classification: classificationValue,
summary,
retryable,
approvalRequired,
destructiveSideEffects,
};
}
function boundedSummary(value: string): string {
const trimmed = value.trim();
return (trimmed || 'Run failed without a diagnostic summary.').slice(0, 2_000);
}
function clampInteger(value: number, minimum: number, maximum: number): number {
if (!Number.isFinite(value)) return minimum;
return Math.max(minimum, Math.min(maximum, Math.trunc(value)));
}

View file

@ -8,11 +8,14 @@ import path from 'path';
import { nanoid } from 'nanoid';
import {
buildWorkflowPipelineSummary,
DEFAULT_ROUTING_CONFIG,
ZERO_AGENT_BUDGET_USAGE,
type AgentBudgetDecision,
type AgentBudgetPolicy,
type AgentBudgetThresholdEvent,
type AgentBudgetUsage,
type AgentType,
type RunRecoveryRecord,
type WorkflowPipelineRoleStatusPatch,
type WorkflowSubagentRunStatus,
type WorkflowSubagentTelemetry,
@ -29,7 +32,11 @@ import { SqliteWorkflowRunRepository } from '../storage/sqlite/workflow-reposito
import { getConfigService } from './config-service.js';
import { getAgentBudgetService } from './agent-budget-service.js';
import { getGovernanceTraceService } from './governance-trace-service.js';
import { NotFoundError, ValidationError } from '../middleware/error-handler.js';
import { ConflictError, NotFoundError, ValidationError } from '../middleware/error-handler.js';
import { RunRecoveryPolicyService } from './run-recovery-policy-service.js';
import { getAgentRoutingService } from './agent-routing-service.js';
import { atomicWriteFile } from '../storage/fs-helpers.js';
import { withFileLock } from './file-lock.js';
const log = createLogger('workflow-run');
@ -38,6 +45,10 @@ const MAX_CONCURRENT_RUNS = 10;
/** Default maximum cross-step reroutes per run before exhaustion policy fires (#780) */
const MAX_REROUTES_DEFAULT = 10;
let activeRunCount = 0;
const scheduledWorkflowRecoveries = new Map<
string,
{ stepId: string; timer: ReturnType<typeof setTimeout> }
>();
const RUN_ID_PATTERN = /^run_\d{10,}_[a-zA-Z0-9_-]{6,}$/;
const RESERVED_CONTEXT_KEYS = new Set([
'task',
@ -57,11 +68,13 @@ export class WorkflowRunService {
private readonly repository: SqliteWorkflowRunRepository | null = null;
private readonly sqliteDatabase: SqliteDatabase | null = null;
private readonly ownsSqliteDatabase: boolean = false;
private readonly runRecoveryPolicy: RunRecoveryPolicyService;
constructor(options: string | WorkflowRunServiceOptions = {}) {
const resolvedOptions = typeof options === 'string' ? { runsDir: options } : options;
this.runsDir = resolvedOptions.runsDir || getWorkflowRunsDir();
this.workflowService = resolvedOptions.workflowService ?? getWorkflowService();
this.runRecoveryPolicy = resolvedOptions.runRecoveryPolicy ?? new RunRecoveryPolicyService();
this.stepExecutor = new WorkflowStepExecutor(resolvedOptions.runsDir, {
persistRun: (run) => this.saveRun(run),
});
@ -453,8 +466,18 @@ export class WorkflowRunService {
broadcastWorkflowStatus(run);
const stepRun = existingStepRun;
stepRun.agent ??= step.agent;
stepRun.status = 'running';
stepRun.startedAt = new Date().toISOString();
if (stepRun.runRetry?.state === 'launching') {
stepRun.runRetry = {
...stepRun.runRetry,
state: 'launched',
launchedAt: stepRun.startedAt,
launchedRunId: `${run.id}:${step.id}:${stepRun.runRetry.sequence}`,
selectedAgent: stepRun.agent ?? stepRun.runRetry.selectedAgent,
};
}
this.syncPipelineSummary(run, workflow);
await this.saveRun(run);
@ -525,12 +548,32 @@ export class WorkflowRunService {
broadcastWorkflowStatus(run);
// Handle failure policy
const handled = await this.handleStepFailure(step, stepRun, stepQueue, workflow, run);
const handled = await this.handleStepFailure(
step,
stepRun,
stepQueue,
workflow,
run,
err
);
if (!handled) {
// No retry policy — fail the entire workflow
throw err;
}
if (stepRun.runRetry?.state === 'scheduled') {
log.info(
{
runId: run.id,
stepId: step.id,
action: stepRun.runRetry.action,
notBefore: stepRun.runRetry.notBefore,
},
'Workflow recovery scheduled'
);
return;
}
if ((run.status as WorkflowRun['status']) === 'blocked') {
log.info({ runId: run.id, stepId: step.id }, 'Workflow run blocked — awaiting resume');
return;
@ -576,37 +619,120 @@ export class WorkflowRunService {
stepRun: StepRun,
stepQueue: string[],
workflow: WorkflowDefinition,
run: WorkflowRun
run: WorkflowRun,
error: unknown
): Promise<boolean> {
const policy = step.on_fail;
if (!policy) return false;
const config = await getConfigService().getConfig();
const routingPolicy = config.agentRouting ?? DEFAULT_ROUTING_CONFIG;
const failure = this.runRecoveryPolicy.classifyError(error);
const selectedAgent = stepRun.agent ?? step.agent ?? step.id;
const previousSequence = stepRun.runRetry?.sequence ?? stepRun.retries;
const maxRetries = policy?.retry ?? (policy?.retry_step ? 0 : routingPolicy.maxRetries);
const explicitFallback = policy?.escalate_to?.startsWith('agent:')
? policy.escalate_to.slice('agent:'.length)
: undefined;
const fallbackOnFailure = Boolean(
explicitFallback || (!policy?.retry_step && routingPolicy.fallbackOnFailure)
);
let fallbackAgent: string | undefined = explicitFallback;
let fallbackEligible: boolean | undefined;
let fallbackReason: string | undefined;
// Strategy 1: Retry the same step
if (policy.retry && stepRun.retries < policy.retry) {
stepRun.retries++;
stepRun.status = 'pending';
stepRun.error = undefined;
// Phase 2: Apply retry delay if specified (#113)
if (policy.retry_delay_ms && policy.retry_delay_ms > 0) {
log.info(
{ stepId: step.id, retry: stepRun.retries, delayMs: policy.retry_delay_ms },
'Delaying retry'
);
await new Promise((resolve) => setTimeout(resolve, policy.retry_delay_ms));
if (
failure.retryable &&
previousSequence >= maxRetries &&
!stepRun.runRetry?.fallbackUsed &&
fallbackOnFailure
) {
if (!fallbackAgent && run.taskId && step.agent) {
const task = await getTaskService().getTask(run.taskId);
if (task) {
const fallback = await getAgentRoutingService().getFallback(
task,
step.agent as AgentType
);
if (fallback && workflow.agents.some((agent) => agent.id === fallback.agent)) {
fallbackAgent = fallback.agent;
fallbackReason = fallback.reason;
}
}
}
if (fallbackAgent) {
try {
await this.stepExecutor.validateFallbackAgent(step, run, fallbackAgent);
fallbackEligible = true;
} catch (fallbackError) {
fallbackEligible = false;
fallbackReason =
fallbackError instanceof Error ? fallbackError.message : String(fallbackError);
}
}
}
// Re-queue this step at the front
stepQueue.unshift(step.id);
const decision = this.runRecoveryPolicy.decide(failure, {
rootRunId: stepRun.runRetry?.rootRunId ?? `${run.id}:${step.id}:0`,
parentRunId: `${run.id}:${step.id}:${previousSequence}`,
selectedAgent,
routingDecision: explicitFallback
? `Workflow failure policy selected fallback ${explicitFallback}.`
: `Workflow step ${step.id} uses workspace recovery policy.`,
...(stepRun.providerRuntimeManifest?.digest
? { sourceManifestDigest: stepRun.providerRuntimeManifest.digest }
: {}),
requiredRuntimeCapabilities: [...(stepRun.requiredRuntimeCapabilities ?? [])],
cumulativeBudget: run.budget?.usage ?? { ...ZERO_AGENT_BUDGET_USAGE },
previousSequence,
fallbackUsed: stepRun.runRetry?.fallbackUsed,
maxRetries,
fallbackOnFailure,
...(fallbackAgent ? { fallbackAgent } : {}),
...(fallbackEligible !== undefined ? { fallbackEligible } : {}),
...(fallbackReason ? { fallbackReason } : {}),
...(policy?.retry_delay_ms !== undefined
? { baseBackoffMs: Math.max(100, policy.retry_delay_ms) }
: {}),
});
stepRun.runRetry = decision;
if (decision.state === 'scheduled') {
stepRun.retries = decision.sequence;
stepRun.status = 'failed';
stepRun.error = decision.reason;
if (decision.action === 'fallback') {
stepRun.agent = decision.selectedAgent;
}
run.status = 'pending';
run.error = decision.reason;
this.syncPipelineSummary(run, workflow);
await this.saveRun(run);
log.info({ stepId: step.id, retry: stepRun.retries }, 'Retrying step');
this.scheduleWorkflowRecovery(run.id, step.id, decision);
return true;
}
// Strategy 2: Retry a different step
if (policy.retry_step) {
if (decision.state === 'approval-required') {
if (policy?.escalate_to === 'skip') {
stepRun.status = 'skipped';
this.syncPipelineSummary(run, workflow);
await this.saveRun(run);
return true;
}
run.status = 'blocked';
run.error =
policy?.escalate_to === 'human'
? policy.escalate_message || `Step ${step.id} failed`
: decision.handoff?.summary || decision.reason;
this.syncPipelineSummary(run, workflow);
await this.saveRun(run);
return true;
}
// Strategy 2: Retry a different step after automatic recovery is exhausted.
if (
policy?.retry_step &&
!failure.approvalRequired &&
failure.classification !== 'cancellation'
) {
const retryStep = workflow.steps.find((s) => s.id === policy.retry_step);
if (!retryStep) {
throw new Error(`retry_step references unknown step: ${policy.retry_step}`);
@ -686,7 +812,7 @@ export class WorkflowRunService {
}
// Strategy 3: Escalation
if (policy.escalate_to === 'human') {
if (policy?.escalate_to === 'human') {
run.status = 'blocked';
run.error = policy.escalate_message || `Step ${step.id} failed`;
this.syncPipelineSummary(run, workflow);
@ -696,7 +822,7 @@ export class WorkflowRunService {
return true; // Handled (blocked, not failed)
}
if (policy.escalate_to === 'skip') {
if (policy?.escalate_to === 'skip') {
stepRun.status = 'skipped';
this.syncPipelineSummary(run, workflow);
await this.saveRun(run);
@ -704,14 +830,201 @@ export class WorkflowRunService {
return true;
}
if (policy.escalate_to?.startsWith('agent:')) {
// Delegate to another agent (future feature)
throw new Error('Agent escalation not yet implemented');
if (explicitFallback) {
throw new Error(decision.reason);
}
return false; // No policy matched — fail the workflow
}
async reconcilePendingRecoveries(): Promise<void> {
const runs = (await this.listRuns()).filter(
(run) => run.status === 'pending' || run.status === 'running'
);
let scheduledCount = 0;
for (const run of runs) {
const stepRun = run.steps.find((step) => step.stepId === run.currentStep);
const recovery = stepRun?.runRetry;
if (!stepRun || !recovery) {
continue;
}
if (recovery.state === 'launched' && run.status === 'running') {
stepRun.runRetry = {
...recovery,
state: 'approval-required',
action: 'approval',
reason:
'The server restarted after recovery launch and cannot prove the provider terminal state.',
backoffMs: 0,
handoff: {
summary: 'Workflow recovery requires operator reconciliation after restart.',
nextActions: [
'Inspect the provider session and persisted step output.',
'Confirm no provider work remains before resuming or replacing the run.',
],
},
};
run.status = 'blocked';
run.error = stepRun.runRetry.handoff?.summary ?? stepRun.runRetry.reason;
await this.saveRun(run);
broadcastWorkflowStatus(run);
continue;
}
if (!['scheduled', 'launching'].includes(recovery.state)) continue;
const reconciledRecovery: RunRecoveryRecord =
recovery.state === 'launching'
? {
...recovery,
state: 'scheduled',
notBefore: new Date().toISOString(),
reason: `${recovery.reason} Re-queued after server restart before provider launch.`,
}
: recovery;
if (recovery.state === 'launching') {
stepRun.runRetry = reconciledRecovery;
stepRun.status = 'failed';
stepRun.error = reconciledRecovery.reason;
run.status = 'pending';
run.error = reconciledRecovery.reason;
await this.saveRun(run);
}
this.scheduleWorkflowRecovery(run.id, stepRun.stepId, reconciledRecovery);
scheduledCount += 1;
}
if (scheduledCount > 0) {
log.info({ scheduledCount }, 'Workflow retry/fallback reconciliation complete');
}
}
async cancelPendingRecovery(
runId: string,
stepId: string,
parentRunId: string,
actor: string
): Promise<WorkflowRun> {
const run = await this.getRun(runId);
if (!run) throw new NotFoundError(`Run ${runId} not found`);
const stepRun = run.steps.find((step) => step.stepId === stepId);
const recovery = stepRun?.runRetry;
if (!stepRun || !recovery || recovery.parentRunId !== parentRunId) {
throw new ConflictError('Recovery cancellation does not match the pending workflow step');
}
if (!['scheduled', 'launching'].includes(recovery.state)) {
throw new ConflictError(`Workflow recovery is not pending (state: ${recovery.state})`);
}
const cancelled: RunRecoveryRecord = {
...recovery,
state: 'cancelled',
action: 'cancelled',
reason: 'Automatic workflow recovery was cancelled by an operator.',
backoffMs: 0,
cancelledAt: new Date().toISOString(),
cancelledBy: actor,
handoff: {
summary: 'Automatic workflow recovery was cancelled.',
nextActions: ['Resume or restart the workflow explicitly if it should continue.'],
},
};
stepRun.runRetry = cancelled;
run.status = 'blocked';
run.error = cancelled.handoff?.summary ?? cancelled.reason;
await this.saveRun(run);
this.clearScheduledWorkflowRecovery(runId, stepId);
broadcastWorkflowStatus(run);
return run;
}
private scheduleWorkflowRecovery(
runId: string,
stepId: string,
recovery: RunRecoveryRecord
): void {
if (recovery.state !== 'scheduled') return;
this.clearScheduledWorkflowRecovery(runId);
const notBefore = recovery.notBefore ? Date.parse(recovery.notBefore) : Date.now();
const delay = Math.max(0, Math.min(2_147_483_647, notBefore - Date.now()));
const timer = setTimeout(() => {
const scheduled = scheduledWorkflowRecoveries.get(runId);
if (!scheduled || scheduled.stepId !== stepId) return;
scheduledWorkflowRecoveries.delete(runId);
void this.resumeScheduledWorkflowRecovery(runId, stepId).catch((error) => {
log.error({ err: error, runId, stepId }, 'Scheduled workflow recovery failed');
});
}, delay);
timer.unref?.();
scheduledWorkflowRecoveries.set(runId, { stepId, timer });
}
private clearScheduledWorkflowRecovery(runId: string, expectedStepId?: string): void {
const scheduled = scheduledWorkflowRecoveries.get(runId);
if (!scheduled || (expectedStepId && scheduled.stepId !== expectedStepId)) return;
clearTimeout(scheduled.timer);
scheduledWorkflowRecoveries.delete(runId);
}
private async resumeScheduledWorkflowRecovery(runId: string, stepId: string): Promise<void> {
let recoveryToReschedule: RunRecoveryRecord | undefined;
const run = await this.getRun(runId);
const stepRun = run?.steps.find((step) => step.stepId === stepId);
const recovery = stepRun?.runRetry;
if (!run || !stepRun || recovery?.state !== 'scheduled' || run.status !== 'pending') {
return;
}
if (recovery.notBefore && Date.parse(recovery.notBefore) > Date.now()) {
recoveryToReschedule = recovery;
}
if (recoveryToReschedule) {
this.scheduleWorkflowRecovery(runId, stepId, recoveryToReschedule);
return;
}
const launchedAt = new Date().toISOString();
stepRun.runRetry = {
...recovery,
state: 'launched',
launchedAt,
launchedRunId: `${run.id}:${stepId}:${recovery.sequence}`,
selectedAgent: stepRun.agent ?? recovery.selectedAgent,
};
stepRun.status = 'pending';
stepRun.error = undefined;
run.status = 'running';
run.error = undefined;
try {
await this.saveRun(run);
} catch (error) {
if (error instanceof ConflictError) {
log.info({ runId, stepId }, 'Workflow recovery claim lost to another process');
return;
}
throw error;
}
const workflow = await this.workflowService.loadWorkflow(run.workflowId);
if (!workflow) {
stepRun.runRetry = {
...stepRun.runRetry,
state: 'exhausted',
action: 'terminal',
reason: `Workflow ${run.workflowId} no longer exists.`,
backoffMs: 0,
handoff: {
summary: 'The persisted workflow recovery cannot be resumed.',
nextActions: ['Restore the workflow definition or start a replacement run.'],
},
};
run.status = 'failed';
run.error = stepRun.runRetry.reason;
run.completedAt = new Date().toISOString();
await this.saveRun(run);
broadcastWorkflowStatus(run);
return;
}
void this.executeRun(run, workflow).catch((error) => {
log.error({ err: error, runId, stepId }, 'Workflow recovery execution failed');
});
}
/**
* Get a workflow run by ID
*/
@ -1190,11 +1503,21 @@ export class WorkflowRunService {
* Phase 2: Updates lastCheckpoint timestamp on every save
*/
private async saveRun(run: WorkflowRun): Promise<void> {
// Update checkpoint timestamp
run.lastCheckpoint = new Date().toISOString();
const expectedRevision = run.revision ?? 0;
const nextRun: WorkflowRun = {
...run,
revision: expectedRevision + 1,
lastCheckpoint: new Date().toISOString(),
};
if (this.repository) {
this.repository.save(run);
if (!this.repository.save(nextRun, expectedRevision)) {
throw new ConflictError('Workflow run changed during persistence', {
runId: run.id,
expectedRevision,
});
}
Object.assign(run, nextRun);
return;
}
@ -1202,7 +1525,29 @@ export class WorkflowRunService {
await fs.mkdir(runDir, { recursive: true });
const runPath = path.join(runDir, 'run.json');
await fs.writeFile(runPath, JSON.stringify(run, null, 2), 'utf-8');
await withFileLock(runPath, async () => {
let current: WorkflowRun | null = null;
try {
current = JSON.parse(await fs.readFile(runPath, 'utf-8')) as WorkflowRun;
} catch (error) {
if (!error || typeof error !== 'object' || !('code' in error) || error.code !== 'ENOENT') {
throw error;
}
}
const currentRevision = current?.revision ?? 0;
if (
(current && currentRevision !== expectedRevision) ||
(!current && expectedRevision !== 0)
) {
throw new ConflictError('Workflow run changed during persistence', {
runId: run.id,
expectedRevision,
currentRevision: current?.revision,
});
}
await atomicWriteFile(runPath, JSON.stringify(nextRun, null, 2));
});
Object.assign(run, nextRun);
}
/**
@ -1235,6 +1580,7 @@ export interface WorkflowRunServiceOptions {
sqliteDatabase?: SqliteDatabase;
sqliteConnectionOptions?: SqliteConnectionOptions;
workflowService?: ReturnType<typeof getWorkflowService>;
runRecoveryPolicy?: RunRecoveryPolicyService;
}
// Singleton

View file

@ -101,21 +101,77 @@ export class WorkflowStepExecutor {
*/
async executeStep(step: WorkflowStep, run: WorkflowRun): Promise<StepExecutionResult> {
log.info({ runId: run.id, stepId: step.id, type: step.type }, 'Executing step');
const selectedAgent = run.steps.find((candidate) => candidate.stepId === step.id)?.agent;
const effectiveStep = selectedAgent ? { ...step, agent: selectedAgent } : step;
switch (step.type) {
switch (effectiveStep.type) {
case 'agent':
return this.executeAgentStep(step, run);
return this.executeAgentStep(effectiveStep, run);
case 'loop':
return this.executeLoopStep(step, run);
return this.executeLoopStep(effectiveStep, run);
case 'gate':
return this.executeGateStep(step, run);
return this.executeGateStep(effectiveStep, run);
case 'parallel':
return this.executeParallelStep(step, run);
return this.executeParallelStep(effectiveStep, run);
default:
throw new Error(`Unknown step type: ${step.type}`);
throw new Error(`Unknown step type: ${effectiveStep.type}`);
}
}
/**
* Probe a fallback through the same capability and sandbox gates used by a
* real workflow launch, without creating a provider session.
*/
async validateFallbackAgent(
step: WorkflowStep,
run: WorkflowRun,
agentId: string
): Promise<ProviderRuntimeManifest> {
const effectiveStep = { ...step, agent: agentId };
let agentDef = this.getAgentDefinition(run, agentId);
if (!agentDef) {
throw new Error(`Workflow fallback agent ${agentId} is not defined in the workflow`);
}
if (run.budget?.modelOverride) {
agentDef = { ...agentDef, model: run.budget.modelOverride };
}
const workflowConfig = run.context.workflow as
{ config?: { fresh_session_default?: boolean } } | undefined;
const sessionConfig = this.buildSessionConfig(effectiveStep, run, workflowConfig?.config);
const toolPolicyFilter = await this.getToolPolicyForAgent(agentDef);
const runtimeProvider = this.resolveWorkflowProvider(effectiveStep, agentDef);
const runtimeManifest = await this.resolveRuntimeManifest(
effectiveStep,
agentDef,
runtimeProvider
);
const requiredRuntimeCapabilities = this.requiredRuntimeCapabilities(
effectiveStep,
run,
agentDef,
runtimeProvider,
sessionConfig,
toolPolicyFilter
);
assertProviderRuntimeCapabilities(
runtimeManifest,
requiredRuntimeCapabilities,
`workflow fallback ${agentId} for step ${step.id}`
);
const sandboxPolicy = await getSandboxPolicyService().dryRunWithTrace({
presetId: agentDef.sandboxPresetId,
provider: runtimeManifest.provider,
workspacePath: this.expandPath(this.getWorkflowWorkingDirectory(run)),
providerRuntimeManifest: runtimeManifest,
});
if (sandboxPolicy.result.decision === 'block') {
throw new Error(
`Workflow fallback ${agentId} cannot enforce sandbox preset ${sandboxPolicy.result.preset.id}`
);
}
return runtimeManifest;
}
/**
* Execute an agent step through its configured provider.
* Integrated features: #108 (progress), #110 (tool policies), #111 (session management)
@ -165,7 +221,7 @@ export class WorkflowStepExecutor {
requiredRuntimeCapabilities,
`workflow step ${step.id} launch`
);
await this.recordRuntimeManifest(run, step, runtimeManifest);
await this.recordRuntimeManifest(run, step, runtimeManifest, requiredRuntimeCapabilities);
const sandboxPolicy = await getSandboxPolicyService().dryRunWithTrace({
presetId: agentDef?.sandboxPresetId,
provider: runtimeManifest.provider,
@ -487,12 +543,23 @@ export class WorkflowStepExecutor {
private async recordRuntimeManifest(
run: WorkflowRun,
step: WorkflowStep,
manifest: ProviderRuntimeManifest
manifest: ProviderRuntimeManifest,
requiredRuntimeCapabilities: ProviderRuntimeCapabilityId[]
): Promise<void> {
const stepRun = run.steps.find((candidate) => candidate.stepId === step.id);
if (stepRun) {
stepRun.providerRuntimeManifest = manifest;
stepRun.requiredRuntimeCapabilities = [...new Set(requiredRuntimeCapabilities)].sort(
(left, right) => left.localeCompare(right)
);
stepRun.runtimeControls = providerRuntimeControls(manifest);
if (stepRun.runRetry?.state === 'launched') {
stepRun.runRetry = {
...stepRun.runRetry,
launchedManifestDigest: manifest.digest,
requiredRuntimeCapabilities: [...stepRun.requiredRuntimeCapabilities],
};
}
}
await this.persistRun(run);
}

View file

@ -303,41 +303,80 @@ export class SqliteWorkflowRunRepository {
}));
}
save(run: WorkflowRun): void {
this.database
.getConnection()
save(run: WorkflowRun, expectedRevision = 0): boolean {
const connection = this.database.getConnection();
const row = connection
.prepare(
`
INSERT INTO workflow_runs (
id,
workspace_id,
workflow_id,
workflow_version,
task_id,
status,
current_step,
run_json,
started_at,
completed_at,
last_checkpoint,
error
)
VALUES (?, 'local', ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(id) DO UPDATE SET
workflow_id = excluded.workflow_id,
workflow_version = excluded.workflow_version,
task_id = excluded.task_id,
status = excluded.status,
current_step = excluded.current_step,
run_json = excluded.run_json,
started_at = excluded.started_at,
completed_at = excluded.completed_at,
last_checkpoint = excluded.last_checkpoint,
error = excluded.error
SELECT run_json
FROM workflow_runs
WHERE workspace_id = 'local'
AND id = ?
`
)
.get(run.id) as WorkflowRunRow | undefined;
if (!row) {
if (expectedRevision !== 0) return false;
const inserted = connection
.prepare(
`
INSERT OR IGNORE INTO workflow_runs (
id,
workspace_id,
workflow_id,
workflow_version,
task_id,
status,
current_step,
run_json,
started_at,
completed_at,
last_checkpoint,
error
)
VALUES (?, 'local', ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
`
)
.run(
run.id,
run.workflowId,
run.workflowVersion,
run.taskId ?? null,
run.status,
run.currentStep ?? null,
JSON.stringify(run),
run.startedAt,
run.completedAt ?? null,
run.lastCheckpoint ?? null,
run.error ?? null
);
return inserted.changes === 1;
}
const current = JSON.parse(row.run_json) as WorkflowRun;
if ((current.revision ?? 0) !== expectedRevision) return false;
const updated = connection
.prepare(
`
UPDATE workflow_runs
SET workflow_id = ?,
workflow_version = ?,
task_id = ?,
status = ?,
current_step = ?,
run_json = ?,
started_at = ?,
completed_at = ?,
last_checkpoint = ?,
error = ?
WHERE workspace_id = 'local'
AND id = ?
AND run_json = ?
`
)
.run(
run.id,
run.workflowId,
run.workflowVersion,
run.taskId ?? null,
@ -347,8 +386,11 @@ export class SqliteWorkflowRunRepository {
run.startedAt,
run.completedAt ?? null,
run.lastCheckpoint ?? null,
run.error ?? null
run.error ?? null,
run.id,
row.run_json
);
return updated.changes === 1;
}
saveWorkflowSnapshot(runId: string, workflow: WorkflowDefinition): void {

View file

@ -197,6 +197,8 @@ export type StepRunStatus = 'pending' | 'running' | 'completed' | 'failed' | 'sk
export interface WorkflowRun {
id: string; // run_<timestamp>_<nanoid>
/** Monotonic persistence revision used for compare-and-set workflow mutations. */
revision?: number;
workflowId: string;
workflowVersion: number;
taskId?: string; // Optional task association
@ -225,7 +227,11 @@ export interface StepRun {
output?: string; // Path to output file
error?: string;
providerRuntimeManifest?: import('@veritas-kanban/shared').ProviderRuntimeManifest;
/** Exact runtime capabilities required by the recorded provider launch. */
requiredRuntimeCapabilities?: import('@veritas-kanban/shared').ProviderRuntimeCapabilityId[];
runtimeControls?: import('@veritas-kanban/shared').ProviderRuntimeControlSet;
/** Durable retry/fallback decision for this workflow step. */
runRetry?: import('@veritas-kanban/shared').RunRecoveryRecord;
// Loop-specific state
loopState?: {

View file

@ -55,6 +55,7 @@ export * from './time-breakdown.types.js';
export * from './task-envelope.types.js';
export * from './run-launch-manifest.types.js';
export * from './run-event.types.js';
export * from './run-recovery.types.js';
export * from './runtime-hook.types.js';
export * from './auth.types.js';
export * from './credential-broker.types.js';

View file

@ -68,6 +68,10 @@ export interface RunLaunchRouting {
reason: string;
fallbackAgent: string | null;
fallbackAllowed: boolean;
/** Captured recovery policy. Optional only for legacy v1 manifests. */
fallbackOnFailure?: boolean;
/** Captured retry bound. Optional only for legacy v1 manifests. */
maxRetries?: number;
}
export interface RunLaunchProfileReference {

View file

@ -0,0 +1,86 @@
import type { AgentBudgetUsage } from './agent-budget.types.js';
export const RUN_RECOVERY_SCHEMA_VERSION = 'run-recovery/v1' as const;
export const RUN_FAILURE_CLASSES = [
'transient-transport',
'provider-unavailable',
'rate-limit',
'invalid-request',
'task-failure',
'verification-failure',
'policy-block',
'cancellation',
'timeout',
'partial-side-effect',
'unknown',
] as const;
export type RunFailureClass = (typeof RUN_FAILURE_CLASSES)[number];
export const RUN_RECOVERY_ACTIONS = [
'retry',
'fallback',
'approval',
'terminal',
'cancelled',
] as const;
export type RunRecoveryAction = (typeof RUN_RECOVERY_ACTIONS)[number];
export const RUN_RECOVERY_STATES = [
'scheduled',
'launching',
'launched',
'approval-required',
'exhausted',
'cancelled',
] as const;
export type RunRecoveryState = (typeof RUN_RECOVERY_STATES)[number];
export interface RunFailureClassification {
classification: RunFailureClass;
summary: string;
retryable: boolean;
approvalRequired: boolean;
destructiveSideEffects: boolean;
}
export interface RunRecoveryHandoff {
summary: string;
nextActions: string[];
}
/**
* Durable causal record for a retry or fallback decision.
*
* Task attempts and workflow steps both persist this shape so recovery can be
* reconciled after restart without relying on an in-memory timer.
*/
export interface RunRecoveryRecord {
schemaVersion: typeof RUN_RECOVERY_SCHEMA_VERSION;
rootRunId: string;
parentRunId: string;
sequence: number;
fallbackUsed: boolean;
state: RunRecoveryState;
action: RunRecoveryAction;
failure: RunFailureClassification;
reason: string;
backoffMs: number;
scheduledAt?: string;
notBefore?: string;
launchedAt?: string;
launchedRunId?: string;
cancelledAt?: string;
cancelledBy?: string;
selectedAgent: string;
fallbackAgent?: string;
routingDecision: string;
sourceManifestDigest?: string;
launchedManifestDigest?: string;
requiredRuntimeCapabilities: string[];
cumulativeBudget: AgentBudgetUsage;
handoff?: RunRecoveryHandoff;
}

View file

@ -81,6 +81,7 @@ export interface TaskAttempt {
runLaunchManifestTraceId?: string;
runLaunchParentAttemptId?: string;
runLaunchManifestDrift?: import('./run-launch-manifest.types.js').RunLaunchManifestDriftResult;
runRetry?: import('./run-recovery.types.js').RunRecoveryRecord;
completionResult?: import('./task-envelope.types.js').CompletionResult;
conversation?: import('./conversation-lifecycle.types.js').ConversationLifecycleRecord;
}

View file

@ -225,6 +225,8 @@ export type StepRunStatus = 'pending' | 'running' | 'completed' | 'failed' | 'sk
export interface WorkflowRun {
id: string; // run_<timestamp>_<nanoid>
/** Monotonic persistence revision used for compare-and-set workflow mutations. */
revision?: number;
workflowId: string;
workflowVersion: number;
taskId?: string; // Optional task association
@ -252,7 +254,11 @@ export interface StepRun {
error?: string;
/** Exact provider evidence snapshot used for this step's launch and controls. */
providerRuntimeManifest?: import('./provider-runtime.types.js').ProviderRuntimeManifest;
/** Exact runtime capabilities required by the recorded provider launch. */
requiredRuntimeCapabilities?: import('./provider-runtime.types.js').ProviderRuntimeCapabilityId[];
runtimeControls?: import('./provider-runtime.types.js').ProviderRuntimeControlSet;
/** Durable retry/fallback decision for this workflow step. */
runRetry?: import('./run-recovery.types.js').RunRecoveryRecord;
// Loop-specific state
loopState?: {