Harden collaborative save against malformed values (#4816)

Build on the crash fixes with broader validation, FK hardening and a
refactor of the save_workflow rescue path:

- Harden collaborative save against malformed values
- Extend UUID validation to all binary_id FKs; dedup validator
- Make cron-cursor FK compound + same-workflow
- Single advisory dangling-reference reconciler + boundary tests
- Scope save_workflow rescue to the transaction only
- Drop dangling-reference reconciler from the save path
- Extract shared Validators.valid_uuid?/1
- Rewrite save_workflow rescue helpers
- Extract save_workflow helpers to reduce complexity
- Tidy save_workflow rescue helpers + cover CastError
- Reuse shared store-test helpers in reconcile test

Author: stuartc

.claude/guidelines/store-structure.md

@@ -325,6 +325,22 @@ export const createMyStore = () => {
 5. **useSyncExternalStore** — All stores implement React 18's external store contract
 6. **Immutability** — All state updates via Immer's `produce()`
 
+### Dangling-reference reconciliation: server authoritative, client advisory
+
+Referential invariants (e.g. a cron trigger's `cron_cursor_job_id` must point at a
+live job in the same workflow) are enforced **authoritatively on the server** — the
+compound foreign key (`ON DELETE SET NULL (cron_cursor_job_id)`) plus the
+`Workflows.save_workflow/3` rescue. The client has exactly ONE advisory cleanup
+function, `adapters/reconcileDanglingReferences.ts`, invoked from every structural
+mutation that can orphan a reference (`removeJob`, `YAMLStateToYDoc.applyToYDoc`,
+and defensively before save in `saveWorkflow`/`saveAndSyncWorkflow`). It is a UX
+fast-path only: it cannot close the concurrent-editor race (a collaborator's delete
+that has not yet merged into this client's doc), and it is NOT the correctness
+guarantee. **Do not add per-path client cursor cleanup** — if a new structural
+mutation can orphan a reference, call `reconcileDanglingReferences` from it (pass
+`{ inTransaction: true }` when already inside a `ydoc.transact`); do not reimplement
+the check.
+
 ---
 
 ## Common Anti-Patterns

CHANGELOG.md

@@ -24,6 +24,20 @@ and this project adheres to
   are now logged once, in `Lightning.Credentials.Resolver`, at `info`/`warning`
   instead of `error`; only a genuinely missing project still logs at `error`.
   [#4814](https://github.com/OpenFn/lightning/issues/4814)
+- Extend UUID format validation to all `:binary_id` foreign keys on jobs,
+  triggers, edges and workflows so a malformed id surfaces as a changeset error
+  instead of an `Ecto.ChangeError` at insert; de-duplicate the validator by
+  routing `Channels.SearchParams` onto the shared
+  `Lightning.Validators.validate_uuid`.
+  [#4816](https://github.com/OpenFn/lightning/issues/4816)
+- The cron-trigger cursor (`cron_cursor_job_id`) foreign key is now compound and
+  same-workflow, matching workflow edges: a trigger's cursor may only reference
+  a job in its own workflow. Cross-workflow cursors — previously accepted
+  silently by the single-column FK — are now rejected with a changeset error on
+  save and on provisioning/import. A migration nulls any pre-existing
+  cross-workflow cursors (the cron lookup falls back to final-run state when the
+  cursor is nil); this nilification is not reversible.
+  [#4816](https://github.com/OpenFn/lightning/issues/4816)
 
 ### Fixed
 
@@ -62,16 +76,28 @@ and this project adheres to
 - Workflow channel raises an exception when fetching trigger auth methods for an
   unpersisted trigger [#4819](https://github.com/OpenFn/lightning/issues/4819)
 - Collaborative session no longer crashes when saving a cron trigger whose
-  `cron_cursor_job_id` references a job that no longer exists. The `Trigger`
-  changeset now declares the foreign key constraint, and `removeJob` clears any
-  cron cursor pointing at a deleted job, so the violation surfaces as a
-  validation error instead of an `Ecto.ConstraintError`.
+  `cron_cursor_job_id` references a deleted job. Two independent mechanisms now
+  cooperate, with the server authoritative and the client advisory: server-side,
+  the compound cron-cursor foreign key nulls the cursor when its job is deleted
+  (and rejects cross-workflow cursors), and `save_workflow/3` rescues the
+  resulting constraint error into a changeset error so the session stays up;
+  client-side, a single advisory `reconcileDanglingReferences` pass nulls
+  orphaned cursors before save as a UX fast-path. The client cleanup does not by
+  itself produce the validation error and cannot close the concurrent-editor
+  race — the server resolves that case authoritatively.
   [#4816](https://github.com/OpenFn/lightning/issues/4816)
 - Collaborative session no longer crashes when a workflow payload contains a
   malformed UUID (e.g. an unsubstituted template placeholder) for a job,
   trigger, or edge id. These ids are now validated in the changesets, so the bad
   value returns a changeset error instead of raising an `Ecto.ChangeError`
   during insert. [#4816](https://github.com/OpenFn/lightning/issues/4816)
+- Collaborative workflow saves no longer crash the session/channel when the
+  payload contains a malformed reference or value: `validate_uuid` now checks
+  with `Ecto.UUID.dump/1` (the function that runs at insert) so 16-byte non-hex
+  placeholders are rejected as changeset errors, and `Workflows.save_workflow/3`
+  rescues a typed allow-list of Ecto exceptions (`Ecto.ChangeError`,
+  `Ecto.Query.CastError`, `Ecto.ConstraintError`) and returns a changeset error
+  instead of raising. [#4816](https://github.com/OpenFn/lightning/issues/4816)
 
 ## [2.16.6] - 2026-05-27
 

assets/js/collaborative-editor/adapters/YAMLStateToYDoc.ts

@@ -8,6 +8,8 @@ import type {
 } from '../../yaml/types';
 import type { Session } from '../types/session';
 
+import { reconcileDanglingReferences } from './reconcileDanglingReferences';
+
 /**
  * YAMLStateToYDoc
  *
@@ -156,6 +158,14 @@ export class YAMLStateToYDoc {
           positionsMap.set(id, pos);
         });
       }
+
+      // 6. Reconcile dangling references introduced by the bulk replace.
+      // transformTrigger copies cron_cursor_job_id verbatim, so an imported /
+      // AI-applied workflow whose cron cursor references a job absent from the
+      // new jobs set would land dangling. Runs inside this single transaction —
+      // do NOT open a new one (Yjs forbids nesting). This is advisory; the
+      // server FK + save_workflow/3 rescue remain authoritative.
+      reconcileDanglingReferences(ydoc, { inTransaction: true });
     });
   }
 }

assets/js/collaborative-editor/adapters/reconcileDanglingReferences.ts

@@ -0,0 +1,67 @@
+import type * as Y from 'yjs';
+
+import type { Session } from '../types/session';
+
+/**
+ * Reconcile dangling references in the Y.Doc. Currently: null any cron trigger
+ * `cron_cursor_job_id` that points at a job no longer present in the jobs array.
+ *
+ * ADVISORY ONLY. This is a client-side UX fast-path so a structural mutation that
+ * orphans a cron cursor does not leave the editor in a state that fails server
+ * validation on save. It is NOT the correctness guarantee — the server-side
+ * compound foreign key (`ON DELETE SET NULL (cron_cursor_job_id)`) and the
+ * `Workflows.save_workflow/3` rescue are. In particular it CANNOT close the
+ * concurrent-editor race: if User A deletes Job B and User B saves before A's
+ * deletion has merged into B's doc, B's `cron_cursor_job_id` is still live in B's
+ * view and this function leaves it alone. The server resolves that case
+ * authoritatively.
+ *
+ * This is the single client-side home for dangling-reference reconciliation. Any
+ * new structural mutation that can orphan a reference (e.g. a future bulk
+ * job-removal or trigger-removal command) MUST call this function rather than
+ * re-implementing per-path cleanup.
+ *
+ * Pattern 1 (Y.Doc → observer → Immer → notify): callers rely on the existing
+ * observeDeep handlers to propagate the nulled cursor to React. No notify() here.
+ *
+ * @param ydoc the workflow document
+ * @param options.inTransaction true when called from within an already-open
+ *   `ydoc.transact` (e.g. `removeJob`, `applyToYDoc`); avoids nesting a second
+ *   transaction, which Yjs forbids. When omitted/false the function opens its own
+ *   transaction.
+ */
+export function reconcileDanglingReferences(
+  ydoc: Session.WorkflowDoc,
+  options: { inTransaction?: boolean } = {}
+): void {
+  const jobsArray = ydoc.getArray('jobs');
+  const triggersArray = ydoc.getArray('triggers');
+
+  // Reads first (outside the write closure) so the transaction body does only
+  // writes — mirrors removeJob's read-then-transact shape.
+  const jobs = jobsArray.toArray() as Y.Map<unknown>[];
+  const triggers = triggersArray.toArray() as Y.Map<unknown>[];
+
+  const jobIds = new Set(jobs.map(job => job.get('id') as string));
+
+  const danglingTriggers = triggers.filter(trigger => {
+    const cursor = trigger.get('cron_cursor_job_id');
+    return typeof cursor === 'string' && !jobIds.has(cursor);
+  });
+
+  // Early-return when nothing dangles so we never emit an empty transaction on
+  // every save.
+  if (danglingTriggers.length === 0) return;
+
+  const apply = () => {
+    danglingTriggers.forEach(trigger => {
+      trigger.set('cron_cursor_job_id', null);
+    });
+  };
+
+  if (options.inTransaction) {
+    apply();
+  } else {
+    ydoc.transact(apply);
+  }
+}

assets/js/collaborative-editor/stores/createWorkflowStore.ts

@@ -137,6 +137,7 @@ import { z } from 'zod';
 import _logger from '#/utils/logger';
 
 import type { WorkflowState as YAMLWorkflowState } from '../../yaml/types';
+import { reconcileDanglingReferences } from '../adapters/reconcileDanglingReferences';
 import { YAMLStateToYDoc } from '../adapters/YAMLStateToYDoc';
 import { channelRequest } from '../hooks/useChannel';
 import { notifications } from '../lib/notifications';
@@ -273,7 +274,7 @@ export const createWorkflowStore = () => {
    * @throws {Error} If Y.Doc is not initialized
    * @returns Y.Doc instance
    */
-  const ensureYDoc = (): Y.Doc => {
+  const ensureYDoc = (): Session.WorkflowDoc => {
     if (!ydoc) {
       throw new Error(
         'Cannot modify workflow: Y.Doc not initialized. ' +
@@ -941,26 +942,20 @@ export const createWorkflowStore = () => {
       // Find all incoming edges (where this job is the target)
       const incomingEdgeIndices = getIncomingEdgeIndices(edges, id);
 
-      // Find triggers whose cron cursor points at the job being removed.
-      const triggersArray = ydoc.getArray('triggers');
-      const triggers = triggersArray.toArray() as Y.Map<unknown>[];
-      const cursorTriggers = triggers.filter(
-        trigger => trigger.get('cron_cursor_job_id') === id
-      );
-
       ydoc.transact(() => {
         // Delete incoming edges first (highest index to lowest)
         incomingEdgeIndices.forEach(edgeIndex => {
           edgesArray.delete(edgeIndex, 1);
         });
 
-        // Clear any cron cursor reference to the job being removed
-        cursorTriggers.forEach(trigger => {
-          trigger.set('cron_cursor_job_id', null);
-        });
-
         // Then delete the job
         jobsArray.delete(jobIndex, 1);
+
+        // Reconcile any cron cursor that pointed at the now-deleted job. The
+        // reconciler reads the jobs array inside this open transaction, so the
+        // deleted job is already absent. This is the single advisory owner for
+        // dangling-reference cleanup — see adapters/reconcileDanglingReferences.
+        reconcileDanglingReferences(ydoc, { inTransaction: true });
       });
     }
     // Observer handles: Y.Doc → Immer → notify
@@ -1023,6 +1018,12 @@ export const createWorkflowStore = () => {
     // Observer handles: Y.Doc → Immer → notify
   };
 
+  // NOTE: there is intentionally no removeTrigger / bulk job-removal command
+  // today. If one is ever added, it MUST call
+  // reconcileDanglingReferences(ydoc, { inTransaction: true }) inside its
+  // transaction (after the structural delete) so it cannot leave a dangling cron
+  // cursor. Do not re-implement per-path cursor cleanup — see
+  // adapters/reconcileDanglingReferences and store-structure.md.
   const updateTrigger = (id: string, updates: Partial<Session.Trigger>) => {
     const ydoc = ensureYDoc();
 

assets/test/collaborative-editor/adapters/YAMLStateToYDoc.test.ts

@@ -244,6 +244,70 @@ describe('YAMLStateToYDoc', () => {
 
       expect(id).toBe('');
     });
+
+    test('reconciles a cron cursor pointing at a job absent from the imported set', () => {
+      const workflowState: YAMLWorkflowState = {
+        id: 'wf-1',
+        name: 'Imported',
+        jobs: [
+          {
+            id: 'job-a',
+            name: 'A',
+            adaptor: '@openfn/language-common@latest',
+            body: 'fn()',
+          },
+        ],
+        triggers: [
+          {
+            id: 'trigger-1',
+            type: 'cron',
+            enabled: true,
+            cron_expression: '* * * * *',
+            // Cursor points at a job NOT in `jobs` above — the uncovered path.
+            cron_cursor_job_id: 'job-ghost',
+          },
+        ],
+        edges: [],
+        positions: null,
+      };
+
+      YAMLStateToYDoc.applyToYDoc(ydoc, workflowState);
+
+      const triggersArray = ydoc.getArray('triggers');
+      const triggerMap = triggersArray.get(0) as Y.Map<unknown>;
+      expect(triggerMap.get('cron_cursor_job_id')).toBeNull();
+    });
+
+    test('leaves a valid cron cursor untouched on import', () => {
+      const workflowState: YAMLWorkflowState = {
+        id: 'wf-1',
+        name: 'Imported',
+        jobs: [
+          {
+            id: 'job-a',
+            name: 'A',
+            adaptor: '@openfn/language-common@latest',
+            body: 'fn()',
+          },
+        ],
+        triggers: [
+          {
+            id: 'trigger-1',
+            type: 'cron',
+            enabled: true,
+            cron_expression: '* * * * *',
+            cron_cursor_job_id: 'job-a',
+          },
+        ],
+        edges: [],
+        positions: null,
+      };
+
+      YAMLStateToYDoc.applyToYDoc(ydoc, workflowState);
+
+      const triggerMap = ydoc.getArray('triggers').get(0) as Y.Map<unknown>;
+      expect(triggerMap.get('cron_cursor_job_id')).toBe('job-a');
+    });
   });
 
   describe('Edge transformations', () => {