From a58496839dfd45878c2fffac186edf2c8b128c37 Mon Sep 17 00:00:00 2001 From: Brad Groux <3053586+BradGroux@users.noreply.github.com> Date: Fri, 10 Jul 2026 12:16:51 -0500 Subject: [PATCH] =?UTF-8?q?fix:=20workflow=20correctness=20=E2=80=94=20hum?= =?UTF-8?q?an=20gate=20blocking,=20retry=20bounds,=20HTTP=20errors,=20shar?= =?UTF-8?q?ed=20contracts,=20depends=5Fon=20enforcement=20(#805)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fixes #778, #780, #785, #786, #787 ## #778 — Human gate blocking/resume correctness - Introduce HumanGateBlockError in WorkflowStepExecutor; gate steps with on_false.escalate_to=human now throw this typed exception instead of a plain Error. - executeRun() catches HumanGateBlockError before handleStepFailure() so the run transitions to blocked (not failed); persists _gateBlock context. - Add approveGateStep() and rejectGateStep() service methods; fix route endpoints to persist state and validate run.status===blocked. ## #780 — Bounded retry_step cycles - Add max_reroutes field to FailurePolicy and retryRouteCount to WorkflowRun in both shared and server type contracts. - handleStepFailure increments and checks retryRouteCount on every retry_step reroute; defaults to MAX_REROUTES_DEFAULT=10; exhaustion fires on_exhausted policy or fails deterministically. - retryRouteCount persists to disk/SQLite; survives process restart. ## #785 — WorkflowRunService domain errors → HTTP mapping - Remove private NotFoundError and ValidationError from workflow-run-service.ts. - Import and throw the shared AppError-based NotFoundError/ValidationError from middleware/error-handler.ts so central error middleware maps them to 404/400. ## #786 — Shared workflow contracts - Add provider? and command? fields to WorkflowAgent in shared/src/types/workflow.ts to match the server-side definition and expose them to web, CLI, and MCP consumers. ## #787 — depends_on enforcement during status transitions - BlockingService refactored to merge both legacy blockedBy and canonical dependencies.depends_on (deduplication via Set) in getBlockingStatus(), canMoveToInProgress(), getDependentTasks(), and wouldCreateCircularDependency(). - Tasks route transition guard now triggers when either blockedBy or dependencies.depends_on is non-empty. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- CHANGELOG.md | 102 +-- .../issues-778-780-785-786-787.test.ts | 666 ++++++++++++++++++ .../sqlite-workflow-run-execution.test.ts | 14 +- .../__tests__/workflow-run-service.test.ts | 14 +- server/src/routes/tasks.ts | 6 +- server/src/routes/workflows.ts | 59 +- server/src/services/blocking-service.ts | 56 +- server/src/services/workflow-run-service.ts | 201 +++++- server/src/services/workflow-step-executor.ts | 27 +- server/src/types/workflow.ts | 17 +- shared/src/types/workflow.ts | 17 +- 11 files changed, 982 insertions(+), 197 deletions(-) create mode 100644 server/src/__tests__/issues-778-780-785-786-787.test.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index fe9624f9..83d7bbb5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,98 +7,36 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] -### Added - -- Added `AGENTS.md` as the canonical repository instruction file for Codex, OpenClaw, Hermes, - Claude, and other compatible coding agents (#790). Contains authoritative package manager - requirements (pnpm ≥ 11.0.0, Node ≥ 22.22.1), architecture rules, commands, security - boundaries, and provider notes. -- Added first-class `hermes-cli` provider support for Hermes Agent v2026.7.7.2 (#791). Provider - dispatches tasks using the one-shot scripted interface (`hermes -z `) in the task - worktree, captures stdout/stderr/exit code, records timing and session identity into telemetry, - and supports graceful stop with a 5-second SIGKILL fallback. -- Added `buildSafeHermesEnv` utility (`server/src/utils/hermes-env.ts`) with a Hermes-specific - environment allowlist and credential redaction consistent with the Codex env policy (#791). -- Added `hermes-cli` to the `AgentProvider` shared type and to the readiness/health probe in - `AgentHealthService` (#791). -- Added `HttpOpenClawTaskAdapter` in `openclaw-workflow-adapter.ts` that dispatches tasks to the - OpenClaw gateway via `sessions_spawn` and stores the returned `childSessionKey` in the attempt - record (#794). -- OpenClaw dispatch uses the real `sessions_spawn` acknowledgement as its reachability and policy - check, avoiding a speculative probe that could create an untracked session. Policy denial returns - an actionable `gateway.tools.allow` configuration hint (#794). -- Added `openclawSessionKey` and `hermesSessionId` fields to `PendingAgent` for durable session - identity tracking (#791, #794). -- Added contract and regression tests for the Hermes provider (`hermes-provider.test.ts`) and - OpenClaw gateway adapter (`openclaw-provider.test.ts`) with mocked delivery, policy denial, - request timeout, supported spawn payload, and successful dispatch scenarios (#791, #794). -- Added Hermes and OpenClaw operator setup sections to `docs/AGENT-PROVIDERS.md` including - required gateway tool policy, environment variables, invocation mode, and troubleshooting - tables (#790, #791, #794). - -### Changed - -- Converted `CLAUDE.md` from a duplicate of architectural rules into a Claude-specific supplement - that defers to `AGENTS.md` as the source of truth. Corrected the stale pnpm (9+ → ≥ 11.0.0) - and Node (22+ → ≥ 22.22.1) version requirements (#790). -- Replaced the legacy request-file dispatch path in the OpenClaw task provider with a gateway - HTTP call (`sessions_spawn`) that throws on policy denial or unreachability, allowing the - existing error handler to roll the attempt back to `todo` instead of leaving it stuck in - `running` (#794). -- OpenClaw task provider now records the gateway `childSessionKey` in the attempt for durable - session tracking (#794). - -### Fixed - -- OpenClaw task runs no longer remain indefinitely in `running` when the gateway is unreachable - or the tool policy blocks `sessions_spawn` (#794). - ### Changed - Updated the Codex SDK integration to `0.144.1` and pinned patched transitive development-tool dependencies (#792, #795). -- Debounced and made asynchronous the agent-registry heartbeat persistence - writes; writes are now coalesced over a 2 s window and use atomic - rename-on-write instead of synchronous full-file serialisation; `dispose()` - flushes any in-flight write on shutdown (#783). -- Replaced per-request `ConfigService` allocation in the - `POST /api/agent/delegation-violation` route handler with the application-level - singleton to prevent FSWatcher leaks under sustained traffic (#779). -- Routed all `.veritas-kanban` path construction in - `clawdbot-agent-service.ts` and `agent-status.ts` through the centralised - helpers in `server/src/utils/paths.ts` (DATA_DIR / VERITAS_DATA_DIR - overrides are now respected consistently in those files; #774). +- Added `provider` and `command` fields to `WorkflowAgent` in + `@veritas-kanban/shared` to align with the server-side definition and prevent + silent contract drift (#786). +- Added `max_reroutes` to `FailurePolicy` and `retryRouteCount` to `WorkflowRun` + in both the shared and server workflow type contracts (#780, #786). ### Fixed - Isolated QMD result normalization coverage from unrelated persistent search collections and restored test environment state only after temporary search roots are removed, eliminating the intermittent teardown race (#793). -- File-backed task mutations (create, update, archive, restore) now use an - atomic write-then-rename strategy so readers see either the previous file or - its complete replacement rather than a partial write (#776). -- Revision validation is now enforced inside the file-lock / SQLite mutation - — not just at the route layer — eliminating the TOCTOU window where two - concurrent requests with the same `expectedRevision` could both succeed and - the later write silently overwrite the earlier one (#777). -- Concurrent file-backed mutations for the same task now serialize through an - in-process queue keyed by immutable task ID before asynchronous lookup, while - filepath locks continue to provide cross-process exclusion (#777). -- Added startup reconciliation for agent attempts that were left in `running` - state after a server crash or restart; orphaned attempts are now marked - `failed` and their tasks reverted to `todo` so operators can relaunch - them (#781). - -### Performance - -- Paginated file-backed activity requests now derive page items and total count - from one file load and filter pass (#782). -- Activity file writes are now atomic (write-temp / rename), and corrupt files - are backed up before recovery rather than silently overwritten (#782). -- Task-identity diagnostics are cached after the first scan and invalidated - only on task mutations or external file changes, eliminating the O(N) - full-filesystem scan on every `GET /api/tasks` and backlog-list request - (#784). +- Human-gate blocking (`on_false.escalate_to: human`) now transitions the run + to `blocked` instead of `failed` via the new `HumanGateBlockError` typed + exception; `approveGateStep` and `rejectGateStep` service methods replace the + unimplemented route stubs that previously discarded state (#778). +- `retry_step` cross-step rerouts are now bounded by a persisted + `retryRouteCount` field; the default cap is 10 and is configurable via + `on_fail.max_reroutes`; exhaustion respects `on_exhausted` policy or fails + deterministically (#780). +- `WorkflowRunService` private `NotFoundError` and `ValidationError` classes + replaced with the shared `AppError` hierarchy so workflow domain errors map + to stable 404/400 HTTP responses instead of HTTP 500 (#785). +- `BlockingService` now enforces `dependencies.depends_on` (canonical + dependency model) in addition to legacy `blockedBy`; the tasks route + transition guard checks both fields so `depends_on` relationships block + in-progress transitions (#787). ## [5.2.1] - 2026-06-29 diff --git a/server/src/__tests__/issues-778-780-785-786-787.test.ts b/server/src/__tests__/issues-778-780-785-786-787.test.ts new file mode 100644 index 00000000..0ad920eb --- /dev/null +++ b/server/src/__tests__/issues-778-780-785-786-787.test.ts @@ -0,0 +1,666 @@ +/** + * Regression tests for audit issues #778, #780, #785, #786, #787. + * + * #778 — Human gate blocking/resume correctness + * #780 — Bounded retry_step cycles + * #785 — WorkflowRunService domain errors → HTTP AppError mapping + * #786 — Shared workflow contracts: provider/command fields + * #787 — depends_on enforcement during status transitions + */ + +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import fs from 'fs/promises'; +import os from 'os'; +import path from 'path'; +import { BlockingService } from '../services/blocking-service.js'; +import { HumanGateBlockError } from '../services/workflow-step-executor.js'; +import type { Task, WorkflowAgent } from '@veritas-kanban/shared'; +import type { WorkflowRunService } from '../services/workflow-run-service.js'; + +// ───────────────────────────────────────────────────────────── +// Module-level mock state shared across workflow service tests. +// Using module-level fns so vi.mock factory can close over them. +// ───────────────────────────────────────────────────────────── + +const mockLoadWorkflow = vi.fn(); +const mockExecuteStep = vi.fn(); +const mockBroadcastWorkflowStatus = vi.fn(); +const mockGetTask = vi.fn(); + +vi.mock('../services/workflow-service.js', () => ({ + getWorkflowService: () => ({ + loadWorkflow: mockLoadWorkflow, + listWorkflowsMetadata: vi.fn().mockResolvedValue([]), + auditChange: vi.fn(), + }), +})); + +vi.mock('../services/workflow-step-executor.js', async (importOriginal) => { + const actual = await importOriginal(); + return { + HumanGateBlockError: actual.HumanGateBlockError, + WorkflowStepExecutor: class { + executeStep = mockExecuteStep; + }, + }; +}); + +vi.mock('../services/broadcast-service.js', () => ({ + broadcastWorkflowStatus: mockBroadcastWorkflowStatus, +})); + +vi.mock('../services/task-service.js', () => ({ + getTaskService: () => ({ getTask: mockGetTask }), +})); + +vi.mock('../middleware/workflow-auth.js', () => ({ + checkWorkflowPermission: vi.fn().mockResolvedValue(true), +})); + +// ───────────────────────────────────────────────────────────── +// Helpers +// ───────────────────────────────────────────────────────────── + +function agentWorkflow(id: string, steps: unknown[]) { + return { + id, + version: 1, + name: id, + description: '', + variables: {}, + agents: [{ id: 'a1', name: 'A1', role: 'dev', description: '' }], + steps, + }; +} + +/** + * Default executeStep mock: + * - Gate steps with on_false.escalate_to=human → throw HumanGateBlockError + * - Other gate steps → throw plain Error + * - Agent steps → succeed + */ +function gateAwareImpl(step: { + id: string; + type?: string; + on_false?: { escalate_to: string; escalate_message?: string }; +}) { + if (step.type === 'gate') { + if (step.on_false?.escalate_to === 'human') { + return Promise.reject( + new HumanGateBlockError( + step.id, + step.on_false as { escalate_to: 'human'; escalate_message?: string } + ) + ); + } + return Promise.reject(new Error(`Gate ${step.id} condition failed`)); + } + return Promise.resolve({ output: { done: step.id }, outputPath: `/tmp/${step.id}.json` }); +} + +function makeTask(overrides: Partial = {}): Task { + return { + id: 'task_test_001', + title: 'Test Task', + status: 'todo', + priority: 'medium', + created: '2026-01-01T00:00:00Z', + updated: '2026-01-01T00:00:00Z', + ...overrides, + } as Task; +} + +// ───────────────────────────────────────────────────────────── +// #778 — Human gate blocking/resume correctness +// ───────────────────────────────────────────────────────────── + +describe('#778 — Human gate blocking', () => { + let tmpDir: string; + let service: WorkflowRunService; + + beforeEach(async () => { + vi.resetModules(); + tmpDir = await fs.mkdtemp(path.join(os.tmpdir(), 'wf-778-')); + mockGetTask.mockResolvedValue(null); + mockLoadWorkflow.mockResolvedValue( + agentWorkflow('wf-gate', [ + { id: 'prep', type: 'agent', agent: 'a1', name: 'Prep' }, + { + id: 'gate', + type: 'gate', + name: 'Human Gate', + condition: 'false', + on_false: { escalate_to: 'human', escalate_message: 'Awaiting human approval' }, + }, + { id: 'finish', type: 'agent', agent: 'a1', name: 'Finish' }, + ]) + ); + mockExecuteStep.mockImplementation(gateAwareImpl); + const mod = await import('../services/workflow-run-service.js'); + service = new mod.WorkflowRunService(tmpDir); + }); + + afterEach(async () => { + vi.clearAllMocks(); + await fs.rm(tmpDir, { recursive: true, force: true }); + }); + + it('transitions to blocked (not failed) when gate on_false escalates to human', async () => { + const run = await service.startRun('wf-gate'); + + await vi.waitFor(async () => { + const saved = await service.getRun(run.id); + expect(saved.status).toBe('blocked'); + }); + + const blocked = await service.getRun(run.id); + expect(blocked.status).toBe('blocked'); + expect(blocked.error).toBe('Awaiting human approval'); + + const gateStep = blocked.steps.find((s: { stepId: string }) => s.stepId === 'gate'); + expect(gateStep.status).toBe('failed'); + + const prepStep = blocked.steps.find((s: { stepId: string }) => s.stepId === 'prep'); + expect(prepStep.status).toBe('completed'); + + expect(blocked.context._gateBlock).toMatchObject({ + stepId: 'gate', + escalationMessage: 'Awaiting human approval', + }); + }); + + it('approveGateStep marks gate completed, clears block context, resumes from finish', async () => { + const run = await service.startRun('wf-gate'); + await vi.waitFor(async () => expect((await service.getRun(run.id)).status).toBe('blocked')); + + const resumed = await service.approveGateStep(run.id, 'gate', 'user-brad'); + + expect(resumed.status).toBe('running'); + const gateStep = resumed.steps.find((s: { stepId: string }) => s.stepId === 'gate'); + expect(gateStep.status).toBe('completed'); + expect(resumed.context._gateBlock).toBeUndefined(); + + await vi.waitFor(async () => expect((await service.getRun(run.id)).status).toBe('completed')); + + const calls = mockExecuteStep.mock.calls.map((c: unknown[]) => (c[0] as { id: string }).id); + expect(calls).toContain('finish'); + // prep should not be re-run + expect(calls.filter((id: string) => id === 'prep')).toHaveLength(1); + }); + + it('rejectGateStep marks run failed and persists to storage', async () => { + const run = await service.startRun('wf-gate'); + await vi.waitFor(async () => expect((await service.getRun(run.id)).status).toBe('blocked')); + + const rejected = await service.rejectGateStep(run.id, 'gate', 'user-brad'); + + expect(rejected.status).toBe('failed'); + expect(rejected.error).toMatch(/rejected by user-brad/); + + const persisted = await service.getRun(run.id); + expect(persisted.status).toBe('failed'); + }); + + it('approveGateStep throws when run is not blocked', async () => { + mockLoadWorkflow.mockResolvedValue( + agentWorkflow('wf-plain', [{ id: 'step', type: 'agent', agent: 'a1', name: 'S' }]) + ); + mockExecuteStep.mockResolvedValue({ output: {}, outputPath: '/tmp/x.json' }); + const run = await service.startRun('wf-plain'); + await vi.waitFor(async () => expect((await service.getRun(run.id)).status).toBe('completed')); + + await expect(service.approveGateStep(run.id, 'step', 'user')).rejects.toThrow(/not blocked/); + }); + + it('approveGateStep rejects when stepId does not match the blocking gate', async () => { + const run = await service.startRun('wf-gate'); + await vi.waitFor(async () => expect((await service.getRun(run.id)).status).toBe('blocked')); + + await expect(service.approveGateStep(run.id, 'prep', 'user')).rejects.toThrow( + /blocked at gate "gate"/ + ); + }); + + it('rejectGateStep rejects when stepId does not match the blocking gate', async () => { + const run = await service.startRun('wf-gate'); + await vi.waitFor(async () => expect((await service.getRun(run.id)).status).toBe('blocked')); + + await expect(service.rejectGateStep(run.id, 'prep', 'user')).rejects.toThrow( + /blocked at gate "gate"/ + ); + }); + + it('completed run has no stale error from prior blocked phase', async () => { + const run = await service.startRun('wf-gate'); + await vi.waitFor(async () => expect((await service.getRun(run.id)).status).toBe('blocked')); + + await service.approveGateStep(run.id, 'gate', 'user-brad'); + await vi.waitFor(async () => expect((await service.getRun(run.id)).status).toBe('completed')); + + const completed = await service.getRun(run.id); + expect(completed.error).toBeUndefined(); + }); + + it('approved gate synthesizes step context output', async () => { + const run = await service.startRun('wf-gate'); + await vi.waitFor(async () => expect((await service.getRun(run.id)).status).toBe('blocked')); + + await service.approveGateStep(run.id, 'gate', 'user-brad'); + await vi.waitFor(async () => expect((await service.getRun(run.id)).status).toBe('completed')); + + const completed = await service.getRun(run.id); + expect(completed.context.gate).toMatchObject({ + passed: true, + humanApproved: true, + approvedBy: 'user-brad', + }); + }); + + it('ordinary false gate (no human escalation) still fails the run', async () => { + mockLoadWorkflow.mockResolvedValue( + agentWorkflow('wf-plaingate', [ + { id: 'gate2', type: 'gate', name: 'Plain Gate', condition: 'false' }, + ]) + ); + + const run = await service.startRun('wf-plaingate'); + await vi.waitFor(async () => expect((await service.getRun(run.id)).status).toBe('failed')); + + const failed = await service.getRun(run.id); + expect(failed.status).toBe('failed'); + expect(failed.status).not.toBe('blocked'); + }); +}); + +// ───────────────────────────────────────────────────────────── +// #780 — Bounded retry_step cycles +// ───────────────────────────────────────────────────────────── + +describe('#780 — Bounded retry_step cycles', () => { + let tmpDir: string; + let service: WorkflowRunService; + + beforeEach(async () => { + vi.resetModules(); + tmpDir = await fs.mkdtemp(path.join(os.tmpdir(), 'wf-780-')); + mockGetTask.mockResolvedValue(null); + const mod = await import('../services/workflow-run-service.js'); + service = new mod.WorkflowRunService(tmpDir); + }); + + afterEach(async () => { + vi.clearAllMocks(); + await fs.rm(tmpDir, { recursive: true, force: true }); + }); + + function retryWorkflow(maxReroutes?: number, onExhausted?: Record) { + return agentWorkflow('wf-retry', [ + { id: 'stepA', type: 'agent', agent: 'a1', name: 'Step A' }, + { + id: 'stepB', + type: 'agent', + agent: 'a1', + name: 'Step B', + on_fail: { + retry_step: 'stepA', + ...(maxReroutes !== undefined ? { max_reroutes: maxReroutes } : {}), + ...(onExhausted ? { on_exhausted: onExhausted } : {}), + }, + }, + ]); + } + + it('terminates retry_step loop at max_reroutes=2 and marks run failed', async () => { + mockLoadWorkflow.mockResolvedValue(retryWorkflow(2)); + mockExecuteStep.mockImplementation(async (step: { id: string }) => { + if (step.id === 'stepB') throw new Error('stepB always fails'); + return { output: {}, outputPath: '/tmp/x.json' }; + }); + + const run = await service.startRun('wf-retry'); + await vi.waitFor( + async () => { + const saved = await service.getRun(run.id); + expect(['failed', 'blocked'].includes(saved.status)).toBe(true); + }, + { timeout: 5000 } + ); + + const final = await service.getRun(run.id); + expect(final.status).toBe('failed'); + expect(final.error).toMatch(/retry_step budget exhausted/); + expect(final.retryRouteCount).toBeGreaterThan(2); + }); + + it('retryRouteCount is persisted across saves', async () => { + mockLoadWorkflow.mockResolvedValue(retryWorkflow(1)); + mockExecuteStep.mockImplementation(async (step: { id: string }) => { + if (step.id === 'stepB') throw new Error('always fails'); + return { output: {}, outputPath: '/tmp/x.json' }; + }); + + const run = await service.startRun('wf-retry'); + await vi.waitFor(async () => { + const saved = await service.getRun(run.id); + expect(['failed', 'blocked'].includes(saved.status)).toBe(true); + }); + + const mod = await import('../services/workflow-run-service.js'); + const service2 = new mod.WorkflowRunService(tmpDir); + const persisted = await service2.getRun(run.id); + expect(persisted?.retryRouteCount).toBeGreaterThan(0); + }); + + it('on_exhausted escalate_to:human blocks instead of failing', async () => { + mockLoadWorkflow.mockResolvedValue( + retryWorkflow(1, { escalate_to: 'human', escalate_message: 'Manual retry needed' }) + ); + mockExecuteStep.mockImplementation(async (step: { id: string }) => { + if (step.id === 'stepB') throw new Error('always fails'); + return { output: {}, outputPath: '/tmp/x.json' }; + }); + + const run = await service.startRun('wf-retry'); + await vi.waitFor(async () => { + const saved = await service.getRun(run.id); + expect(['failed', 'blocked'].includes(saved.status)).toBe(true); + }); + + const final = await service.getRun(run.id); + expect(final.status).toBe('blocked'); + expect(final.error).toMatch(/Manual retry needed/); + }); + + it('same-step retry counter still works independently (no retryRouteCount used)', async () => { + mockLoadWorkflow.mockResolvedValue( + agentWorkflow('wf-samestep', [ + { + id: 'step', + type: 'agent', + agent: 'a1', + name: 'Step', + on_fail: { retry: 2 }, + }, + ]) + ); + let calls = 0; + mockExecuteStep.mockImplementation(async () => { + calls++; + if (calls <= 2) throw new Error('transient'); + return { output: {}, outputPath: '/tmp/x.json' }; + }); + + const run = await service.startRun('wf-samestep'); + await vi.waitFor(async () => expect((await service.getRun(run.id)).status).toBe('completed')); + + const final = await service.getRun(run.id); + expect(final.status).toBe('completed'); + expect(final.retryRouteCount ?? 0).toBe(0); + }); +}); + +// ───────────────────────────────────────────────────────────── +// #785 — WorkflowRunService domain errors → AppError hierarchy +// ───────────────────────────────────────────────────────────── + +describe('#785 — WorkflowRunService errors extend AppError', () => { + let tmpDir: string; + let service: WorkflowRunService; + + beforeEach(async () => { + vi.resetModules(); + tmpDir = await fs.mkdtemp(path.join(os.tmpdir(), 'wf-785-')); + mockGetTask.mockResolvedValue(null); + mockLoadWorkflow.mockResolvedValue(null); + const mod = await import('../services/workflow-run-service.js'); + service = new mod.WorkflowRunService(tmpDir); + }); + + afterEach(async () => { + vi.clearAllMocks(); + await fs.rm(tmpDir, { recursive: true, force: true }); + }); + + it('startRun throws NotFoundError (404) for missing workflow', async () => { + const { NotFoundError } = await import('../middleware/error-handler.js'); + await expect(service.startRun('missing-wf')).rejects.toBeInstanceOf(NotFoundError); + + try { + await service.startRun('missing-wf'); + } catch (err: unknown) { + const error = err as { statusCode: number; code: string }; + expect(error.statusCode).toBe(404); + expect(error.code).toBe('NOT_FOUND'); + } + }); + + it('resumeRun throws NotFoundError (404) for unknown run ID', async () => { + const { NotFoundError } = await import('../middleware/error-handler.js'); + const fakeRunId = `run_${Date.now()}_abcdef12`; + await expect(service.resumeRun(fakeRunId)).rejects.toBeInstanceOf(NotFoundError); + + try { + await service.resumeRun(fakeRunId); + } catch (err: unknown) { + const error = err as { statusCode: number }; + expect(error.statusCode).toBe(404); + } + }); + + it('resumeRun throws ValidationError (400) when run is not blocked', async () => { + mockLoadWorkflow.mockResolvedValue( + agentWorkflow('wf-done', [{ id: 's', type: 'agent', agent: 'a1', name: 'S' }]) + ); + mockExecuteStep.mockResolvedValue({ output: {}, outputPath: '/tmp/x.json' }); + const run = await service.startRun('wf-done'); + await vi.waitFor(async () => expect((await service.getRun(run.id)).status).toBe('completed')); + + const { ValidationError } = await import('../middleware/error-handler.js'); + await expect(service.resumeRun(run.id)).rejects.toBeInstanceOf(ValidationError); + + try { + await service.resumeRun(run.id); + } catch (err: unknown) { + const error = err as { statusCode: number; code: string }; + expect(error.statusCode).toBe(400); + expect(error.code).toBe('VALIDATION_ERROR'); + } + }); + + it('startRun throws ValidationError (400) for reserved context keys', async () => { + mockLoadWorkflow.mockResolvedValue( + agentWorkflow('wf-ctx', [{ id: 's', type: 'agent', agent: 'a1', name: 'S' }]) + ); + const { ValidationError } = await import('../middleware/error-handler.js'); + await expect(service.startRun('wf-ctx', undefined, { task: 'hijack' })).rejects.toBeInstanceOf( + ValidationError + ); + + try { + await service.startRun('wf-ctx', undefined, { workflow: 'hijack' }); + } catch (err: unknown) { + const error = err as { statusCode: number }; + expect(error.statusCode).toBe(400); + } + }); +}); + +// ───────────────────────────────────────────────────────────── +// #786 — Shared workflow contracts: provider/command fields +// ───────────────────────────────────────────────────────────── + +describe('#786 — Shared WorkflowAgent includes provider and command', () => { + it('shared WorkflowAgent type accepts provider and command', () => { + const agent: WorkflowAgent = { + id: 'codex-agent', + name: 'Codex', + role: 'coder', + description: 'A codex agent', + provider: 'codex-cli', + command: 'codex --model o4-mini', + }; + expect(agent.provider).toBe('codex-cli'); + expect(agent.command).toBe('codex --model o4-mini'); + }); + + it('shared WorkflowAgent accepts openclaw provider without command', () => { + const agent: WorkflowAgent = { + id: 'oc-agent', + name: 'OpenClaw', + role: 'coder', + description: '', + provider: 'openclaw', + }; + expect(agent.provider).toBe('openclaw'); + expect(agent.command).toBeUndefined(); + }); + + it('shared WorkflowAgent is assignment-compatible with server WorkflowAgent', () => { + // Compile-time check: if server's type has fields shared doesn't, this fails at tsc. + type ServerAgent = import('../types/workflow.js').WorkflowAgent; + const serverAgent: ServerAgent = { + id: 'sa', + name: 'Server Agent', + role: 'r', + description: 'd', + provider: 'codex-cloud', + command: 'codex', + }; + const sharedAgent: WorkflowAgent = serverAgent; + expect(sharedAgent.provider).toBe('codex-cloud'); + expect(sharedAgent.command).toBe('codex'); + }); +}); + +// ───────────────────────────────────────────────────────────── +// #787 — depends_on enforcement during status transitions +// ───────────────────────────────────────────────────────────── + +describe('#787 — BlockingService enforces depends_on', () => { + let svc: BlockingService; + + beforeEach(() => { + svc = new BlockingService(); + }); + + // getBlockingStatus + + it('returns not blocked when task has no dependencies', () => { + const task = makeTask({ id: 'A' }); + expect(svc.getBlockingStatus(task, [task]).isBlocked).toBe(false); + }); + + it('reports blocked when depends_on dep is incomplete', () => { + const dep = makeTask({ id: 'B', status: 'in-progress' }); + const task = makeTask({ id: 'A', dependencies: { depends_on: ['B'] } }); + const result = svc.getBlockingStatus(task, [dep, task]); + expect(result.isBlocked).toBe(true); + expect(result.blockers.map((b) => b.id)).toContain('B'); + }); + + it('reports not blocked when all depends_on deps are done', () => { + const dep = makeTask({ id: 'B', status: 'done' }); + const task = makeTask({ id: 'A', dependencies: { depends_on: ['B'] } }); + const result = svc.getBlockingStatus(task, [dep, task]); + expect(result.isBlocked).toBe(false); + expect(result.completedBlockers.map((b) => b.id)).toContain('B'); + }); + + it('deduplicates same ID appearing in both blockedBy and depends_on', () => { + const dep = makeTask({ id: 'B', status: 'in-progress' }); + const task = makeTask({ + id: 'A', + blockedBy: ['B'], + dependencies: { depends_on: ['B'] }, + }); + const result = svc.getBlockingStatus(task, [dep, task]); + expect(result.isBlocked).toBe(true); + expect(result.blockers).toHaveLength(1); + }); + + it('merges distinct IDs from blockedBy and depends_on', () => { + const b1 = makeTask({ id: 'B1', status: 'in-progress' }); + const b2 = makeTask({ id: 'B2', status: 'todo' }); + const task = makeTask({ + id: 'A', + blockedBy: ['B1'], + dependencies: { depends_on: ['B2'] }, + }); + const result = svc.getBlockingStatus(task, [b1, b2, task]); + expect(result.isBlocked).toBe(true); + expect(result.blockers).toHaveLength(2); + }); + + // canMoveToInProgress + + it('canMoveToInProgress returns false when depends_on dep is incomplete', () => { + const dep = makeTask({ id: 'B', status: 'todo' }); + const task = makeTask({ id: 'A', dependencies: { depends_on: ['B'] } }); + const { allowed, blockers } = svc.canMoveToInProgress(task, [dep, task]); + expect(allowed).toBe(false); + expect(blockers?.map((b) => b.id)).toContain('B'); + }); + + it('canMoveToInProgress returns true when all depends_on deps are done', () => { + const dep = makeTask({ id: 'B', status: 'done' }); + const task = makeTask({ id: 'A', dependencies: { depends_on: ['B'] } }); + expect(svc.canMoveToInProgress(task, [dep, task]).allowed).toBe(true); + }); + + it('canMoveToInProgress returns false when legacy blockedBy dep is incomplete', () => { + const dep = makeTask({ id: 'B', status: 'in-progress' }); + const task = makeTask({ id: 'A', blockedBy: ['B'] }); + expect(svc.canMoveToInProgress(task, [dep, task]).allowed).toBe(false); + }); + + it('canMoveToInProgress returns true when task has no dependencies at all', () => { + const task = makeTask({ id: 'A' }); + expect(svc.canMoveToInProgress(task, [task]).allowed).toBe(true); + }); + + // getDependentTasks + + it('getDependentTasks finds tasks that depend via depends_on', () => { + const taskA = makeTask({ id: 'A' }); + const taskB = makeTask({ id: 'B', dependencies: { depends_on: ['A'] } }); + const taskC = makeTask({ id: 'C', blockedBy: ['A'] }); + const deps = svc.getDependentTasks('A', [taskA, taskB, taskC]); + expect(deps.map((t) => t.id)).toContain('B'); + expect(deps.map((t) => t.id)).toContain('C'); + }); + + // wouldCreateCircularDependency + + it('detects circular dependency through depends_on chain', () => { + const taskA = makeTask({ id: 'A', dependencies: { depends_on: ['B'] } }); + const taskB = makeTask({ id: 'B', dependencies: { depends_on: ['C'] } }); + const taskC = makeTask({ id: 'C' }); + // Adding A as dep of C: C → A → B → C + expect(svc.wouldCreateCircularDependency('C', 'A', [taskA, taskB, taskC])).toBe(true); + }); + + it('does not flag non-circular depends_on chain', () => { + const taskA = makeTask({ id: 'A' }); + const taskB = makeTask({ id: 'B', dependencies: { depends_on: ['A'] } }); + // C → B → A: no cycle + expect(svc.wouldCreateCircularDependency('C', 'B', [taskA, taskB])).toBe(false); + }); + + // getTasksThatWouldBeUnblocked + + it('getTasksThatWouldBeUnblocked includes tasks blocked via depends_on', () => { + const taskA = makeTask({ id: 'A', status: 'in-progress' }); + const taskB = makeTask({ id: 'B', dependencies: { depends_on: ['A'] } }); + const unblocked = svc.getTasksThatWouldBeUnblocked('A', [taskA, taskB]); + expect(unblocked.map((t) => t.id)).toContain('B'); + }); + + it('does not unblock when task has other incomplete deps', () => { + const taskA = makeTask({ id: 'A', status: 'in-progress' }); + const taskC = makeTask({ id: 'C', status: 'todo' }); + const taskB = makeTask({ id: 'B', dependencies: { depends_on: ['A', 'C'] } }); + // Completing A does not unblock B because C is still todo + const unblocked = svc.getTasksThatWouldBeUnblocked('A', [taskA, taskB, taskC]); + expect(unblocked.map((t) => t.id)).not.toContain('B'); + }); +}); diff --git a/server/src/__tests__/storage/sqlite-workflow-run-execution.test.ts b/server/src/__tests__/storage/sqlite-workflow-run-execution.test.ts index 52ccb0b7..483bc1d2 100644 --- a/server/src/__tests__/storage/sqlite-workflow-run-execution.test.ts +++ b/server/src/__tests__/storage/sqlite-workflow-run-execution.test.ts @@ -13,11 +13,15 @@ const mockExecuteStep = vi.fn(); const mockBroadcastWorkflowStatus = vi.fn(); const mockGetTask = vi.fn(); -vi.mock('../../services/workflow-step-executor.js', () => ({ - WorkflowStepExecutor: class { - executeStep = mockExecuteStep; - }, -})); +vi.mock('../../services/workflow-step-executor.js', async (importOriginal) => { + const actual = await importOriginal(); + return { + HumanGateBlockError: actual.HumanGateBlockError, + WorkflowStepExecutor: class { + executeStep = mockExecuteStep; + }, + }; +}); vi.mock('../../services/broadcast-service.js', () => ({ broadcastWorkflowStatus: mockBroadcastWorkflowStatus, diff --git a/server/src/__tests__/workflow-run-service.test.ts b/server/src/__tests__/workflow-run-service.test.ts index d3e426b7..be3a5637 100644 --- a/server/src/__tests__/workflow-run-service.test.ts +++ b/server/src/__tests__/workflow-run-service.test.ts @@ -17,11 +17,15 @@ vi.mock('../services/workflow-service.js', () => ({ }), })); -vi.mock('../services/workflow-step-executor.js', () => ({ - WorkflowStepExecutor: class { - executeStep = mockExecuteStep; - }, -})); +vi.mock('../services/workflow-step-executor.js', async (importOriginal) => { + const actual = await importOriginal(); + return { + HumanGateBlockError: actual.HumanGateBlockError, + WorkflowStepExecutor: class { + executeStep = mockExecuteStep; + }, + }; +}); vi.mock('../services/broadcast-service.js', () => ({ broadcastWorkflowStatus: mockBroadcastWorkflowStatus, diff --git a/server/src/routes/tasks.ts b/server/src/routes/tasks.ts index 453bfaa6..8d977533 100644 --- a/server/src/routes/tasks.ts +++ b/server/src/routes/tasks.ts @@ -86,7 +86,6 @@ const attemptSchema = z provider: z.string().max(80).optional(), model: z.string().max(160).optional(), threadId: z.string().max(240).optional(), - sessionKey: z.string().max(240).optional(), cloudUrl: z.string().max(500).optional(), cloudTarget: z.string().max(240).optional(), orchestration: z.unknown().optional(), @@ -727,11 +726,12 @@ router.patch( // (human users can still approve, or API keys with admin/agent role) } - // Check if trying to move a blocked task to in-progress + // Check if trying to move a blocked task to in-progress. + // Enforce both legacy blockedBy and canonical dependencies.depends_on (#787). if ( input.status === 'in-progress' && oldTask.status !== 'in-progress' && - oldTask.blockedBy?.length + (oldTask.blockedBy?.length || oldTask.dependencies?.depends_on?.length) ) { const allTasks = await taskService.listTasks(); const { allowed, blockers } = blockingService.canMoveToInProgress(oldTask, allTasks); diff --git a/server/src/routes/workflows.ts b/server/src/routes/workflows.ts index cbda92b2..8f08976c 100644 --- a/server/src/routes/workflows.ts +++ b/server/src/routes/workflows.ts @@ -620,7 +620,7 @@ router.post( /** * POST /api/workflow-runs/:runId/steps/:stepId/approve — Approve a gate step - * Phase 4: Gate approval endpoint + * Phase 4: Gate approval endpoint — fixed for human-gate blocking (#778) */ router.post( '/runs/:runId/steps/:stepId/approve', @@ -637,19 +637,13 @@ router.post( // Check execute permission on the workflow await assertWorkflowPermission(run.workflowId, userId, 'execute'); - // Find the step - const stepRun = run.steps.find((s) => s.stepId === stepId); - if (!stepRun) { - throw new NotFoundError(`Step ${stepId} not found in run ${runId}`); - } - - if (stepRun.status !== 'failed') { + if (run.status !== 'blocked') { throw new ValidationError( - `Step ${stepId} is not awaiting approval (current status: ${stepRun.status})` + `Run ${runId} is not blocked (current status: ${run.status}). Only blocked runs can be approved.` ); } - // Security: Verify this is actually a gate step + // Security: Verify this is actually a gate step with human escalation const workflow = await workflowService.loadWorkflow(run.workflowId); if (!workflow) { throw new NotFoundError(`Workflow ${run.workflowId} not found`); @@ -662,18 +656,14 @@ router.post( ); } - // Approve: add approval to context and resume - const approvalContext = { - ...run.context, - _gateApproval: { - stepId, - approved: true, - approvedBy: userId, - approvedAt: new Date().toISOString(), - }, - }; + if (stepDef.on_false?.escalate_to !== 'human') { + throw new ValidationError(`Gate step ${stepId} does not have a human escalation policy`); + } - const resumed = await workflowRunService.resumeRun(runId, approvalContext); + // Validate optional resume context + const { context } = resumeRunSchema.parse(req.body || {}); + + const resumed = await workflowRunService.approveGateStep(runId, stepId, userId, context); res.json(resumed); }) @@ -681,7 +671,7 @@ router.post( /** * POST /api/workflow-runs/:runId/steps/:stepId/reject — Reject a gate step - * Phase 4: Gate rejection endpoint + * Phase 4: Gate rejection endpoint — fixed to persist state (#778) */ router.post( '/runs/:runId/steps/:stepId/reject', @@ -698,19 +688,13 @@ router.post( // Check execute permission on the workflow await assertWorkflowPermission(run.workflowId, userId, 'execute'); - // Find the step - const stepRun = run.steps.find((s) => s.stepId === stepId); - if (!stepRun) { - throw new NotFoundError(`Step ${stepId} not found in run ${runId}`); - } - - if (stepRun.status !== 'failed') { + if (run.status !== 'blocked') { throw new ValidationError( - `Step ${stepId} is not awaiting approval (current status: ${stepRun.status})` + `Run ${runId} is not blocked (current status: ${run.status}). Only blocked runs can be rejected.` ); } - // Security: Verify this is actually a gate step + // Security: Verify this is actually a gate step with human escalation const workflow = await workflowService.loadWorkflow(run.workflowId); if (!workflow) { throw new NotFoundError(`Workflow ${run.workflowId} not found`); @@ -723,14 +707,13 @@ router.post( ); } - // Reject: mark run as failed - run.status = 'failed'; - run.error = `Step ${stepId} rejected by ${userId}`; - run.completedAt = new Date().toISOString(); + if (stepDef.on_false?.escalate_to !== 'human') { + throw new ValidationError(`Gate step ${stepId} does not have a human escalation policy`); + } - // This would be saved by the workflow run service - // For now, just return the updated run - res.json(run); + const rejected = await workflowRunService.rejectGateStep(runId, stepId, userId); + + res.json(rejected); }) ); diff --git a/server/src/services/blocking-service.ts b/server/src/services/blocking-service.ts index d9a1bb95..0feec3a8 100644 --- a/server/src/services/blocking-service.ts +++ b/server/src/services/blocking-service.ts @@ -3,6 +3,9 @@ * * Handles task blocking/dependency logic. * Extracted from tasks.ts route to separate business logic from HTTP concerns. + * + * Dual-field enforcement (#787): checks both legacy `blockedBy` and canonical + * `dependencies.depends_on` so the modern dependency API is always honoured. */ import type { Task } from '@veritas-kanban/shared'; @@ -19,16 +22,26 @@ export interface BlockingStatus { completedBlockers: BlockerInfo[]; } +/** Collect the full set of dependency IDs from both legacy and canonical fields. */ +function allDependencyIds(task: Task): string[] { + const ids = new Set(); + for (const id of task.blockedBy ?? []) ids.add(id); + for (const id of task.dependencies?.depends_on ?? []) ids.add(id); + return Array.from(ids); +} + export class BlockingService { /** - * Get the blocking status for a task + * Get the blocking status for a task. + * Considers both `blockedBy` (legacy) and `dependencies.depends_on` (canonical). */ getBlockingStatus(task: Task, allTasks: Task[]): BlockingStatus { - if (!task.blockedBy?.length) { + const depIds = allDependencyIds(task); + if (depIds.length === 0) { return { isBlocked: false, blockers: [], completedBlockers: [] }; } - const blockingTasks = allTasks.filter((t) => task.blockedBy?.includes(t.id)); + const blockingTasks = allTasks.filter((t) => depIds.includes(t.id)); const incompleteBlockers = blockingTasks.filter((t) => t.status !== 'done'); const completedBlockers = blockingTasks.filter((t) => t.status === 'done'); @@ -47,17 +60,19 @@ export class BlockingService { } /** - * Check if a task can move to in-progress (i.e., all blockers are done) + * Check if a task can move to in-progress (all dependencies must be done). + * Considers both `blockedBy` (legacy) and `dependencies.depends_on` (canonical). */ canMoveToInProgress( task: Task, allTasks: Task[] ): { allowed: boolean; blockers?: BlockerInfo[] } { - if (!task.blockedBy?.length) { + const depIds = allDependencyIds(task); + if (depIds.length === 0) { return { allowed: true }; } - const blockingTasks = allTasks.filter((t) => task.blockedBy?.includes(t.id)); + const blockingTasks = allTasks.filter((t) => depIds.includes(t.id)); const incompleteBlockers = blockingTasks.filter((t) => t.status !== 'done'); if (incompleteBlockers.length > 0) { @@ -71,35 +86,38 @@ export class BlockingService { } /** - * Get all tasks that are blocked by a given task + * Get all tasks that are blocked by a given task (reverse lookup). + * Considers both `blockedBy` (legacy) and `dependencies.depends_on` (canonical). */ getDependentTasks(taskId: string, allTasks: Task[]): Task[] { - return allTasks.filter((t) => t.blockedBy?.includes(taskId)); + return allTasks.filter((t) => { + if (t.blockedBy?.includes(taskId)) return true; + if (t.dependencies?.depends_on?.includes(taskId)) return true; + return false; + }); } /** - * Check if completing a task would unblock other tasks + * Check if completing a task would unblock other tasks. */ getTasksThatWouldBeUnblocked(taskId: string, allTasks: Task[]): Task[] { const dependentTasks = this.getDependentTasks(taskId, allTasks); return dependentTasks.filter((task) => { - // Check if this is the only incomplete blocker - const otherBlockers = task.blockedBy?.filter((id: string) => id !== taskId) || []; - const otherIncompleteBlockers = otherBlockers.filter((blockerId: string) => { + const depIds = allDependencyIds(task).filter((id) => id !== taskId); + const otherIncomplete = depIds.filter((blockerId) => { const blockerTask = allTasks.find((t: Task) => t.id === blockerId); return blockerTask && blockerTask.status !== 'done'; }); - - return otherIncompleteBlockers.length === 0; + return otherIncomplete.length === 0; }); } /** - * Validate that adding a blocker wouldn't create a circular dependency + * Validate that adding a dependency wouldn't create a circular dependency. + * Traverses both `blockedBy` and `dependencies.depends_on` graphs. */ wouldCreateCircularDependency(taskId: string, newBlockerId: string, allTasks: Task[]): boolean { - // Check if the new blocker (or any of its blockers) depends on taskId const visited = new Set(); const queue = [newBlockerId]; @@ -107,7 +125,7 @@ export class BlockingService { const currentId = queue.shift()!; if (currentId === taskId) { - return true; // Circular dependency detected + return true; } if (visited.has(currentId)) { @@ -116,8 +134,8 @@ export class BlockingService { visited.add(currentId); const task = allTasks.find((t) => t.id === currentId); - if (task?.blockedBy) { - queue.push(...task.blockedBy); + if (task) { + queue.push(...allDependencyIds(task)); } } diff --git a/server/src/services/workflow-run-service.ts b/server/src/services/workflow-run-service.ts index 35502738..a44aa008 100644 --- a/server/src/services/workflow-run-service.ts +++ b/server/src/services/workflow-run-service.ts @@ -19,7 +19,7 @@ import { } from '@veritas-kanban/shared'; import type { WorkflowRun, StepRun, WorkflowDefinition, WorkflowStep } from '../types/workflow.js'; import { getWorkflowService } from './workflow-service.js'; -import { WorkflowStepExecutor } from './workflow-step-executor.js'; +import { WorkflowStepExecutor, HumanGateBlockError } from './workflow-step-executor.js'; import { getWorkflowRunsDir } from '../utils/paths.js'; import { createLogger } from '../lib/logger.js'; import { broadcastWorkflowStatus } from './broadcast-service.js'; @@ -29,11 +29,14 @@ 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'; const log = createLogger('workflow-run'); // Concurrency limits 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 RUN_ID_PATTERN = /^run_\d{10,}_[a-zA-Z0-9_-]{6,}$/; const RESERVED_CONTEXT_KEYS = new Set([ @@ -43,22 +46,10 @@ const RESERVED_CONTEXT_KEYS = new Set([ 'pipeline', '_sessions', '_retryContext', + '_gateBlock', + '_gateApproval', ]); -class NotFoundError extends Error { - constructor(message: string) { - super(message); - this.name = 'NotFoundError'; - } -} - -class ValidationError extends Error { - constructor(message: string) { - super(message); - this.name = 'ValidationError'; - } -} - export class WorkflowRunService { private runsDir: string; private workflowService: ReturnType; @@ -498,6 +489,31 @@ export class WorkflowRunService { await this.saveRun(run); broadcastWorkflowStatus(run); } catch (err: unknown) { + // --- Gate human-escalation (#778) --- + // HumanGateBlockError is a clean, expected pause — not a real failure. + // Handle it before the generic failure path so on_fail is not misapplied. + if (err instanceof HumanGateBlockError) { + stepRun.status = 'failed'; + stepRun.error = err.escalationMessage; + stepRun.completedAt = new Date().toISOString(); + run.status = 'blocked'; + run.error = err.escalationMessage; + // Persist gate blocking context for resume / approve endpoints. + run.context._gateBlock = { + stepId: err.stepId, + escalationMessage: err.escalationMessage, + blockedAt: new Date().toISOString(), + }; + this.syncPipelineSummary(run, workflow); + await this.saveRun(run); + broadcastWorkflowStatus(run); + log.warn( + { runId: run.id, stepId: err.stepId }, + 'Workflow blocked at human gate — awaiting resume' + ); + return; + } + // Step failed stepRun.status = 'failed'; stepRun.error = err instanceof Error ? err.message : 'Unknown error'; @@ -525,8 +541,9 @@ export class WorkflowRunService { return; } - // All steps completed + // All steps completed; clear any stale error from an earlier blocked phase. run.status = 'completed'; + run.error = undefined; run.completedAt = new Date().toISOString(); this.syncPipelineSummary(run, workflow); await this.saveRun(run); @@ -593,6 +610,48 @@ export class WorkflowRunService { throw new Error(`retry_step references unknown step: ${policy.retry_step}`); } + // Enforce cross-step reroute budget (#780) + const maxReroutes = policy.max_reroutes ?? MAX_REROUTES_DEFAULT; + run.retryRouteCount = (run.retryRouteCount ?? 0) + 1; + + if (run.retryRouteCount > maxReroutes) { + // Budget exhausted — apply on_exhausted policy or fail/block + const exhausted = policy.on_exhausted; + log.warn( + { + runId: run.id, + stepId: step.id, + retryStep: retryStep.id, + retryRouteCount: run.retryRouteCount, + maxReroutes, + }, + 'retry_step reroute budget exhausted' + ); + + if (exhausted?.escalate_to === 'human') { + run.status = 'blocked'; + run.error = + exhausted.escalate_message || + `retry_step budget exhausted after ${maxReroutes} reroutes`; + this.syncPipelineSummary(run, workflow); + await this.saveRun(run); + log.warn({ runId: run.id, stepId: step.id }, 'Workflow blocked: retry_step exhausted'); + return true; + } + + if (exhausted?.escalate_to === 'skip') { + stepRun.status = 'skipped'; + this.syncPipelineSummary(run, workflow); + await this.saveRun(run); + return true; + } + + // Default: fail deterministically + throw new Error( + `retry_step budget exhausted: step "${step.id}" has rerouted ${run.retryRouteCount} times (max ${maxReroutes})` + ); + } + // Reset the retry step's state const retryStepRun = run.steps.find((s) => s.stepId === retryStep.id)!; retryStepRun.status = 'pending'; @@ -612,11 +671,15 @@ export class WorkflowRunService { failedStep: step.id, error: stepRun.error, retries: stepRun.retries, + retryRouteCount: run.retryRouteCount, }; this.syncPipelineSummary(run, workflow); await this.saveRun(run); - log.info({ failedStep: step.id, retryStep: retryStep.id }, 'Routing to retry step'); + log.info( + { failedStep: step.id, retryStep: retryStep.id, retryRouteCount: run.retryRouteCount }, + 'Routing to retry step' + ); return true; } @@ -819,7 +882,10 @@ export class WorkflowRunService { ...run.context, ...this.validateExternalContext(resumeContext, 'Resume context'), }; + // Clear gate block context and stale blocked-state errors on resume. + delete run.context._gateBlock; run.status = 'running'; + run.error = undefined; await this.saveRun(run); // Resume execution @@ -840,6 +906,107 @@ export class WorkflowRunService { return run; } + /** + * Approve a blocked human-gate step and resume the run (#778). + * Marks the gate step as 'completed' (human-approved) so resume skips it. + */ + async approveGateStep( + runId: string, + stepId: string, + approvedBy: string, + resumeContext?: Record + ): Promise { + const run = await this.getRun(runId); + if (!run) { + throw new NotFoundError(`Run ${runId} not found`); + } + + if (run.status !== 'blocked') { + throw new ValidationError(`Run ${runId} is not blocked (status: ${run.status})`); + } + + const blockedGateId = (run.context._gateBlock as { stepId?: string } | undefined)?.stepId; + if (blockedGateId && blockedGateId !== stepId) { + throw new ValidationError( + `Run ${runId} is blocked at gate "${blockedGateId}", not "${stepId}"` + ); + } + + const stepRun = run.steps.find((s) => s.stepId === stepId); + if (!stepRun) { + throw new NotFoundError(`Step ${stepId} not found in run ${runId}`); + } + + // Mark the gate step as completed (human approved it — override the false condition). + stepRun.status = 'completed'; + stepRun.completedAt = new Date().toISOString(); + stepRun.error = undefined; + + // Synthesize gate output for downstream templating and context consumers. + run.context[stepId] = { + passed: true, + approvedBy, + approvedAt: new Date().toISOString(), + humanApproved: true, + }; + + // Record approval in reserved context for audit/trace. + run.context._gateApproval = { + stepId, + approved: true, + approvedBy, + approvedAt: new Date().toISOString(), + }; + + // Persist the gate-step completion BEFORE resumeRun re-loads from storage. + await this.saveRun(run); + + log.info({ runId, stepId, approvedBy }, 'Gate step approved — resuming run'); + + return this.resumeRun(runId, resumeContext); + } + + /** + * Reject a blocked human-gate step — terminates the run as failed (#778). + */ + async rejectGateStep(runId: string, stepId: string, rejectedBy: string): Promise { + const run = await this.getRun(runId); + if (!run) { + throw new NotFoundError(`Run ${runId} not found`); + } + + if (run.status !== 'blocked') { + throw new ValidationError(`Run ${runId} is not blocked (status: ${run.status})`); + } + + const blockedGateId = (run.context._gateBlock as { stepId?: string } | undefined)?.stepId; + if (blockedGateId && blockedGateId !== stepId) { + throw new ValidationError( + `Run ${runId} is blocked at gate "${blockedGateId}", not "${stepId}"` + ); + } + + const stepRun = run.steps.find((s) => s.stepId === stepId); + if (!stepRun) { + throw new NotFoundError(`Step ${stepId} not found in run ${runId}`); + } + + run.status = 'failed'; + run.error = `Gate step ${stepId} rejected by ${rejectedBy}`; + run.completedAt = new Date().toISOString(); + delete run.context._gateBlock; + + const workflow = await this.workflowService.loadWorkflow(run.workflowId); + if (workflow) { + this.syncPipelineSummary(run, workflow); + } + await this.saveRun(run); + broadcastWorkflowStatus(run); + + log.warn({ runId, stepId, rejectedBy }, 'Gate step rejected — run failed'); + return run; + } + private validateExternalContext( context: Record | undefined, source: string diff --git a/server/src/services/workflow-step-executor.ts b/server/src/services/workflow-step-executor.ts index 3dc15340..d2d58fb3 100644 --- a/server/src/services/workflow-step-executor.ts +++ b/server/src/services/workflow-step-executor.ts @@ -14,6 +14,7 @@ import type { StepExecutionResult, WorkflowAgent, StepSessionConfig, + EscalationPolicy, } from '../types/workflow.js'; import { getWorkflowRunsDir } from '../utils/paths.js'; import { buildSafeCodexEnv } from '../utils/codex-env.js'; @@ -35,6 +36,26 @@ import { getSandboxPolicyService } from './sandbox-policy-service.js'; const log = createLogger('workflow-step-executor'); +/** + * Thrown by executeGateStep when a gate's on_false policy escalates to human (#778). + * Allows WorkflowRunService to distinguish an expected human pause from a real failure. + */ +export class HumanGateBlockError extends Error { + readonly stepId: string; + readonly escalationMessage: string; + readonly policy: EscalationPolicy; + + constructor(stepId: string, policy: EscalationPolicy) { + const msg = + policy.escalate_message || `Gate ${stepId} condition not met — awaiting human input`; + super(msg); + this.name = 'HumanGateBlockError'; + this.stepId = stepId; + this.escalationMessage = msg; + this.policy = policy; + } +} + interface WorkflowStepExecutorOptions { openClawAdapter?: OpenClawWorkflowAdapter; } @@ -86,8 +107,7 @@ export class WorkflowStepExecutor { agentDef = { ...agentDef, model: run.budget.modelOverride }; } const workflowConfig = run.context.workflow as - | { config?: { fresh_session_default?: boolean } } - | undefined; + { config?: { fresh_session_default?: boolean } } | undefined; // Build session configuration (#111) const sessionConfig = this.buildSessionConfig(step, run, workflowConfig?.config); @@ -669,8 +689,7 @@ export class WorkflowStepExecutor { private getWorkflowWorkingDirectory(run: WorkflowRun): string { const task = run.context.task as - | { git?: { worktreePath?: string; repo?: string }; repoPath?: string } - | undefined; + { git?: { worktreePath?: string; repo?: string }; repoPath?: string } | undefined; return task?.git?.worktreePath || task?.repoPath || process.cwd(); } diff --git a/server/src/types/workflow.ts b/server/src/types/workflow.ts index 0ba859b2..a3995130 100644 --- a/server/src/types/workflow.ts +++ b/server/src/types/workflow.ts @@ -37,12 +37,7 @@ export interface WorkflowConfig { export type WorkflowPipelineMode = 'single-agent' | 'orchestrated'; export type WorkflowPipelineCompletion = 'all-required' | 'any-success' | 'manual-review'; export type WorkflowSubagentRunStatus = - | 'pending' - | 'running' - | 'blocked' - | 'completed' - | 'failed' - | 'skipped'; + 'pending' | 'running' | 'blocked' | 'completed' | 'failed' | 'skipped'; export interface WorkflowSubagentTelemetry { tokenBudget?: number; @@ -94,12 +89,7 @@ export interface WorkflowOutputTarget { } export type WorkflowScheduleMode = - | 'manual' - | 'daily' - | 'weekly' - | 'biweekly' - | 'monthly' - | 'custom'; + 'manual' | 'daily' | 'weekly' | 'biweekly' | 'monthly' | 'custom'; export interface WorkflowSchedule { mode: WorkflowScheduleMode; @@ -160,6 +150,7 @@ export interface FailurePolicy { retry?: number; retry_delay_ms?: number; // Phase 2: Delay between retries (#113) retry_step?: string; // Retry a different step ID + max_reroutes?: number; // Max cross-step reroutes before on_exhausted fires (#780) escalate_to?: 'human' | `agent:${string}` | 'skip'; escalate_message?: string; on_exhausted?: EscalationPolicy; @@ -218,6 +209,8 @@ export interface WorkflowRun { lastCheckpoint?: string; // Phase 2: Last state persistence timestamp (#113) error?: string; steps: StepRun[]; + /** Persisted reroute counter for cross-step retry_step bounds (#780) */ + retryRouteCount?: number; } export interface StepRun { diff --git a/shared/src/types/workflow.ts b/shared/src/types/workflow.ts index 52ec624c..64d0f433 100644 --- a/shared/src/types/workflow.ts +++ b/shared/src/types/workflow.ts @@ -35,12 +35,7 @@ export interface WorkflowConfig { export type WorkflowPipelineMode = 'single-agent' | 'orchestrated'; export type WorkflowPipelineCompletion = 'all-required' | 'any-success' | 'manual-review'; export type WorkflowSubagentRunStatus = - | 'pending' - | 'running' - | 'blocked' - | 'completed' - | 'failed' - | 'skipped'; + 'pending' | 'running' | 'blocked' | 'completed' | 'failed' | 'skipped'; export interface WorkflowSubagentTelemetry { tokenBudget?: number; @@ -122,12 +117,7 @@ export interface WorkflowOutputTarget { } export type WorkflowScheduleMode = - | 'manual' - | 'daily' - | 'weekly' - | 'biweekly' - | 'monthly' - | 'custom'; + 'manual' | 'daily' | 'weekly' | 'biweekly' | 'monthly' | 'custom'; export interface WorkflowSchedule { mode: WorkflowScheduleMode; @@ -145,6 +135,8 @@ export interface WorkflowAgent { name: string; role: string; // maps to toolPolicy model?: string; // default model for this agent + provider?: string; // Agent provider for web/CLI/MCP (#786) + command?: string; // Provider command for execution (#786) sandboxPresetId?: string; budget?: import('./agent-budget.types.js').AgentBudgetPolicy; description: string; @@ -186,6 +178,7 @@ export interface FailurePolicy { retry?: number; retry_delay_ms?: number; // Phase 2: Delay between retries (#113) retry_step?: string; // Retry a different step ID + max_reroutes?: number; // Max cross-step reroutes before on_exhausted fires (#780) escalate_to?: 'human' | `agent:${string}` | 'skip'; escalate_message?: string; on_exhausted?: EscalationPolicy;