perf(warp-core): run a single execution worker on the calling thread

CHANGELOG.md

@@ -95,6 +95,12 @@
 
 ### Changed
 
+- With one execution worker (the engine default), a tick now runs its rules
+  on the thread that commits it instead of spawning a thread and waiting for
+  it. Results, their order, and panic payloads are unchanged; two or more
+  workers still each get a scoped thread. Native rules run on the caller's
+  stack and must not rely on a fresh thread-local environment per tick.
+
 - Provider semantic declarations now support explicitly bounded Core byte
   aliases, including exact 32-byte identifiers. Byte aliases participate in
   bounded record graphs and generate canonical lawpack type definitions.

crates/warp-core/src/engine_impl.rs

@@ -344,6 +344,10 @@ impl<S> EngineBuilder<S> {
     /// - Workers are capped at `NUM_SHARDS` (256) internally
     /// - Values less than 1 are treated as 1
     /// - Use explicit worker counts in benchmarks to compare parallelism levels
+    /// - With one worker (including when the cap leaves one), rules execute on
+    ///   the thread that calls commit, on its stack; no thread is spawned. With
+    ///   two or more, each worker runs on its own scoped thread. Native rules
+    ///   must not rely on a fresh thread-local environment per tick.
     #[must_use]
     pub fn workers(mut self, n: usize) -> Self {
         self.worker_count = n.max(1);

crates/warp-core/src/parallel/exec.rs

@@ -1028,6 +1028,18 @@ pub fn build_work_units(
 /// This is the **only** spawn site for cross-warp execution. Workers claim
 /// units atomically and execute items serially within each unit.
 ///
+/// # Threads
+///
+/// With `workers >= 2`, each worker runs on its own scoped thread. With
+/// `workers == 1` (the engine default) no thread is spawned: the single
+/// worker runs on the caller's thread and stack, and returns before this
+/// function does. Rule executors therefore cannot rely on a fresh thread or
+/// thread-local environment per tick. Thread-local values a rule sets persist
+/// on the caller's thread across ticks (as they already do per worker thread
+/// with two or more workers), and a deeply recursive rule uses the caller's
+/// stack rather than a spawned thread's default stack. Results, their order,
+/// and panic payloads are the same either way.
+///
 /// # Footprint Enforcement (cfg-gated)
 ///
 /// When enforcement is active, the worker loop:
@@ -1060,7 +1072,10 @@ pub fn build_work_units(
 ///
 /// # Panics
 ///
-/// Panics if `workers == 0` or if any worker thread panics.
+/// Panics if `workers == 0`, or if a worker panics outside the per-item
+/// `catch_unwind` (for example, the enforcement invariant on guards). With one
+/// worker the panic unwinds directly from the caller's thread; with more it is
+/// re-raised from the join with the same payload.
 pub fn execute_work_queue<'state, F>(
     units: &[WorkUnit],
     workers: usize,
@@ -1079,50 +1094,19 @@ where
 
     let next_unit = AtomicUsize::new(0);
 
+    // One worker gains nothing from a thread: it would run the same loop
+    // while this thread waits to join it. Run it here instead. Rules then run
+    // on the caller's thread and stack (see "Threads" above).
+    if workers == 1 {
+        return vec![drain_units(units, &next_unit, &resolve_store)];
+    }
+
     std::thread::scope(|s| {
         let handles: Vec<_> = (0..workers)
             .map(|_| {
-                let units = &units;
                 let next_unit = &next_unit;
                 let resolve_store = &resolve_store;
-
-                s.spawn(move || -> WorkerResult {
-                    let mut delta = TickDelta::new();
-                    let mut evidence = Vec::new();
-
-                    // Work-stealing loop: claim units until none remain
-                    loop {
-                        let unit_idx = next_unit.fetch_add(1, Ordering::Relaxed);
-                        if unit_idx >= units.len() {
-                            break;
-                        }
-
-                        let unit = &units[unit_idx];
-
-                        // Resolve view for this warp (per-unit, NOT cached across units)
-                        let Some(store) = resolve_store(&unit.warp_id) else {
-                            return WorkerResult::MissingStore(unit.warp_id);
-                        };
-
-                        // Execute items SERIALLY (no nested threading!)
-                        for (idx, item) in unit.items.iter().enumerate() {
-                            match execute_item_enforced(store, item, idx, unit, delta) {
-                                Ok((next_delta, record)) => {
-                                    delta = next_delta;
-                                    evidence.push(record);
-                                }
-                                Err(mut poisoned) => {
-                                    poisoned.prepend_evidence(evidence);
-                                    return WorkerResult::Poisoned(poisoned);
-                                }
-                            }
-                        }
-
-                        // View dropped here - no long-lived borrows across warps
-                    }
-
-                    WorkerResult::Success(WorkerExecution::new(delta, evidence))
-                })
+                s.spawn(move || drain_units(units, next_unit, resolve_store))
             })
             .collect();
 
@@ -1136,6 +1120,53 @@ where
     })
 }
 
+/// One worker's loop: claims units until none remain, executing each unit's
+/// items serially against its warp's store.
+fn drain_units<'state, F>(
+    units: &[WorkUnit],
+    next_unit: &AtomicUsize,
+    resolve_store: &F,
+) -> WorkerResult
+where
+    F: Fn(&WarpId) -> Option<&'state GraphStore> + Sync,
+{
+    let mut delta = TickDelta::new();
+    let mut evidence = Vec::new();
+
+    // Work-stealing loop: claim units until none remain
+    loop {
+        let unit_idx = next_unit.fetch_add(1, Ordering::Relaxed);
+        if unit_idx >= units.len() {
+            break;
+        }
+
+        let unit = &units[unit_idx];
+
+        // Resolve view for this warp (per-unit, NOT cached across units)
+        let Some(store) = resolve_store(&unit.warp_id) else {
+            return WorkerResult::MissingStore(unit.warp_id);
+        };
+
+        // Execute items SERIALLY (no nested threading!)
+        for (idx, item) in unit.items.iter().enumerate() {
+            match execute_item_enforced(store, item, idx, unit, delta) {
+                Ok((next_delta, record)) => {
+                    delta = next_delta;
+                    evidence.push(record);
+                }
+                Err(mut poisoned) => {
+                    poisoned.prepend_evidence(evidence);
+                    return WorkerResult::Poisoned(poisoned);
+                }
+            }
+        }
+
+        // View dropped here - no long-lived borrows across warps
+    }
+
+    WorkerResult::Success(WorkerExecution::new(delta, evidence))
+}
+
 struct ItemExecutionOutcome {
     delta: TickDelta,
     evidence: ExecutionFootprintEvidence,
@@ -1837,4 +1868,44 @@ mod tests {
             (ParallelExecutionPolicy::DYNAMIC_PER_SHARD, worker_hint(4),)
         );
     }
+
+    /// A missing store surfaces identically whether the single worker runs
+    /// inline on the caller's thread or workers run on scoped threads: exactly
+    /// one `MissingStore` naming the unresolvable warp, and every other worker
+    /// an empty success. The engine maps that to `EngineError::UnknownWarp`.
+    #[test]
+    fn missing_store_propagates_identically_for_every_worker_count() {
+        use super::{execute_work_queue, WorkUnit, WorkerResult};
+        use crate::make_warp_id;
+
+        let missing = make_warp_id("exec-test/missing-store");
+        let units = [WorkUnit {
+            warp_id: missing,
+            items: Vec::new(),
+            #[cfg(any(debug_assertions, feature = "footprint_enforce_release"))]
+            #[cfg(not(feature = "unsafe_graph"))]
+            guards: Vec::new(),
+        }];
+
+        for workers in [1_usize, 2, 4, 8] {
+            let results = execute_work_queue(&units, workers, |_| None);
+            assert_eq!(results.len(), workers, "one result per worker");
+            let missing_stores: Vec<_> = results
+                .iter()
+                .filter_map(|result| match result {
+                    WorkerResult::MissingStore(warp_id) => Some(*warp_id),
+                    _ => None,
+                })
+                .collect();
+            assert_eq!(missing_stores, vec![missing], "{workers} workers");
+            let empty_successes = results
+                .iter()
+                .filter(|result| {
+                    matches!(result, WorkerResult::Success(execution)
+                        if execution.delta.is_empty() && execution.evidence.is_empty())
+                })
+                .count();
+            assert_eq!(empty_successes, workers - 1, "{workers} workers");
+        }
+    }
 }

crates/warp-core/tests/parallel_engine_worker_invariance.rs

@@ -589,3 +589,103 @@ fn multiwarp_worker_count_invariance_stress() {
         }
     }
 }
