fix(orchestrations): pin a run to the graph it started on

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;
 

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;
   })

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;
+}

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';

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_',

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<typeof db.Orchestration>;
+  projectId: number;
+  state: Record<string, unknown>;
+  artifacts: Record<string, unknown>;
+  input?: Record<string, unknown>;
+  triggerId?: string;
+  principal?: RequestPrincipal;
+  authHeader?: string;
+  wait?: boolean;
+}): Promise<InstanceType<typeof db.OrchestrationRun>> => {
+  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.<name>" }`) 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<string, unknown> = { input: runInput };
   const artifacts: Record<string, unknown> = {};
 
-  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<string, unknown>) };
   const artifacts = { ...((run.artifacts ?? {}) as Record<string, unknown>) };
@@ -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<string, unknown>) };
   const artifacts = { ...((run.artifacts ?? {}) as Record<string, unknown>) };
@@ -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<string, unknown>) };
   const artifacts = { ...((run.artifacts ?? {}) as Record<string, unknown>) };