From 850491afbfa21d078b018ad67ee67e9f07b8e955 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 8 Aug 2026 03:05:46 +0000 Subject: [PATCH] fix(orchestrations): pin a run to the graph it started on MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Orchestration graphs had no versioning and runs were not pinned, so all four execution entry points re-read the live `Orchestration` row: `driveQueuedRun`, `wakeRun`, `resumeOrchestrationRunExecution` and `redriveRun`. An `update-orchestration` therefore re-shaped runs already executing — including runs parked for days in `sleeping` or `awaiting_input` — with no error and no record of which topology actually ran. Adds `OrchestrationVersion` on the shared archive engine from #880 and resolves every execution through the run's pinned version: - `orchestrationVersionSnapshot.ts` projects the graph (`nodes`, `edges`, `state_schema`, `input_schema`) and owns the write side. Name and description stay metadata, as for a guardrail: two version numbers must not denote the same topology. - `orchestrationVersions.ts` supplies the archive adapters — list, get, and a restore that appends rather than rewinding the counter. - `orchestrationRunGraph.ts` is the single resolution seam. The four call sites each used to do `orch.nodes as OrchestrationNode[]`, so fixing three of them would have looked correct in review and left the bug in the path a run actually parks in. - `OrchestrationRun.orchestrationVersion` records the pin, stamped at `start-orchestration-run` and exposed as `orchestration_version`. Runs created before pinning existed carry a null version and keep executing the live graph — the only graph they ever had. A pinned version whose archive row is missing degrades the same way rather than losing the run's work. Per #883, this skips the run-level JSONB snapshot #877 proposed as step 1: that was justified by the shared engine not existing yet, and #877's own step 3 already turned the snapshot into a version reference. No release/canary layer, per #883's third step. `releaseAssignment.ts` is already extracted and pure for when something asks for it. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01PLPL6R4W6p4aLQ2rpiojSm --- .../postgresdb/src/models/Orchestration.ts | 17 + .../postgresdb/src/models/OrchestrationRun.ts | 16 + .../src/models/OrchestrationVersion.ts | 110 ++++ packages/postgresdb/src/models/index.ts | 1 + packages/postgresdb/src/utils/publicId.ts | 1 + .../server/src/lib/orchestrationEngine.ts | 89 ++- .../server/src/lib/orchestrationRunGraph.ts | 121 ++++ .../src/lib/orchestrationVersionSnapshot.ts | 76 +++ .../server/src/lib/orchestrationVersions.ts | 185 ++++++ packages/server/src/lib/orchestrations.ts | 132 ++++- .../src/permissions/orchestrations.json | 15 + .../src/rest/openapi/v1/orchestrations.yaml | 259 ++++++++ .../src/rest/v1/orchestrationVersions.ts | 129 ++++ packages/server/src/rest/v1/orchestrations.ts | 11 + .../src/rest/v1/orchestrationsRequestBody.ts | 8 + .../tests/lib/orchestrationRunPinning.test.ts | 398 +++++++++++++ .../tests/lib/orchestrationScheduler.test.ts | 1 + .../tests/rest/orchestrationVersions.test.ts | 560 ++++++++++++++++++ .../website/docs/modules/orchestrations.md | 75 +++ tests/smoke-tests.sh | 57 +- 20 files changed, 2202 insertions(+), 59 deletions(-) create mode 100644 packages/postgresdb/src/models/OrchestrationVersion.ts create mode 100644 packages/server/src/lib/orchestrationRunGraph.ts create mode 100644 packages/server/src/lib/orchestrationVersionSnapshot.ts create mode 100644 packages/server/src/lib/orchestrationVersions.ts create mode 100644 packages/server/src/rest/v1/orchestrationVersions.ts create mode 100644 packages/server/tests/unit/tests/lib/orchestrationRunPinning.test.ts create mode 100644 packages/server/tests/unit/tests/rest/orchestrationVersions.test.ts diff --git a/packages/postgresdb/src/models/Orchestration.ts b/packages/postgresdb/src/models/Orchestration.ts index ff445ec0..8e8430fa 100644 --- a/packages/postgresdb/src/models/Orchestration.ts +++ b/packages/postgresdb/src/models/Orchestration.ts @@ -10,6 +10,7 @@ import { import { generatePublicId, PUBLIC_ID_PREFIXES } from '../utils/publicId'; import { OrchestrationRun } from './OrchestrationRun'; +import { OrchestrationVersion } from './OrchestrationVersion'; import { Project } from './Project'; @Table({ @@ -53,6 +54,17 @@ export class Orchestration extends Model { @Column({ type: DataType.TEXT, allowNull: true }) declare description: string | null; + /** + * Incremented on every write that changes the graph (`nodes`, `edges`, + * `stateSchema`, `inputSchema`); each version is archived as an + * `OrchestrationVersion`. A run pins the version it started on, so editing the + * graph never re-shapes a run already in flight (#872) — which makes these + * columns a *draft* for runs started from now on, not a live rewrite of the + * ones already executing. Metadata-only edits leave it untouched. + */ + @Column({ type: DataType.INTEGER, allowNull: false, defaultValue: 1 }) + declare version: number; + @Column({ type: DataType.JSONB, allowNull: false, defaultValue: [] }) declare nodes: object[]; @@ -70,6 +82,11 @@ export class Orchestration extends Model { }) declare runs: OrchestrationRun[]; + @HasMany(() => { + return OrchestrationVersion; + }) + declare versions: OrchestrationVersion[]; + @Column({ type: DataType.DATE }) declare createdAt: Date; diff --git a/packages/postgresdb/src/models/OrchestrationRun.ts b/packages/postgresdb/src/models/OrchestrationRun.ts index 8481933e..0ccbaec8 100644 --- a/packages/postgresdb/src/models/OrchestrationRun.ts +++ b/packages/postgresdb/src/models/OrchestrationRun.ts @@ -51,6 +51,22 @@ export class OrchestrationRun extends Model { }) declare orchestration: Orchestration; + // The orchestration version this run executes, stamped at start and never + // changed afterwards (#872). Every execution entry point — the first drive of + // a queued run, a wake from `sleeping`, a human/approval resume, and a redrive + // after a lease expiry — resolves the graph through this number rather than + // reading the live `Orchestration` row, so editing the orchestration cannot + // re-shape a run already in flight, including one parked for days. + // + // A version *number* rather than a foreign key to `orchestration_versions`: + // the number is what the run response exposes and what an audit reader cites, + // it matches `Generation.agentVersion`, and the archive row is reachable from + // it with no join. Null only for runs created before pinning existed, which + // fall back to the live row — the pre-#872 behavior, and the only thing there + // is to fall back to. + @Column({ type: DataType.INTEGER, allowNull: true }) + declare orchestrationVersion: number | null; + @ForeignKey(() => { return Project; }) diff --git a/packages/postgresdb/src/models/OrchestrationVersion.ts b/packages/postgresdb/src/models/OrchestrationVersion.ts new file mode 100644 index 00000000..a6935183 --- /dev/null +++ b/packages/postgresdb/src/models/OrchestrationVersion.ts @@ -0,0 +1,110 @@ +import { + BelongsTo, + Column, + DataType, + ForeignKey, + Model, + Table, +} from '@ttoss/postgresdb'; + +import { generatePublicId, PUBLIC_ID_PREFIXES } from '../utils/publicId'; +import { Orchestration } from './Orchestration'; +import { User } from './User'; + +/** + * Immutable archive of an orchestration's graph at a given version. A new row is + * written by the shared lib write path on every write that actually changes the + * graph; existing rows are never mutated, so a run can be pinned to the exact + * topology it started on (issue #872) and `node_executions` always reference node + * ids from a graph that still exists. A restore appends a new version rather than + * rewinding the counter, so there is no `updatedAt`. + * + * Shares its column layout — and the lib engine that reads and writes it — with + * `AgentVersion` and `GuardrailVersion` (`src/lib/resourceVersions.ts`). The + * table stays separate so the foreign key to `orchestrations` is a real one. + */ +@Table({ + tableName: 'orchestration_versions', + indexes: [ + { + name: 'orchestration_versions_public_id_unique', + unique: true, + fields: ['public_id'], + }, + // Serves both the point lookup of one version — which every background + // execution of a pinned run performs — and the newest-first paginated + // listing (`WHERE orchestration_id = ? ORDER BY version DESC`), so no + // separate index on `created_at` is needed. + { + name: 'orchestration_versions_orchestration_id_version_unique', + unique: true, + fields: ['orchestration_id', 'version'], + }, + ], + updatedAt: false, + hooks: { + beforeValidate: (instance: OrchestrationVersion) => { + if (!instance.publicId) { + instance.publicId = generatePublicId( + PUBLIC_ID_PREFIXES.orchestrationVersion + ); + } + }, + }, +}) +export class OrchestrationVersion extends Model { + @Column({ + type: DataType.STRING(32), + allowNull: false, + }) + declare publicId: string; + + @ForeignKey(() => { + return Orchestration; + }) + @Column({ type: DataType.INTEGER, allowNull: false }) + declare orchestrationId: number; + + @BelongsTo(() => { + return Orchestration; + }) + declare orchestration: Orchestration; + + @Column({ type: DataType.INTEGER, allowNull: false }) + declare version: number; + + /** + * The orchestration's versioned surface, stored in the wire (snake_case) shape + * the orchestrations OpenAPI spec documents: `{ nodes, edges, state_schema, + * input_schema }`. + * + * Only the graph is versioned. Name and description are metadata — bumping the + * version when one of them changes would make two version numbers denote the + * same topology, and the version number is exactly what a run cites to say + * which topology it executed. + */ + @Column({ type: DataType.JSONB, allowNull: false }) + declare config: object; + + /** Optional human tag for this version, e.g. `pre-rewire`. */ + @Column({ type: DataType.STRING, allowNull: true }) + declare label: string | null; + + /** + * The user whose action produced this version. Null for writes with no + * request user behind them. + */ + @ForeignKey(() => { + return User; + }) + @Column({ type: DataType.INTEGER, allowNull: true }) + declare createdByUserId: number | null; + + @BelongsTo(() => { + return User; + }) + declare createdBy: User | null; + + @Column({ type: DataType.DATE }) + declare createdAt: Date; +} diff --git a/packages/postgresdb/src/models/index.ts b/packages/postgresdb/src/models/index.ts index 9c2c5a5f..7241a73c 100644 --- a/packages/postgresdb/src/models/index.ts +++ b/packages/postgresdb/src/models/index.ts @@ -36,6 +36,7 @@ export { OrchestrationCheckpoint } from './OrchestrationCheckpoint'; export { OrchestrationNodeExecution } from './OrchestrationNodeExecution'; export { OrchestrationRun } from './OrchestrationRun'; export { OrchestrationRunTask } from './OrchestrationRunTask'; +export { OrchestrationVersion } from './OrchestrationVersion'; export { Policy } from './Policy'; export { PriceBook } from './PriceBook'; export { Project } from './Project'; diff --git a/packages/postgresdb/src/utils/publicId.ts b/packages/postgresdb/src/utils/publicId.ts index 8577415f..77bacd89 100644 --- a/packages/postgresdb/src/utils/publicId.ts +++ b/packages/postgresdb/src/utils/publicId.ts @@ -28,6 +28,7 @@ export const PUBLIC_ID_PREFIXES = { memoryEntry: 'mem_entry_', modelRoute: 'route_', orchestration: 'orch_', + orchestrationVersion: 'orch_ver_', orchestrationRun: 'orch_run_', orchestrationRunTask: 'orch_task_', formation: 'form_', diff --git a/packages/server/src/lib/orchestrationEngine.ts b/packages/server/src/lib/orchestrationEngine.ts index 6c3c879d..29be404f 100644 --- a/packages/server/src/lib/orchestrationEngine.ts +++ b/packages/server/src/lib/orchestrationEngine.ts @@ -25,6 +25,7 @@ import { } from './orchestrationNodeRecorder'; import { writeNodeArtifact } from './orchestrationNodesNamespace'; import { recordHumanInputResumption } from './orchestrationPauseRecords'; +import { resolveRunGraph } from './orchestrationRunGraph'; import type { PersistedWakeContext } from './orchestrationRunHelpers'; import { applyHumanInputToState, @@ -382,6 +383,46 @@ const resolveRunPrincipal = (args: { }; }; +/** Writes the run row a `start-orchestration-run` produces. */ +const createRunRecord = async (args: { + orchestration: InstanceType; + projectId: number; + state: Record; + artifacts: Record; + input?: Record; + triggerId?: string; + principal?: RequestPrincipal; + authHeader?: string; + wait?: boolean; +}): Promise> => { + return db.OrchestrationRun.create({ + orchestrationId: args.orchestration.id as number, + // Pin the run to the graph it starts on (#872). Every later execution of + // this run resolves its topology through this number, so an + // `update-orchestration` that lands while the run is queued, sleeping or + // awaiting input cannot re-shape it. + orchestrationVersion: args.orchestration.version, + projectId: args.projectId, + // Synchronous mode enters `running` immediately (it drives in-process); + // async mode enters `queued` — the run is enqueued and a worker picks it up. + status: args.wait ? 'running' : 'queued', + state: args.state, + activeNodes: [], + artifacts: args.artifacts, + input: args.input ?? null, + triggerId: args.triggerId ?? null, + ...resolveRunPrincipal({ + principal: args.principal, + authHeader: args.authHeader, + }), + startedAt: new Date(), + // In `wait` mode the run is `running` immediately, so acquire a lease so the + // reaper can reclaim it if this driver crashes before the first checkpoint. + // A `queued` run holds no lease until a worker claims and drives it. + leaseExpiresAt: args.wait ? newLeaseExpiry() : null, + }); +}; + export const startOrchestrationRun = async (args: { orchestrationPublicId: string; projectId?: number; @@ -420,8 +461,6 @@ export const startOrchestrationRun = async (args: { orchestrationProjectId: orch.projectId as number, }); - const nodes = orch.nodes as OrchestrationNode[]; - const edges = orch.edges as OrchestrationEdge[]; // Seed the run input under the `input` namespace only, matching the // pipeline/formation convention (`{ "var": "input." }`) so a graph // reads run input the same way everywhere in the platform. Earlier releases @@ -431,28 +470,16 @@ export const startOrchestrationRun = async (args: { const state: Record = { input: runInput }; const artifacts: Record = {}; - const runPrincipal = resolveRunPrincipal({ - principal: args.principal, - authHeader: args.authHeader, - }); - - const runRecord = await db.OrchestrationRun.create({ - orchestrationId: orch.id as number, + const runRecord = await createRunRecord({ + orchestration: orch, projectId: effectiveProjectId, - // Synchronous mode enters `running` immediately (it drives in-process); - // async mode enters `queued` — the run is enqueued and a worker picks it up. - status: args.wait ? 'running' : 'queued', state, - activeNodes: [], artifacts, - input: args.input ?? null, - triggerId: args.triggerId ?? null, - ...runPrincipal, - startedAt: new Date(), - // In `wait` mode the run is `running` immediately, so acquire a lease so the - // reaper can reclaim it if this driver crashes before the first checkpoint. - // A `queued` run holds no lease until a worker claims and drives it. - leaseExpiresAt: args.wait ? newLeaseExpiry() : null, + input: args.input, + triggerId: args.triggerId, + principal: args.principal, + authHeader: args.authHeader, + wait: args.wait, }); const startMapped = await mapRunWithIncludes(runRecord.id as number); @@ -473,6 +500,14 @@ export const startOrchestrationRun = async (args: { // Synchronous (compatibility) mode: block until the run reaches a terminal or // awaiting_input state, sleeping through any delay/poll waits in-process. if (args.wait) { + // Resolved through the pinned version rather than the row just read, so the + // inline drive and every later background drive of this run are guaranteed + // to execute the same graph even if an edit lands in between. + const { nodes, edges } = await resolveRunGraph({ + run: runRecord, + orchestration: orch, + }); + return driveRunToRest({ runRecord, nodes, @@ -543,8 +578,7 @@ export const driveQueuedRun = async (args: { return; } - const nodes = orch.nodes as OrchestrationNode[]; - const edges = orch.edges as OrchestrationEdge[]; + const { nodes, edges } = await resolveRunGraph({ run, orchestration: orch }); // Clone so mutations produce a fresh reference (see wakeRun). const state = { ...((run.state ?? {}) as Record) }; const artifacts = { ...((run.artifacts ?? {}) as Record) }; @@ -595,8 +629,7 @@ export const wakeRun = async (args: { return; } - const nodes = orch.nodes as OrchestrationNode[]; - const edges = orch.edges as OrchestrationEdge[]; + const { nodes, edges } = await resolveRunGraph({ run, orchestration: orch }); // Clone so mutations produce a fresh object reference — Sequelize does not // reliably detect in-place mutation of a JSONB attribute, so reusing // run.state directly can cause the final update to skip persisting it. @@ -823,8 +856,7 @@ export const resumeOrchestrationRunExecution = async (args: { `Orchestration for run not found.` ); - const nodes = orch.nodes as OrchestrationNode[]; - const edges = orch.edges as OrchestrationEdge[]; + const { nodes, edges } = await resolveRunGraph({ run, orchestration: orch }); // Clone so mutations produce a fresh reference (see wakeRun). const state = { ...((run.state ?? {}) as Record) }; const artifacts = { ...((run.artifacts ?? {}) as Record) }; @@ -965,8 +997,7 @@ export const redriveRun = async (args: { return; } - const nodes = orch.nodes as OrchestrationNode[]; - const edges = orch.edges as OrchestrationEdge[]; + const { nodes, edges } = await resolveRunGraph({ run, orchestration: orch }); // Clone so mutations produce a fresh reference (see wakeRun). const state = { ...((run.state ?? {}) as Record) }; const artifacts = { ...((run.artifacts ?? {}) as Record) }; diff --git a/packages/server/src/lib/orchestrationRunGraph.ts b/packages/server/src/lib/orchestrationRunGraph.ts new file mode 100644 index 00000000..6cfeeaed --- /dev/null +++ b/packages/server/src/lib/orchestrationRunGraph.ts @@ -0,0 +1,121 @@ +import createDebug from 'debug'; + +import { db } from '../db'; +import { parseOrchestrationGraph } from './orchestrationGraphWire'; +import type { OrchestrationEdge, OrchestrationNode } from './orchestrations'; + +const log = createDebug('soat:orchestrations'); + +/** + * Resolves the graph a run executes (issue #872). + * + * A run is pinned to an orchestration version at `start-orchestration-run` and + * every later execution — the first drive of a queued run, a wake from + * `sleeping`, a human or approval resume, a redrive after a lease expiry — + * resolves its topology through this module rather than reading the live + * `Orchestration` row. That is the whole fix: `update-orchestration` can rewire + * or delete nodes freely, and a run parked for days still finishes on the graph + * it started on. + * + * The single seam matters as much as the pinning. Before this, four call sites + * each did `orch.nodes as OrchestrationNode[]`, so getting one of them right and + * missing another would look correct in review and still leave the bug in the + * path that matters least often — which is exactly the path a run parks in. + */ + +export type RunGraph = { + nodes: OrchestrationNode[]; + edges: OrchestrationEdge[]; +}; + +/** The live graph, as persisted on the orchestration row (already camelCase). */ +const liveGraph = ( + orchestration: InstanceType +): RunGraph => { + return { + nodes: orchestration.nodes as OrchestrationNode[], + edges: orchestration.edges as OrchestrationEdge[], + }; +}; + +/** + * The graph at one archived version of an orchestration, or `null` when that + * version was never archived. + * + * The archive stores the graph wire-shaped, so it comes back through the same + * snake_case → camelCase boundary an inbound request uses. + */ +const findArchivedGraph = async (args: { + orchestrationDbId: number; + version: number; +}): Promise => { + const archived = await db.OrchestrationVersion.findOne({ + where: { orchestrationId: args.orchestrationDbId, version: args.version }, + }); + if (!archived) return null; + + const config = archived.config; + /* istanbul ignore next -- the column is JSONB NOT NULL and only ever written + from buildOrchestrationConfigSnapshot, so no entry point can produce a + non-object here; the narrowing exists to keep the return type honest. */ + if (typeof config !== 'object' || config === null || Array.isArray(config)) { + return null; + } + + const { nodes, edges } = config as Record; + return parseOrchestrationGraph({ nodes, edges }); +}; + +/** + * The graph a run must execute: its pinned version's, falling back to the live + * row when there is no pinned version to resolve. + * + * The fallback covers exactly one real case — a run created before pinning + * existed, whose `orchestrationVersion` is null. Those runs behave as they did + * before #872 because the live row is the only graph they ever had; there is + * nothing better to give them, and refusing to drive them would strand every + * run that was in flight across the deploy. + * + * A pinned version whose archive row is missing degrades the same way rather + * than failing the run. It is unreachable through the API — a version is + * archived before any run can pin it, and versions are only deleted with their + * orchestration, which deletes its runs in the same transaction — so this is a + * guard against an out-of-band deletion, not a supported state. Failing the run + * instead would turn a bookkeeping inconsistency into lost work, and the log + * line names it either way. + */ +export const resolveRunGraph = async (args: { + run: InstanceType; + orchestration: InstanceType; +}): Promise => { + const version = args.run.orchestrationVersion; + + if (version === null || version === undefined) { + log( + 'resolveRunGraph: run=%s has no pinned version, executing the live graph', + args.run.publicId + ); + return liveGraph(args.orchestration); + } + + const archived = await findArchivedGraph({ + orchestrationDbId: args.orchestration.id as number, + version, + }); + + if (!archived) { + log( + 'resolveRunGraph: run=%s pinned to missing version=%d, executing the live graph', + args.run.publicId, + version + ); + return liveGraph(args.orchestration); + } + + log( + 'resolveRunGraph: run=%s executing version=%d', + args.run.publicId, + version + ); + return archived; +}; diff --git a/packages/server/src/lib/orchestrationVersionSnapshot.ts b/packages/server/src/lib/orchestrationVersionSnapshot.ts new file mode 100644 index 00000000..04a6e25f --- /dev/null +++ b/packages/server/src/lib/orchestrationVersionSnapshot.ts @@ -0,0 +1,76 @@ +import { db } from '../db'; +import type { MappedOrchestration } from './orchestrations'; +import { + type ConfigSnapshot, + makeVersionStore, + projectConfigSnapshot, +} from './resourceVersions'; + +/** + * How an orchestration's configuration is projected into the shared version + * archive (`resourceVersions.ts`). Everything generic — the projection + * mechanics, the change detection, the equality check — lives there. + * + * This module holds the archive's **write side** so that `orchestrations.ts` can + * reach it without importing `orchestrationVersions.ts`, which imports + * `orchestrations.ts` back for `updateOrchestration`. + */ + +/** + * An orchestration's archived configuration, in the wire (snake_case) shape the + * orchestrations OpenAPI spec documents. + */ +export type OrchestrationConfigSnapshot = ConfigSnapshot; + +/** + * The keys of an orchestration response that are **not** configuration: its + * identity, its version bookkeeping, its timestamps — and its name and + * description. + * + * Name and description are metadata, as they are for a guardrail: bumping the + * version when one of them changes would make two version numbers denote the + * same topology, and the version number is exactly what a run cites to say which + * topology it executed. What remains — `nodes`, `edges`, `state_schema`, + * `input_schema` — is the graph the engine executes and nothing else. + * + * Stated as an exclusion rather than an allowlist on purpose; see + * `projectConfigSnapshot` for why. `orchestrationVersions.test.ts` pins the exact + * key set the projection produces, so adding an orchestration field forces a + * deliberate choice here. + */ +const NON_CONFIG_ORCHESTRATION_FIELDS: ReadonlySet = new Set([ + 'id', + 'project_id', + 'name', + 'description', + 'version', + 'created_at', + 'updated_at', +]); + +/** + * Projects an orchestration response down to its graph. + * + * The graph is copied as a **value**. `nodes` and `edges` carry author-authored + * JSON Logic (`expression`, `input_mapping`, `state_mapping`, `exit_condition`) + * whose `var` paths must round-trip byte-for-byte, and the two schemas are + * caller-authored JSON Schema — nothing here descends into any of them or + * rewrites a key (`.claude/rules/case-convention.md`). + */ +export const buildOrchestrationConfigSnapshot = ( + orchestration: MappedOrchestration +): OrchestrationConfigSnapshot => { + return projectConfigSnapshot({ + resource: orchestration, + nonConfigFields: NON_CONFIG_ORCHESTRATION_FIELDS, + }); +}; + +/** The write side of the orchestration graph archive. */ +export const orchestrationVersionStore = makeVersionStore({ + resourceLabel: 'Orchestration', + versionModel: () => { + return db.OrchestrationVersion; + }, + foreignKey: 'orchestrationId', +}); diff --git a/packages/server/src/lib/orchestrationVersions.ts b/packages/server/src/lib/orchestrationVersions.ts new file mode 100644 index 00000000..09760748 --- /dev/null +++ b/packages/server/src/lib/orchestrationVersions.ts @@ -0,0 +1,185 @@ +import createDebug from 'debug'; + +import { db } from '../db'; +import { DomainError } from '../errors'; +import { parseOrchestrationGraph } from './orchestrationGraphWire'; +import { + type MappedOrchestration, + updateOrchestration, +} from './orchestrations'; +import { orchestrationVersionStore } from './orchestrationVersionSnapshot'; +import { + type ArchivedVersionRow, + configObject, + makeVersionArchive, + mapArchivedVersionFields, + type VersionedResourceRef, +} from './resourceVersions'; + +const log = createDebug('soat:orchestrations'); + +/** + * Orchestration graph version history (issue #872). + * + * The archive mechanics live in `resourceVersions.ts` and are shared with agents + * and guardrails; this module supplies the orchestration-specific adapters. + * Versions are never written from here — they are archived by the shared write + * path in `orchestrations.ts`, so a REST edit and a formation apply leave + * identical history. + * + * Orchestrations have no release/canary layer: a run is pinned at + * `start-orchestration-run` and stays on that version for its whole life, which + * can be days. Splitting *new* runs across two graphs is a coherent idea but + * nothing has asked for it (#883), and the mechanism is already extracted and + * pure in `releaseAssignment.ts` for when something does. + */ + +type OrchestrationInstance = InstanceType<(typeof db)['Orchestration']>; + +// ── Mapping ────────────────────────────────────────────────────────────── + +export const mapOrchestrationVersion = ( + version: ArchivedVersionRow, + orchestrationPublicId: string +) => { + return { + orchestration_id: orchestrationPublicId, + ...mapArchivedVersionFields(version), + }; +}; + +// ── Lookup helpers ─────────────────────────────────────────────────────── + +const findOrchestrationInstance = async (args: { + projectIds?: number[]; + id: string; +}): Promise => { + const where: Record = { publicId: args.id }; + if (args.projectIds !== undefined) where.projectId = args.projectIds; + + const orchestration = await db.Orchestration.findOne({ where }); + // Cross-project access resolves here as "not found" rather than a 403, so an + // orchestration's existence never leaks across a tenant boundary. + if (!orchestration) { + throw new DomainError( + 'ORCHESTRATION_NOT_FOUND', + `Orchestration '${args.id}' not found.` + ); + } + return orchestration as OrchestrationInstance; +}; + +const toResourceRef = ( + orchestration: OrchestrationInstance +): VersionedResourceRef => { + return { + dbId: orchestration.id as number, + publicId: orchestration.publicId, + version: orchestration.version, + }; +}; + +/** + * The orchestration adapter over the shared archive. `applyConfig` routes through + * `updateOrchestration` rather than touching columns, so a restored graph goes + * through the same static validation as an authored one and is archived by the + * same choke point as any other edit — and a graph identical to the live one is + * recognised as a no-op. + * + * Unlike a guardrail document, that validation cannot start failing over time: + * `assertOrchestrationValid` is a pure check on the graph's shape, and a node's + * resource references (`agent_id`, `tool_id`, `orchestration_id`) resolve when a + * run reaches the node. A target deleted since the snapshot was taken therefore + * restores cleanly and surfaces as a failed run — the same place it surfaces when + * the graph is authored that way in the first place. + */ +const orchestrationVersionArchive = makeVersionArchive({ + store: orchestrationVersionStore, + loadResource: async (args) => { + return toResourceRef(await findOrchestrationInstance(args)); + }, + mapVersion: mapOrchestrationVersion, + applyConfig: async (args): Promise => { + // The archived graph is wire-shaped, so it goes back through the same + // snake_case → camelCase boundary an inbound request does. + const graph = parseOrchestrationGraph({ + nodes: args.config.nodes, + edges: args.config.edges, + }); + + return updateOrchestration({ + projectIds: args.projectIds, + id: args.id, + nodes: graph.nodes, + edges: graph.edges, + // A version replaces the whole graph, so an absent schema means "cleared", + // never "leave as is". + stateSchema: configObject(args.config.state_schema), + inputSchema: configObject(args.config.input_schema), + versionLabel: args.label, + createdByUserId: args.createdByUserId, + }); + }, +}); + +// ── Read endpoints ─────────────────────────────────────────────────────── + +export const listOrchestrationVersions = async (args: { + projectIds?: number[]; + orchestrationId: string; + limit?: number; + offset?: number; +}) => { + log('listOrchestrationVersions: orchestrationId=%s', args.orchestrationId); + + return orchestrationVersionArchive.listVersions({ + projectIds: args.projectIds, + resourceId: args.orchestrationId, + limit: args.limit, + offset: args.offset, + }); +}; + +export const getOrchestrationVersion = async (args: { + projectIds?: number[]; + orchestrationId: string; + version: number; +}) => { + log( + 'getOrchestrationVersion: orchestrationId=%s version=%d', + args.orchestrationId, + args.version + ); + + return orchestrationVersionArchive.getVersion({ + projectIds: args.projectIds, + resourceId: args.orchestrationId, + version: args.version, + }); +}; + +export const restoreOrchestrationVersion = async (args: { + projectIds?: number[]; + orchestrationId: string; + version: number; + label?: string | null; + createdByUserId?: number | null; +}): Promise => { + log( + 'restoreOrchestrationVersion: orchestrationId=%s version=%d', + args.orchestrationId, + args.version + ); + + // Appends a new version rather than rewinding the counter, so a run pinned to + // any version in between still resolves the graph it started on. Runs already + // in flight are untouched: a restore is an ordinary graph edit, and pinning is + // what keeps it from reaching them. + return orchestrationVersionArchive.restoreVersion({ + projectIds: args.projectIds, + resourceId: args.orchestrationId, + version: args.version, + label: args.label, + createdByUserId: args.createdByUserId, + }); +}; diff --git a/packages/server/src/lib/orchestrations.ts b/packages/server/src/lib/orchestrations.ts index 94abb783..80356c73 100644 --- a/packages/server/src/lib/orchestrations.ts +++ b/packages/server/src/lib/orchestrations.ts @@ -12,6 +12,10 @@ import { assertOrchestrationUpdateValid, assertOrchestrationValid, } from './orchestrationValidation'; +import { + buildOrchestrationConfigSnapshot, + orchestrationVersionStore, +} from './orchestrationVersionSnapshot'; import { paginatedList, type PaginatedResult, @@ -132,11 +136,22 @@ export type OrchestrationEdge = { activationCondition?: 'all' | 'any'; }; +/** + * The authorship a write attaches to the version it archives. Optional + * throughout: a write with no request user behind it (a scheduler-driven apply, + * an internal repair) archives a version with a null author rather than none. + */ +export type OrchestrationVersionAuthorship = { + createdByUserId?: number | null; + versionLabel?: string | null; +}; + export type MappedOrchestration = { id: string; project_id: string; name: string; description: string | null; + version: number; nodes: ReturnType[]; edges: ReturnType[]; state_schema: object | null; @@ -161,6 +176,10 @@ export type MappedNodeExecution = { export type MappedOrchestrationRun = { id: string; orchestration_id: string; + // The orchestration version this run executes, fixed when the run started. + // Null for runs created before pinning existed (#872), which execute the live + // graph — the only thing there is to fall back to. + orchestration_version: number | null; project_id: string; status: | 'queued' @@ -208,6 +227,7 @@ const mapOrchestration = ( project_id: orch.project.publicId, name: orch.name, description: orch.description, + version: orch.version, ...mapOrchestrationGraph({ nodes: orch.nodes as OrchestrationNode[], edges: orch.edges as OrchestrationEdge[], @@ -283,6 +303,7 @@ export const mapOrchestrationRun = ( return { id: run.publicId, orchestration_id: run.orchestration.publicId, + orchestration_version: run.orchestrationVersion, project_id: run.project.publicId, status: run.status, state: run.state as Record, @@ -327,15 +348,17 @@ export const nodeExecutionsInclude = (): object => { // ── CRUD: Orchestrations ────────────────────────────────────────────────── -export const createOrchestration = async (args: { - projectId: number; - name: string; - description?: string | null; - nodes: OrchestrationNode[]; - edges: OrchestrationEdge[]; - stateSchema?: object | null; - inputSchema?: object | null; -}): Promise => { +export const createOrchestration = async ( + args: { + projectId: number; + name: string; + description?: string | null; + nodes: OrchestrationNode[]; + edges: OrchestrationEdge[]; + stateSchema?: object | null; + inputSchema?: object | null; + } & OrchestrationVersionAuthorship +): Promise => { log('createOrchestration %o', { projectId: args.projectId, name: args.name }); assertOrchestrationValid({ @@ -348,6 +371,7 @@ export const createOrchestration = async (args: { projectId: args.projectId, name: args.name, description: args.description ?? null, + version: 1, nodes: args.nodes, edges: args.edges, stateSchema: args.stateSchema ?? null, @@ -359,11 +383,23 @@ export const createOrchestration = async (args: { include: [{ model: db.Project, as: 'project' }], }); - return mapOrchestration( + const mapped = mapOrchestration( created as InstanceType & { project: InstanceType; } ); + + // Version 1 is archived on create, so the very first run has a pinned graph to + // resolve rather than falling back to the live row. + await orchestrationVersionStore.writeVersion({ + resourceDbId: orch.id as number, + version: 1, + config: buildOrchestrationConfigSnapshot(mapped), + label: args.versionLabel, + createdByUserId: args.createdByUserId, + }); + + return mapped; }; export const listOrchestrations = async (args: { @@ -419,28 +455,44 @@ export const findOrchestration = async (args: { ); }; -export const updateOrchestration = async (args: { - id: string; - projectIds?: number[]; - name?: string; - description?: string | null; - nodes?: OrchestrationNode[]; - edges?: OrchestrationEdge[]; - stateSchema?: object | null; - inputSchema?: object | null; -}): Promise => { +export const updateOrchestration = async ( + args: { + id: string; + projectIds?: number[]; + name?: string; + description?: string | null; + nodes?: OrchestrationNode[]; + edges?: OrchestrationEdge[]; + stateSchema?: object | null; + inputSchema?: object | null; + } & OrchestrationVersionAuthorship +): Promise => { log('updateOrchestration %o', { id: args.id }); const where: Record = { publicId: args.id }; if (args.projectIds) where['projectId'] = args.projectIds; - const orch = await db.Orchestration.findOne({ where }); + const orch = await db.Orchestration.findOne({ + where, + include: [{ model: db.Project, as: 'project' }], + }); if (!orch) throw new DomainError( 'ORCHESTRATION_NOT_FOUND', `Orchestration '${args.id}' not found.` ); + // `orch` is loaded with its project so it can be mapped directly, before and + // after the write: `update` mutates the instance in place, so the same + // reference yields the pre-write config here and the post-write one below, + // with no second query and no chance of the two views disagreeing. + const asMappable = orch as InstanceType & { + project: InstanceType; + }; + const beforeConfig = buildOrchestrationConfigSnapshot( + mapOrchestration(asMappable) + ); + assertOrchestrationUpdateValid({ update: { nodes: args.nodes, @@ -464,16 +516,31 @@ export const updateOrchestration = async (args: { await orch.update(updates); - const updated = await db.Orchestration.findOne({ - where: { id: orch.id as number }, - include: [{ model: db.Project, as: 'project' }], + // A graph write bumps the version and archives the new graph, so a run pinned + // to any earlier version still resolves the topology it started on. + // Metadata-only edits (name / description) leave the version untouched — as + // does re-writing the graph the orchestration already holds, which is what + // makes restoring the live graph a genuine no-op rather than an endless + // version chain. + await orchestrationVersionStore.archiveConfigChange({ + resourceDbId: orch.id as number, + currentVersion: orch.version, + before: beforeConfig, + after: buildOrchestrationConfigSnapshot(mapOrchestration(asMappable)), + label: args.versionLabel, + createdByUserId: args.createdByUserId, + bumpVersion: async (nextVersion) => { + await orch.update({ version: nextVersion }); + log( + 'updateOrchestration: id=%s bumped to version=%d', + args.id, + nextVersion + ); + }, }); - return mapOrchestration( - updated as InstanceType & { - project: InstanceType; - } - ); + // Mapped last, so the response carries the version the archive just wrote. + return mapOrchestration(asMappable); }; export const deleteOrchestration = async (args: { @@ -524,6 +591,13 @@ export const deleteOrchestration = async (args: { }); } + // Archived versions are owned by the orchestration; remove them before the + // parent so no orphan version rows are left behind. + await orchestrationVersionStore.deleteVersions({ + resourceDbId: orch.id as number, + transaction: t, + }); + await orch.destroy({ transaction: t }); }); }; diff --git a/packages/server/src/permissions/orchestrations.json b/packages/server/src/permissions/orchestrations.json index edf0e11f..9fb3dfa0 100644 --- a/packages/server/src/permissions/orchestrations.json +++ b/packages/server/src/permissions/orchestrations.json @@ -31,6 +31,21 @@ "action": "orchestrations:DeleteOrchestration", "description": "Delete an orchestration" }, + { + "operationId": "listOrchestrationVersions", + "action": "orchestrations:ListOrchestrationVersions", + "description": "List an orchestration's archived graph versions" + }, + { + "operationId": "getOrchestrationVersion", + "action": "orchestrations:GetOrchestrationVersion", + "description": "Fetch one archived orchestration graph version" + }, + { + "operationId": "restoreOrchestrationVersion", + "action": "orchestrations:RestoreOrchestrationVersion", + "description": "Restore an archived orchestration graph as a new version" + }, { "operationId": "startRun", "action": "orchestrations:StartRun", diff --git a/packages/server/src/rest/openapi/v1/orchestrations.yaml b/packages/server/src/rest/openapi/v1/orchestrations.yaml index e20ab1df..87113921 100644 --- a/packages/server/src/rest/openapi/v1/orchestrations.yaml +++ b/packages/server/src/rest/openapi/v1/orchestrations.yaml @@ -236,6 +236,153 @@ paths: '404': description: Not found + /api/v1/orchestrations/{orchestration_id}/versions: + get: + tags: + - Orchestrations + summary: List an orchestration's graph versions + description: > + Returns the orchestration's archived graphs, newest first. A version is + written on create and on every subsequent write that changes the graph + (`nodes`, `edges`, `state_schema`, `input_schema`) — through the REST API + or a formation apply alike. Metadata-only edits (name, description) do not + archive a version. See + [Versioning](/docs/modules/orchestrations#versioning). + operationId: listOrchestrationVersions + parameters: + - $ref: '#/components/parameters/orchestration_id' + - name: limit + in: query + required: false + description: Maximum number of results to return + schema: + type: integer + default: 50 + - name: offset + in: query + required: false + description: Number of results to skip + schema: + type: integer + default: 0 + responses: + '200': + description: List of orchestration versions, newest first + content: + application/json: + schema: + type: object + required: + - data + - total + - limit + - offset + properties: + data: + type: array + items: + $ref: '#/components/schemas/OrchestrationVersion' + total: + type: integer + limit: + type: integer + offset: + type: integer + '401': + description: Unauthorized + '403': + description: Forbidden + '404': + description: Orchestration not found + + /api/v1/orchestrations/{orchestration_id}/versions/{version}: + get: + tags: + - Orchestrations + summary: Fetch an archived orchestration version + description: > + Returns the exact graph a given version describes. Every run records the + version it started on in `orchestration_version` and executes that graph + for its whole life, so this is how you read the topology a run actually + took — including a run whose orchestration has been rewired since. + operationId: getOrchestrationVersion + parameters: + - $ref: '#/components/parameters/orchestration_id' + - name: version + in: path + required: true + description: The archived version number + schema: + type: integer + minimum: 1 + responses: + '200': + description: Archived orchestration version + content: + application/json: + schema: + $ref: '#/components/schemas/OrchestrationVersion' + '400': + description: Bad Request — version is not a positive integer + '401': + description: Unauthorized + '403': + description: Forbidden + '404': + description: Not found + + /api/v1/orchestrations/{orchestration_id}/versions/{version}/restore: + post: + tags: + - Orchestrations + summary: Restore an archived orchestration graph + description: > + Writes an archived version's graph back as the orchestration's live + definition, which archives it again as a **new** version rather than + rewinding the counter — so a run pinned to any version in between still + resolves the graph it started on. + + + The restore runs through the ordinary update path, so the archived graph + goes through the same static validation as an authored one. Node resource + references (`agent_id`, `tool_id`, `orchestration_id`) resolve when a run + reaches the node, so a target deleted since the snapshot was taken restores + cleanly and surfaces as a failed run rather than a `400`. Restoring the + graph the orchestration already holds is a no-op and archives nothing. Runs + already in flight are unaffected either way — a restore is an ordinary + edit, and pinning is what keeps it from reaching them. + operationId: restoreOrchestrationVersion + parameters: + - $ref: '#/components/parameters/orchestration_id' + - name: version + in: path + required: true + description: The archived version number + schema: + type: integer + minimum: 1 + requestBody: + required: false + content: + application/json: + schema: + $ref: '#/components/schemas/RestoreOrchestrationVersionRequest' + responses: + '200': + description: The orchestration, at its new version + content: + application/json: + schema: + $ref: '#/components/schemas/Orchestration' + '400': + description: Bad Request — version is not a positive integer + '401': + description: Unauthorized + '403': + description: Forbidden + '404': + description: Not found + /api/v1/orchestration-runs: post: tags: @@ -755,6 +902,7 @@ components: - id - project_id - name + - version - nodes - edges - created_at @@ -774,6 +922,14 @@ components: type: string nullable: true description: Optional description. + version: + type: integer + description: > + Incremented on every write that changes the graph; prior versions are + archived. A run pins the version it started on, so these fields are a + draft for runs started from now on rather than a live rewrite of the + ones already executing. + example: 1 nodes: type: array items: @@ -828,6 +984,11 @@ components: input_schema: type: object nullable: true + version_label: + type: string + description: >- + Optional tag for the version this create archives, e.g. `initial`. + example: initial UpdateOrchestrationRequest: type: object @@ -851,6 +1012,89 @@ components: input_schema: type: object nullable: true + version_label: + type: string + description: >- + Optional tag for the version this write archives, e.g. `pre-rewire`. + Ignored when the write changes no graph field, since no version is + archived. + example: pre-rewire + + OrchestrationVersion: + type: object + description: >- + An immutable archive of an orchestration's graph at one version. + properties: + id: + type: string + description: Public ID of the archived version + example: orch_ver_V1StGXR8Z5jdHi6B + orchestration_id: + x-soat-ref: orchestrations + type: string + description: Public ID of the orchestration this version belongs to + example: orch_V1StGXR8Z5jdHi6B + version: + type: integer + description: The archived version number + example: 1 + config: + type: object + additionalProperties: true + description: >- + The orchestration's versioned surface as it stood at this version: + `nodes`, `edges`, `state_schema` and `input_schema`. Name and + description are metadata — bumping the version when one of them changes + would make two version numbers denote the same topology, which is + exactly what a run cites. + + + Deliberately open rather than a fixed schema: an archive written by an + earlier release of SOAT reflects the orchestration surface **of its own + time**, so it may carry fields the current API no longer documents. + properties: + nodes: + type: array + items: + $ref: '#/components/schemas/OrchestrationNode' + edges: + type: array + items: + $ref: '#/components/schemas/OrchestrationEdge' + state_schema: + type: object + nullable: true + input_schema: + type: object + nullable: true + label: + type: string + nullable: true + description: >- + Optional human tag for this version, e.g. `pre-rewire`. Set from the + `version_label` field of a write, the `label` field of a restore, or + generated for one. + example: restored from v2 + created_by: + x-soat-ref: users + type: string + nullable: true + description: >- + Public ID of the user whose action produced this version. Null for + writes with no request user behind them. + created_at: + type: string + format: date-time + + RestoreOrchestrationVersionRequest: + type: object + properties: + label: + type: string + description: >- + Optional tag for the version the restore creates. Defaults to + `restored from v`. + example: rollback to pre-incident graph OrchestrationRun: type: object @@ -872,6 +1116,21 @@ components: x-soat-ref: orchestrations type: string description: Public ID of the parent orchestration. + orchestration_version: + type: integer + nullable: true + description: > + The orchestration version this run executes, fixed when the run + started. Every later step of the run — the first drive, a wake from + `sleeping`, a human or approval resume, a redrive after a crash — + resolves the graph from this version, so editing the orchestration + never re-shapes a run already in flight. Fetch the graph it names at + `GET /api/v1/orchestrations/{orchestration_id}/versions/{version}`. + + + Null for runs created before pinning existed, which execute the live + graph. + example: 3 project_id: x-soat-ref: projects type: string diff --git a/packages/server/src/rest/v1/orchestrationVersions.ts b/packages/server/src/rest/v1/orchestrationVersions.ts new file mode 100644 index 00000000..05a74515 --- /dev/null +++ b/packages/server/src/rest/v1/orchestrationVersions.ts @@ -0,0 +1,129 @@ +import { Router } from '@ttoss/http-server'; +import type { Context } from 'src/Context'; +import { DomainError } from 'src/errors'; +import { + getOrchestrationVersion, + listOrchestrationVersions, + restoreOrchestrationVersion, +} from 'src/lib/orchestrationVersions'; + +import { parsePagination } from './helpers'; + +/** + * Orchestration graph version history (issue #872). + * + * Versions are never written through this router: they are archived by the + * shared orchestration write path, so this surface is read-only apart from + * `restore`, which expresses itself as an ordinary orchestration update carrying + * an archived graph. + * + * A separate router, mounted onto `orchestrationsRouter`, mirroring how + * `agentVersions.ts` hangs off the agents router. + */ +export const orchestrationVersionsRouter = new Router(); + +/** Path-param `{version}` is a version *number*, not a public ID. */ +const parseVersionParam = (raw: string): number => { + const version = Number(raw); + if (!Number.isInteger(version) || version < 1) { + throw new DomainError( + 'VALIDATION_FAILED', + 'version must be a positive integer.' + ); + } + return version; +}; + +const checkOrchestrationAccess = async ( + ctx: Context, + action: string +): Promise => { + if (!ctx.authUser) { + ctx.status = 401; + ctx.body = { error: 'Unauthorized' }; + return null; + } + const projectIds = await ctx.authUser.resolveProjectIds({ + action, + resourceType: 'orchestration', + }); + if (projectIds === null) { + ctx.status = 403; + ctx.body = { error: 'Forbidden' }; + return null; + } + return projectIds ?? undefined; +}; + +/** + * @openapi + * /api/v1/orchestrations/{orchestration_id}/versions: + * get: + * $ref: 'openapi/v1/orchestrations.yaml#/paths/~1api~1v1~1orchestrations~1{orchestration_id}~1versions/get' + */ +orchestrationVersionsRouter.get( + '/orchestrations/:orchestration_id/versions', + async (ctx: Context) => { + const projectIds = await checkOrchestrationAccess( + ctx, + 'orchestrations:ListOrchestrationVersions' + ); + if (projectIds === null) return; + + ctx.body = await listOrchestrationVersions({ + projectIds, + orchestrationId: ctx.params['orchestration_id'] as string, + ...parsePagination(ctx), + }); + } +); + +/** + * @openapi + * /api/v1/orchestrations/{orchestration_id}/versions/{version}: + * get: + * $ref: 'openapi/v1/orchestrations.yaml#/paths/~1api~1v1~1orchestrations~1{orchestration_id}~1versions~1{version}/get' + */ +orchestrationVersionsRouter.get( + '/orchestrations/:orchestration_id/versions/:version', + async (ctx: Context) => { + const projectIds = await checkOrchestrationAccess( + ctx, + 'orchestrations:GetOrchestrationVersion' + ); + if (projectIds === null) return; + + ctx.body = await getOrchestrationVersion({ + projectIds, + orchestrationId: ctx.params['orchestration_id'] as string, + version: parseVersionParam(ctx.params['version'] as string), + }); + } +); + +/** + * @openapi + * /api/v1/orchestrations/{orchestration_id}/versions/{version}/restore: + * post: + * $ref: 'openapi/v1/orchestrations.yaml#/paths/~1api~1v1~1orchestrations~1{orchestration_id}~1versions~1{version}~1restore/post' + */ +orchestrationVersionsRouter.post( + '/orchestrations/:orchestration_id/versions/:version/restore', + async (ctx: Context) => { + const projectIds = await checkOrchestrationAccess( + ctx, + 'orchestrations:RestoreOrchestrationVersion' + ); + if (projectIds === null) return; + + const body = (ctx.request.body ?? {}) as { label?: unknown }; + + ctx.body = await restoreOrchestrationVersion({ + projectIds, + orchestrationId: ctx.params['orchestration_id'] as string, + version: parseVersionParam(ctx.params['version'] as string), + label: typeof body.label === 'string' ? body.label : undefined, + createdByUserId: ctx.authUser?.id, + }); + } +); diff --git a/packages/server/src/rest/v1/orchestrations.ts b/packages/server/src/rest/v1/orchestrations.ts index dfdff4e7..373e906b 100644 --- a/packages/server/src/rest/v1/orchestrations.ts +++ b/packages/server/src/rest/v1/orchestrations.ts @@ -25,12 +25,15 @@ import { import { parseRunInput, parseUpdateBody, + parseVersionLabel, type RawCreateBody, type RawUpdateBody, validateCreateBody, } from './orchestrationsRequestBody'; +import { orchestrationVersionsRouter } from './orchestrationVersions'; export const orchestrationsRouter = new Router(); + const resolveAuth = async ( ctx: Context, action: string, @@ -123,6 +126,8 @@ orchestrationsRouter.post('/orchestrations', async (ctx: Context) => { body.input_schema != null && typeof body.input_schema === 'object' ? body.input_schema : undefined, + versionLabel: parseVersionLabel(body.version_label), + createdByUserId: ctx.authUser?.id, }); ctx.status = 201; @@ -239,6 +244,7 @@ orchestrationsRouter.patch( id: orchestrationId, projectIds: projectIds ?? undefined, ...parseUpdateBody(body), + createdByUserId: ctx.authUser?.id, }); ctx.body = result; @@ -468,3 +474,8 @@ orchestrationsRouter.post( ctx.body = result; } ); + +// The version-history surface lives in its own file and hangs off this router, +// mirroring `agentVersions.ts` under the agents router. +orchestrationsRouter.use(orchestrationVersionsRouter.routes()); +orchestrationsRouter.use(orchestrationVersionsRouter.allowedMethods()); diff --git a/packages/server/src/rest/v1/orchestrationsRequestBody.ts b/packages/server/src/rest/v1/orchestrationsRequestBody.ts index c12ee8bc..a42157cb 100644 --- a/packages/server/src/rest/v1/orchestrationsRequestBody.ts +++ b/packages/server/src/rest/v1/orchestrationsRequestBody.ts @@ -19,6 +19,7 @@ export type RawCreateBody = { edges?: unknown; state_schema?: unknown; input_schema?: unknown; + version_label?: unknown; }; export type RawUpdateBody = { @@ -28,6 +29,12 @@ export type RawUpdateBody = { edges?: unknown; state_schema?: unknown; input_schema?: unknown; + version_label?: unknown; +}; + +/** The tag to attach to the version a write archives, when one was given. */ +export const parseVersionLabel = (raw: unknown): string | undefined => { + return typeof raw === 'string' ? raw : undefined; }; export const validateCreateBody = ( @@ -64,6 +71,7 @@ export const parseUpdateBody = (body: RawUpdateBody) => { body.input_schema !== undefined ? (body.input_schema as object | null) : undefined, + versionLabel: parseVersionLabel(body.version_label), }; }; diff --git a/packages/server/tests/unit/tests/lib/orchestrationRunPinning.test.ts b/packages/server/tests/unit/tests/lib/orchestrationRunPinning.test.ts new file mode 100644 index 00000000..956359b7 --- /dev/null +++ b/packages/server/tests/unit/tests/lib/orchestrationRunPinning.test.ts @@ -0,0 +1,398 @@ +import { db } from 'src/db'; +import { driveQueuedRun } from 'src/lib/orchestrationEngine'; +import { resumeOrchestrationRun } from 'src/lib/orchestrationRunActions'; +import { reapOrphanedRuns, wakeDueRuns } from 'src/lib/orchestrationScheduler'; + +import { setupProjectWithUsers } from '../../fixtures/bootstrap'; +import { authenticatedTestClient } from '../../testClient'; + +/** + * A run executes the graph it started on, not the graph the orchestration holds + * now (issue #872). + * + * Each test drives one of the four execution entry points against the real + * database, with the orchestration edited **after** the run was created: + * + * | Entry point | Reached through | + * |---|---| + * | `wakeRun` | `wakeDueRuns` — the scheduler, for a run parked on a timer | + * | `redriveRun` | `reapOrphanedRuns` — the reaper, for a run whose lease expired | + * | `resumeOrchestrationRunExecution` | `POST /orchestration-runs/:id/resume` | + * | `driveQueuedRun` | the worker's own entry point, called directly | + * + * Every graph's second node writes a marker into state, so the assertion names + * *which topology ran* rather than merely that the run finished. Before pinning + * these all resolved the live `Orchestration` row and every one of them read + * `v2` — including the wake path, where the edit can land days after the run + * started. + */ + +let userToken: string; +let projectPublicId: string; +let projectPk: number; +let orchSeq = 0; + +/** The node that reports which graph executed, via its state mapping. */ +const markerNode = (marker: string) => { + return { + id: 'answer', + type: 'transform', + expression: marker, + state_mapping: { 'state.answer': { var: 'output.result' } }, + }; +}; + +const createOrchestration = async (nodes: unknown[], edges: unknown[]) => { + orchSeq += 1; + const res = await authenticatedTestClient(userToken) + .post('/api/v1/orchestrations') + .send({ + project_id: projectPublicId, + name: `Pinning ${orchSeq}`, + nodes, + edges, + }); + expect(res.status).toBe(201); + return res.body as { id: string; version: number }; +}; + +/** Rewires the marker node to report `v2`, bumping the orchestration's version. */ +const rewireToV2 = async (args: { + orchestrationId: string; + nodes: unknown[]; + edges: unknown[]; +}) => { + const res = await authenticatedTestClient(userToken) + .patch(`/api/v1/orchestrations/${args.orchestrationId}`) + .send({ nodes: args.nodes, edges: args.edges }); + expect(res.status).toBe(200); + expect(res.body.version).toBe(2); +}; + +const orchPk = async (publicId: string): Promise => { + const orch = await db.Orchestration.findOne({ where: { publicId } }); + return orch?.id as number; +}; + +/** + * Polls a run row until it reaches one of `statuses`. Uses no timer APIs, so each + * real DB round-trip yields to the event loop and lets the scheduler's detached + * wake/redrive work progress (the shape `orchestrationScheduler.test.ts` uses). + */ +const waitForRunStatus = async ( + orchestrationRunId: number, + statuses: string[] +): Promise> => { + for (let i = 0; i < 3000; i += 1) { + const run = await db.OrchestrationRun.findByPk(orchestrationRunId); + if (run && statuses.includes(run.status)) return run; + } + throw new Error( + `run ${orchestrationRunId} never reached ${statuses.join('/')}` + ); +}; + +const runState = ( + run: InstanceType +): Record => { + return run.state as Record; +}; + +beforeAll(async () => { + const setup = await setupProjectWithUsers({ + prefix: 'orchpin', + policyActions: [ + 'orchestrations:CreateOrchestration', + 'orchestrations:UpdateOrchestration', + 'orchestrations:GetRun', + 'orchestrations:ResumeRun', + ], + createNoPermUser: false, + }); + userToken = setup.userToken; + projectPublicId = setup.projectId; + const project = await db.Project.findOne({ + where: { publicId: projectPublicId }, + }); + projectPk = project?.id as number; +}); + +describe('a run woken from `sleeping`', () => { + // The issue's own reproduction: a run parked on a `delay` whose graph is + // rewired while it sleeps. + const DELAY_NODE = { + id: 'delay', + type: 'delay', + duration: '1s', + state_mapping: { 'state.waited': { var: 'output.waited' } }, + }; + const EDGES = [{ from: 'delay', to: 'answer' }]; + + const parkSleepingRun = async (args: { + orchestrationPk: number; + orchestrationVersion: number | null; + }) => { + return db.OrchestrationRun.create({ + orchestrationId: args.orchestrationPk, + orchestrationVersion: args.orchestrationVersion, + projectId: projectPk, + status: 'sleeping', + state: {}, + activeNodes: ['delay'], + artifacts: {}, + input: {}, + startedAt: new Date(), + wakeAt: new Date(Date.now() - 1000), + wakeContext: { + nodeId: 'delay', + resume: { kind: 'delay', artifact: { waited: '1s' } }, + }, + }); + }; + + test('executes the graph it went to sleep on, not the edited one', async () => { + const orch = await createOrchestration( + [DELAY_NODE, markerNode('v1')], + EDGES + ); + const pk = await orchPk(orch.id); + const run = await parkSleepingRun({ + orchestrationPk: pk, + orchestrationVersion: orch.version, + }); + + await rewireToV2({ + orchestrationId: orch.id, + nodes: [DELAY_NODE, markerNode('v2')], + edges: EDGES, + }); + + await wakeDueRuns(); + const settled = await waitForRunStatus(run.id as number, [ + 'succeeded', + 'failed', + ]); + + expect(settled.status).toBe('succeeded'); + expect(runState(settled).answer).toBe('v1'); + }); + + test('still runs a node the edit deleted', async () => { + // The sharper form of the same bug: with the successor gone from the live + // graph, an unpinned run resolves no next node and settles having silently + // skipped the work it was created to do. + const orch = await createOrchestration( + [DELAY_NODE, markerNode('v1')], + EDGES + ); + const pk = await orchPk(orch.id); + const run = await parkSleepingRun({ + orchestrationPk: pk, + orchestrationVersion: orch.version, + }); + + await rewireToV2({ + orchestrationId: orch.id, + nodes: [DELAY_NODE], + edges: [], + }); + + await wakeDueRuns(); + const settled = await waitForRunStatus(run.id as number, [ + 'succeeded', + 'failed', + ]); + + expect(settled.status).toBe('succeeded'); + expect(runState(settled).answer).toBe('v1'); + }); + + test('a run with no pinned version executes the live graph', async () => { + // Runs created before pinning existed carry a null version. The live row is + // the only graph they ever had, so they keep the pre-#872 behaviour rather + // than being stranded. + const orch = await createOrchestration( + [DELAY_NODE, markerNode('v1')], + EDGES + ); + const pk = await orchPk(orch.id); + const run = await parkSleepingRun({ + orchestrationPk: pk, + orchestrationVersion: null, + }); + + await rewireToV2({ + orchestrationId: orch.id, + nodes: [DELAY_NODE, markerNode('v2')], + edges: EDGES, + }); + + await wakeDueRuns(); + const settled = await waitForRunStatus(run.id as number, [ + 'succeeded', + 'failed', + ]); + + expect(settled.status).toBe('succeeded'); + expect(runState(settled).answer).toBe('v2'); + }); + + test('a pinned version whose archive is gone falls back to the live graph', async () => { + // Only reachable by deleting the archive row out of band — the API deletes + // versions with their orchestration, which takes the runs with it. The run + // degrades to the live graph rather than losing the work it has done. + const orch = await createOrchestration( + [DELAY_NODE, markerNode('v1')], + EDGES + ); + const pk = await orchPk(orch.id); + const run = await parkSleepingRun({ + orchestrationPk: pk, + orchestrationVersion: orch.version, + }); + + await rewireToV2({ + orchestrationId: orch.id, + nodes: [DELAY_NODE, markerNode('v2')], + edges: EDGES, + }); + await db.OrchestrationVersion.destroy({ + where: { orchestrationId: pk, version: orch.version }, + }); + + await wakeDueRuns(); + const settled = await waitForRunStatus(run.id as number, [ + 'succeeded', + 'failed', + ]); + + expect(settled.status).toBe('succeeded'); + expect(runState(settled).answer).toBe('v2'); + }); +}); + +describe('a run redriven after its lease expired', () => { + const FIRST_NODE = { id: 'first', type: 'transform', expression: 'start' }; + const EDGES = [{ from: 'first', to: 'answer' }]; + + test('resumes the frontier on the graph it crashed on', async () => { + const orch = await createOrchestration( + [FIRST_NODE, markerNode('v1')], + EDGES + ); + const pk = await orchPk(orch.id); + + // `first` already produced an artifact, so the redrive frontier is `answer`. + const run = await db.OrchestrationRun.create({ + orchestrationId: pk, + orchestrationVersion: orch.version, + projectId: projectPk, + status: 'running', + state: {}, + activeNodes: [], + artifacts: { first: { result: 'start' } }, + input: {}, + startedAt: new Date(), + leaseExpiresAt: new Date(Date.now() - 60_000), + }); + + await rewireToV2({ + orchestrationId: orch.id, + nodes: [FIRST_NODE, markerNode('v2')], + edges: EDGES, + }); + + await reapOrphanedRuns(); + const settled = await waitForRunStatus(run.id as number, [ + 'succeeded', + 'failed', + ]); + + expect(settled.status).toBe('succeeded'); + expect(runState(settled).answer).toBe('v1'); + }); +}); + +describe('a run resumed from `awaiting_input`', () => { + const HUMAN_NODE = { id: 'human', type: 'human', prompt: 'Approve?' }; + const EDGES = [{ from: 'human', to: 'answer' }]; + + test('finishes on the graph it parked on', async () => { + const orch = await createOrchestration( + [HUMAN_NODE, markerNode('v1')], + EDGES + ); + const pk = await orchPk(orch.id); + + const run = await db.OrchestrationRun.create({ + orchestrationId: pk, + orchestrationVersion: orch.version, + projectId: projectPk, + status: 'awaiting_input', + state: {}, + // The human node is done; the run resumes from its successor. + activeNodes: ['answer'], + artifacts: { human: { answer: 'yes' } }, + input: {}, + startedAt: new Date(), + }); + + await rewireToV2({ + orchestrationId: orch.id, + nodes: [HUMAN_NODE, markerNode('v2')], + edges: EDGES, + }); + + const resumed = await resumeOrchestrationRun({ + runPublicId: run.publicId as string, + projectIds: [projectPk], + }); + + expect(resumed.status).toBe('succeeded'); + expect(resumed.state.answer).toBe('v1'); + expect(resumed.orchestration_version).toBe(orch.version); + }); +}); + +describe('a queued run driven for the first time', () => { + const START_NODE = { id: 'first', type: 'transform', expression: 'start' }; + const EDGES = [{ from: 'first', to: 'answer' }]; + + test('drives the graph it was enqueued on', async () => { + const orch = await createOrchestration( + [START_NODE, markerNode('v1')], + EDGES + ); + const pk = await orchPk(orch.id); + + const run = await db.OrchestrationRun.create({ + orchestrationId: pk, + orchestrationVersion: orch.version, + projectId: projectPk, + status: 'queued', + state: {}, + activeNodes: [], + artifacts: {}, + input: {}, + startedAt: new Date(), + }); + + await rewireToV2({ + orchestrationId: orch.id, + nodes: [START_NODE, markerNode('v2')], + edges: EDGES, + }); + + // The worker's own entry point. Called directly rather than through a queue + // drain because the drain claims whatever else is enqueued in this database, + // which would make the assertion depend on unrelated tests' runs. + await driveQueuedRun({ run }); + + const settled = await waitForRunStatus(run.id as number, [ + 'succeeded', + 'failed', + ]); + expect(settled.status).toBe('succeeded'); + expect(runState(settled).answer).toBe('v1'); + }); +}); diff --git a/packages/server/tests/unit/tests/lib/orchestrationScheduler.test.ts b/packages/server/tests/unit/tests/lib/orchestrationScheduler.test.ts index 9eaaca15..2e3b834a 100644 --- a/packages/server/tests/unit/tests/lib/orchestrationScheduler.test.ts +++ b/packages/server/tests/unit/tests/lib/orchestrationScheduler.test.ts @@ -21,6 +21,7 @@ import { authenticatedTestClient } from '../../testClient'; const fakeRun: MappedOrchestrationRun = { id: 'orch_run_fake', orchestration_id: 'orch_fake', + orchestration_version: 1, project_id: 'prj_fake', status: 'succeeded', state: {}, diff --git a/packages/server/tests/unit/tests/rest/orchestrationVersions.test.ts b/packages/server/tests/unit/tests/rest/orchestrationVersions.test.ts new file mode 100644 index 00000000..57bcb792 --- /dev/null +++ b/packages/server/tests/unit/tests/rest/orchestrationVersions.test.ts @@ -0,0 +1,560 @@ +import { setupProjectWithUsers } from '../../fixtures/bootstrap'; +import { authenticatedTestClient, testClient } from '../../testClient'; + +const ORCHESTRATION_VERSION_ACTIONS = [ + 'orchestrations:CreateOrchestration', + 'orchestrations:GetOrchestration', + 'orchestrations:UpdateOrchestration', + 'orchestrations:DeleteOrchestration', + 'orchestrations:ListOrchestrationVersions', + 'orchestrations:GetOrchestrationVersion', + 'orchestrations:RestoreOrchestrationVersion', + 'orchestrations:StartRun', + 'orchestrations:GetRun', +]; + +/** + * Orchestration version history, on the shared archive engine (issue #872). + * + * Every assertion drives the REST entry point: versions are written by the + * shared lib choke point in `orchestrations.ts`, so a create, a `PATCH` and a + * restore are indistinguishable from here — and a formation apply, which goes + * through the same `updateOrchestration`, leaves identical history. + */ +describe('Orchestration versions', () => { + let adminToken: string; + let userToken: string; + let userId: string; + let projectId: string; + let otherProjectId: string; + let noPermToken: string; + + const NODES_V1 = [ + { id: 'a', type: 'transform', expression: 'v1' }, + { id: 'b', type: 'transform', expression: 'downstream' }, + ]; + const EDGES_V1 = [{ from: 'a', to: 'b' }]; + + const createOrchestration = async (body: Record) => { + const res = await authenticatedTestClient(userToken) + .post('/api/v1/orchestrations') + .send({ + project_id: projectId, + nodes: NODES_V1, + edges: EDGES_V1, + ...body, + }); + expect(res.status).toBe(201); + return res.body; + }; + + const patchOrchestration = async ( + id: string, + body: Record + ): Promise> => { + const res = await authenticatedTestClient(userToken) + .patch(`/api/v1/orchestrations/${id}`) + .send(body); + expect(res.status).toBe(200); + return res.body; + }; + + beforeAll(async () => { + const setup = await setupProjectWithUsers({ + prefix: 'orchver', + policyActions: ORCHESTRATION_VERSION_ACTIONS, + createOtherProject: true, + }); + + adminToken = setup.adminToken; + userToken = setup.userToken; + userId = setup.userId; + projectId = setup.projectId; + otherProjectId = setup.otherProjectId as string; + noPermToken = setup.noPermToken as string; + }); + + describe('archiving', () => { + test('creating an orchestration archives version 1', async () => { + const orch = await createOrchestration({ + name: 'Archive On Create', + version_label: 'initial', + }); + expect(orch.version).toBe(1); + + const res = await authenticatedTestClient(userToken).get( + `/api/v1/orchestrations/${orch.id}/versions/1` + ); + + expect(res.status).toBe(200); + expect(res.body.id).toMatch(/^orch_ver_/); + expect(res.body.orchestration_id).toBe(orch.id); + expect(res.body.version).toBe(1); + expect(res.body.config.nodes).toEqual(NODES_V1); + expect(res.body.config.edges).toEqual(EDGES_V1); + expect(res.body.label).toBe('initial'); + expect(res.body.created_by).toBe(userId); + expect(res.body.created_at).toBeDefined(); + }); + + test('the archived config carries the graph and nothing else', async () => { + const orch = await createOrchestration({ + name: 'Config Shape', + description: 'metadata that is not versioned', + state_schema: { type: 'object' }, + input_schema: { type: 'object', properties: { topic: {} } }, + }); + + const res = await authenticatedTestClient(userToken).get( + `/api/v1/orchestrations/${orch.id}/versions/1` + ); + + // Pinned deliberately. Only the graph is versioned: bumping the version + // when a name or description changes would make two version numbers + // denote the same topology, and the version number is exactly what a run + // cites to say which topology it executed. + expect(Object.keys(res.body.config).sort()).toEqual([ + 'edges', + 'input_schema', + 'nodes', + 'state_schema', + ]); + }); + + test('a graph change bumps the version and archives the new graph', async () => { + const orch = await createOrchestration({ name: 'Bumps' }); + + const updated = await patchOrchestration(orch.id, { + nodes: [{ id: 'a', type: 'transform', expression: 'v2' }], + edges: [], + version_label: 'rewired', + }); + expect(updated.version).toBe(2); + + const v1 = await authenticatedTestClient(userToken).get( + `/api/v1/orchestrations/${orch.id}/versions/1` + ); + const v2 = await authenticatedTestClient(userToken).get( + `/api/v1/orchestrations/${orch.id}/versions/2` + ); + + expect(v1.body.config.nodes).toHaveLength(2); + expect(v2.body.config.nodes).toEqual([ + { id: 'a', type: 'transform', expression: 'v2' }, + ]); + expect(v2.body.label).toBe('rewired'); + }); + + test('a metadata-only edit archives no version', async () => { + const orch = await createOrchestration({ name: 'Metadata Only' }); + + const updated = await patchOrchestration(orch.id, { + name: 'Metadata Only, Renamed', + description: 'still the same graph', + }); + + expect(updated.version).toBe(1); + const versions = await authenticatedTestClient(userToken).get( + `/api/v1/orchestrations/${orch.id}/versions` + ); + expect(versions.body.total).toBe(1); + }); + + test('re-writing the graph the orchestration already holds archives no version', async () => { + const orch = await createOrchestration({ name: 'No-op Write' }); + + const updated = await patchOrchestration(orch.id, { + nodes: NODES_V1, + edges: EDGES_V1, + }); + + expect(updated.version).toBe(1); + const versions = await authenticatedTestClient(userToken).get( + `/api/v1/orchestrations/${orch.id}/versions` + ); + expect(versions.body.total).toBe(1); + }); + + test('a JSON Logic expression round-trips through the archive verbatim', async () => { + // The graph is author-authored data the platform does not own: a `var` + // path must survive archiving byte-for-byte, underscores included + // (.claude/rules/case-convention.md). + const nodes = [ + { + id: 'gate', + type: 'condition', + expression: { '>': [{ var: 'state.max_daily_budget' }, 100] }, + }, + ]; + + const orch = await createOrchestration({ + name: 'Author Payloads', + nodes, + edges: [], + }); + + const res = await authenticatedTestClient(userToken).get( + `/api/v1/orchestrations/${orch.id}/versions/1` + ); + expect(res.body.config.nodes).toEqual(nodes); + }); + }); + + describe('pinning a run to the version it started on', () => { + test('starting a run records the orchestration version it runs', async () => { + const orch = await createOrchestration({ name: 'Pins At Start' }); + await patchOrchestration(orch.id, { + nodes: [{ id: 'a', type: 'transform', expression: 'v2' }], + edges: [], + }); + + const res = await authenticatedTestClient(userToken) + .post('/api/v1/orchestration-runs') + .send({ orchestration_id: orch.id, wait: true }); + + expect(res.status).toBe(201); + expect(res.body.orchestration_version).toBe(2); + }); + }); + + describe('GET /api/v1/orchestrations/:orchestration_id/versions', () => { + let listedId: string; + + beforeAll(async () => { + const orch = await createOrchestration({ name: 'Listed' }); + listedId = orch.id; + await patchOrchestration(listedId, { + nodes: [{ id: 'a', type: 'transform', expression: 'second' }], + edges: [], + }); + await patchOrchestration(listedId, { + nodes: [{ id: 'a', type: 'transform', expression: 'third' }], + edges: [], + }); + }); + + test('lists versions newest first', async () => { + const res = await authenticatedTestClient(userToken).get( + `/api/v1/orchestrations/${listedId}/versions` + ); + + expect(res.status).toBe(200); + expect(res.body.total).toBe(3); + expect( + res.body.data.map((row: { version: number }) => { + return row.version; + }) + ).toEqual([3, 2, 1]); + expect(res.body.data[0].config.nodes[0].expression).toBe('third'); + }); + + test('honors limit and offset', async () => { + const res = await authenticatedTestClient(userToken).get( + `/api/v1/orchestrations/${listedId}/versions?limit=1&offset=1` + ); + + expect(res.status).toBe(200); + expect(res.body.total).toBe(3); + expect(res.body.limit).toBe(1); + expect(res.body.offset).toBe(1); + expect(res.body.data).toHaveLength(1); + expect(res.body.data[0].version).toBe(2); + }); + + test('unknown orchestration returns 404', async () => { + const res = await authenticatedTestClient(userToken).get( + '/api/v1/orchestrations/orch_missing/versions' + ); + expect(res.status).toBe(404); + }); + + test('an orchestration in another project resolves as not found', async () => { + const other = await authenticatedTestClient(userToken) + .post('/api/v1/orchestrations') + .send({ + project_id: otherProjectId, + name: 'Other Project', + nodes: NODES_V1, + edges: EDGES_V1, + }); + expect(other.status).toBe(201); + + const res = await authenticatedTestClient(noPermToken).get( + `/api/v1/orchestrations/${other.body.id}/versions` + ); + expect(res.status).toBe(404); + }); + + test('unauthenticated request returns 401', async () => { + const res = await testClient.get( + `/api/v1/orchestrations/${listedId}/versions` + ); + expect(res.status).toBe(401); + }); + }); + + describe('GET /api/v1/orchestrations/:orchestration_id/versions/:version', () => { + test('unknown version returns 404', async () => { + const orch = await createOrchestration({ name: 'Missing Version Get' }); + + const res = await authenticatedTestClient(userToken).get( + `/api/v1/orchestrations/${orch.id}/versions/99` + ); + expect(res.status).toBe(404); + }); + + test('non-integer version returns 400', async () => { + const orch = await createOrchestration({ name: 'Bad Version Get' }); + + const res = await authenticatedTestClient(userToken).get( + `/api/v1/orchestrations/${orch.id}/versions/abc` + ); + expect(res.status).toBe(400); + }); + }); + + describe('POST /api/v1/orchestrations/:orchestration_id/versions/:version/restore', () => { + test('restore appends a new version rather than rewinding the counter', async () => { + const orch = await createOrchestration({ name: 'Restorable' }); + await patchOrchestration(orch.id, { + nodes: [{ id: 'a', type: 'transform', expression: 'v2' }], + edges: [], + }); + + const res = await authenticatedTestClient(userToken).post( + `/api/v1/orchestrations/${orch.id}/versions/1/restore` + ); + + expect(res.status).toBe(200); + expect(res.body.version).toBe(3); + expect(res.body.nodes).toEqual(NODES_V1); + expect(res.body.edges).toEqual(EDGES_V1); + + // Version 2 — the graph that was rolled back — is still retrievable, so a + // run that executed it does not dangle. + const v2 = await authenticatedTestClient(userToken).get( + `/api/v1/orchestrations/${orch.id}/versions/2` + ); + expect(v2.status).toBe(200); + expect(v2.body.config.nodes[0].expression).toBe('v2'); + + const v3 = await authenticatedTestClient(userToken).get( + `/api/v1/orchestrations/${orch.id}/versions/3` + ); + expect(v3.body.config.nodes).toEqual(NODES_V1); + expect(v3.body.label).toBe('restored from v1'); + expect(v3.body.created_by).toBe(userId); + }); + + test('an explicit label annotates the version the restore creates', async () => { + const orch = await createOrchestration({ name: 'Labelled Restore' }); + await patchOrchestration(orch.id, { + nodes: [{ id: 'a', type: 'transform', expression: 'v2' }], + edges: [], + }); + + const res = await authenticatedTestClient(userToken) + .post(`/api/v1/orchestrations/${orch.id}/versions/1/restore`) + .send({ label: 'rollback to pre-incident graph' }); + + expect(res.status).toBe(200); + const v3 = await authenticatedTestClient(userToken).get( + `/api/v1/orchestrations/${orch.id}/versions/3` + ); + expect(v3.body.label).toBe('rollback to pre-incident graph'); + }); + + test('restoring the live graph is a no-op that creates no version', async () => { + const orch = await createOrchestration({ name: 'Restore Current' }); + + const res = await authenticatedTestClient(userToken).post( + `/api/v1/orchestrations/${orch.id}/versions/1/restore` + ); + + expect(res.status).toBe(200); + expect(res.body.version).toBe(1); + + const versions = await authenticatedTestClient(userToken).get( + `/api/v1/orchestrations/${orch.id}/versions` + ); + expect(versions.body.total).toBe(1); + }); + + test('restore leaves metadata untouched — only the graph rolls back', async () => { + const orch = await createOrchestration({ name: 'Metadata Survives' }); + await patchOrchestration(orch.id, { + nodes: [{ id: 'a', type: 'transform', expression: 'v2' }], + edges: [], + }); + await patchOrchestration(orch.id, { + name: 'Renamed After The Edit', + description: 'set after version 2', + }); + + const res = await authenticatedTestClient(userToken).post( + `/api/v1/orchestrations/${orch.id}/versions/1/restore` + ); + + expect(res.status).toBe(200); + expect(res.body.nodes).toEqual(NODES_V1); + expect(res.body.name).toBe('Renamed After The Edit'); + expect(res.body.description).toBe('set after version 2'); + }); + + test('a restored graph keeps a node reference whose target is gone', async () => { + // Deliberately a 200, not a 400. An orchestration's node references + // (`agent_id`, `tool_id`, `orchestration_id`) resolve when a run reaches + // the node, not when the graph is written — `assertOrchestrationValid` is a + // static check, so restoring a graph is exactly as legal as authoring it + // was. The dangling reference surfaces as a failed run, which is where it + // surfaces on the create path too. + const target = await createOrchestration({ name: 'Restore Target' }); + const orch = await createOrchestration({ + name: 'Dangling After Delete', + nodes: [ + { id: 'a', type: 'sub_orchestration', orchestration_id: target.id }, + ], + edges: [], + }); + await patchOrchestration(orch.id, { + nodes: [{ id: 'a', type: 'transform', expression: 'v2' }], + edges: [], + }); + + const del = await authenticatedTestClient(userToken).delete( + `/api/v1/orchestrations/${target.id}` + ); + expect(del.status).toBe(204); + + const res = await authenticatedTestClient(userToken).post( + `/api/v1/orchestrations/${orch.id}/versions/1/restore` + ); + expect(res.status).toBe(200); + expect(res.body.version).toBe(3); + expect(res.body.nodes[0].orchestration_id).toBe(target.id); + }); + + test('unknown version returns 404', async () => { + const orch = await createOrchestration({ + name: 'Missing Version Restore', + }); + + const res = await authenticatedTestClient(userToken).post( + `/api/v1/orchestrations/${orch.id}/versions/99/restore` + ); + expect(res.status).toBe(404); + }); + + test('non-integer version returns 400', async () => { + const orch = await createOrchestration({ name: 'Bad Version Restore' }); + + const res = await authenticatedTestClient(userToken).post( + `/api/v1/orchestrations/${orch.id}/versions/abc/restore` + ); + expect(res.status).toBe(400); + }); + + test('unauthenticated request returns 401', async () => { + const res = await testClient.post( + '/api/v1/orchestrations/orch_whatever/versions/1/restore' + ); + expect(res.status).toBe(401); + }); + + test('a user without RestoreOrchestrationVersion is refused', async () => { + const orch = await createOrchestration({ + name: 'No Restore Permission', + }); + + const res = await authenticatedTestClient(noPermToken).post( + `/api/v1/orchestrations/${orch.id}/versions/1/restore` + ); + // `noPermToken` resolves to an empty project list, so the orchestration is + // invisible rather than forbidden — the same shape as a cross-tenant read. + expect(res.status).toBe(404); + }); + }); + + describe('deletion', () => { + test('deleting an orchestration removes its archived versions', async () => { + const orch = await createOrchestration({ + name: 'Deleted With Versions', + }); + await patchOrchestration(orch.id, { + nodes: [{ id: 'a', type: 'transform', expression: 'v2' }], + edges: [], + }); + + const del = await authenticatedTestClient(userToken).delete( + `/api/v1/orchestrations/${orch.id}` + ); + expect(del.status).toBe(204); + + const versions = await authenticatedTestClient(userToken).get( + `/api/v1/orchestrations/${orch.id}/versions` + ); + expect(versions.status).toBe(404); + }); + }); + + describe('a restricted API key', () => { + /** + * A project-scoped API key whose policy excludes `excludedAction`. Unlike + * `noPermToken` — which resolves to an empty project list and 404s — this + * reaches the route with a resolvable project and exercises the 403 branch. + */ + const createRestrictedApiKey = async (excludedAction: string) => { + const allowedActions = ORCHESTRATION_VERSION_ACTIONS.filter((action) => { + return action !== excludedAction; + }); + const policyRes = await authenticatedTestClient(adminToken) + .post('/api/v1/policies') + .send({ + document: { + statement: [{ effect: 'Allow', action: allowedActions }], + }, + }); + expect(policyRes.status).toBe(201); + + const keyRes = await authenticatedTestClient(userToken) + .post('/api/v1/api-keys') + .send({ + project_id: projectId, + name: `No ${excludedAction} Key`, + policy_ids: [policyRes.body.id], + }); + expect(keyRes.status).toBe(201); + return keyRes.body.key as string; + }; + + test('without ListOrchestrationVersions returns 403', async () => { + const rawKey = await createRestrictedApiKey( + 'orchestrations:ListOrchestrationVersions' + ); + const res = await authenticatedTestClient(rawKey).get( + '/api/v1/orchestrations/orch_anything/versions' + ); + expect(res.status).toBe(403); + }); + + test('without GetOrchestrationVersion returns 403', async () => { + const rawKey = await createRestrictedApiKey( + 'orchestrations:GetOrchestrationVersion' + ); + const res = await authenticatedTestClient(rawKey).get( + '/api/v1/orchestrations/orch_anything/versions/1' + ); + expect(res.status).toBe(403); + }); + + test('without RestoreOrchestrationVersion returns 403', async () => { + const rawKey = await createRestrictedApiKey( + 'orchestrations:RestoreOrchestrationVersion' + ); + const res = await authenticatedTestClient(rawKey).post( + '/api/v1/orchestrations/orch_anything/versions/1/restore' + ); + expect(res.status).toBe(403); + }); + }); +}); diff --git a/packages/website/docs/modules/orchestrations.md b/packages/website/docs/modules/orchestrations.md index d6252cb9..9cc88bbb 100644 --- a/packages/website/docs/modules/orchestrations.md +++ b/packages/website/docs/modules/orchestrations.md @@ -69,6 +69,7 @@ To run an orchestration automatically — on a cron schedule, in response to an | `project_id` | string | Owning project | | `name` | string | Human-readable name | | `description` | string \| null | Optional description | +| `version` | integer | Incremented on every write that changes the graph; prior versions are archived (see [Versioning](#versioning)) | | `nodes` | array | Ordered list of node definitions | | `edges` | array | Directed connections between nodes | | `state_schema` | object | Optional JSON Schema describing the run state | @@ -82,6 +83,7 @@ To run an orchestration automatically — on a cron schedule, in response to an | ------------------ | -------------- | ----------------------------------------------------------------- | | `id` | string | Public ID (`orch_run_` prefix) | | `orchestration_id` | string | Parent orchestration | +| `orchestration_version` | integer \| null | The orchestration version this run executes, fixed when the run started (see [Versioning](#versioning)). `null` for runs created before pinning existed, which execute the live graph | | `project_id` | string | Owning project | | `status` | string | `queued` \| `running` \| `sleeping` \| `awaiting_input` \| `succeeded` \| `failed` \| `cancelled` \| `expired` | | `state` | object | Current mutable execution state | @@ -520,6 +522,79 @@ soat validate-orchestration \ # → { "valid": true, "errors": [], "warnings": [] } ``` +### Versioning + +An orchestration's graph is versioned by the same append-only archive that backs +[agent versions](./agents.md#versioning-and-staged-rollout) and +[guardrail versions](./guardrails.md#versioning). Version 1 is written on create, +and every subsequent write that **changes** the graph increments `version` and +archives the new graph as an `OrchestrationVersion`. The versioned surface is +`nodes`, `edges`, `state_schema` and `input_schema` — the graph the engine +executes, and nothing else. + +**A run executes the version it started on.** `start-orchestration-run` stamps +the orchestration's current `version` onto the run as `orchestration_version`, and +every later step of that run — the first drive out of the queue, a wake from +`sleeping`, a human or approval resume, a redrive after a worker crash — resolves +its topology from that version rather than from the live orchestration. Editing an +orchestration therefore never re-shapes a run already in flight, including one +parked for days on a `delay`, a `poll`, or an `awaiting_input` pause. The live +columns are a **draft** for runs started from now on. + +Before this, every resume path re-read the live row, so deleting a node could +leave a sleeping run skipping the work it was created to do, and `node_executions` +could reference node ids from a graph that no longer existed. + +Three writes archive nothing, and all three follow from the same rule — a version +exists to name a distinct graph: + +- a metadata-only edit (`name`, `description`); +- re-writing the graph the orchestration already holds (compared structurally, so + key order does not matter); +- restoring the version that is already live. + +`version_label` on a create or update annotates the version that write archives. +It is not stored on the orchestration and is not part of the config, so labelling +a change is never itself a change. + +| Operation | Endpoint | +| --- | --- | +| List versions, newest first | `GET /api/v1/orchestrations/{orchestration_id}/versions` | +| Fetch one version | `GET /api/v1/orchestrations/{orchestration_id}/versions/{version}` | +| Roll back to a version | `POST /api/v1/orchestrations/{orchestration_id}/versions/{version}/restore` | + +To read the topology a given run actually took, fetch the version its +`orchestration_version` names: + +```bash +soat get-orchestration-run --orchestration-run-id "$RUN_ID" +# → { "orchestration_version": 3, ... } + +soat get-orchestration-version --orchestration-id "$ORCH_ID" --version 3 +``` + +**Restore appends, it does not rewind.** Restoring v1 of an orchestration at v2 +writes v1's graph back as **v3**, so a run pinned to v2 still resolves the graph +it started on. Only the graph rolls back: `name` and `description` are left +exactly as they are. Runs already in flight are unaffected — a restore is an +ordinary graph edit, and pinning is what keeps it from reaching them. + +A restored graph goes through the same static validation as an authored one. Node +resource references (`agent_id`, `tool_id`, `orchestration_id`) resolve when a run +reaches the node, not when the graph is written, so restoring a graph whose target +has since been deleted succeeds and surfaces as a failed run — the same place it +surfaces when the graph is authored that way in the first place. + +Orchestrations have no release/canary layer, unlike agents. A run is pinned for +its whole life, which can be days; splitting *new* runs across two graphs is a +coherent idea that nothing has asked for yet. + +Pinning is per run, and a `loop` or `sub_orchestration` node starts a **new** run +of the child orchestration. That child pins the child's current version at the +moment it starts, not the version its parent was pinned to — so editing a +sub-orchestration does reach iterations that have not started yet. Version the +parent and the child together if you need a whole nested pipeline frozen. + ### Node Executions Every time a node runs, the engine persists an entry in the run's `node_executions` array capturing the resolved `input_mapping` it received, the `output` artifact it produced, its `status`, and — on failure — the structured `error`. The record is written even when a node throws, so a failed run is fully debuggable: `get-orchestration-run` shows **which** node failed, **what** input it received, and **why**, instead of only the final state plus a single error message. diff --git a/tests/smoke-tests.sh b/tests/smoke-tests.sh index d827f774..70a8af4f 100755 --- a/tests/smoke-tests.sh +++ b/tests/smoke-tests.sh @@ -1231,7 +1231,7 @@ echo "=== Orchestrations ===" echo "--- Creating orchestration-scoped auth ---" ORCH_POLICY_RESP=$($SOAT_CLI create-policy \ --name smoke-orchestration-policy \ - --document '{"statement":[{"effect":"Allow","action":["orchestrations:CreateOrchestration","orchestrations:ValidateOrchestration","orchestrations:ListOrchestrations","orchestrations:GetOrchestration","orchestrations:UpdateOrchestration","orchestrations:DeleteOrchestration","orchestrations:StartRun","orchestrations:ListRuns","orchestrations:GetRun","orchestrations:CancelRun","orchestrations:SubmitHumanInput","orchestrations:ResumeRun"]}]}' ) + --document '{"statement":[{"effect":"Allow","action":["orchestrations:CreateOrchestration","orchestrations:ValidateOrchestration","orchestrations:ListOrchestrations","orchestrations:GetOrchestration","orchestrations:UpdateOrchestration","orchestrations:DeleteOrchestration","orchestrations:ListOrchestrationVersions","orchestrations:GetOrchestrationVersion","orchestrations:RestoreOrchestrationVersion","orchestrations:StartRun","orchestrations:ListRuns","orchestrations:GetRun","orchestrations:CancelRun","orchestrations:SubmitHumanInput","orchestrations:ResumeRun"]}]}' ) ORCH_POLICY_ID=$(printf '%s\n' "$ORCH_POLICY_RESP" | jq -r '.id') if [ -z "$ORCH_POLICY_ID" ] || [ "$ORCH_POLICY_ID" = "null" ]; then echo "Failed to create orchestration policy" @@ -1363,6 +1363,54 @@ if ! printf '%s\n' "$ORCH_UPDATE_RESP" | jq -e '.description == "Smoke orchestra fi echo "Update orchestration: OK" +# Graph versioning (#872). The version endpoints are exercised here; that a +# parked run keeps executing its pinned graph is covered by the unit suite, +# which can park a run without waiting on a real timer. +echo "--- Listing orchestration versions ---" +ORCH_VER_LIST_RESP=$(SOAT_TOKEN="$ORCH_API_KEY_RAW" $SOAT_CLI list-orchestration-versions --orchestration-id "$ORCH_ID") +# The description-only update above changed no graph field, so the +# orchestration is still at version 1 with exactly one archived version. +if ! printf '%s\n' "$ORCH_VER_LIST_RESP" | jq -e '.total == 1 and .data[0].version == 1' >/dev/null 2>&1; then + echo "list-orchestration-versions did not report the create-time version" + printf '%s\n' "$ORCH_VER_LIST_RESP" + exit 1 +fi +echo "List orchestration versions: OK" + +echo "--- Getting an archived orchestration version ---" +ORCH_VER_GET_RESP=$(SOAT_TOKEN="$ORCH_API_KEY_RAW" $SOAT_CLI get-orchestration-version \ + --orchestration-id "$ORCH_ID" \ + --version 1) +if ! printf '%s\n' "$ORCH_VER_GET_RESP" | jq -e --arg id "$ORCH_ID" '.orchestration_id == $id and .version == 1 and (.config.nodes | length) == 2' >/dev/null 2>&1; then + echo "get-orchestration-version returned unexpected response" + printf '%s\n' "$ORCH_VER_GET_RESP" + exit 1 +fi +echo "Get orchestration version: OK" + +echo "--- Bumping the graph and restoring version 1 ---" +ORCH_BUMP_RESP=$(SOAT_TOKEN="$ORCH_API_KEY_RAW" $SOAT_CLI update-orchestration \ + --orchestration-id "$ORCH_ID" \ + --nodes '[{"id":"seed","type":"transform","expression":{"var":"input.theme"},"state_mapping":{"state.theme":{"var":"output.result"}}}]' \ + --edges '[]' \ + --version-label smoke-rewire) +if ! printf '%s\n' "$ORCH_BUMP_RESP" | jq -e '.version == 2 and (.nodes | length) == 1' >/dev/null 2>&1; then + echo "update-orchestration did not bump the version on a graph change" + printf '%s\n' "$ORCH_BUMP_RESP" + exit 1 +fi +# A restore appends rather than rewinding, so v1's graph comes back as v3. +ORCH_RESTORE_RESP=$(SOAT_TOKEN="$ORCH_API_KEY_RAW" $SOAT_CLI restore-orchestration-version \ + --orchestration-id "$ORCH_ID" \ + --version 1 \ + --label smoke-rollback) +if ! printf '%s\n' "$ORCH_RESTORE_RESP" | jq -e '.version == 3 and (.nodes | length) == 2' >/dev/null 2>&1; then + echo "restore-orchestration-version did not append the restored graph" + printf '%s\n' "$ORCH_RESTORE_RESP" + exit 1 +fi +echo "Restore orchestration version: OK" + echo "--- Starting completed run (synchronous wait) ---" ORCH_RUN_RESP=$(SOAT_TOKEN="$ORCH_API_KEY_RAW" $SOAT_CLI start-orchestration-run \ --orchestration-id "$ORCH_ID" \ @@ -1371,11 +1419,18 @@ ORCH_RUN_RESP=$(SOAT_TOKEN="$ORCH_API_KEY_RAW" $SOAT_CLI start-orchestration-run ORCH_RUN_ID=$(printf '%s\n' "$ORCH_RUN_RESP" | jq -r '.id') ORCH_RUN_STATUS=$(printf '%s\n' "$ORCH_RUN_RESP" | jq -r '.status') ORCH_RUN_TITLE=$(printf '%s\n' "$ORCH_RUN_RESP" | jq -r '.state.title') +ORCH_RUN_VERSION=$(printf '%s\n' "$ORCH_RUN_RESP" | jq -r '.orchestration_version') if [ "$ORCH_RUN_STATUS" != "succeeded" ] || [ "$ORCH_RUN_TITLE" != "orchestration sonnet" ]; then echo "start-orchestration-run did not complete as expected" printf '%s\n' "$ORCH_RUN_RESP" exit 1 fi +# The run is pinned to the graph it started on — version 3 after the restore. +if [ "$ORCH_RUN_VERSION" != "3" ]; then + echo "start-orchestration-run did not pin the run to the current version (got: $ORCH_RUN_VERSION)" + printf '%s\n' "$ORCH_RUN_RESP" + exit 1 +fi echo "Completed run: OK" # Worker-fleet coverage: the API tier runs with ORCHESTRATION_WORKER_DISABLED,