+
+/// A rule panic surfaces identically whether the single worker runs inline on
+/// the caller's thread (`workers(1)`) or workers run on scoped threads.
+///
+/// One of twenty independent scopes panics in its executor. For every worker
+/// count the commit must unwind with the executor's own payload, commit
+/// nothing, retain no evidence, keep the poisoned item's evidence as a
+/// failure artifact, and leave the state root unchanged.
+#[test]
+fn rule_panic_propagates_identically_for_every_worker_count() {
+    use std::panic::{catch_unwind, AssertUnwindSafe};
+    use warp_core::{ApplyResult, EngineBuilder, NodeRecord};
+
+    const RULE_NAME: &str = "t16p/touch";
+    const PAYLOAD: &str = "t16p: executor failure on the poisoned scope";
+
+    let node_ty = warp_core::make_type_id("t16p/node");
+    let mut base_store = warp_core::GraphStore::default();
+    let root = warp_core::make_node_id("t16p/root");
+    base_store.insert_node(root, NodeRecord { ty: node_ty });
+    let mut scopes = vec![root];
+    for i in 1..20 {
+        let scope = warp_core::make_node_id(&format!("t16p/node{i}"));
+        base_store.insert_node(scope, NodeRecord { ty: node_ty });
+        scopes.push(scope);
+    }
+    let poisoned = warp_core::make_node_id("t16p/node7");
+
+    let make_rule = || {
+        let mut rule = make_touch_rule!(RULE_NAME, "t16p/marker", b"touched-t16p");
+        rule.executor = warp_core::RuleExecutor::observed(
+            |view: &mut warp_core::ExecutionGraphView<'_, '_>, scope, delta| {
+                assert!(
+                    *scope != warp_core::make_node_id("t16p/node7"),
+                    "t16p: executor failure on the poisoned scope"
+                );
+                let key = warp_core::AttachmentKey::node_alpha(warp_core::NodeKey {
+                    warp_id: view.warp_id(),
+                    local_id: *scope,
+                });
+                let payload = warp_core::AtomPayload::new(
+                    warp_core::make_type_id("t16p/marker"),
+                    bytes::Bytes::from_static(b"touched-t16p"),
+                );
+                delta.push(warp_core::WarpOp::SetAttachment {
+                    key,
+                    value: Some(warp_core::AttachmentValue::Atom(payload)),
+                });
+            },
+        );
+        rule
+    };
+
+    let run = |workers: usize| -> warp_core::Hash {
+        let mut engine = EngineBuilder::new(base_store.clone(), root)
+            .workers(workers)
+            .build();
+        engine.register_rule(make_rule()).expect("register rule");
+        let before = engine.snapshot().state_root;
+        let tx = engine.begin();
+        for scope in &scopes {
+            assert!(matches!(
+                engine.apply(tx, RULE_NAME, scope),
+                Ok(ApplyResult::Applied)
+            ));
+        }
+
+        let Err(payload) = catch_unwind(AssertUnwindSafe(|| engine.commit(tx))) else {
+            panic!("commit must unwind with {workers} workers");
+        };
+        assert_eq!(
+            payload.downcast_ref::<&str>().copied(),
+            Some(PAYLOAD),
+            "panic payload changed with {workers} workers"
+        );
+        assert!(
+            engine.last_execution_footprints().is_empty(),
+            "a failed tick retains no evidence ({workers} workers)"
+        );
+        assert!(
+            engine
+                .last_aborted_execution_footprints()
+                .iter()
+                .any(|record| record.scope() == poisoned),
+            "the poisoned item's evidence is kept as a failure artifact ({workers} workers)"
+        );
+        let after = engine.snapshot().state_root;
+        assert_eq!(before, after, "state changed with {workers} workers");
+        after
+    };
+
+    let baseline = run(1);
+    for &workers in WORKER_COUNTS {
+        assert_eq!(
+            run(workers),
+            baseline,
+            "outcome differs for {workers} workers"
+        );
+    }
+}