fix(engine): fence long-running claims and restore regression coverage

docs/README.md

@@ -125,6 +125,7 @@ open-item lists.
 - [Changelog](../CHANGELOG.md) — release history and unreleased changes.
 - [Rust review](RUST_REVIEW.md) — July 2026 best-practice review record.
 - [Security audit](SECURITY_AUDIT.md) — second Rust security audit record.
+- [Long-running engine review](reviews/2026-10-18-long-running-engine.md) — October 2026 lease, lifecycle, and recovery audit.
 
 ## Accuracy contract
 

docs/reviews/2026-10-18-long-running-engine.md

@@ -0,0 +1,34 @@
+# Long-running engine review — 2026-10-18
+
+## Scope and method
+
+Reviewed the scheduler's claim, heartbeat, retry, preemption, shutdown, and terminal paths; the Postgres, SQLite, and encrypted storage implementations; the shared lifecycle transition API; targeted storage contracts; ignored regressions; and CI/dependency checks. The repository has roughly 488 Rust source files and 174 SQL migrations, so this is a focused reliability review of the long-running execution path, not a line-by-line proof of the whole codebase.
+
+## Fixed in this review
+
+| Severity | Finding | Resolution |
+| --- | --- | --- |
+| Critical | A stale worker could use a state-only compare-and-swap after its lease was reclaimed, changing a replacement owner's `Running` instance to `Failed` or `Scheduled`. The heartbeat-then-update handback also had a check/write race. | Added a required atomic state-and-claim-token transition in both databases and the encrypted wrapper; used it in scheduler error, retry, handback, setup rollback, concurrency deferral, and preemption paths. |
+| Critical | Terminal transitions from a claimed run could complete or fail a replacement owner's row and enqueue webhooks for the wrong run. | Added transactional claim-fenced terminal transitions with outbox rows and routed claimed scheduler and step execution paths through them. |
+| High | Tree handlers used `updated_at` as a lease token after context writes had changed that field, so delayed work could park in `Waiting` instead of `Scheduled`. | Carried the original claim token through evaluator dispatch to step and loop handlers; the far-future cancellation regression now exercises the delayed tree path. |
+| High | A fixed/virtual clock could issue the same `claimed_at` token to consecutive owners, defeating fencing. A node clock ahead of Postgres could also delay stale-lease recovery. | Claim timestamps now advance at least one microsecond beyond the prior token. Postgres uses its own clock; SQLite uses wall time. Added a fixed-clock replacement-owner test for all four storage modes. |
+| High | Rows explicitly created as `Running` or `Waiting` had no claim token, so claim-fenced transitions could not process them. | Canonical instance inserts now initialize the token from the row timestamp for active states. Added a backend contract for explicitly created `Running` rows. |
+| Medium | `InstanceStore::conditional_update_instance_state` had an unconditional default that let new backends silently violate its compare-and-swap contract. | Made the method required for every backend implementation. |
+| High | A governed memory record could shed `schema_version` and `content_sha256`, then be accepted as legacy shared memory after its content changed. | Governed records now require both fields. Truly ungoverned legacy records remain readable only from instance-local KV. |
+| Medium | Six regression tests stayed ignored; five had already been fixed, while the memory test still exposed a bug. | Re-enabled tests for API capability grants, cross-tenant push acknowledgements, federation digest normalization, executor RPC exposure, native field collisions, and memory content commitment. |
+
+Existing records that have a governance envelope but lack its version or content hash will now be unreadable. Recover them only from a trusted source; accepting the partial envelope would reopen the tampering bypass.
+
+## Residual risks and recommended work
+
+1. **Hook delivery under overload (high).** `HookTasks` drops cancel/escalation hooks when its 1,024-item in-memory queue is full and aborts remaining hooks when the shutdown grace expires (`orch8-engine/src/scheduler.rs`). A durable hook queue with retry/recovery is needed if these hooks are contractual. Until then, alert on `orch8_hooks_dropped_total` and exercise overload and restart cases in an integration test.
+2. **In-flight writes after lease loss (high).** State and webhook transitions are fenced, but a handler already in progress can still save a block output, execution node, or context update after another worker reclaims the instance. Add a claim or run epoch to those write contracts and test a paused handler against reaper recovery.
+3. **External effects during lease loss (high).** An already-running handler can also finish a remote effect after its heartbeat fails. The engine should pass a stable idempotency key to every external worker/connector and test recovery during a slow effect. At-least-once execution remains the contract until that is enforced end to end.
+4. **Cross-backend failure testing (medium).** Add a fault-injection test that pauses a handler, reclaims its instance, and proves the original task cannot publish outputs or terminal events after the new claim begins. The storage contract now tests atomic state and outbox fencing, but does not simulate every in-flight handler side effect.
+5. **Dependency advisory (medium).** The current audit reports [RUSTSEC-2026-0367](https://rustsec.org/advisories/RUSTSEC-2026-0367.html), [0368](https://rustsec.org/advisories/RUSTSEC-2026-0368.html), and [0369](https://rustsec.org/advisories/RUSTSEC-2026-0369.html) for `zip 8.6.0`, a build dependency of Swagger UI generation. `cargo deny check` passes with the repository's configured exceptions. Reassess when the upstream build dependency has a usable patched version; do not classify the advisory as an engine runtime exploit without an affected call path.
+
+## Verification
+
+The new `claim_fence` contract exercises a replacement owner with an unchanged scheduler clock, stale failure/reschedule attempts, a stale terminal webhook, and a valid current-owner terminal webhook on SQLite, Postgres, and both encrypted wrappers. CI should continue to run these legs with `DATABASE_URL` set. The six restored tests should remain enabled in the workspace suite.
+
+Re-review triggers: a new storage backend; changes to claim or reaper SQL; lifecycle/step terminal paths; hook dispatch persistence; or the next `zip` dependency update.

orch8-engine/src/dataflow.rs

@@ -1937,7 +1937,6 @@ mod tests {
     }
 
     #[test]
-    #[ignore = "BUG: `order-id` and `order_id` both sanitize to `order_id`, emitting duplicate native fields — fixed on test/bugs"]
     fn colliding_sanitized_field_names_get_distinct_native_identifiers() {
         let schema = json!({
             "type": "object",

orch8-engine/src/evaluator.rs

@@ -73,6 +73,8 @@ pub enum EvalOutcome {
 struct EvalContext {
     tree: Vec<ExecutionNode>,
     instance: TaskInstance,
+    /// Stable lease token; ordinary context writes can change `updated_at`.
+    claim_token: chrono::DateTime<chrono::Utc>,
     clock: orch8_types::clock::SharedClock,
     /// When true, the tree must be re-read from storage before the next use.
     tree_stale: bool,
@@ -556,14 +558,36 @@ pub async fn evaluate_with_clock(
 /// set (the scheduler's lease-lost flag), so no further node is dispatched
 /// after the instance's lease is presumed lost. Stopping early reports
 /// [`EvalOutcome::MoreWork`].
-#[allow(clippy::too_many_lines)]
 pub async fn evaluate_fenced(
     storage: &Arc<dyn StorageBackend>,
     handlers: &HandlerRegistry,
     instance: &TaskInstance,
     sequence: &SequenceDefinition,
     clock: &orch8_types::clock::SharedClock,
     stop: &std::sync::atomic::AtomicBool,
+) -> Result<EvalOutcome, EngineError> {
+    evaluate_with_claim_token(
+        storage,
+        handlers,
+        instance,
+        sequence,
+        clock,
+        instance.updated_at,
+        stop,
+    )
+    .await
+}
+
+/// Evaluate work claimed by the scheduler using its immutable lease token.
+#[allow(clippy::too_many_lines)]
+pub(crate) async fn evaluate_with_claim_token(
+    storage: &Arc<dyn StorageBackend>,
+    handlers: &HandlerRegistry,
+    instance: &TaskInstance,
+    sequence: &SequenceDefinition,
+    clock: &orch8_types::clock::SharedClock,
+    claim_token: chrono::DateTime<chrono::Utc>,
+    stop: &std::sync::atomic::AtomicBool,
 ) -> Result<EvalOutcome, EngineError> {
     // Merge sequence blocks with any dynamically injected blocks.
     let blocks = merged_blocks(storage.as_ref(), instance.id, sequence).await?;
@@ -587,6 +611,7 @@ pub async fn evaluate_fenced(
     let mut ctx = EvalContext {
         tree: initial_tree,
         instance: initial_instance,
+        claim_token,
         clock: clock.clone(),
         tree_stale: false,
         instance_stale: false,
@@ -907,6 +932,7 @@ async fn phase_root_activation(
             sequence.interceptors.as_ref(),
             outputs_snapshot,
             &ctx.clock,
+            ctx.claim_token,
         )
         .await?;
         ctx.tree_stale = true;
@@ -969,6 +995,7 @@ async fn phase_running_steps(
                     sequence.interceptors.as_ref(),
                     outputs_snapshot,
                     &ctx.clock,
+                    ctx.claim_token,
                 ),
                 dispatch_block(
                     storage,
@@ -980,6 +1007,7 @@ async fn phase_running_steps(
                     sequence.interceptors.as_ref(),
                     outputs_snapshot,
                     &ctx.clock,
+                    ctx.claim_token,
                 ),
             )
         })
@@ -1015,6 +1043,7 @@ async fn phase_running_steps(
         sequence.interceptors.as_ref(),
         outputs_snapshot,
         &ctx.clock,
+        ctx.claim_token,
     )
     .await?;
     ctx.tree_stale = true;
@@ -1137,6 +1166,7 @@ async fn phase_composite_reevaluation(
             sequence.interceptors.as_ref(),
             outputs_snapshot,
             &ctx.clock,
+            ctx.claim_token,
         )
         .await?;
         if may_mutate_instance(block) {

orch8-engine/src/evaluator/dispatch.rs

@@ -218,6 +218,7 @@ pub(super) async fn dispatch_block(
     interceptors: Option<&orch8_types::interceptor::InterceptorDef>,
     outputs: &OutputsSnapshot,
     clock: &orch8_types::clock::SharedClock,
+    claim_token: chrono::DateTime<chrono::Utc>,
 ) -> Result<bool, EngineError> {
     // Mark node as running.
     if node.state == NodeState::Pending {
@@ -241,7 +242,14 @@ pub(super) async fn dispatch_block(
             // Boxed: the step-dispatch future (templating, placement,
             // effect guard, worker enqueue) is large; keep `evaluate` small.
             let result = Box::pin(crate::handlers::step_block::execute_step_node_with_clock(
-                storage, handlers, instance, node, step_def, outputs, clock,
+                storage,
+                handlers,
+                instance,
+                node,
+                step_def,
+                outputs,
+                clock,
+                claim_token,
             ))
             .await;
             // Interceptor: after_step
@@ -297,6 +305,7 @@ pub(super) async fn dispatch_block(
                 loop_def,
                 tree,
                 clock,
+                claim_token,
             )
             .await
         }

orch8-engine/src/handlers/loop_block.rs

@@ -73,7 +73,16 @@ pub async fn execute_loop(
     tree: &[ExecutionNode],
 ) -> Result<bool, EngineError> {
     let clock = orch8_types::clock::SharedClock::default();
-    execute_loop_with_clock(storage, instance, node, loop_def, tree, &clock).await
+    execute_loop_with_clock(
+        storage,
+        instance,
+        node,
+        loop_def,
+        tree,
+        &clock,
+        instance.updated_at,
+    )
+    .await
 }
 
 #[allow(clippy::too_many_lines, clippy::too_many_arguments)]
@@ -84,6 +93,7 @@ pub(crate) async fn execute_loop_with_clock(
     loop_def: &LoopDef,
     tree: &[ExecutionNode],
     clock: &orch8_types::clock::SharedClock,
+    claim_token: chrono::DateTime<chrono::Utc>,
 ) -> Result<bool, EngineError> {
     // Empty body: nothing to run. Completing immediately prevents a
     // condition-truthy loop from spinning on a no-op body forever.
@@ -267,8 +277,9 @@ pub(crate) async fn execute_loop_with_clock(
     if let Some(interval_secs) = loop_def.poll_interval {
         let next_fire = poll_fire_at(clock.now(), interval_secs);
         let transitioned = storage
-            .conditional_update_instance_state(
+            .conditional_update_claimed_instance_state(
                 instance.id,
+                claim_token,
                 InstanceState::Running,
                 InstanceState::Scheduled,
                 Some(next_fire),

orch8-engine/src/handlers/memory.rs

@@ -935,15 +935,15 @@ fn memory_content_sha256(content: &Value) -> Result<String, StepError> {
 }
 
 fn matches_content_commitment(record: &Value, governance: &Value) -> bool {
-    let legacy = match governance.get("schema_version") {
-        None => true,
-        Some(version) if version.as_u64() == Some(1) => false,
-        // Unknown governed schemas must not be interpreted as this version.
-        Some(_) => return false,
-    };
+    // An ungoverned record is handled separately by `governed_records` for
+    // instance-local compatibility. Once a governance envelope exists, its
+    // version and commitment are both mandatory: treating a missing version
+    // as legacy lets a tampered record remove both fields and bypass hashing.
+    if governance.get("schema_version").and_then(Value::as_u64) != Some(1) {
+        return false;
+    }
     let Some(expected) = governance.get("content_sha256") else {
-        // Only unversioned legacy records may omit the commitment.
-        return legacy;
+        return false;
     };
     let Some(expected) = expected.as_str() else {
         return false;
@@ -1827,7 +1827,6 @@ mod tests {
         }
 
         #[test]
-        #[ignore = "BUG: governed record with schema_version and content_sha256 stripped is accepted as legacy (T4 #6) — fixed on test/bugs"]
         fn stripping_the_schema_version_does_not_bypass_the_content_commitment() {
             let mut record = governed_record(&authorization(), now() + Duration::seconds(10));
             let governance = record["governance"].as_object_mut().unwrap();

orch8-engine/src/handlers/step_block.rs

@@ -337,7 +337,17 @@ pub async fn execute_step_node(
     outputs: &OutputsSnapshot,
 ) -> Result<bool, EngineError> {
     let clock = orch8_types::clock::SharedClock::default();
-    execute_step_node_with_clock(storage, handlers, instance, node, step_def, outputs, &clock).await
+    execute_step_node_with_clock(
+        storage,
+        handlers,
+        instance,
+        node,
+        step_def,
+        outputs,
+        &clock,
+        instance.updated_at,
+    )
+    .await
 }
 
 #[allow(clippy::too_many_lines, clippy::too_many_arguments)]
@@ -349,6 +359,7 @@ pub(crate) async fn execute_step_node_with_clock(
     step_def: &StepDef,
     outputs: &OutputsSnapshot,
     clock: &orch8_types::clock::SharedClock,
+    claim_token: chrono::DateTime<chrono::Utc>,
 ) -> Result<bool, EngineError> {
     // Conditional guard: if `when` is set, evaluate the expression against
     // the current context + outputs. Skip the step entirely if falsy. Strict
@@ -394,7 +405,13 @@ pub(crate) async fn execute_step_node_with_clock(
         crate::scheduler::step_preamble_deferral(storage.as_ref(), instance, step_def, clock)
             .await?
     {
-        crate::scheduler::park_tree_instance_until(storage.as_ref(), instance.id, fire_at).await?;
+        crate::scheduler::park_tree_instance_until(
+            storage.as_ref(),
+            instance.id,
+            claim_token,
+            fire_at,
+        )
+        .await?;
         return Ok(false);
     }
 
@@ -431,12 +448,13 @@ pub(crate) async fn execute_step_node_with_clock(
             patched.context.runtime.current_step_started_at = Some(node_started);
             patched
         });
-        let deferred = crate::scheduler::check_human_input_at(
+        let deferred = crate::scheduler::check_human_input_at_with_claim(
             storage.as_ref(),
             gate_instance.as_ref().unwrap_or(instance),
             step_def,
             human_def,
             clock.now(),
+            claim_token,
         )
         .await?;
         if deferred {
@@ -554,8 +572,9 @@ pub(crate) async fn execute_step_node_with_clock(
         BreakerDecision::Proceed(cow) => cow,
         BreakerDecision::Defer { fire_at } => {
             storage
-                .conditional_update_instance_state(
+                .conditional_update_claimed_instance_state(
                     instance.id,
+                    claim_token,
                     orch8_types::instance::InstanceState::Running,
                     orch8_types::instance::InstanceState::Scheduled,
                     Some(fire_at),
@@ -882,10 +901,11 @@ pub(crate) async fn execute_step_node_with_clock(
                     message = %message,
                     "tree path: retryable failure, re-scheduling instance with backoff"
                 );
-                crate::lifecycle::transition_instance(
+                crate::lifecycle::transition_claimed_instance(
                     storage.as_ref(),
                     instance.id,
                     Some(&instance.tenant_id),
+                    claim_token,
                     orch8_types::instance::InstanceState::Running,
                     orch8_types::instance::InstanceState::Scheduled,
                     Some(fire_at),