diff --git a/docs-web/architecture/node-flow-durable-execution.md b/docs-web/architecture/node-flow-durable-execution.md new file mode 100644 index 0000000000..e5bc80f167 --- /dev/null +++ b/docs-web/architecture/node-flow-durable-execution.md @@ -0,0 +1,7 @@ +# Node Flow Durable Execution + +Node flows execute immutable published snapshots. A run explicitly pins a published version or follows the latest published version; later edits cannot change a pinned run. + +Runs are durably queued and leased with bounded global and project concurrency. Node attempts retain attempt number, executor and invocation identity, artifact digest, redacted payloads, credential ids, failure class, and retry decision. Retryable failures use bounded exponential backoff and jitter, while cancellation and timeout signals propagate to provider and HTTP work. + +On restart, expired pre-invocation work is safely requeued. Work with an external invocation and an unknown outcome moves to `attention_required` and is never silently replayed. Credential values are resolved only for the active node and are not retained in run history or diagnostics. diff --git a/docs-web/architecture/node-flows.md b/docs-web/architecture/node-flows.md index 493c84b123..f3f15bfffa 100644 --- a/docs-web/architecture/node-flows.md +++ b/docs-web/architecture/node-flows.md @@ -14,3 +14,5 @@ Node flows are project-owned, versioned Graph v2 workflows. | `output` | Selects the result. | These are the only executable definitions. Trigger, agent-router, task, condition, notification, and other palette concepts are planned entries without runtime handlers. Graph v1 migration preserves the legacy snapshot and appends deterministic v2. + +Execution uses immutable publications rather than the mutable editor row. Runs select a pinned publication or the latest published version, then use durable queue claims, leases, bounded quotas, timeout/cancellation propagation, and numbered retry attempts. Expired external attempts with unknown outcomes require operator attention and are not silently replayed. See [Node Flow Durable Execution](./node-flow-durable-execution.md). diff --git a/docs-web/content/docs/architecture-node-flow-durable-execution.mdx b/docs-web/content/docs/architecture-node-flow-durable-execution.mdx new file mode 100644 index 0000000000..e5bc80f167 --- /dev/null +++ b/docs-web/content/docs/architecture-node-flow-durable-execution.mdx @@ -0,0 +1,7 @@ +# Node Flow Durable Execution + +Node flows execute immutable published snapshots. A run explicitly pins a published version or follows the latest published version; later edits cannot change a pinned run. + +Runs are durably queued and leased with bounded global and project concurrency. Node attempts retain attempt number, executor and invocation identity, artifact digest, redacted payloads, credential ids, failure class, and retry decision. Retryable failures use bounded exponential backoff and jitter, while cancellation and timeout signals propagate to provider and HTTP work. + +On restart, expired pre-invocation work is safely requeued. Work with an external invocation and an unknown outcome moves to `attention_required` and is never silently replayed. Credential values are resolved only for the active node and are not retained in run history or diagnostics. diff --git a/docs-web/content/docs/architecture-node-flows.mdx b/docs-web/content/docs/architecture-node-flows.mdx index 493c84b123..31bf1274aa 100644 --- a/docs-web/content/docs/architecture-node-flows.mdx +++ b/docs-web/content/docs/architecture-node-flows.mdx @@ -14,3 +14,5 @@ Node flows are project-owned, versioned Graph v2 workflows. | `output` | Selects the result. | These are the only executable definitions. Trigger, agent-router, task, condition, notification, and other palette concepts are planned entries without runtime handlers. Graph v1 migration preserves the legacy snapshot and appends deterministic v2. + +Execution uses immutable publications rather than the mutable editor row. Runs select a pinned publication or the latest published version, then use durable queue claims, leases, bounded quotas, timeout/cancellation propagation, and numbered retry attempts. Expired external attempts with unknown outcomes require operator attention and are not silently replayed. See [Node Flow Durable Execution](/docs/architecture-node-flow-durable-execution). diff --git a/docs-web/content/docs/registry.ts b/docs-web/content/docs/registry.ts index 522ed80858..637afe7f3a 100644 --- a/docs-web/content/docs/registry.ts +++ b/docs-web/content/docs/registry.ts @@ -108,6 +108,7 @@ export type DocsSlug = | 'user-dashboard-custom-dashboards' | 'architecture-custom-dashboard-foundation' | 'architecture-managed-container-runtime' + | 'architecture-node-flow-durable-execution' | 'architecture-node-flow-foundation' | 'architecture-node-flows' | 'architecture-speech-input' @@ -858,6 +859,13 @@ export const docsRegistry: Record = { title: "Managed Container Runtime", description: "The managed container runtime removes first-invocation Docker builds while keeping provider binaries local to each user's Docker host.", }, + 'architecture-node-flow-durable-execution': { + id: 'architecture-node-flow-durable-execution', + path: '/docs/architecture-node-flow-durable-execution', + section: 'Architecture', + title: "Node Flow Durable Execution", + description: "Node flows execute immutable published snapshots. A run explicitly pins a published version or follows the latest published version; later edits cannot change a pinned run.", + }, 'architecture-node-flow-foundation': { id: 'architecture-node-flow-foundation', path: '/docs/architecture-node-flow-foundation', @@ -1001,6 +1009,7 @@ export const orderedDocs: DocsRegistryEntry[] = [ docsRegistry['user-dashboard-custom-dashboards'], docsRegistry['architecture-custom-dashboard-foundation'], docsRegistry['architecture-managed-container-runtime'], + docsRegistry['architecture-node-flow-durable-execution'], docsRegistry['architecture-node-flow-foundation'], docsRegistry['architecture-node-flows'], docsRegistry['architecture-speech-input'], diff --git a/docs-web/content/docs/user-dashboard-scheduler.mdx b/docs-web/content/docs/user-dashboard-scheduler.mdx index aeda9453e7..3331436bb2 100644 --- a/docs-web/content/docs/user-dashboard-scheduler.mdx +++ b/docs-web/content/docs/user-dashboard-scheduler.mdx @@ -22,11 +22,13 @@ Each scheduler entry has a **target** — the thing that runs when it fires: | **Message** | Posts a project message (for example, a recurring planning or status prompt). | | **Memory remediation** | Runs the long-term memory cleanup workflow on a schedule. | -Node-flow entries store `nodeFlowTarget = { flowId, input?, flowVersion? }` inside the existing +Node-flow entries store `nodeFlowTarget = { flowId, input?, versionSelection }` inside the existing target JSON payload, validate that the flow belongs to the selected project, and run through the node-flow runtime with scheduler trigger metadata when due. Blank dashboard input is omitted, and supplied input must be a JSON object. +Choose a pinned published version when every occurrence must execute the same immutable snapshot, or latest published when each occurrence should pick up the newest publication. A legacy `flowVersion` is treated as a pinned version and affects execution, not just audit metadata. + The backend scheduler contract also supports agent-created wakeups. Agent wakeups are stored in the target JSON payload with `origin` and `source` set to `agent_scheduler`, plus `createdByAgentId` when the creating agent provides it. Agent wakeups post diff --git a/docs-web/routes/docs.architecture-node-flow-durable-execution.lazy.tsx b/docs-web/routes/docs.architecture-node-flow-durable-execution.lazy.tsx new file mode 100644 index 0000000000..033f986814 --- /dev/null +++ b/docs-web/routes/docs.architecture-node-flow-durable-execution.lazy.tsx @@ -0,0 +1,11 @@ +import { createLazyFileRoute } from '@tanstack/react-router' +import ArchitectureNodeFlowDurableExecutionContent from '../content/docs/architecture-node-flow-durable-execution.mdx' +import { DocsPage } from '../components/docs/DocsPage' + +export const Route = createLazyFileRoute('/docs/architecture-node-flow-durable-execution')({ + component: () => ( + + + + ) +}) diff --git a/docs-web/user/dashboard/scheduler.md b/docs-web/user/dashboard/scheduler.md index c2ebc68e24..34481144bb 100644 --- a/docs-web/user/dashboard/scheduler.md +++ b/docs-web/user/dashboard/scheduler.md @@ -22,11 +22,13 @@ Each scheduler entry has a **target** — the thing that runs when it fires: | **Message** | Posts a project message (for example, a recurring planning or status prompt). | | **Memory remediation** | Runs the long-term memory cleanup workflow on a schedule. | -Node-flow entries store `nodeFlowTarget = { flowId, input?, flowVersion? }` inside the existing +Node-flow entries store `nodeFlowTarget = { flowId, input?, versionSelection }` inside the existing target JSON payload, validate that the flow belongs to the selected project, and run through the node-flow runtime with scheduler trigger metadata when due. Blank dashboard input is omitted, and supplied input must be a JSON object. +Choose a pinned published version when every occurrence must execute the same immutable snapshot, or latest published when each occurrence should pick up the newest publication. A legacy `flowVersion` is treated as a pinned version and affects execution, not just audit metadata. + The backend scheduler contract also supports agent-created wakeups. Agent wakeups are stored in the target JSON payload with `origin` and `source` set to `agent_scheduler`, plus `createdByAgentId` when the creating agent provides it. Agent wakeups post diff --git a/docs/SUMMARY.md b/docs/SUMMARY.md index 5e0a632d23..933e7c124d 100644 --- a/docs/SUMMARY.md +++ b/docs/SUMMARY.md @@ -39,6 +39,7 @@ - [Agent Knowledge Base](./architecture/agent-knowledge-base.md) - [Node Flow Foundation](./architecture/node-flow-foundation.md) - [Node Flows](./architecture/node-flows.md) +- [Node Flow Durable Execution](./architecture/node-flow-durable-execution.md) - [Custom Dashboard Foundation](./architecture/custom-dashboard-foundation.md) - [Memory Claims and Evidence](./architecture/memory-claims.md) - [Speech Input Architecture](./architecture/speech-input.md) diff --git a/docs/architecture/node-flow-durable-execution.md b/docs/architecture/node-flow-durable-execution.md new file mode 100644 index 0000000000..334447c9de --- /dev/null +++ b/docs/architecture/node-flow-durable-execution.md @@ -0,0 +1,15 @@ +# Node Flow Durable Execution + +Node-flow execution is publication based. Saving a flow appends an immutable version and publication containing the normalized graph and an immutable execution-policy snapshot. Manual, MCP, and scheduled callers select either `{ mode: "pinned", version: N }` or `{ mode: "latest_published" }`; the runtime never executes the mutable `node_flows` graph. + +## Durable lifecycle + +Runs move through `queued`, `running`, `approval_waiting`, `retry_waiting`, `attention_required`, and terminal `succeeded`, `failed`, or `cancelled` states. A queue claim assigns an executor, lease expiry, and heartbeat. Global and per-project limits bound active claims. Cancellation and node timeouts propagate through `AbortSignal`. + +Every node execution creates a numbered attempt with executor identity, optional execution invocation id, SHA-256 output digest, redacted input/output, credential ids, failure classification, and retry decision. Retryable timeout, quota, and transient failures use the publication policy's bounded exponential backoff and jitter. Credential values are resolved only at the node boundary and are never written to run, attempt, invocation, or diagnostic records. + +## Recovery contract + +Startup recovery scans queued and waiting work plus running work with expired leases. A pre-invocation attempt can be requeued safely without inserting a duplicate attempt. An expired attempt with an invocation id has an unknown externally observable outcome and moves to `attention_required`; Code UX does not silently replay it. Approval- and retry-waiting runs retain their durable state until their prerequisite becomes actionable. + +The relevant tables are `node_flow_publications`, `node_flow_runs`, `node_flow_node_runs`, and `node_flow_node_attempts`. Attempt history is available at `GET /api/node-flow-runs/:runId/attempts` and contains only redacted payloads and credential identifiers. diff --git a/docs/architecture/node-flows.md b/docs/architecture/node-flows.md index 6268a53066..bc355cf310 100644 --- a/docs/architecture/node-flows.md +++ b/docs/architecture/node-flows.md @@ -11,10 +11,12 @@ Node-flow persistence is owned by `NodeFlowRepository` and stored in SQLite: | Table | Purpose | | --- | --- | | `node_flows` | Current project-scoped flow record: id, project id, title, description, normalized `graph_json`, current version, and timestamps. | -| `node_flow_versions` | Immutable snapshots written on create and every update. Versions keep the graph saved at that point, even though current runtime execution uses the latest flow record. | +| `node_flow_versions` | Immutable edit snapshots written on create and every update. | +| `node_flow_publications` | Immutable executable graph and execution-policy snapshots selected by pinned or latest-published runs. | | `node_flow_agent_skills` | Agent attachment table keyed by flow and agent preset. It stores the skill display name and description used when exposing the flow as a repeatable agent capability. | | `node_flow_runs` | Flow run records with status, version, trigger type, redacted trigger payload, redacted input/output, error message, timestamps, and optional `execution_invocation_id`. | | `node_flow_node_runs` | Per-node run records with status, node id, redacted input/output, error message, timestamps, and optional `execution_invocation_id`. | +| `node_flow_node_attempts` | Numbered attempts with executor/invocation identity, artifact digest, credential ids, redacted payloads, failure class, and retry decision. | All graphs, widget schemas, run inputs, outputs, and trigger payloads are stored as JSON text and hydrated into typed contracts at the repository boundary. Flow, version, run, and attachment records belong to a project. Agent attachment operations verify that the target agent preset belongs to the same project as the flow. @@ -37,7 +39,7 @@ Dashboard-only editable canvas state lives in `dashboard/src/v2/lib/nodes-canvas ## Runtime -`NodeFlowRuntimeService.runFlow(projectId, flowId, input, options)` revalidates the saved graph, checks project ownership, and then executes nodes in the validator's topological order. +`NodeFlowRuntimeService.runFlow(projectId, flowId, input, options)` resolves an explicit pinned or latest-published snapshot, revalidates that immutable graph, claims a durable lease, and executes nodes in topological order. See [Node Flow Durable Execution](./node-flow-durable-execution.md) for queue, retry, lease, recovery, quota, and redaction guarantees. Runtime-supported node types are: @@ -75,7 +77,7 @@ Cancellation records cancelled node rows for the current and remaining nodes. At ## Scheduling -Scheduler entries with `targetType: "node_flow"` persist `nodeFlowTarget = { flowId, input?, flowVersion? }` inside `scheduler_entries.target_json`. Ownership is validated when entries are created or updated and again before due-run execution. +Scheduler entries with `targetType: "node_flow"` persist an explicit `versionSelection`: pinned schedules continue to execute version N after N+1 is published, while latest-published schedules resolve the newest publication at dispatch time. Legacy `flowVersion` values normalize to pinned selection and are executable semantics, not audit-only metadata. Ownership is validated when entries are created or updated and again before due-run execution. Due runs call `NodeFlowRuntimeService.runFlow` with `triggerType = "scheduler"` and trigger payload metadata for the scheduler entry id, scheduled occurrence time, target type, and persisted flow version when present. Node-flow schedules advance only when `runFlow` returns a run status of `succeeded`. Returned `failed` or `cancelled` runs mark the scheduler entry `failed` with the run error and still count the attempted occurrence in `lastRunAt` and `runCount`; runtime startup rejections mark failure without creating a false successful schedule run. diff --git a/docs/dashboard/scheduler.md b/docs/dashboard/scheduler.md index b7f31b8dcd..583234204c 100644 --- a/docs/dashboard/scheduler.md +++ b/docs/dashboard/scheduler.md @@ -78,7 +78,7 @@ The target payload keys are: - `chatTarget`: `{ bodyMarkdown, threadId?, title?, connectionId? }` - `memoryRemediationTarget`: `{ mode, source? }` - `taskTarget`: `{ taskId, provider?, origin: "agent_scheduler", source: "agent_scheduler", createdByAgentId? }` -- `nodeFlowTarget`: `{ flowId, input?, flowVersion? }` +- `nodeFlowTarget`: `{ flowId, input?, versionSelection }`; legacy `flowVersion` normalizes to pinned selection - `agentWakeupTarget`: `{ bodyMarkdown, threadId?, title?, connectionId?, origin: "agent_scheduler", source: "agent_scheduler", createdByAgentId? }` `node_flow` entries keep their flow id and optional input in `target_json`; ownership is checked when entries are created or updated and again before due-run execution. The persisted `flowVersion` is target metadata and is passed in scheduler trigger payloads for auditability; the current runtime executes through the latest node-flow runtime API. Due-run handling treats the returned node-flow run status as authoritative: only `succeeded` advances the schedule as successful, while `failed` and `cancelled` mark the scheduler entry `failed`, persist the run error, and record the attempted occurrence in `lastRunAt` and `runCount`. `agent_wakeup` and `task` entries always normalize `origin` and `source` to `agent_scheduler` in `target_json`. When the creator supplies `createdByAgentId`, it is preserved with the target payload for later authorization, audit, and notification work. Existing sprint, quicksprint, chat, memory remediation, recurrence, pause/resume, and `after_sprint_end` anchor rows continue to hydrate from the same JSON payload without a schema migration. @@ -156,7 +156,7 @@ For sprint targets, failures from either automatic planning or direct orchestrat ### Node-Flow Schedules -Node-flow schedules use `targetType: "node_flow"` and `nodeFlowTarget = { flowId, input?, flowVersion? }`. +Node-flow schedules use `targetType: "node_flow"` and `nodeFlowTarget = { flowId, input?, versionSelection }`. A pinned selection always executes that published graph and policy snapshot after newer versions are published; `latest_published` resolves the newest publication per occurrence. Behavior: diff --git a/docs/index.md b/docs/index.md index 201c31ab40..21d4ab0ebf 100644 --- a/docs/index.md +++ b/docs/index.md @@ -66,6 +66,7 @@ Use this page as the main entrypoint. 28. [Agent Knowledge Base](./architecture/agent-knowledge-base.md) 29. [Node Flow Foundation](./architecture/node-flow-foundation.md) 30. [Node Flows](./architecture/node-flows.md) +31. [Node Flow Durable Execution](./architecture/node-flow-durable-execution.md) 31. [Custom Dashboard Foundation](./architecture/custom-dashboard-foundation.md) 32. [Memory Claims and Evidence](./architecture/memory-claims.md) 33. [Speech Input Architecture](./architecture/speech-input.md) @@ -158,6 +159,7 @@ Use this page as the main entrypoint. - [Agent Knowledge Base](./architecture/agent-knowledge-base.md) - [Node Flow Foundation](./architecture/node-flow-foundation.md) - [Node Flows](./architecture/node-flows.md) +- [Node Flow Durable Execution](./architecture/node-flow-durable-execution.md) - [Custom Dashboard Foundation](./architecture/custom-dashboard-foundation.md) - [Memory Claims and Evidence](./architecture/memory-claims.md) - [Speech Input Architecture](./architecture/speech-input.md) diff --git a/src/app/dependency-factory/dashboard-factory.ts b/src/app/dependency-factory/dashboard-factory.ts index af70fdec29..da4d85983f 100644 --- a/src/app/dependency-factory/dashboard-factory.ts +++ b/src/app/dependency-factory/dashboard-factory.ts @@ -31,6 +31,7 @@ import { SpeechSynthesisService } from "../../services/speech-synthesis-service. import { SpeechModelManager } from "../../services/speech-model-manager.js"; import { NodeFlowRuntimeService } from "../../services/node-flow-runtime-service.js"; import { NodeFlowService } from "../../services/node-flow-service.js"; +import { NodeFlowRecoveryService } from "../../services/node-flows/node-flow-recovery-service.js"; import { resolveEffectiveDashboardSettings } from "../../services/settings-resolution-service.js"; export interface DashboardDependencies { @@ -230,6 +231,9 @@ export function createDashboardDependencies( credentialBroker: coreDeps.credentialBroker, getDashboardSettings: (projectId) => resolveDashboardSettings({ projectId }), }); + if (coreDeps.nodeFlowRepository) { + new NodeFlowRecoveryService(coreDeps.nodeFlowRepository).recover(); + } const nodeFlowService = new NodeFlowService(coreDeps.nodeFlowRepository, nodeFlowRuntimeService); const activityCacheService = new ActivityCacheService( diff --git a/src/contracts/node-flow-execution-policy-types.ts b/src/contracts/node-flow-execution-policy-types.ts new file mode 100644 index 0000000000..151a316e9e --- /dev/null +++ b/src/contracts/node-flow-execution-policy-types.ts @@ -0,0 +1,45 @@ +export type NodeFlowVersionSelection = + | { mode: "latest_published" } + | { mode: "pinned"; version: number }; + +export type NodeFlowFailureClassification = + | "cancelled" + | "timeout" + | "quota" + | "validation" + | "credential" + | "transient" + | "permanent" + | "unknown_side_effect"; + +export interface NodeFlowRetryPolicySnapshot { + maxAttempts: number; + backoffMs: number; + maxBackoffMs: number; + jitterRatio: number; + retryableClasses: NodeFlowFailureClassification[]; +} + +export interface NodeFlowExecutionPolicySnapshot { + maxConcurrentRuns: number; + maxConcurrentRunsPerProject: number; + leaseDurationMs: number; + heartbeatIntervalMs: number; + defaultTimeoutMs: number; + retry: NodeFlowRetryPolicySnapshot; +} + +export const DEFAULT_NODE_FLOW_EXECUTION_POLICY: Readonly = Object.freeze({ + maxConcurrentRuns: 4, + maxConcurrentRunsPerProject: 2, + leaseDurationMs: 30_000, + heartbeatIntervalMs: 10_000, + defaultTimeoutMs: 60_000, + retry: Object.freeze({ + maxAttempts: 1, + backoffMs: 500, + maxBackoffMs: 30_000, + jitterRatio: 0.2, + retryableClasses: Object.freeze(["timeout", "quota", "transient"]) as NodeFlowFailureClassification[], + }), +}); diff --git a/src/contracts/node-flow-types.ts b/src/contracts/node-flow-types.ts index 8e19d7ab7a..cc6471c28b 100644 --- a/src/contracts/node-flow-types.ts +++ b/src/contracts/node-flow-types.ts @@ -160,6 +160,17 @@ export interface NodeFlowVersionRecord { createdAt: string; } +export interface NodeFlowPublicationRecord { + id: string; + flowId: string; + projectId: string; + version: number; + graph: NodeFlowGraph; + policy: import("./node-flow-execution-policy-types.js").NodeFlowExecutionPolicySnapshot; + publishedBy: string; + createdAt: string; +} + export interface CreateNodeFlowInput { id?: string; title: string; @@ -202,15 +213,25 @@ export interface AttachNodeFlowSkillInput { description?: string; } -export type NodeFlowRunStatus = "queued" | "running" | "succeeded" | "failed" | "cancelled"; -export type NodeFlowNodeRunStatus = "pending" | "running" | "succeeded" | "failed" | "skipped" | "cancelled"; +export type NodeFlowRunStatus = + | "queued" | "running" | "approval_waiting" | "retry_waiting" + | "attention_required" | "succeeded" | "failed" | "cancelled"; +export type NodeFlowNodeRunStatus = + | "pending" | "running" | "retry_waiting" | "attention_required" + | "succeeded" | "failed" | "skipped" | "cancelled"; export interface NodeFlowRunRecord { id: string; flowId: string; projectId: string; version: number; + publicationId: string | null; status: NodeFlowRunStatus; + policy: import("./node-flow-execution-policy-types.js").NodeFlowExecutionPolicySnapshot; + leaseOwner: string | null; + leaseExpiresAt: string | null; + heartbeatAt: string | null; + cancelRequestedAt: string | null; executionInvocationId: string | null; triggerType: string; triggerPayload: NodeFlowJsonObject | null; @@ -240,6 +261,27 @@ export interface NodeFlowNodeRunRecord { updatedAt: string; } +export interface NodeFlowNodeAttemptRecord { + id: string; + runId: string; + nodeRunId: string; + nodeId: string; + attemptNumber: number; + status: NodeFlowNodeRunStatus; + executorId: string; + invocationId: string | null; + artifactDigest: string | null; + input: NodeFlowJsonObject | null; + output: NodeFlowJsonObject | null; + credentialIds: string[]; + failureClassification: import("./node-flow-execution-policy-types.js").NodeFlowFailureClassification | null; + retryDecision: "retry" | "stop" | "attention_required" | null; + errorMessage: string | null; + startedAt: string; + finishedAt: string | null; + createdAt: string; +} + export interface NodeFlowListResponse { flows: NodeFlowRecord[]; } @@ -256,6 +298,8 @@ export interface CreateNodeFlowRunInput { flowId: string; projectId: string; version: number; + publicationId?: string | null; + policy?: import("./node-flow-execution-policy-types.js").NodeFlowExecutionPolicySnapshot; status?: NodeFlowRunStatus; executionInvocationId?: string | null; triggerType?: string; @@ -274,6 +318,10 @@ export interface UpdateNodeFlowRunInput { errorMessage?: string | null; startedAt?: string | null; finishedAt?: string | null; + leaseOwner?: string | null; + leaseExpiresAt?: string | null; + heartbeatAt?: string | null; + cancelRequestedAt?: string | null; } export interface CreateNodeFlowNodeRunInput { @@ -304,10 +352,13 @@ export interface RunNodeFlowOptions { triggerType?: string; triggerPayload?: NodeFlowJsonObject; signal?: AbortSignal; + versionSelection?: import("./node-flow-execution-policy-types.js").NodeFlowVersionSelection; + executorId?: string; } export interface NodeFlowRunSummaryResponse { run: NodeFlowRunRecord; nodeRuns: NodeFlowNodeRunRecord[]; + attempts?: NodeFlowNodeAttemptRecord[]; output: NodeFlowJsonObject | null; } diff --git a/src/contracts/scheduler-types.ts b/src/contracts/scheduler-types.ts index 366f3d9968..12bcc88f64 100644 --- a/src/contracts/scheduler-types.ts +++ b/src/contracts/scheduler-types.ts @@ -1,6 +1,7 @@ import type { QuicksprintExecutionInput } from "./quicksprint-types.js"; import type { ProviderId } from "./app-types.js"; import type { NodeFlowJsonObject } from "./node-flow-types.js"; +import type { NodeFlowVersionSelection } from "./node-flow-execution-policy-types.js"; export type ScheduleTargetType = "sprint" | "quicksprint" | "chat" | "memory_remediation" | "agent_wakeup" | "task" | "node_flow"; export type ScheduleStatus = "scheduled" | "paused" | "completed" | "failed" | "cancelled"; @@ -79,6 +80,7 @@ export interface ScheduleNodeFlowTarget { flowId: string; input?: NodeFlowJsonObject; flowVersion?: number; + versionSelection?: NodeFlowVersionSelection; } export interface SchedulerEntryRecord { diff --git a/src/mcp/management/node-flow-actions.ts b/src/mcp/management/node-flow-actions.ts index cdb6cd138e..a35437f0b5 100644 --- a/src/mcp/management/node-flow-actions.ts +++ b/src/mcp/management/node-flow-actions.ts @@ -16,6 +16,7 @@ import type { NodeFlowService } from "../../services/node-flow-service.js"; import { managementValidationError, parseOptionalObject, + parseOptionalNumber, parseOptionalString, parseRequiredObject, parseRequiredString, @@ -139,8 +140,12 @@ export class NodeFlowActions { const projectId = parseRequiredString(payload, "projectId"); const flowId = parseRequiredString(payload, "flowId"); const input = parseOptionalObject(payload, "input") ?? {}; + const flowVersion = parseOptionalNumber(payload, "flowVersion", 1); const result = await this.nodeFlowService.runFlow(projectId, flowId, input, { triggerType: "mcp_management", + versionSelection: flowVersion === undefined + ? { mode: "latest_published" } + : { mode: "pinned", version: Math.floor(flowVersion) }, }); return { result: formatRunSummary(result) }; } diff --git a/src/repositories/db/app-db-migrations.ts b/src/repositories/db/app-db-migrations.ts index da475c1306..d54064de0e 100644 --- a/src/repositories/db/app-db-migrations.ts +++ b/src/repositories/db/app-db-migrations.ts @@ -257,7 +257,54 @@ export function ensureNodeFlowTables(db: DatabaseAdapter): void { ) `); + db.exec(` + CREATE TABLE IF NOT EXISTS node_flow_publications ( + id TEXT PRIMARY KEY, + flow_id TEXT NOT NULL, + project_id TEXT NOT NULL, + version INTEGER NOT NULL, + graph_json TEXT NOT NULL, + policy_json TEXT NOT NULL, + published_by TEXT NOT NULL, + created_at TEXT NOT NULL, + FOREIGN KEY (flow_id) REFERENCES node_flows(id) ON DELETE CASCADE, + FOREIGN KEY (project_id) REFERENCES projects(id) ON DELETE CASCADE, + UNIQUE (flow_id, version) + ) + `); + db.exec(` + CREATE TABLE IF NOT EXISTS node_flow_node_attempts ( + id TEXT PRIMARY KEY, + run_id TEXT NOT NULL, + node_run_id TEXT NOT NULL, + node_id TEXT NOT NULL, + attempt_number INTEGER NOT NULL, + status TEXT NOT NULL, + executor_id TEXT NOT NULL, + invocation_id TEXT, + artifact_digest TEXT, + input_json TEXT, + output_json TEXT, + credential_ids_json TEXT NOT NULL DEFAULT '[]', + failure_classification TEXT, + retry_decision TEXT, + error_message TEXT, + started_at TEXT NOT NULL, + finished_at TEXT, + created_at TEXT NOT NULL, + FOREIGN KEY (run_id) REFERENCES node_flow_runs(id) ON DELETE CASCADE, + FOREIGN KEY (node_run_id) REFERENCES node_flow_node_runs(id) ON DELETE CASCADE, + UNIQUE (run_id, node_id, attempt_number) + ) + `); + ensureColumn(db, "node_flow_runs", "execution_invocation_id", "TEXT"); + ensureColumn(db, "node_flow_runs", "publication_id", "TEXT"); + ensureColumn(db, "node_flow_runs", "policy_json", "TEXT NOT NULL DEFAULT '{}'"); + ensureColumn(db, "node_flow_runs", "lease_owner", "TEXT"); + ensureColumn(db, "node_flow_runs", "lease_expires_at", "TEXT"); + ensureColumn(db, "node_flow_runs", "heartbeat_at", "TEXT"); + ensureColumn(db, "node_flow_runs", "cancel_requested_at", "TEXT"); ensureColumn(db, "node_flow_node_runs", "execution_invocation_id", "TEXT"); ensureIndex(db, "idx_node_flows_project_updated", "node_flows", "project_id, updated_at DESC"); @@ -266,6 +313,10 @@ export function ensureNodeFlowTables(db: DatabaseAdapter): void { ensureIndex(db, "idx_node_flow_runs_flow_created", "node_flow_runs", "flow_id, created_at DESC"); ensureIndex(db, "idx_node_flow_runs_project_created", "node_flow_runs", "project_id, created_at DESC"); ensureIndex(db, "idx_node_flow_node_runs_run_created", "node_flow_node_runs", "run_id, created_at ASC"); + ensureIndex(db, "idx_node_flow_publications_latest", "node_flow_publications", "flow_id, version DESC"); + ensureIndex(db, "idx_node_flow_runs_queue", "node_flow_runs", "status, lease_expires_at, created_at ASC"); + ensureIndex(db, "idx_node_flow_runs_project_status", "node_flow_runs", "project_id, status"); + ensureIndex(db, "idx_node_flow_attempts_run_node", "node_flow_node_attempts", "run_id, node_id, attempt_number"); } interface LegacyNodeFlowRow { diff --git a/src/repositories/node-flow-repository.ts b/src/repositories/node-flow-repository.ts index faf2e5ae48..c3648e3606 100644 --- a/src/repositories/node-flow-repository.ts +++ b/src/repositories/node-flow-repository.ts @@ -12,6 +12,8 @@ import type { NodeFlowJsonObject, NodeFlowJsonValue, NodeFlowNodeRunRecord, + NodeFlowNodeAttemptRecord, + NodeFlowPublicationRecord, NodeFlowRecord, NodeFlowRunRecord, NodeFlowSkillAttachment, @@ -20,6 +22,8 @@ import type { UpdateNodeFlowInput, UpdateNodeFlowRunInput, } from "../contracts/node-flow-types.js"; +import { DEFAULT_NODE_FLOW_EXECUTION_POLICY } from "../contracts/node-flow-execution-policy-types.js"; +import type { NodeFlowExecutionPolicySnapshot, NodeFlowFailureClassification } from "../contracts/node-flow-execution-policy-types.js"; import { migratePersistedNodeFlowGraphs } from "./db/app-db-migrations.js"; interface NodeFlowRow { @@ -59,7 +63,13 @@ interface NodeFlowRunRow { flow_id: string; project_id: string; version: number | string; + publication_id: string | null; + policy_json: string; status: string; + lease_owner: string | null; + lease_expires_at: string | null; + heartbeat_at: string | null; + cancel_requested_at: string | null; execution_invocation_id: string | null; trigger_type: string; trigger_payload_json: string | null; @@ -72,6 +82,19 @@ interface NodeFlowRunRow { updated_at: string; } +interface NodeFlowPublicationRow { + id: string; flow_id: string; project_id: string; version: number | string; + graph_json: string; policy_json: string; published_by: string; created_at: string; +} + +interface NodeFlowAttemptRow { + id: string; run_id: string; node_run_id: string; node_id: string; attempt_number: number | string; + status: string; executor_id: string; invocation_id: string | null; artifact_digest: string | null; + input_json: string | null; output_json: string | null; credential_ids_json: string; + failure_classification: string | null; retry_decision: string | null; error_message: string | null; + started_at: string; finished_at: string | null; created_at: string; +} + interface NodeFlowNodeRunRow { id: string; run_id: string; @@ -98,6 +121,7 @@ export class NodeFlowRepository { ) { this.db = storage.getDatabase(); migratePersistedNodeFlowGraphs(this.db); + this.backfillPublications(); } listFlows(projectId: string): NodeFlowRecord[] { @@ -142,6 +166,7 @@ export class NodeFlowRepository { graphJson, createdAt: now, }); + this.insertPublication(id, projectId, 1, graphJson, DEFAULT_NODE_FLOW_EXECUTION_POLICY, "system"); }); const created = this.requireFlow(id); @@ -173,6 +198,7 @@ export class NodeFlowRepository { graphJson, createdAt: now, }); + this.insertPublication(flowId, current.projectId, nextVersion, graphJson, DEFAULT_NODE_FLOW_EXECUTION_POLICY, "system"); }); const updated = this.requireFlow(flowId); @@ -207,6 +233,26 @@ export class NodeFlowRepository { return row ? this.mapVersionRow(row) : null; } + listPublications(flowId: string): NodeFlowPublicationRecord[] { + this.requireFlow(flowId); + return (this.db.prepare(`SELECT * FROM node_flow_publications WHERE flow_id = ? ORDER BY version DESC`).all(flowId) as unknown as NodeFlowPublicationRow[]) + .map((row) => this.mapPublicationRow(row)); + } + + getPublication(flowId: string, version?: number): NodeFlowPublicationRecord | null { + const row = version === undefined + ? this.db.prepare(`SELECT * FROM node_flow_publications WHERE flow_id = ? ORDER BY version DESC LIMIT 1`).get(flowId) + : this.db.prepare(`SELECT * FROM node_flow_publications WHERE flow_id = ? AND version = ?`).get(flowId, Math.floor(version)); + return row ? this.mapPublicationRow(row as NodeFlowPublicationRow) : null; + } + + publishVersion(flowId: string, version: number, policy: NodeFlowExecutionPolicySnapshot = DEFAULT_NODE_FLOW_EXECUTION_POLICY, publishedBy = "system"): NodeFlowPublicationRecord { + const snapshot = this.getVersion(flowId, version); + if (!snapshot) throw new EntityNotFoundError(`Node flow version not found: ${flowId}@${version}`); + this.insertPublication(flowId, snapshot.projectId, snapshot.version, this.serializeJson(snapshot.graph), policy, publishedBy); + return requireRecord(this.getPublication(flowId, version), "Node flow publication", `${flowId}@${version}`); + } + attachToAgent(flowId: string, input: AttachNodeFlowSkillInput): NodeFlowSkillAttachment { const flow = this.requireFlow(flowId); const agentPresetId = input.agentPresetId?.trim(); @@ -317,14 +363,16 @@ export class NodeFlowRepository { const id = randomUUID(); this.db.prepare(` INSERT INTO node_flow_runs ( - id, flow_id, project_id, version, status, execution_invocation_id, trigger_type, + id, flow_id, project_id, version, publication_id, policy_json, status, execution_invocation_id, trigger_type, trigger_payload_json, input_json, output_json, error_message, started_at, finished_at, created_at, updated_at - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) `).run( id, input.flowId, input.projectId, Math.max(1, Math.floor(input.version)), + input.publicationId ?? null, + this.serializeJson(input.policy ?? DEFAULT_NODE_FLOW_EXECUTION_POLICY), input.status || "running", input.executionInvocationId ?? null, input.triggerType?.trim() || "manual", @@ -350,7 +398,7 @@ export class NodeFlowRepository { output_json = ?, error_message = ?, started_at = ?, - finished_at = ?, + finished_at = ?, lease_owner = ?, lease_expires_at = ?, heartbeat_at = ?, cancel_requested_at = ?, updated_at = ? WHERE id = ? `).run( @@ -360,12 +408,62 @@ export class NodeFlowRepository { input.errorMessage === undefined ? current.errorMessage : input.errorMessage, input.startedAt === undefined ? current.startedAt : input.startedAt, input.finishedAt === undefined ? current.finishedAt : input.finishedAt, + input.leaseOwner === undefined ? current.leaseOwner : input.leaseOwner, + input.leaseExpiresAt === undefined ? current.leaseExpiresAt : input.leaseExpiresAt, + input.heartbeatAt === undefined ? current.heartbeatAt : input.heartbeatAt, + input.cancelRequestedAt === undefined ? current.cancelRequestedAt : input.cancelRequestedAt, now, runId, ); return requireRecord(this.getRun(runId), "Node flow run", runId); } + claimQueuedRun(runId: string, executorId: string, leaseDurationMs: number, now = new Date()): NodeFlowRunRecord | null { + const expiresAt = new Date(now.getTime() + leaseDurationMs).toISOString(); + const result = this.db.prepare(`UPDATE node_flow_runs SET status = 'running', lease_owner = ?, lease_expires_at = ?, heartbeat_at = ?, started_at = COALESCE(started_at, ?), updated_at = ? WHERE id = ? AND status IN ('queued','retry_waiting') AND (lease_expires_at IS NULL OR lease_expires_at <= ?)`) + .run(executorId, expiresAt, now.toISOString(), now.toISOString(), now.toISOString(), runId, now.toISOString()); + return result.changes > 0 ? this.getRun(runId) : null; + } + + heartbeatRun(runId: string, executorId: string, leaseDurationMs: number, now = new Date()): boolean { + const result = this.db.prepare(`UPDATE node_flow_runs SET heartbeat_at = ?, lease_expires_at = ?, updated_at = ? WHERE id = ? AND status = 'running' AND lease_owner = ?`) + .run(now.toISOString(), new Date(now.getTime() + leaseDurationMs).toISOString(), now.toISOString(), runId, executorId); + return result.changes > 0; + } + + countActiveRuns(projectId?: string): number { + const row = projectId + ? this.db.prepare(`SELECT COUNT(*) AS count FROM node_flow_runs WHERE project_id = ? AND status = 'running'`).get(projectId) + : this.db.prepare(`SELECT COUNT(*) AS count FROM node_flow_runs WHERE status = 'running'`).get(); + return toNumber((row as { count: number | string }).count); + } + + listRecoverableRuns(nowIso = new Date().toISOString()): NodeFlowRunRecord[] { + return (this.db.prepare(`SELECT * FROM node_flow_runs WHERE status IN ('queued','retry_waiting','approval_waiting') OR (status = 'running' AND lease_expires_at <= ?) ORDER BY created_at ASC`).all(nowIso) as unknown as NodeFlowRunRow[]).map((row) => this.mapRunRow(row)); + } + + requestCancellation(runId: string): NodeFlowRunRecord { + return this.updateRun(runId, { cancelRequestedAt: new Date().toISOString() }); + } + + createNodeAttempt(input: Omit): NodeFlowNodeAttemptRecord { + const id = randomUUID(); const now = new Date().toISOString(); + this.db.prepare(`INSERT INTO node_flow_node_attempts (id, run_id, node_run_id, node_id, attempt_number, status, executor_id, invocation_id, artifact_digest, input_json, output_json, credential_ids_json, failure_classification, retry_decision, error_message, started_at, finished_at, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`) + .run(id, input.runId, input.nodeRunId, input.nodeId, input.attemptNumber, input.status, input.executorId, input.invocationId, input.artifactDigest, this.serializeNullableJson(input.input), this.serializeNullableJson(input.output), JSON.stringify(input.credentialIds), input.failureClassification, input.retryDecision, input.errorMessage, input.startedAt, input.finishedAt, now); + return requireRecord(this.getNodeAttempt(id), "Node flow node attempt", id); + } + + updateNodeAttempt(id: string, input: Partial>): NodeFlowNodeAttemptRecord { + const current = requireRecord(this.getNodeAttempt(id), "Node flow node attempt", id); + this.db.prepare(`UPDATE node_flow_node_attempts SET status = ?, invocation_id = ?, artifact_digest = ?, output_json = ?, failure_classification = ?, retry_decision = ?, error_message = ?, finished_at = ? WHERE id = ?`) + .run(input.status ?? current.status, input.invocationId === undefined ? current.invocationId : input.invocationId, input.artifactDigest === undefined ? current.artifactDigest : input.artifactDigest, input.output === undefined ? this.serializeNullableJson(current.output) : this.serializeNullableJson(input.output), input.failureClassification === undefined ? current.failureClassification : input.failureClassification, input.retryDecision === undefined ? current.retryDecision : input.retryDecision, input.errorMessage === undefined ? current.errorMessage : input.errorMessage, input.finishedAt === undefined ? current.finishedAt : input.finishedAt, id); + return requireRecord(this.getNodeAttempt(id), "Node flow node attempt", id); + } + + listNodeAttempts(runId: string): NodeFlowNodeAttemptRecord[] { + return (this.db.prepare(`SELECT * FROM node_flow_node_attempts WHERE run_id = ? ORDER BY node_id, attempt_number`).all(runId) as unknown as NodeFlowAttemptRow[]).map((row) => this.mapAttemptRow(row)); + } + createNodeRun(input: CreateNodeFlowNodeRunInput): NodeFlowNodeRunRecord { const run = requireRecord(this.getRun(input.runId), "Node flow run", input.runId); if (run.flowId !== input.flowId || run.projectId !== input.projectId) { @@ -450,6 +548,19 @@ export class NodeFlowRepository { ); } + private insertPublication(flowId: string, projectId: string, version: number, graphJson: string, policy: NodeFlowExecutionPolicySnapshot, publishedBy: string): void { + this.db.prepare(`INSERT OR IGNORE INTO node_flow_publications (id, flow_id, project_id, version, graph_json, policy_json, published_by, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`) + .run(randomUUID(), flowId, projectId, version, graphJson, this.serializeJson(policy), publishedBy, new Date().toISOString()); + } + + private backfillPublications(): void { + const versions = this.db.prepare(`SELECT flow_id, project_id, version, graph_json, created_at FROM node_flow_versions`).all() as Array<{ flow_id: string; project_id: string; version: number | string; graph_json: string; created_at: string }>; + for (const version of versions) { + this.db.prepare(`INSERT OR IGNORE INTO node_flow_publications (id, flow_id, project_id, version, graph_json, policy_json, published_by, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`) + .run(randomUUID(), version.flow_id, version.project_id, toNumber(version.version), version.graph_json, this.serializeJson(DEFAULT_NODE_FLOW_EXECUTION_POLICY), "migration", version.created_at); + } + } + private requireProject(projectId: string): void { requireRecord(this.db.prepare(`SELECT id FROM projects WHERE id = ?`).get(projectId), "Project", projectId); } @@ -489,6 +600,11 @@ export class NodeFlowRepository { return row ? this.mapNodeRunRow(row) : null; } + private getNodeAttempt(id: string): NodeFlowNodeAttemptRecord | null { + const row = this.db.prepare(`SELECT * FROM node_flow_node_attempts WHERE id = ?`).get(id) as NodeFlowAttemptRow | undefined; + return row ? this.mapAttemptRow(row) : null; + } + private requireTitle(title: string | undefined): string { const normalized = title?.trim(); if (!normalized) { @@ -527,6 +643,11 @@ export class NodeFlowRepository { } } + private parsePolicy(value: string): NodeFlowExecutionPolicySnapshot { + try { return { ...DEFAULT_NODE_FLOW_EXECUTION_POLICY, ...JSON.parse(value) as NodeFlowExecutionPolicySnapshot }; } + catch { return { ...DEFAULT_NODE_FLOW_EXECUTION_POLICY }; } + } + private isJsonValue(value: unknown): value is NodeFlowJsonValue { if (value === null) { return true; @@ -594,7 +715,13 @@ export class NodeFlowRepository { flowId: row.flow_id, projectId: row.project_id, version: toNumber(row.version), + publicationId: row.publication_id, status: row.status as NodeFlowRunRecord["status"], + policy: this.parsePolicy(row.policy_json), + leaseOwner: row.lease_owner, + leaseExpiresAt: row.lease_expires_at, + heartbeatAt: row.heartbeat_at, + cancelRequestedAt: row.cancel_requested_at, executionInvocationId: row.execution_invocation_id, triggerType: row.trigger_type, triggerPayload: this.parseObject(row.trigger_payload_json), @@ -627,6 +754,16 @@ export class NodeFlowRepository { }; } + private mapPublicationRow(row: NodeFlowPublicationRow): NodeFlowPublicationRecord { + return { id: row.id, flowId: row.flow_id, projectId: row.project_id, version: toNumber(row.version), graph: this.parseGraph(row.graph_json), policy: this.parsePolicy(row.policy_json), publishedBy: row.published_by, createdAt: row.created_at }; + } + + private mapAttemptRow(row: NodeFlowAttemptRow): NodeFlowNodeAttemptRecord { + let credentialIds: string[] = []; + try { const parsed = JSON.parse(row.credential_ids_json) as unknown; if (Array.isArray(parsed)) credentialIds = parsed.filter((item): item is string => typeof item === "string"); } catch { /* legacy row */ } + return { id: row.id, runId: row.run_id, nodeRunId: row.node_run_id, nodeId: row.node_id, attemptNumber: toNumber(row.attempt_number), status: row.status as NodeFlowNodeRunRecord["status"], executorId: row.executor_id, invocationId: row.invocation_id, artifactDigest: row.artifact_digest, input: this.parseObject(row.input_json), output: this.parseObject(row.output_json), credentialIds, failureClassification: row.failure_classification as NodeFlowFailureClassification | null, retryDecision: row.retry_decision as NodeFlowNodeAttemptRecord["retryDecision"], errorMessage: row.error_message, startedAt: row.started_at, finishedAt: row.finished_at, createdAt: row.created_at }; + } + private publishProjectStructureRefresh(projectId: string): void { this.realtimeNotifier?.scheduleProjectStructureRefresh(projectId, { includeProjects: false }); } diff --git a/src/repositories/scheduler-repository.ts b/src/repositories/scheduler-repository.ts index fff9c56e63..a073fa9159 100644 --- a/src/repositories/scheduler-repository.ts +++ b/src/repositories/scheduler-repository.ts @@ -364,8 +364,17 @@ export class SchedulerRepository { target.input = normalizedInput; } const flowVersion = this.normalizeOptionalPositiveInteger(input.nodeFlowTarget?.flowVersion, "nodeFlowTarget.flowVersion"); - if (flowVersion !== undefined) { + const versionSelection = input.nodeFlowTarget?.versionSelection; + if (versionSelection?.mode === "pinned") { + target.versionSelection = { mode: "pinned", version: this.normalizeOptionalPositiveInteger(versionSelection.version, "nodeFlowTarget.versionSelection.version")! }; + target.flowVersion = target.versionSelection.version; + } else if (versionSelection?.mode === "latest_published") { + target.versionSelection = { mode: "latest_published" }; + } else if (flowVersion !== undefined) { target.flowVersion = flowVersion; + target.versionSelection = { mode: "pinned", version: flowVersion }; + } else { + target.versionSelection = { mode: "latest_published" }; } return { nodeFlowTarget: target }; } diff --git a/src/server/node-flow-routes.ts b/src/server/node-flow-routes.ts index 4d9a3805cd..a608686dd4 100644 --- a/src/server/node-flow-routes.ts +++ b/src/server/node-flow-routes.ts @@ -66,6 +66,7 @@ export function registerNodeFlowRoutes(app: Express, deps: DashboardDependencies input?: Record; triggerType?: string; triggerPayload?: NodeFlowJsonObject; + flowVersion?: number; }; const result = await requireNodeFlowService(deps).runFlow( requireTrimmedString(body.projectId, "projectId"), @@ -74,6 +75,9 @@ export function registerNodeFlowRoutes(app: Express, deps: DashboardDependencies { triggerType: body.triggerType, triggerPayload: body.triggerPayload, + versionSelection: body.flowVersion === undefined + ? { mode: "latest_published" } + : { mode: "pinned", version: body.flowVersion }, }, ); res.status(201).json(result); @@ -121,4 +125,7 @@ export function registerNodeFlowRoutes(app: Express, deps: DashboardDependencies app.get("/api/node-flow-runs/:runId/node-runs", syncRoute((req, res) => { res.json(requireNodeFlowService(deps).listNodeRuns(requireTrimmedString(req.params.runId, "runId"))); })); + app.get("/api/node-flow-runs/:runId/attempts", syncRoute((req, res) => { + res.json(requireNodeFlowService(deps).listNodeAttempts(requireTrimmedString(req.params.runId, "runId"))); + })); } diff --git a/src/services/node-flow-runtime-service.ts b/src/services/node-flow-runtime-service.ts index 92c16e1afb..57a8f95138 100644 --- a/src/services/node-flow-runtime-service.ts +++ b/src/services/node-flow-runtime-service.ts @@ -11,6 +11,11 @@ import type { ProviderExecutionService } from "./provider-execution-service.js"; import type { CliProviderId } from "../infrastructure/providers/cli/provider-command-specs.js"; import type { ProviderRunResult } from "../infrastructure/providers/cli/provider-runner.js"; import type { CredentialBroker } from "./credentials/credential-broker.js"; +import { NodeFlowPublicationService } from "./node-flows/node-flow-publication-service.js"; +import { NodeFlowQueueService } from "./node-flows/node-flow-queue-service.js"; +import { NodeFlowAttemptService } from "./node-flows/node-flow-attempt-service.js"; +import { NodeFlowLeaseService } from "./node-flows/node-flow-lease-service.js"; +import type { NodeFlowFailureClassification } from "../contracts/node-flow-execution-policy-types.js"; import { buildProviderInvocationWorkspaceOptions } from "../infrastructure/providers/cli/invocation-workspace-preparer.js"; import type { DashboardSettings, @@ -55,6 +60,8 @@ interface RuntimeContext { predecessors: Map; descendants: Map>; options: RunNodeFlowOptions; + executorId: string; + currentAttemptId?: string; } interface NodeExecutionResult { @@ -79,7 +86,9 @@ export class NodeFlowRuntimeService { throw new ValidationError("Node flow does not belong to the requested project."); } - const { graph, executionOrder } = normalizeNodeFlowGraph(flow.graph); + const selection = options.versionSelection ?? { mode: "latest_published" }; + const publication = new NodeFlowPublicationService(this.deps.nodeFlowRepository).resolve(flow.id, selection); + const { graph, executionOrder } = normalizeNodeFlowGraph(publication.graph); this.requireSupportedNodes(graph); const sanitizedInput = maskSecrets(input); const startedAt = new Date().toISOString(); @@ -92,29 +101,39 @@ export class NodeFlowRuntimeService { }); this.deps.executionRepository.appendExecutionInvocationMessage(parentInvocation.id, { role: "system", - contentMarkdown: `Node flow run started for flow ${flow.id} at version ${flow.version}.`, + contentMarkdown: `Node flow run started for flow ${flow.id} at published version ${publication.version}.`, metadata: { flowId: flow.id, - flowVersion: flow.version, + flowVersion: publication.version, + publicationId: publication.id, }, }); const run = this.deps.nodeFlowRepository.createRun({ flowId: flow.id, projectId, - version: flow.version, - status: "running", + version: publication.version, + publicationId: publication.id, + policy: publication.policy, + status: "queued", executionInvocationId: parentInvocation.id, triggerType: options.triggerType, triggerPayload: options.triggerPayload ? maskSecrets(options.triggerPayload) : null, input: sanitizedInput, - startedAt, + startedAt: null, }); + const executorId = options.executorId?.trim() || `node-flow-runtime:${process.pid}:${randomUUID()}`; + const claimedRun = new NodeFlowQueueService(this.deps.nodeFlowRepository).claim(run, executorId); + const leaseService = new NodeFlowLeaseService(this.deps.nodeFlowRepository); + const heartbeatTimer = setInterval(() => { + leaseService.heartbeat(claimedRun.id, executorId, publication.policy.leaseDurationMs); + }, publication.policy.heartbeatIntervalMs); + heartbeatTimer.unref?.(); const context: RuntimeContext = { projectId, flowId: flow.id, - runId: run.id, + runId: claimedRun.id, graph, order: executionOrder, input, @@ -122,6 +141,7 @@ export class NodeFlowRuntimeService { predecessors: buildPredecessors(graph), descendants: buildDescendants(graph), options, + executorId, }; const blockedNodes = new Set(); @@ -136,7 +156,7 @@ export class NodeFlowRuntimeService { } if (options.signal?.aborted) { terminalStatus = "cancelled"; - terminalError = "Node flow run was cancelled."; + terminalError ??= "Node flow run was cancelled."; await this.persistSkippedNode(context, node, "cancelled", terminalError); for (const remainingNodeId of executionOrder.slice(executionOrder.indexOf(nodeId) + 1)) { const remaining = graph.nodes.find((candidate) => candidate.id === remainingNodeId); @@ -155,28 +175,59 @@ export class NodeFlowRuntimeService { continue; } + const nodeInput = maskSecrets(this.buildNodeInput(context, node.id)); const nodeRun = this.deps.nodeFlowRepository.createNodeRun({ runId: run.id, flowId: flow.id, projectId, nodeId: node.id, status: "running", - input: maskSecrets(this.buildNodeInput(context, node.id)), + input: nodeInput, startedAt: new Date().toISOString(), }); - try { + const attemptService = new NodeFlowAttemptService(this.deps.nodeFlowRepository); + const retryPolicy = { + ...publication.policy.retry, + ...(node.policy?.retry ?? {}), + }; + let attemptNumber = 0; + while (attemptNumber < retryPolicy.maxAttempts) { + attemptNumber += 1; + const attempt = attemptService.start(nodeRun, executorId, nodeInput, (node.credentialBindings ?? []).map((binding) => binding.credentialId)); + context.currentAttemptId = attempt.id; + const timeoutMs = node.policy?.timeout?.timeoutMs ?? publication.policy.defaultTimeoutMs; + const timeoutController = new AbortController(); + const parentAbort = (): void => timeoutController.abort(options.signal?.reason); + options.signal?.addEventListener("abort", parentAbort, { once: true }); + const timeout = setTimeout(() => timeoutController.abort(new Error(`Node ${node.id} timed out after ${timeoutMs}ms.`)), timeoutMs); + const previousOptions = context.options; + context.options = { ...options, signal: timeoutController.signal }; + try { const result = await this.executeNode(context, node, nodeRun); context.outputs.set(node.id, result.output); + attemptService.succeed(attempt, maskSecrets(result.output), result.invocationId); this.deps.nodeFlowRepository.updateNodeRun(nodeRun.id, { status: "succeeded", executionInvocationId: result.invocationId ?? nodeRun.executionInvocationId, output: maskSecrets(result.output), finishedAt: new Date().toISOString(), }); + clearTimeout(timeout); options.signal?.removeEventListener("abort", parentAbort); context.options = previousOptions; + break; } catch (error) { + clearTimeout(timeout); options.signal?.removeEventListener("abort", parentAbort); context.options = previousOptions; const message = error instanceof Error ? error.message : String(error); - const wasCancelled = options.signal?.aborted === true; + const classification = classifyFailure(error, options.signal?.aborted === true, timeoutController.signal.aborted); + const wasCancelled = classification === "cancelled"; + const retryable = retryPolicy.retryableClasses.includes(classification) && attemptNumber < retryPolicy.maxAttempts; + attemptService.fail(attempt, classification, message, retryable, this.deps.nodeFlowRepository.listNodeAttempts(run.id).find((candidate) => candidate.id === attempt.id)?.invocationId); + if (retryable) { + this.deps.nodeFlowRepository.updateNodeRun(nodeRun.id, { status: "retry_waiting", errorMessage: message }); + await delay(retryDelay(retryPolicy.backoffMs, retryPolicy.maxBackoffMs ?? retryPolicy.backoffMs, retryPolicy.jitterRatio ?? 0, attemptNumber), options.signal); + this.deps.nodeFlowRepository.updateNodeRun(nodeRun.id, { status: "running", errorMessage: null }); + continue; + } const continueOnError = node.data?.continueOnError === true; const failureOutput = { error: message }; context.outputs.set(node.id, failureOutput); @@ -186,7 +237,10 @@ export class NodeFlowRuntimeService { errorMessage: message, finishedAt: new Date().toISOString(), }); - if (wasCancelled) { + if (classification === "unknown_side_effect") { + terminalStatus = "attention_required"; + terminalError = message; + } else if (wasCancelled) { terminalStatus = "cancelled"; terminalError ??= message || "Node flow run was cancelled."; for (const remainingNodeId of executionOrder.slice(nodeIndex + 1)) { @@ -204,6 +258,11 @@ export class NodeFlowRuntimeService { blockedNodes.add(descendant); } } + break; + } + } + if (terminalStatus === "cancelled" || terminalStatus === "attention_required") { + break; } } @@ -214,9 +273,11 @@ export class NodeFlowRuntimeService { output: maskSecrets(output), errorMessage: terminalError, finishedAt, + leaseOwner: null, + leaseExpiresAt: null, }); this.deps.executionRepository.updateExecutionInvocation(parentInvocation.id, { - status: terminalStatus === "succeeded" ? "completed" : terminalStatus, + status: terminalStatus === "succeeded" ? "completed" : terminalStatus === "attention_required" ? "failed" : terminalStatus, errorMessage: terminalError, finishedAt, }); @@ -232,9 +293,11 @@ export class NodeFlowRuntimeService { }, }); + clearInterval(heartbeatTimer); return { run: updatedRun, nodeRuns: this.deps.nodeFlowRepository.listNodeRuns(run.id), + attempts: this.deps.nodeFlowRepository.listNodeAttempts(run.id), output: updatedRun.output, }; } @@ -263,6 +326,9 @@ export class NodeFlowRuntimeService { startedAt: new Date().toISOString(), }); this.deps.nodeFlowRepository.updateNodeRun(nodeRun.id, { executionInvocationId: invocation.id }); + if (context.currentAttemptId) { + this.deps.nodeFlowRepository.updateNodeAttempt(context.currentAttemptId, { invocationId: invocation.id }); + } try { const result = node.type === "provider_prompt" ? await this.executeProviderPromptNode(context, node, invocation.id) @@ -761,6 +827,32 @@ function redactUrl(url: URL): string { return clone.toString(); } +function classifyFailure(error: unknown, parentAborted: boolean, attemptAborted: boolean): NodeFlowFailureClassification { + if (parentAborted) return "cancelled"; + const message = error instanceof Error ? error.message : String(error); + if (attemptAborted || /timed? out|timeout/i.test(message)) return "timeout"; + if (/quota|rate.?limit|429/i.test(message)) return "quota"; + if (/credential|secret|access denied/i.test(message)) return "credential"; + if (error instanceof ValidationError || /requires|unsupported|must /i.test(message)) return "validation"; + if (/ECONNRESET|ECONNREFUSED|temporar|unavailable|502|503|504/i.test(message)) return "transient"; + return "permanent"; +} + +function retryDelay(baseMs: number, maxMs: number, jitterRatio: number, attemptNumber: number): number { + const exponential = Math.min(maxMs, Math.max(0, baseMs) * (2 ** Math.max(0, attemptNumber - 1))); + const jitter = exponential * Math.max(0, Math.min(1, jitterRatio)); + return Math.max(0, Math.round(exponential - jitter + (Math.random() * jitter * 2))); +} + +async function delay(ms: number, signal?: AbortSignal): Promise { + if (ms <= 0) return; + await new Promise((resolve, reject) => { + const timer = setTimeout(resolve, ms); + const abort = (): void => { clearTimeout(timer); reject(new Error("Node flow run was cancelled.")); }; + signal?.addEventListener("abort", abort, { once: true }); + }); +} + function providerFailureMessage(result: ProviderRunResult): string { const output = [result.stderr, result.stdout] .map((stream) => stream?.trim()) diff --git a/src/services/node-flow-service.ts b/src/services/node-flow-service.ts index e6462ceb72..066e300c57 100644 --- a/src/services/node-flow-service.ts +++ b/src/services/node-flow-service.ts @@ -107,6 +107,10 @@ export class NodeFlowService { return { nodeRuns: this.repository.listNodeRuns(runId) }; } + listNodeAttempts(runId: string) { + return { attempts: this.repository.listNodeAttempts(runId) }; + } + async runFlow( projectId: string, flowId: string, diff --git a/src/services/node-flows/node-flow-attempt-service.ts b/src/services/node-flows/node-flow-attempt-service.ts new file mode 100644 index 0000000000..cad3afb9f6 --- /dev/null +++ b/src/services/node-flows/node-flow-attempt-service.ts @@ -0,0 +1,21 @@ +import { createHash } from "crypto"; +import type { NodeFlowRepository } from "../../repositories/node-flow-repository.js"; +import type { NodeFlowJsonObject, NodeFlowNodeAttemptRecord, NodeFlowNodeRunRecord } from "../../contracts/node-flow-types.js"; +import type { NodeFlowFailureClassification } from "../../contracts/node-flow-execution-policy-types.js"; + +export class NodeFlowAttemptService { + constructor(private readonly repository: NodeFlowRepository) {} + + start(nodeRun: NodeFlowNodeRunRecord, executorId: string, input: NodeFlowJsonObject, credentialIds: string[]): NodeFlowNodeAttemptRecord { + const attemptNumber = this.repository.listNodeAttempts(nodeRun.runId).filter((attempt) => attempt.nodeId === nodeRun.nodeId).length + 1; + return this.repository.createNodeAttempt({ runId: nodeRun.runId, nodeRunId: nodeRun.id, nodeId: nodeRun.nodeId, attemptNumber, status: "running", executorId, invocationId: null, artifactDigest: null, input, output: null, credentialIds, failureClassification: null, retryDecision: null, errorMessage: null, startedAt: new Date().toISOString(), finishedAt: null }); + } + + succeed(attempt: NodeFlowNodeAttemptRecord, output: NodeFlowJsonObject, invocationId?: string | null): NodeFlowNodeAttemptRecord { + return this.repository.updateNodeAttempt(attempt.id, { status: "succeeded", invocationId: invocationId ?? null, output, artifactDigest: createHash("sha256").update(JSON.stringify(output)).digest("hex"), retryDecision: "stop", finishedAt: new Date().toISOString() }); + } + + fail(attempt: NodeFlowNodeAttemptRecord, classification: NodeFlowFailureClassification, errorMessage: string, retry: boolean, invocationId?: string | null): NodeFlowNodeAttemptRecord { + return this.repository.updateNodeAttempt(attempt.id, { status: classification === "cancelled" ? "cancelled" : "failed", invocationId: invocationId ?? attempt.invocationId, failureClassification: classification, retryDecision: classification === "unknown_side_effect" ? "attention_required" : retry ? "retry" : "stop", errorMessage, finishedAt: new Date().toISOString() }); + } +} diff --git a/src/services/node-flows/node-flow-lease-service.ts b/src/services/node-flows/node-flow-lease-service.ts new file mode 100644 index 0000000000..721b058e43 --- /dev/null +++ b/src/services/node-flows/node-flow-lease-service.ts @@ -0,0 +1,8 @@ +import type { NodeFlowRepository } from "../../repositories/node-flow-repository.js"; + +export class NodeFlowLeaseService { + constructor(private readonly repository: NodeFlowRepository) {} + heartbeat(runId: string, executorId: string, leaseDurationMs: number): boolean { + return this.repository.heartbeatRun(runId, executorId, leaseDurationMs); + } +} diff --git a/src/services/node-flows/node-flow-publication-service.ts b/src/services/node-flows/node-flow-publication-service.ts new file mode 100644 index 0000000000..78433d8c58 --- /dev/null +++ b/src/services/node-flows/node-flow-publication-service.ts @@ -0,0 +1,19 @@ +import { EntityNotFoundError } from "../../repositories/repository-utils.js"; +import type { NodeFlowRepository } from "../../repositories/node-flow-repository.js"; +import type { NodeFlowPublicationRecord } from "../../contracts/node-flow-types.js"; +import type { NodeFlowVersionSelection } from "../../contracts/node-flow-execution-policy-types.js"; + +export class NodeFlowPublicationService { + constructor(private readonly repository: NodeFlowRepository) {} + + resolve(flowId: string, selection: NodeFlowVersionSelection): NodeFlowPublicationRecord { + const publication = selection.mode === "pinned" + ? this.repository.getPublication(flowId, selection.version) + : this.repository.getPublication(flowId); + if (!publication) { + const suffix = selection.mode === "pinned" ? ` at version ${selection.version}` : ""; + throw new EntityNotFoundError(`Published node flow not found: ${flowId}${suffix}`); + } + return publication; + } +} diff --git a/src/services/node-flows/node-flow-queue-service.ts b/src/services/node-flows/node-flow-queue-service.ts new file mode 100644 index 0000000000..f96241ebfc --- /dev/null +++ b/src/services/node-flows/node-flow-queue-service.ts @@ -0,0 +1,18 @@ +import type { NodeFlowRepository } from "../../repositories/node-flow-repository.js"; +import type { NodeFlowRunRecord } from "../../contracts/node-flow-types.js"; + +export class NodeFlowQuotaExceededError extends Error {} + +export class NodeFlowQueueService { + constructor(private readonly repository: NodeFlowRepository) {} + + claim(run: NodeFlowRunRecord, executorId: string): NodeFlowRunRecord { + if (this.repository.countActiveRuns() >= run.policy.maxConcurrentRuns + || this.repository.countActiveRuns(run.projectId) >= run.policy.maxConcurrentRunsPerProject) { + throw new NodeFlowQuotaExceededError("Node flow concurrency quota is exhausted."); + } + const claimed = this.repository.claimQueuedRun(run.id, executorId, run.policy.leaseDurationMs); + if (!claimed) throw new Error("Node flow run could not be claimed."); + return claimed; + } +} diff --git a/src/services/node-flows/node-flow-recovery-service.ts b/src/services/node-flows/node-flow-recovery-service.ts new file mode 100644 index 0000000000..2af71113cb --- /dev/null +++ b/src/services/node-flows/node-flow-recovery-service.ts @@ -0,0 +1,27 @@ +import type { NodeFlowRepository } from "../../repositories/node-flow-repository.js"; +import type { NodeFlowRunRecord } from "../../contracts/node-flow-types.js"; + +export class NodeFlowRecoveryService { + constructor(private readonly repository: NodeFlowRepository) {} + + recover(now = new Date()): NodeFlowRunRecord[] { + const recoverable = this.repository.listRecoverableRuns(now.toISOString()); + return recoverable.map((run) => { + if (run.status !== "running") return run; + const attempts = this.repository.listNodeAttempts(run.id); + const active = attempts.find((attempt) => attempt.status === "running"); + if (active) { + const nextStatus = active.invocationId ? "attention_required" : "queued"; + return this.repository.updateRun(run.id, { + status: nextStatus, + errorMessage: active.invocationId + ? "An externally observable attempt lost its lease; its outcome is unknown and requires attention." + : "Lease expired before an external invocation began; the run was safely requeued.", + leaseOwner: null, + leaseExpiresAt: null, + }); + } + return this.repository.updateRun(run.id, { status: "queued", leaseOwner: null, leaseExpiresAt: null }); + }); + } +} diff --git a/src/services/scheduler-service.ts b/src/services/scheduler-service.ts index 70d8ab7c9e..8fc5dd55f8 100644 --- a/src/services/scheduler-service.ts +++ b/src/services/scheduler-service.ts @@ -38,7 +38,7 @@ export interface SchedulerServiceDeps { taskRerunService?: TaskRerunService; memoryRemediationService?: MemoryRemediationService; nodeFlowRuntimeService?: NodeFlowRuntimeService; - nodeFlowRepository?: Pick; + nodeFlowRepository?: Pick; logger: Logger; tickIntervalMs?: number; } @@ -403,6 +403,9 @@ export class SchedulerService { target.input ?? {}, { triggerType: "scheduler", + versionSelection: target.versionSelection ?? (target.flowVersion !== undefined + ? { mode: "pinned", version: target.flowVersion } + : { mode: "latest_published" }), triggerPayload: { schedulerEntryId: entry.id, scheduledFor: occurrenceIso, @@ -440,6 +443,12 @@ export class SchedulerService { throw new Error("nodeFlowTarget.flowId is required."); } this.validateNodeFlowTargetOwnership(projectId, flowId); + const selection = input.nodeFlowTarget?.versionSelection + ?? (input.nodeFlowTarget?.flowVersion !== undefined ? { mode: "pinned" as const, version: input.nodeFlowTarget.flowVersion } : { mode: "latest_published" as const }); + if (selection.mode === "pinned" && typeof this.deps.nodeFlowRepository?.getPublication === "function" + && !this.deps.nodeFlowRepository.getPublication(flowId, selection.version)) { + throw new Error("Scheduled node flow version must reference a published version."); + } return; } diff --git a/tests/backend/mcp/management-node-flow-actions.test.ts b/tests/backend/mcp/management-node-flow-actions.test.ts index 54711b8284..dbd69abd52 100644 --- a/tests/backend/mcp/management-node-flow-actions.test.ts +++ b/tests/backend/mcp/management-node-flow-actions.test.ts @@ -112,6 +112,7 @@ describe("manage_node_flows", () => { expect(nodeFlowService.runFlow).toHaveBeenCalledWith("project-1", "flow-1", { prompt: "Ship" }, { triggerType: "mcp_management", + versionSelection: { mode: "latest_published" }, }); expect(parsed.result.run.id).toBe("run-1"); expect(parsed.result.output).toEqual({ ok: true }); diff --git a/tests/backend/repositories/scheduler-repository.test.ts b/tests/backend/repositories/scheduler-repository.test.ts index 1bac634aa3..6a3b1cbf78 100644 --- a/tests/backend/repositories/scheduler-repository.test.ts +++ b/tests/backend/repositories/scheduler-repository.test.ts @@ -390,6 +390,7 @@ describe("SchedulerRepository", () => { flowId: "flow-1", input: { prompt: "Ship it", count: 2, nested: { ok: true } }, flowVersion: 3, + versionSelection: { mode: "pinned", version: 3 }, }); expect(schedulerRepository.getEntry(entry.id)?.nodeFlowTarget).toEqual(entry.nodeFlowTarget); @@ -411,6 +412,7 @@ describe("SchedulerRepository", () => { expect(updated.nodeFlowTarget).toEqual({ flowId: "flow-2", input: { next: true }, + versionSelection: { mode: "latest_published" }, }); }); diff --git a/tests/backend/server/node-flow-routes.test.ts b/tests/backend/server/node-flow-routes.test.ts index add15bec4d..662fa498a1 100644 --- a/tests/backend/server/node-flow-routes.test.ts +++ b/tests/backend/server/node-flow-routes.test.ts @@ -138,6 +138,7 @@ describe("node flow routes", () => { expect(nodeFlowService.runFlow).toHaveBeenCalledWith("project-1", "flow-1", { prompt: "Ship" }, { triggerType: "manual", triggerPayload: undefined, + versionSelection: { mode: "latest_published" }, }); }); }); diff --git a/tests/backend/services/node-flow-recovery-service.test.ts b/tests/backend/services/node-flow-recovery-service.test.ts new file mode 100644 index 0000000000..918f0e4719 --- /dev/null +++ b/tests/backend/services/node-flow-recovery-service.test.ts @@ -0,0 +1,59 @@ +import { afterEach, describe, expect, it } from "vitest"; +import * as fs from "fs/promises"; +import * as os from "os"; +import * as path from "path"; +import { AppDbStorage } from "../../../src/repositories/app-db-storage.js"; +import { NodeFlowRepository } from "../../../src/repositories/node-flow-repository.js"; +import { ProjectManagementRepository } from "../../../src/repositories/project-management-repository.js"; +import { NodeFlowRecoveryService } from "../../../src/services/node-flows/node-flow-recovery-service.js"; +import { NodeFlowQueueService, NodeFlowQuotaExceededError } from "../../../src/services/node-flows/node-flow-queue-service.js"; +import { DEFAULT_NODE_FLOW_EXECUTION_POLICY } from "../../../src/contracts/node-flow-execution-policy-types.js"; + +const dirs: string[] = []; +afterEach(async () => Promise.all(dirs.splice(0).map((dir) => fs.rm(dir, { recursive: true, force: true })))); + +describe("NodeFlowRecoveryService", () => { + it("enforces the immutable project concurrency quota before claiming", async () => { + const dir = await fs.mkdtemp(path.join(os.tmpdir(), "flow-quota-")); dirs.push(dir); + const storage = new AppDbStorage(path.join(dir, "app.db")); + const projects = new ProjectManagementRepository(storage); const repository = new NodeFlowRepository(storage); + const project = projects.createProject({ name: "Quota", sourceType: "local", sourceRef: dir }); + const flow = repository.createFlow(project.id, { title: "Quota", graph: { nodes: [{ id: "input", type: "input", title: "Input" }], edges: [] } }); + const run = repository.createRun({ flowId: flow.id, projectId: project.id, version: 1, status: "queued", policy: { ...DEFAULT_NODE_FLOW_EXECUTION_POLICY, maxConcurrentRunsPerProject: 0 } }); + + expect(() => new NodeFlowQueueService(repository).claim(run, "executor")).toThrow(NodeFlowQuotaExceededError); + expect(repository.getRun(run.id)?.status).toBe("queued"); + }); + + it("requeues an expired pre-invocation attempt without creating a duplicate attempt", async () => { + const dir = await fs.mkdtemp(path.join(os.tmpdir(), "flow-recovery-")); dirs.push(dir); + const storage = new AppDbStorage(path.join(dir, "app.db")); + const projects = new ProjectManagementRepository(storage); const repository = new NodeFlowRepository(storage); + const project = projects.createProject({ name: "Recovery", sourceType: "local", sourceRef: dir }); + const flow = repository.createFlow(project.id, { title: "Recover", graph: { nodes: [{ id: "input", type: "input", title: "Input" }], edges: [] } }); + const run = repository.createRun({ flowId: flow.id, projectId: project.id, version: 1, status: "running" }); + repository.updateRun(run.id, { leaseOwner: "dead", leaseExpiresAt: "2020-01-01T00:00:00.000Z" }); + const nodeRun = repository.createNodeRun({ runId: run.id, flowId: flow.id, projectId: project.id, nodeId: "input", status: "running" }); + repository.createNodeAttempt({ runId: run.id, nodeRunId: nodeRun.id, nodeId: "input", attemptNumber: 1, status: "running", executorId: "dead", invocationId: null, artifactDigest: null, input: {}, output: null, credentialIds: [], failureClassification: null, retryDecision: null, errorMessage: null, startedAt: "2020-01-01T00:00:00.000Z", finishedAt: null }); + + const [recovered] = new NodeFlowRecoveryService(repository).recover(); + expect(recovered?.status).toBe("queued"); + expect(repository.listNodeAttempts(run.id)).toHaveLength(1); + }); + + it("requires attention when an expired attempt has an external invocation", async () => { + const dir = await fs.mkdtemp(path.join(os.tmpdir(), "flow-recovery-")); dirs.push(dir); + const storage = new AppDbStorage(path.join(dir, "app.db")); + const projects = new ProjectManagementRepository(storage); const repository = new NodeFlowRepository(storage); + const project = projects.createProject({ name: "Recovery External", sourceType: "local", sourceRef: dir }); + const flow = repository.createFlow(project.id, { title: "Recover", graph: { nodes: [{ id: "http", type: "http_request", title: "HTTP" }], edges: [] } }); + const run = repository.createRun({ flowId: flow.id, projectId: project.id, version: 1, status: "running" }); + repository.updateRun(run.id, { leaseOwner: "dead", leaseExpiresAt: "2020-01-01T00:00:00.000Z" }); + const nodeRun = repository.createNodeRun({ runId: run.id, flowId: flow.id, projectId: project.id, nodeId: "http", status: "running" }); + repository.createNodeAttempt({ runId: run.id, nodeRunId: nodeRun.id, nodeId: "http", attemptNumber: 1, status: "running", executorId: "dead", invocationId: "external-1", artifactDigest: null, input: {}, output: null, credentialIds: [], failureClassification: null, retryDecision: null, errorMessage: null, startedAt: "2020-01-01T00:00:00.000Z", finishedAt: null }); + + const [recovered] = new NodeFlowRecoveryService(repository).recover(); + expect(recovered?.status).toBe("attention_required"); + expect(recovered?.errorMessage).toMatch(/outcome is unknown/i); + }); +}); diff --git a/tests/backend/services/node-flow-runtime-service.test.ts b/tests/backend/services/node-flow-runtime-service.test.ts index 072726e075..6a94053222 100644 --- a/tests/backend/services/node-flow-runtime-service.test.ts +++ b/tests/backend/services/node-flow-runtime-service.test.ts @@ -46,6 +46,51 @@ afterEach(async () => { }); describe("NodeFlowRuntimeService", () => { + it("executes an explicitly pinned publication while latest selection follows the newest publication", async () => { + const { dir, projectRepository, nodeFlowRepository, runtime } = await createRuntime(); + const project = projectRepository.createProject({ name: "Version Project", sourceType: "local", sourceRef: dir }); + const flow = nodeFlowRepository.createFlow(project.id, { title: "Versioned", graph: { nodes: [{ id: "set", type: "set_fields", title: "Set", data: { fields: { release: "v1" } } }], edges: [] } }); + nodeFlowRepository.updateFlow(flow.id, { graph: { nodes: [{ id: "set", type: "set_fields", title: "Set", data: { fields: { release: "v2" } } }], edges: [] } }); + + const pinned = await runtime.runFlow(project.id, flow.id, {}, { versionSelection: { mode: "pinned", version: 1 } }); + const latest = await runtime.runFlow(project.id, flow.id, {}, { versionSelection: { mode: "latest_published" } }); + + expect(pinned.run.version).toBe(1); + expect(pinned.output).toEqual({ release: "v1" }); + expect(latest.run.version).toBe(2); + expect(latest.output).toEqual({ release: "v2" }); + }); + + it("retries classified transient failures and persists redacted numbered attempts", async () => { + const executeProvider = vi.fn() + .mockRejectedValueOnce(new Error("503 temporarily unavailable")) + .mockResolvedValue({ ok: true, stdout: "ok", stderr: "", code: 0, text: "ok", nativeSessionId: null, usageTelemetry: { transcriptText: "ok", inputTokens: 0, cachedInputTokens: 0, outputTokens: 0, reasoningOutputTokens: 0, totalTokens: 0, usageSource: "reported", rawUsageJson: null } }); + const { dir, projectRepository, nodeFlowRepository, runtime } = await createRuntime({ executeProvider } as Partial); + const project = projectRepository.createProject({ name: "Retry Project", sourceType: "local", sourceRef: dir }); + const flow = nodeFlowRepository.createFlow(project.id, { title: "Retry", graph: { nodes: [{ id: "prompt", type: "provider_prompt", title: "Prompt", data: { provider: "mockup-cli", prompt: "{{input.apiToken}}" }, policy: { retry: { maxAttempts: 2, backoffMs: 0, maxBackoffMs: 0 } } }], edges: [] } }); + + const result = await runtime.runFlow(project.id, flow.id, { apiToken: "never-store" }); + + expect(result.run.status).toBe("succeeded"); + expect(executeProvider).toHaveBeenCalledTimes(2); + expect(result.attempts?.map((attempt) => [attempt.attemptNumber, attempt.retryDecision])).toEqual([[1, "retry"], [2, "stop"]]); + expect(JSON.stringify(result.attempts)).not.toContain("never-store"); + }); + + it("propagates node timeouts to the executor and classifies the attempt", async () => { + const executeProvider = vi.fn().mockImplementation(({ signal }: { signal?: AbortSignal }) => new Promise((_resolve, reject) => { + signal?.addEventListener("abort", () => reject(signal.reason), { once: true }); + })); + const { dir, projectRepository, nodeFlowRepository, runtime } = await createRuntime({ executeProvider } as Partial); + const project = projectRepository.createProject({ name: "Timeout Project", sourceType: "local", sourceRef: dir }); + const flow = nodeFlowRepository.createFlow(project.id, { title: "Timeout", graph: { nodes: [{ id: "prompt", type: "provider_prompt", title: "Prompt", data: { provider: "mockup-cli", prompt: "wait" }, policy: { timeout: { timeoutMs: 5 } } }], edges: [] } }); + + const result = await runtime.runFlow(project.id, flow.id, {}); + + expect(result.run.status).toBe("failed"); + expect(result.attempts?.[0]).toMatchObject({ failureClassification: "timeout", retryDecision: "stop" }); + }); + it("executes deterministic nodes in topological order and persists the succeeded run", async () => { const { dir, projectRepository, nodeFlowRepository, runtime } = await createRuntime(); const project = projectRepository.createProject({ name: "Runtime Project", sourceType: "local", sourceRef: dir }); diff --git a/tests/backend/services/scheduler-service.test.ts b/tests/backend/services/scheduler-service.test.ts index 4bac0cd5b2..cf21d93cb2 100644 --- a/tests/backend/services/scheduler-service.test.ts +++ b/tests/backend/services/scheduler-service.test.ts @@ -533,6 +533,7 @@ describe("SchedulerService", () => { expect(nodeFlowRuntimeService.runFlow).toHaveBeenCalledWith("project-1", "flow-1", { prompt: "Ship" }, { triggerType: "scheduler", + versionSelection: { mode: "pinned", version: 2 }, triggerPayload: { schedulerEntryId: "entry-1", scheduledFor: "2026-05-18T09:00:00.000Z",