grafos_std/cpu_shared/

host_mock.rs

//! Shared-memory tasklet SDK surface — host mock (Phase 48.3 P3, Phase 48.14).
//!
//! This module exposes the *programmer-facing* types for the shared-memory
//! tasklet execution mode: a builder, a coordinator context, a worker
//! context, typed views over the shared region and per-worker scratch, and
//! partition helpers.
//!
//! # Source portability (Phase 48.14)
//!
//! The host mock and the wasm32 guest ([`super::guest`]) deliberately do
//! NOT present a single source-portable surface. They present **three
//! levels** of parity, documented in detail in
//! [`docs/grafos/shared-memory-tasklet-programming-model.md`][portability]:
//!
//! 1. **Behavioral portability** is the primary promise. The fidelity
//!    test `shared_memory_host_wasm_fidelity_e2e` guarantees that the
//!    same logical workload produces byte-identical output on both
//!    targets.
//! 2. **Source portability** is a narrower documented subset. Methods on
//!    the safe-list compile and behave identically on both targets. The
//!    cfg-list and behavior-divergent sub-list do NOT.
//! 3. **`#[cfg]` is the honest tool outside that subset.** Methods
//!    tagged with `**Target-specific.**` below require per-target source
//!    when used portably.
//!
//! Methods carrying the `**Target-specific.**` rustdoc tag are on the
//! cfg-list. Methods carrying `**Behavior diverges across targets.**`
//! compile on both targets but mean different things (see
//! [`CoordinatorCtx::barrier`] and [`WorkerCtx::barrier`]).
//!
//! [portability]: https://example.invalid
//! (See `docs/grafos/shared-memory-tasklet-programming-model.md#source-portability-between-host-mock-and-wasm32-guest`)
//!
//! ## Coordinator-lane programming model
//!
//! Phase 48.3 — the host mock implements the same coordinator-lane model
//! as the wasm32 guest path in [`super::guest`]: worker 0 is the
//! coordinator lane and runs only the coordinator closure;
//! [`CoordinatorCtx::parallel_for_workers`] spawns the worker body on
//! data worker lanes `1..worker_count()` (skipping worker 0). Source
//! compiled for either target produces identical observable behavior
//! under the `SharedMemory` execution mode, so a programmer can
//! prototype against the host mock and deploy unchanged on a real
//! fabricBIOS runtime. The `data_worker_index() / data_worker_count() /
//! data_scratch(i)` helpers expose the data-plane lane space for
//! partitioning, while the raw `worker_index() / worker_count() /
//! scratch(i)` accessors keep "raw wasm worker space" semantics for
//! advanced cases.
//!
//! ## What this is
//!
//! - The types developers will use to write a shared-memory tasklet.
//! - A host-mock launch path so the SDK is unit-testable without a runtime.
//! - The companion ABI declaration lives in [`crate::grafos_worker_v0`].
//!
//! ## What this is NOT
//!
//! - This is not the runtime. The wasmtime/wasmi-backed runtime that
//!   actually executes shared-memory tasklets is delivered as part of the
//!   P1 track. Here, `launch()` runs the coordinator and worker closures
//!   directly on the host using a `std::thread::scope` plus a real
//!   `std::sync::Barrier`, purely so the SDK abstractions can be exercised
//!   in unit tests.
//! - The programmer-facing API never mentions wasmi, wasmtime, or "threads".
//!   The abstraction is *coordinator + workers* with explicit shared and
//!   per-worker state.
//!
//! ## Programming model
//!
//! ```no_run
//! use grafos_std::cpu::{CpuBuilder, CoordinatorCtx, WorkerCtx};
//! use std::sync::atomic::{AtomicU32, Ordering};
//!
//! #[derive(Default)]
//! struct Shared {
//!     counter: AtomicU32,
//! }
//! #[derive(Default, Clone)]
//! struct Scratch {
//!     local: u32,
//! }
//!
//! # grafos_std::host::reset_mock();
//! let lease = CpuBuilder::new().cores(4).acquire().unwrap();
//! let _ = lease.cpu().shared_memory_tasklet::<Shared, Scratch>()
//!     .cores(4)
//!     .workers(4)
//!     .shared_bytes(1024)
//!     .scratch_bytes_per_worker(64)
//!     .fuel(1_000_000)
//!     .launch(|coord: &mut CoordinatorCtx<Shared, Scratch>| {
//!         // Coordinator: load input, orchestrate phases, emit output.
//!         coord.parallel_for_workers(|w| {
//!             w.scratch_mut().local = w.worker_index() as u32;
//!             w.shared().counter.fetch_add(1, Ordering::SeqCst);
//!         }).unwrap();
//!         let n = coord.shared().counter.load(Ordering::SeqCst);
//!         coord.set_output(&n.to_le_bytes());
//!         Ok(())
//!     });
//! ```

extern crate alloc;
use alloc::sync::Arc;
use alloc::vec::Vec;
use core::marker::PhantomData;
use core::sync::atomic::{AtomicBool, AtomicU64, Ordering};

use crate::cpu::FabricCpu;
use crate::error::FabricError;
use crate::grafos_worker_v0::mock::{self, with_worker_slot, WorkerSlot};

/// Phase 48.13 W2b — minimum precharge slice per worker. Mirrors the
/// wasmtime runtime constant in the shared-memory dispatcher so the host
/// mock and the real runtime agree on the precharge math.
const MIN_INITIAL_SLICE: u64 = 4096;

/// Result of a completed shared-memory tasklet.
#[derive(Debug, Clone)]
pub struct SharedTaskletResult {
    /// Exit code (0 = success).
    pub exit_code: u64,
    /// Output bytes emitted by the coordinator.
    pub output: Vec<u8>,
}

/// Errors produced by the shared-memory tasklet SDK surface.
///
/// These are SDK-side validation errors, distinct from the underlying
/// transport errors in [`FabricError`]. A successful submission may still
/// surface a [`FabricError`] from the runtime.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum TaskletError {
    /// `max_threads > num_cores`, or either is zero.
    InvalidWorkerCount,
    /// Memory envelope `shared + max_threads * scratch_per_worker` overflowed
    /// or exceeded the configured tasklet linear-memory budget.
    MemoryEnvelopeInvalid,
    /// A required builder field was not set.
    Missing(&'static str),
    /// The whole tasklet was cancelled (revoke / expiry / explicit cancel).
    Cancelled,
    /// The lease-wide shared fuel pool is exhausted (Phase 48.14 — folds
    /// Phase 48.3 follow-up #35). Produced by `?`-propagation from
    /// [`CoordinatorCtx::fuel_checkpoint`] / [`WorkerCtx::fuel_checkpoint`]
    /// via the [`From<FuelExhausted>`] impl below, so programmer source
    /// can write `ctx.fuel_checkpoint(n)?` and propagate a typed fuel
    /// error without collapsing it into [`TaskletError::Cancelled`].
    FuelExhausted,
    /// Underlying transport / host error.
    Fabric(FabricError),
}

impl From<FabricError> for TaskletError {
    fn from(e: FabricError) -> Self {
        TaskletError::Fabric(e)
    }
}

impl From<FuelExhausted> for TaskletError {
    fn from(_: FuelExhausted) -> Self {
        TaskletError::FuelExhausted
    }
}

impl From<Cancelled> for TaskletError {
    fn from(_: Cancelled) -> Self {
        TaskletError::Cancelled
    }
}

/// Returned by [`WorkerCtx::barrier`] / [`CoordinatorCtx::barrier`] when
/// the tasklet is cancelled while waiting on the barrier.
///
/// This mirrors [`super::guest::Cancelled`] on the wasm32 guest so that
/// source code written as `w.barrier()?` compiles unchanged against
/// either target. The host mock currently models cancellation with the
/// same `Arc<AtomicBool>` that backs [`CoordinatorCtx::cancel`] and
/// [`WorkerCtx::cancelled`]: after the underlying `std::sync::Barrier`
/// releases, we re-check the flag and surface `Err(Cancelled)` if set.
/// This is not a fully faithful simulation of a racing cancel mid-wait
/// — the mock cannot unstick a `std::sync::Barrier` that never
/// quorums — but it is sufficient for the "programmer writes
/// `barrier()?` and the type checks" source-parity goal.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Cancelled;

/// Returned by [`WorkerCtx::fuel_checkpoint`] /
/// [`CoordinatorCtx::fuel_checkpoint`] when the lease-wide shared fuel
/// pool is exhausted (or, defensively, when called by a V1 worker that
/// has no shared pool installed).
///
/// This mirrors [`super::guest::FuelExhausted`] on the wasm32 guest so
/// source written as `ctx.fuel_checkpoint(n)?` compiles unchanged
/// against either target. Phase 48.13 W2b: workers MUST stop computing
/// immediately on `Err(FuelExhausted)` — there is no recovery path; a
/// subsequent fuel-consuming instruction will trap fail-closed.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct FuelExhausted;

impl core::fmt::Display for FuelExhausted {
    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
        f.write_str("shared fuel pool exhausted")
    }
}

// ---------------------------------------------------------------------------
// Builder
// ---------------------------------------------------------------------------

/// Builder for a shared-memory tasklet.
///
/// Construct via [`FabricCpu::shared_memory_tasklet`]. The builder is
/// generic over the programmer's shared-state type `S` and per-worker
/// scratch type `W`. These are the types that the coordinator and workers
/// see in their respective contexts.
///
/// The builder is intentionally distinct from [`crate::cpu::TaskletBuilder`]
/// (the single-threaded path) so that "more cores" never silently means
/// "more threads": shared-memory execution is an explicit, named mode.
pub struct SharedTaskletBuilder<'a, S, W>
where
    S: Default + Send + Sync + 'static,
    W: Default + Send + 'static,
{
    _cpu: &'a FabricCpu,
    num_cores: Option<u16>,
    max_threads: Option<u16>,
    shared_bytes: Option<u32>,
    scratch_bytes_per_worker: Option<u32>,
    max_linear_memory: u32,
    fuel: u64,
    /// Phase 48.13 W2b — optional V2 shared-pool max fuel. When set, the
    /// host mock constructs an `Arc<AtomicU64>` shared pool initialized
    /// to this value, precharges each worker's slice the same way the
    /// wasmtime runtime does, and installs the pool into every worker's
    /// `WorkerSlot`. When `None`, no pool is installed and the
    /// `fuel_checkpoint` SDK methods return `Err(FuelExhausted)` (V1
    /// behavior).
    max_fuel: Option<u64>,
    input: Vec<u8>,
    _phantom: PhantomData<(S, W)>,
}

impl<'a, S, W> SharedTaskletBuilder<'a, S, W>
where
    S: Default + Send + Sync + 'static,
    W: Default + Send + 'static,
{
    fn new(cpu: &'a FabricCpu) -> Self {
        Self {
            _cpu: cpu,
            num_cores: None,
            max_threads: None,
            shared_bytes: None,
            scratch_bytes_per_worker: None,
            max_linear_memory: u32::MAX,
            fuel: 1_000_000,
            max_fuel: None,
            input: Vec::new(),
            _phantom: PhantomData,
        }
    }

    /// Number of CPU cores reserved for this tasklet (`num_cores`).
    pub fn cores(mut self, n: u16) -> Self {
        self.num_cores = Some(n);
        self
    }

    /// Number of worker lanes the runtime will launch (`max_threads`).
    /// Must satisfy `workers <= cores`.
    pub fn workers(mut self, n: u16) -> Self {
        self.max_threads = Some(n);
        self
    }

    /// Size of the shared region in bytes.
    pub fn shared_bytes(mut self, n: u32) -> Self {
        self.shared_bytes = Some(n);
        self
    }

    /// Size of each worker's private scratch region in bytes.
    pub fn scratch_bytes_per_worker(mut self, n: u32) -> Self {
        self.scratch_bytes_per_worker = Some(n);
        self
    }

    /// Override the per-tasklet linear-memory budget. Defaults to `u32::MAX`.
    /// `shared + workers * scratch_per_worker` must fit within this.
    pub fn max_linear_memory(mut self, n: u32) -> Self {
        self.max_linear_memory = n;
        self
    }

    /// Total fuel budget shared across coordinator and all workers.
    pub fn fuel(mut self, n: u64) -> Self {
        self.fuel = n;
        self
    }

    /// Phase 48.13 W2b — opt this builder into the V2 shared-pool fuel
    /// model with `max_fuel` units in the lease-wide pool.
    ///
    /// When set, [`launch`](Self::launch) constructs an
    /// `Arc<AtomicU64>` initialized to `max_fuel` and installs it into
    /// each spawned worker's `WorkerSlot`, plus the coordinator's slot
    /// during the launch lifetime. The
    /// [`fuel_checkpoint`](WorkerCtx::fuel_checkpoint) SDK methods then
    /// CAS-decrement the pool the same way the wasmtime runtime does.
    ///
    /// When unset (the default), no pool is installed and
    /// `fuel_checkpoint` returns `Err(FuelExhausted)` — this preserves
    /// the V1 behavior of all existing tests, which never call
    /// `fuel_checkpoint` and so observe identical behavior.
    pub fn with_max_fuel(mut self, max_fuel: u64) -> Self {
        self.max_fuel = Some(max_fuel);
        self
    }

    /// Input bytes the coordinator will read from shared memory.
    pub fn input(mut self, bytes: Vec<u8>) -> Self {
        self.input = bytes;
        self
    }

    fn validate(&self) -> core::result::Result<SharedMemoryEnvelope, TaskletError> {
        let cores = self.num_cores.ok_or(TaskletError::Missing("cores"))?;
        let workers = self.max_threads.ok_or(TaskletError::Missing("workers"))?;
        let shared = self
            .shared_bytes
            .ok_or(TaskletError::Missing("shared_bytes"))?;
        let scratch = self
            .scratch_bytes_per_worker
            .ok_or(TaskletError::Missing("scratch_bytes_per_worker"))?;

        if cores == 0 || workers == 0 || workers > cores {
            return Err(TaskletError::InvalidWorkerCount);
        }
        // Phase 48.3 — shared-memory tasklets need a coordinator lane
        // (worker 0) AND at least one data worker lane (worker 1..N).
        // A single-lane shared-memory tasklet has no data workers and
        // can't make progress; reject up front.
        if workers < 2 {
            return Err(TaskletError::InvalidWorkerCount);
        }
        // shared + workers * scratch  with overflow check
        let scratch_total = (workers as u64).checked_mul(scratch as u64);
        let total = scratch_total.and_then(|s| s.checked_add(shared as u64));
        let total = total.ok_or(TaskletError::MemoryEnvelopeInvalid)?;
        if total > self.max_linear_memory as u64 {
            return Err(TaskletError::MemoryEnvelopeInvalid);
        }
        Ok(SharedMemoryEnvelope {
            workers,
            shared_bytes: shared,
            scratch_bytes_per_worker: scratch,
        })
    }

    /// **Target-specific.** Launch the tasklet on the host mock. This
    /// entry point is cfg-list: the wasm32 guest's equivalent is
    /// [`super::guest::run_shared_memory_tasklet`], which has a different
    /// shape (it is called BY the runtime, not BY the program, and it
    /// takes both a coordinator closure and a worker closure because
    /// every worker lane re-enters `tasklet_run`).
    ///
    /// Takes a single coordinator closure. The host mock drives data
    /// workers from inside
    /// [`CoordinatorCtx::parallel_for_workers`], so there is no separate
    /// worker closure at launch time. Phase 48.14 removed the previous
    /// vestigial `_worker` parameter — it was never called on host and
    /// only masked the genuine cross-target shape difference.
    ///
    /// The `coordinator` closure signature
    /// (`FnOnce(&mut CoordinatorCtx<S, W>) -> Result<(), TaskletError>`)
    /// is source-portable with the coordinator closure passed to
    /// `run_shared_memory_tasklet` on wasm32, so the closure *body* can
    /// live in a shared module while the two outer entry points live
    /// under `#[cfg(target_arch = "wasm32")]` gates. See the
    /// "Cross-target source patterns" section of
    /// `docs/grafos/shared-memory-tasklet-build-guide.md`.
    pub fn launch<FCoord>(
        self,
        coordinator: FCoord,
    ) -> core::result::Result<SharedTaskletResult, TaskletError>
    where
        FCoord: FnOnce(&mut CoordinatorCtx<S, W>) -> core::result::Result<(), TaskletError>,
    {
        let env = self.validate()?;

        // Construct the shared state and per-worker scratch on the host.
        // The runtime will eventually back these by linear memory; here we
        // back them by host-side typed values so SDK tests can exercise the
        // ctx APIs end to end.
        let shared = Arc::new(S::default());
        let mut scratches: Vec<W> = (0..env.workers).map(|_| W::default()).collect();
        let cancelled = Arc::new(AtomicBool::new(false));
        let input: Arc<Vec<u8>> = Arc::new(self.input);

        // Phase 48.13 W2b — V2 shared-pool fuel construction. If the
        // builder opted in via `with_max_fuel`, build the pool now and
        // precharge each worker's slice the same way the wasmtime
        // runtime does in `fabricbiosd`'s shared-memory dispatcher:
        //
        //     slice = max(max_fuel / max_threads / 4, MIN_INITIAL_SLICE)
        //
        // The slice is debited from the pool atomically. The mock
        // doesn't actually consume host-side native fuel — it only
        // tracks the logical pool state — but the precharge math
        // matches so SDK fidelity tests can compare against the runtime.
        let shared_fuel_pool: Option<Arc<AtomicU64>> = self.max_fuel.map(|max_fuel| {
            let pool = Arc::new(AtomicU64::new(max_fuel));
            let workers = env.workers as u64;
            let slice = core::cmp::max(max_fuel / workers / 4, MIN_INITIAL_SLICE);
            // Precharge per worker. Saturating at the pool floor preserves
            // the runtime invariant that we never debit below zero.
            for _ in 0..workers {
                let mut current = pool.load(Ordering::SeqCst);
                loop {
                    let take = core::cmp::min(slice, current);
                    if take == 0 {
                        break;
                    }
                    match pool.compare_exchange_weak(
                        current,
                        current - take,
                        Ordering::SeqCst,
                        Ordering::SeqCst,
                    ) {
                        Ok(_) => break,
                        Err(observed) => current = observed,
                    }
                }
            }
            pool
        });

        let mut coord = CoordinatorCtx {
            shared: shared.clone(),
            scratches: &mut scratches,
            workers: env.workers,
            shared_bytes: env.shared_bytes,
            scratch_bytes_per_worker: env.scratch_bytes_per_worker,
            cancelled: cancelled.clone(),
            input,
            output: Vec::new(),
            shared_fuel_pool: shared_fuel_pool.clone(),
        };

        // Install a coordinator-thread WorkerSlot for the duration of
        // the coordinator closure so `coord.fuel_checkpoint(n)` can
        // observe the shared pool through the same `fb_fuel_checkpoint`
        // shim that workers use. This mirrors wasm32 where worker 0 (the
        // coordinator lane) has its own runtime slot.
        let coord_slot = WorkerSlot {
            worker_index: 0,
            worker_count: env.workers as i32,
            cancelled: cancelled.clone(),
            barrier: None,
            shared_ptr: 0,
            shared_len: 0,
            scratch_ptr: 0,
            scratch_len: 0,
            shared_fuel_pool: shared_fuel_pool.clone(),
        };
        let coord_result = with_worker_slot(coord_slot, || coordinator(&mut coord));
        coord_result?;
        if cancelled.load(Ordering::SeqCst) {
            return Err(TaskletError::Cancelled);
        }
        let output = core::mem::take(&mut coord.output);
        Ok(SharedTaskletResult {
            exit_code: 0,
            output,
        })
    }
}

#[derive(Clone, Copy)]
struct SharedMemoryEnvelope {
    workers: u16,
    shared_bytes: u32,
    scratch_bytes_per_worker: u32,
}

impl FabricCpu {
    /// Begin building a shared-memory tasklet.
    ///
    /// This is a distinct entry point from [`FabricCpu::submit`]. Reserving
    /// more CPU cores via [`crate::cpu::CpuBuilder::cores`] does not by
    /// itself produce a multi-worker tasklet — only this builder does.
    pub fn shared_memory_tasklet<S, W>(&self) -> SharedTaskletBuilder<'_, S, W>
    where
        S: Default + Send + Sync + 'static,
        W: Default + Send + 'static,
    {
        SharedTaskletBuilder::new(self)
    }
}

// ---------------------------------------------------------------------------
// CoordinatorCtx
// ---------------------------------------------------------------------------

/// Programmer-facing coordinator context.
///
/// The coordinator is the only role that may perform authority-bearing
/// hostcalls (FBMU, FBBU, leases, services, logging). Workers may compute
/// on shared state and their own scratch, but must not call into those
/// host modules.
pub struct CoordinatorCtx<'a, S, W>
where
    S: Send + Sync + 'static,
    W: Send + 'static,
{
    shared: Arc<S>,
    scratches: &'a mut Vec<W>,
    workers: u16,
    shared_bytes: u32,
    scratch_bytes_per_worker: u32,
    cancelled: Arc<AtomicBool>,
    input: Arc<Vec<u8>>,
    /// Output bytes written by the coordinator via `set_output`. Phase
    /// 48.3 — cross-target API parity with the wasm32 guest: programmer
    /// writes `coord.set_output(&bytes); Ok(())` on both targets, and the
    /// host mock's `launch` surfaces this vec to the caller.
    output: Vec<u8>,
    /// Phase 48.13 W2b — V2 lease-wide shared fuel pool. `None` for V1
    /// (per-worker-slice) launches; `Some` when the builder opted in via
    /// `with_max_fuel`. Cloned into each worker's WorkerSlot in
    /// `parallel_for_workers`.
    shared_fuel_pool: Option<Arc<AtomicU64>>,
}

impl<'a, S, W> CoordinatorCtx<'a, S, W>
where
    S: Send + Sync + 'static,
    W: Send + 'static,
{
    /// Number of worker lanes that will run.
    pub fn worker_count(&self) -> u16 {
        self.workers
    }

    /// Whether the tasklet has been cancelled.
    pub fn cancelled(&self) -> bool {
        self.cancelled.load(Ordering::SeqCst)
    }

    /// **Target-specific.** Bytes available in the shared region (host
    /// mock only; the wasm32 guest exposes length only through
    /// [`SharedRegion::len_bytes`]).
    pub fn shared_bytes(&self) -> u32 {
        self.shared_bytes
    }

    /// **Target-specific.** Bytes available in each worker's scratch
    /// region (host mock only).
    pub fn scratch_bytes_per_worker(&self) -> u32 {
        self.scratch_bytes_per_worker
    }

    /// Input bytes provided to the tasklet at submission time.
    pub fn input(&self) -> &[u8] {
        &self.input
    }

    /// Read access to shared state.
    pub fn shared(&self) -> &S {
        &self.shared
    }

    /// Read-only access to a specific worker's scratch slot.
    ///
    /// # Safety / discipline
    ///
    /// Only call this AFTER a barrier that the worker has reached. Reading
    /// a worker's scratch before that worker has finished writing to it is
    /// a race the SDK cannot detect. Typical pattern: workers write into
    /// their scratch slots and then call [`WorkerCtx::barrier`]; the
    /// coordinator calls its own barrier and only then reads
    /// `coord.scratch(i)` for each worker.
    ///
    /// Worker scratch is NOT a general shared-memory channel between
    /// workers. For ongoing communication during a phase, use the `Shared`
    /// region with atomics. Worker scratch is for results that get
    /// collected by the coordinator at well-defined phase boundaries.
    ///
    /// Takes `u32` to match the wasm32 guest signature so the same
    /// source compiles unchanged against both targets.
    pub fn scratch(&self, worker_index: u32) -> &W {
        &self.scratches[worker_index as usize]
    }

    /// **Target-specific.** Mutable access to a single worker's scratch
    /// (coordinator-only). The host mock takes a worker index because
    /// the coordinator owns `&mut` references to every lane's scratch
    /// slot; the wasm32 guest's `CoordinatorCtx::scratch_mut` takes no
    /// arguments and returns worker 0's scratch slot only. Portable
    /// source that needs cross-lane mutable scratch access must
    /// `#[cfg]`-gate this call.
    pub fn scratch_mut(&mut self, worker_index: u32) -> &mut W {
        &mut self.scratches[worker_index as usize]
    }

    /// Number of data workers (excluding the coordinator lane).
    ///
    /// Equal to `worker_count() - 1`. Mirrors the wasm32 guest helper
    /// of the same name; programmer code calling this compiles and runs
    /// identically on both targets.
    pub fn data_worker_count(&self) -> u32 {
        (self.workers as u32) - 1
    }

    /// Read-only access to a specific data worker's scratch slot.
    ///
    /// `i` is the data worker index in `0..data_worker_count()`.
    /// Internally this maps to wasm worker index `i + 1` (skipping the
    /// coordinator's own scratch slot at wasm index 0, which is unused
    /// under the coordinator-lane model).
    ///
    /// Same barrier discipline as [`CoordinatorCtx::scratch`]: only call
    /// after a barrier the target data worker has reached.
    pub fn data_scratch(&self, i: u32) -> &W {
        self.scratch(i + 1)
    }

    /// **Target-specific.** Mark the tasklet as cancelled. Host mock
    /// only — on wasm32 the runtime drives cancellation via
    /// `fb_tasklet_cancelled()` and the guest has no `cancel()` lever.
    /// Workers will observe this on their next
    /// [`WorkerCtx::cancelled`] check or barrier wait.
    pub fn cancel(&self) {
        self.cancelled.store(true, Ordering::SeqCst);
    }

    /// **Behavior diverges across targets.** See the "Source portability"
    /// section of `docs/grafos/shared-memory-tasklet-programming-model.md`.
    ///
    /// On the **host mock** this call:
    ///
    /// 1. Returns `Err(Cancelled)` if cancellation has already been
    ///    committed at the call site.
    /// 2. Does NOT actually block on anything. The coordinator lane is
    ///    not a participant in the `std::sync::Barrier` used inside
    ///    [`CoordinatorCtx::parallel_for_workers`] (that barrier sizes
    ///    to the data-worker count only).
    ///
    /// On **wasm32** ([`super::guest::CoordinatorCtx::barrier`]) this
    /// same call IS a real cross-lane barrier with the data workers,
    /// because wasm32 worker 0 re-enters `tasklet_run` alongside
    /// every other worker and all lanes rendezvous at
    /// `fb_barrier_wait()`.
    ///
    /// The recommended portable pattern: use
    /// [`CoordinatorCtx::parallel_for_workers`] as the phase boundary
    /// on host (which already performs the join), and use
    /// `coord.barrier()?` between phases on wasm32. A single source
    /// body gated with `#[cfg(target_arch = "wasm32")]` is the honest
    /// tool here — the SDK does not paper this over.
    ///
    /// Phase 48.14 closes Phase 48.3 follow-up #18 (cancellable-barrier
    /// upgrade) as WONTFIX: the divergence is documented as an
    /// intentional target-specific behavior, not slated for semantic
    /// unification.
    pub fn barrier(&self) -> core::result::Result<(), Cancelled> {
        if self.cancelled.load(Ordering::SeqCst) {
            Err(Cancelled)
        } else {
            Ok(())
        }
    }

    /// Write the coordinator's output bytes. Mirrors the wasm32 guest
    /// `CoordinatorCtx::set_output`: programmer calls this to emit the
    /// tasklet output and then returns `Ok(())` from the coordinator
    /// closure.
    ///
    /// On wasm32 the runtime copies `bytes` into the host-managed output
    /// region and the coordinator's `tasklet_run` returns the number of
    /// bytes written. Here on the host mock, `launch` collects the
    /// recorded output and returns it inside [`SharedTaskletResult`].
    pub fn set_output(&mut self, bytes: &[u8]) {
        self.output.clear();
        self.output.extend_from_slice(bytes);
    }

    /// Cooperatively reconcile fuel with the lease-wide shared pool.
    ///
    /// Returns `Ok(())` on a successful debit. Returns
    /// `Err(FuelExhausted)` when the shared pool is exhausted, in which
    /// case the coordinator MUST stop computing immediately — there is
    /// no recovery path; a subsequent fuel-consuming instruction will
    /// trap fail-closed.
    ///
    /// V2 (shared-pool) workers MUST call this at safe points within
    /// any compute loop that may consume more fuel than the initial
    /// precharge. On V1 (per-worker-slice) launches this method always
    /// returns `Err(FuelExhausted)` because no shared pool is installed
    /// — V1 source should not call it.
    ///
    /// `amount == 0` is a silent no-op returning `Ok(())`.
    ///
    /// Mirrors [`super::guest::CoordinatorCtx::fuel_checkpoint`] —
    /// source compiled for either target observes identical
    /// success/failure semantics.
    pub fn fuel_checkpoint(&mut self, amount: u64) -> core::result::Result<(), FuelExhausted> {
        if amount > i64::MAX as u64 {
            return Err(FuelExhausted);
        }
        let result = mock::fb_fuel_checkpoint(amount as i64);
        if result == 0 {
            Ok(())
        } else {
            Err(FuelExhausted)
        }
    }

    /// **Target-specific.** Run a worker phase: invokes `body` once per **data worker lane**
    /// in parallel, each with its own [`WorkerCtx`]. Returns when every
    /// data worker has completed (the implicit tasklet-wide barrier).
    ///
    /// Phase 48.3 — runs the body on data worker lanes
    /// `1..worker_count()`. Worker 0 is the coordinator lane and does
    /// not run the worker closure on either host mock or wasm32
    /// targets — this matches the wasm32 SDK contract in
    /// [`super::guest::run_shared_memory_tasklet`]. The number of body
    /// invocations is therefore `data_worker_count() == worker_count() - 1`.
    ///
    /// `body` may compute on shared state (via interior mutability such as
    /// atomics) and on its own scratch. It may NOT perform authority-bearing
    /// hostcalls — those are coordinator-only by contract.
    ///
    /// This method is host-mock-only. The wasm32 guest drives workers by
    /// having the runtime call `tasklet_run` once per lane; a program
    /// writing source for both targets must `#[cfg]`-gate the outer
    /// entry-point structure (see "Cross-target source patterns" in
    /// the build guide). The worker closure *body* can still be shared
    /// verbatim between the two entry points.
    pub fn parallel_for_workers<F>(&mut self, body: F) -> core::result::Result<(), TaskletError>
    where
        F: Fn(&mut WorkerCtx<S, W>) + Sync + Send,
    {
        let workers = self.workers;
        // Validated at builder time; defensive assert for the rare path
        // where a caller constructed a CoordinatorCtx by hand.
        assert!(
            workers >= 2,
            "shared-memory tasklet requires at least 2 worker lanes \
             (1 coordinator + >=1 data worker)"
        );
        let shared = self.shared.clone();
        let cancelled = self.cancelled.clone();
        let shared_fuel_pool = self.shared_fuel_pool.clone();
        let input_slice: &[u8] = &self.input;
        // Barrier counts data workers only — the coordinator lane does
        // not participate in the worker barrier.
        let data_count = (workers - 1) as usize;
        let barrier = Arc::new(std::sync::Barrier::new(data_count));

        // Take exclusive references to each scratch slot. We split the Vec
        // into per-element &mut and hand each lane its own. Worker 0's
        // scratch slot is unused under the coordinator-lane model but we
        // still hold its &mut so the index math lines up with wasm32.
        let mut slots: Vec<&mut W> = self.scratches.iter_mut().collect();

        std::thread::scope(|scope| {
            let body = &body;
            let mut handles = Vec::with_capacity(data_count);
            let mut indexed: Vec<(u16, &mut W)> = slots
                .drain(..)
                .enumerate()
                .map(|(i, s)| (i as u16, s))
                .collect();

            // Drop worker 0's slot — the coordinator lane does not run
            // the body. Its scratch slot remains untouched in this
            // phase, matching wasm32 where worker 0's scratch is
            // accessible to the coordinator via `coord.scratch(0)` but
            // never written by `parallel_for_workers`.
            let _coord_slot = indexed.remove(0);

            for _ in 1..workers {
                let (idx, scratch_ref) = indexed.remove(0);
                let shared = shared.clone();
                let cancelled = cancelled.clone();
                let barrier = barrier.clone();
                let input = input_slice;
                let shared_fuel_pool = shared_fuel_pool.clone();
                handles.push(scope.spawn(move || {
                    let slot = WorkerSlot {
                        worker_index: idx as i32,
                        worker_count: workers as i32,
                        cancelled: cancelled.clone(),
                        barrier: Some(barrier.clone()),
                        shared_ptr: 0,
                        shared_len: 0,
                        scratch_ptr: 0,
                        scratch_len: 0,
                        shared_fuel_pool,
                    };
                    with_worker_slot(slot, || {
                        let mut wctx = WorkerCtx {
                            shared,
                            scratch: scratch_ref,
                            worker_index: idx,
                            worker_count: workers,
                            cancelled,
                            barrier,
                            input,
                        };
                        body(&mut wctx);
                    });
                }));
            }
            for h in handles {
                let _ = h.join();
            }
        });

        if self.cancelled.load(Ordering::SeqCst) {
            return Err(TaskletError::Cancelled);
        }
        Ok(())
    }
}

// ---------------------------------------------------------------------------
// WorkerCtx
// ---------------------------------------------------------------------------

/// Programmer-facing worker context.
///
/// Workers may read shared state, mutate their own scratch, observe
/// cancellation, and synchronize at the implicit tasklet-wide barrier.
/// Workers must not perform authority-bearing hostcalls.
pub struct WorkerCtx<'a, S, W>
where
    S: Send + Sync + 'static,
    W: Send + 'static,
{
    shared: Arc<S>,
    scratch: &'a mut W,
    worker_index: u16,
    worker_count: u16,
    cancelled: Arc<AtomicBool>,
    barrier: Arc<std::sync::Barrier>,
    input: &'a [u8],
}

impl<'a, S, W> WorkerCtx<'a, S, W>
where
    S: Send + Sync + 'static,
    W: Send + 'static,
{
    /// This worker's lane index in `[0, worker_count())`.
    pub fn worker_index(&self) -> u16 {
        self.worker_index
    }

    /// Total worker count for this tasklet.
    pub fn worker_count(&self) -> u16 {
        self.worker_count
    }

    /// Index of this worker among data workers (excluding the
    /// coordinator lane).
    ///
    /// Data workers are wasm worker indices `1..worker_count()`,
    /// exposed here as `0..data_worker_count()`. Use this for
    /// partitioning input across data workers via `partition::range` or
    /// `partition::tiles`. Mirrors the wasm32 guest helper.
    ///
    /// Under the Phase 48.3 coordinator-lane programming model, worker
    /// 0 only runs the coordinator closure, so this method is only ever
    /// called from a worker closure invocation where `worker_index() >=
    /// 1`, and the subtraction is structurally safe.
    pub fn data_worker_index(&self) -> u32 {
        (self.worker_index as u32) - 1
    }

    /// Number of data workers (worker lanes excluding the coordinator
    /// lane).
    ///
    /// Equal to `worker_count() - 1`. Always at least 1 for a valid
    /// shared-memory tasklet (the SDK builder requires `workers >= 2`).
    pub fn data_worker_count(&self) -> u32 {
        (self.worker_count as u32) - 1
    }

    /// **Behavior diverges across targets.** See the "Source portability"
    /// section of `docs/grafos/shared-memory-tasklet-programming-model.md`.
    ///
    /// Wait at the implicit tasklet-wide barrier. Returns
    /// [`Err(Cancelled)`] if the tasklet was cancelled by the time this
    /// worker releases from the barrier.
    ///
    /// On the **host mock** this IS a real barrier among the data
    /// workers: the underlying `std::sync::Barrier` is sized to
    /// `data_worker_count()` (not `worker_count()`), and the
    /// coordinator lane is not a participant. That means host-side
    /// `w.barrier()` rendezvous with other data workers, but NOT with
    /// the coordinator.
    ///
    /// On **wasm32** ([`super::guest::WorkerCtx::barrier`]) this
    /// rendezvous with the coordinator lane as well, because worker 0
    /// re-enters `tasklet_run` and participates in `fb_barrier_wait()`.
    ///
    /// The asymmetry in a single sentence: on host mock, `coord.barrier()`
    /// is a no-op-with-cancel-check while `w.barrier()` is a real
    /// data-worker rendezvous. On wasm32 both are real full-lane
    /// barriers. Source that relies on cross-lane coordinator/worker
    /// rendezvous must `#[cfg]`-gate this call.
    pub fn barrier(&self) -> core::result::Result<(), Cancelled> {
        self.barrier.wait();
        if self.cancelled.load(Ordering::SeqCst) {
            Err(Cancelled)
        } else {
            Ok(())
        }
    }

    /// Whether the tasklet has been cancelled.
    pub fn cancelled(&self) -> bool {
        self.cancelled.load(Ordering::SeqCst)
    }

    /// Read-only access to shared state. For shared mutation, the program
    /// uses interior mutability inside `S` (e.g. atomics).
    pub fn shared(&self) -> &S {
        &self.shared
    }

    /// Input bytes provided to the tasklet at submission time.
    ///
    /// Mirrors [`CoordinatorCtx::input`]: every worker sees the same
    /// bytes read-only. Workers must not mutate these bytes; for
    /// cross-worker communication use the `Shared` region with atomics.
    ///
    /// The returned slice is tied to the context's lifetime rather
    /// than to `&self`, so callers can hold it across a
    /// `scratch_mut()` borrow.
    pub fn input(&self) -> &'a [u8] {
        self.input
    }

    /// Mutable access to *this* worker's private scratch. The type system
    /// guarantees that no other worker can touch this slot.
    pub fn scratch_mut(&mut self) -> &mut W {
        self.scratch
    }

    /// **Target-specific.** Read-only access to *this* worker's scratch.
    /// Host mock only — the wasm32 guest's `WorkerCtx` exposes
    /// `scratch_mut(&mut self)` only and has no read-only accessor.
    pub fn scratch(&self) -> &W {
        self.scratch
    }

    /// Cooperatively reconcile fuel with the lease-wide shared pool.
    ///
    /// Returns `Ok(())` on a successful debit. Returns
    /// `Err(FuelExhausted)` when the shared pool is exhausted, in which
    /// case the worker MUST stop computing immediately — there is no
    /// recovery path; a subsequent fuel-consuming instruction will trap
    /// fail-closed.
    ///
    /// V2 (shared-pool) workers MUST call this at safe points within
    /// any compute loop that may consume more fuel than the initial
    /// precharge. On V1 (per-worker-slice) launches this method always
    /// returns `Err(FuelExhausted)` because no shared pool is installed
    /// — V1 source should not call it.
    ///
    /// `amount == 0` is a silent no-op returning `Ok(())`.
    ///
    /// Mirrors [`super::guest::WorkerCtx::fuel_checkpoint`].
    pub fn fuel_checkpoint(&mut self, amount: u64) -> core::result::Result<(), FuelExhausted> {
        if amount > i64::MAX as u64 {
            return Err(FuelExhausted);
        }
        let result = mock::fb_fuel_checkpoint(amount as i64);
        if result == 0 {
            Ok(())
        } else {
            Err(FuelExhausted)
        }
    }
}

// ---------------------------------------------------------------------------
// SharedRegion / WorkerScratch — typed views over raw bytes.
//
// These exist for programs that want raw byte-addressed access to the
// shared / scratch regions instead of a typed `S` / `W`. The runtime will
// eventually back these with linear-memory slices; in the mock SDK they
// are zero-length placeholders.
// ---------------------------------------------------------------------------

/// Typed wrapper over the shared region of a tasklet.
pub struct SharedRegion<T> {
    _phantom: PhantomData<T>,
    len: u32,
}

impl<T> SharedRegion<T> {
    /// Returns the length of the shared region in bytes.
    pub fn len_bytes(&self) -> u32 {
        self.len
    }
}

/// Typed wrapper over a single worker's scratch region.
///
/// The lifetime of `WorkerScratch` is tied to the worker lane that owns it,
/// so it cannot alias another worker's scratch.
pub struct WorkerScratch<T> {
    _phantom: PhantomData<T>,
    len: u32,
}

impl<T> WorkerScratch<T> {
    /// Returns the length of the scratch region in bytes.
    pub fn len_bytes(&self) -> u32 {
        self.len
    }
}

// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------

#[cfg(test)]
mod tests {
    use super::*;
    use crate::cpu::CpuBuilder;
    use crate::cpu_shared::partition::{range, tiles, Tile};
    use crate::host;
    use std::collections::HashSet;
    use std::sync::atomic::AtomicU32;

    #[derive(Default)]
    struct TestShared {
        counter: AtomicU32,
    }

    #[derive(Default, Clone)]
    struct TestScratch {
        local: u32,
    }

    fn cpu() -> crate::cpu::CpuLease {
        host::reset_mock();
        CpuBuilder::new().cores(4).acquire().expect("acquire")
    }

    #[test]
    fn builder_fluent_api_compiles() {
        let lease = cpu();
        let result = lease
            .cpu()
            .shared_memory_tasklet::<TestShared, TestScratch>()
            .cores(4)
            .workers(4)
            .shared_bytes(1024)
            .scratch_bytes_per_worker(64)
            .fuel(1_000_000)
            .input(b"hi".to_vec())
            .launch(|coord| {
                assert_eq!(coord.worker_count(), 4);
                assert_eq!(coord.input(), b"hi");
                coord.set_output(b"ok");
                Ok(())
            })
            .expect("launch");
        assert_eq!(result.exit_code, 0);
        assert_eq!(&result.output, b"ok");
    }

    #[test]
    fn rejects_workers_gt_cores() {
        let lease = cpu();
        let err = lease
            .cpu()
            .shared_memory_tasklet::<TestShared, TestScratch>()
            .cores(2)
            .workers(4)
            .shared_bytes(0)
            .scratch_bytes_per_worker(0)
            .launch(|_c| Ok(()))
            .unwrap_err();
        assert_eq!(err, TaskletError::InvalidWorkerCount);
    }

    #[test]
    fn rejects_oversize_envelope() {
        let lease = cpu();
        let err = lease
            .cpu()
            .shared_memory_tasklet::<TestShared, TestScratch>()
            .cores(4)
            .workers(4)
            .shared_bytes(100)
            .scratch_bytes_per_worker(50)
            .max_linear_memory(150) // 100 + 4*50 = 300 > 150
            .launch(|_c| Ok(()))
            .unwrap_err();
        assert_eq!(err, TaskletError::MemoryEnvelopeInvalid);
    }

    #[test]
    fn partition_range_divides_evenly() {
        assert_eq!(range(100, 0, 4), 0..25);
        assert_eq!(range(100, 1, 4), 25..50);
        assert_eq!(range(100, 2, 4), 50..75);
        assert_eq!(range(100, 3, 4), 75..100);
    }

    #[test]
    fn partition_range_handles_remainder() {
        let r0 = range(10, 0, 3);
        let r1 = range(10, 1, 3);
        let r2 = range(10, 2, 3);
        // Full coverage, no overlap.
        assert_eq!(r0.start, 0);
        assert_eq!(r0.end, r1.start);
        assert_eq!(r1.end, r2.start);
        assert_eq!(r2.end, 10);
        // First (10 % 3) = 1 worker gets one extra.
        assert_eq!(r0.len(), 4);
        assert_eq!(r1.len(), 3);
        assert_eq!(r2.len(), 3);
    }

    #[test]
    fn partition_tiles_covers_full_image() {
        let mut covered: HashSet<(u32, u32)> = HashSet::new();
        let mut total_pixels = 0usize;
        for wi in 0..4u16 {
            for Tile { x, y, w, h } in tiles(64, 64, 16, 16, wi, 4) {
                for dy in 0..h {
                    for dx in 0..w {
                        let inserted = covered.insert((x + dx, y + dy));
                        assert!(inserted, "pixel covered twice");
                        total_pixels += 1;
                    }
                }
            }
        }
        assert_eq!(total_pixels, 64 * 64);
        assert_eq!(covered.len(), 64 * 64);
    }

    #[test]
    fn coordinator_ctx_exposes_worker_count() {
        let lease = cpu();
        lease
            .cpu()
            .shared_memory_tasklet::<TestShared, TestScratch>()
            .cores(8)
            .workers(8)
            .shared_bytes(0)
            .scratch_bytes_per_worker(0)
            .launch(|coord| {
                assert_eq!(coord.worker_count(), 8);
                Ok(())
            })
            .unwrap();
    }

    #[test]
    fn worker_ctx_exposes_worker_index() {
        // Phase 48.3 — under the coordinator-lane model,
        // parallel_for_workers runs the body on data worker lanes
        // 1..worker_count(). With workers=4, the body sees raw wasm
        // worker indices [1, 2, 3] (NOT [0, 1, 2, 3]).
        let lease = cpu();
        let seen = Arc::new(std::sync::Mutex::new(Vec::<u16>::new()));
        let seen_data = Arc::new(std::sync::Mutex::new(Vec::<u32>::new()));
        let seen_clone = seen.clone();
        let seen_data_clone = seen_data.clone();
        lease
            .cpu()
            .shared_memory_tasklet::<TestShared, TestScratch>()
            .cores(4)
            .workers(4)
            .shared_bytes(0)
            .scratch_bytes_per_worker(0)
            .launch(move |coord| {
                assert_eq!(coord.data_worker_count(), 3);
                let s = seen_clone.clone();
                let sd = seen_data_clone.clone();
                coord
                    .parallel_for_workers(move |w| {
                        assert_eq!(w.worker_count(), 4);
                        assert_eq!(w.data_worker_count(), 3);
                        s.lock().unwrap().push(w.worker_index());
                        sd.lock().unwrap().push(w.data_worker_index());
                    })
                    .unwrap();
                Ok(())
            })
            .unwrap();
        let mut got = seen.lock().unwrap().clone();
        got.sort();
        assert_eq!(got, vec![1, 2, 3]);
        let mut got_data = seen_data.lock().unwrap().clone();
        got_data.sort();
        assert_eq!(got_data, vec![0, 1, 2]);
    }

    #[test]
    fn shared_region_visible_to_all_workers() {
        let lease = cpu();
        let result = lease
            .cpu()
            .shared_memory_tasklet::<TestShared, TestScratch>()
            .cores(4)
            .workers(4)
            .shared_bytes(1024)
            .scratch_bytes_per_worker(64)
            .launch(|coord| {
                coord
                    .parallel_for_workers(|w| {
                        w.shared().counter.fetch_add(1, Ordering::SeqCst);
                        w.scratch_mut().local = w.worker_index() as u32;
                    })
                    .unwrap();
                let n = coord.shared().counter.load(Ordering::SeqCst);
                coord.set_output(&n.to_le_bytes());
                Ok(())
            })
            .unwrap();
        // Phase 48.3 — coordinator lane (worker 0) does NOT run the
        // worker closure, so only `data_worker_count() == 3` lanes
        // increment the counter when launched with workers=4.
        assert_eq!(u32::from_le_bytes(result.output.try_into().unwrap()), 3);
    }

    #[test]
    fn worker_ctx_sees_input_bytes() {
        let lease = cpu();
        let seen = Arc::new(std::sync::Mutex::new(Vec::<Vec<u8>>::new()));
        let seen_clone = seen.clone();
        lease
            .cpu()
            .shared_memory_tasklet::<TestShared, TestScratch>()
            .cores(4)
            .workers(4)
            .shared_bytes(0)
            .scratch_bytes_per_worker(0)
            .input(b"hello workers".to_vec())
            .launch(move |coord| {
                let s = seen_clone.clone();
                assert_eq!(coord.input(), b"hello workers");
                coord
                    .parallel_for_workers(move |w| {
                        assert_eq!(w.input(), b"hello workers");
                        s.lock().unwrap().push(w.input().to_vec());
                    })
                    .unwrap();
                Ok(())
            })
            .unwrap();
        // Phase 48.3 — only the 3 data worker lanes (workers 1..4)
        // run the body; worker 0 is the coordinator lane.
        let got = seen.lock().unwrap().clone();
        assert_eq!(got.len(), 3);
        for v in got {
            assert_eq!(v, b"hello workers");
        }
    }

    #[test]
    fn coordinator_reads_cross_worker_scratch_after_barrier() {
        let lease = cpu();
        lease
            .cpu()
            .shared_memory_tasklet::<TestShared, TestScratch>()
            .cores(4)
            .workers(4)
            .shared_bytes(0)
            .scratch_bytes_per_worker(64)
            .launch(|coord| {
                // Each data worker writes its (index*10) into its
                // scratch. parallel_for_workers joins all data
                // workers before returning, which acts as the
                // barrier.
                coord
                    .parallel_for_workers(|w| {
                        w.scratch_mut().local = (w.worker_index() as u32) * 10;
                    })
                    .unwrap();
                // Now read every worker's scratch from the
                // coordinator. Worker 0 (coordinator lane) was not
                // touched, so its slot stays at the default 0;
                // workers 1, 2, 3 wrote 10, 20, 30. Sum = 60.
                //
                // Coincidence note: the post-Phase-48.3 sum
                // (0 + 10 + 20 + 30) happens to equal the
                // pre-Phase-48.3 sum (0 + 10 + 20 + 30 — same set
                // because worker 0 always wrote 0 anyway). If this
                // assertion's magic numbers are ever changed, the
                // expected value MUST be re-derived against the
                // current "data workers only" model — do NOT
                // assume the old "all lanes participate" math.
                let mut sum: u32 = 0;
                for i in 0..coord.worker_count() {
                    sum += coord.scratch(u32::from(i)).local;
                }
                coord.set_output(&sum.to_le_bytes());
                Ok(())
            })
            .map(|r| {
                assert_eq!(u32::from_le_bytes(r.output.try_into().unwrap()), 60);
            })
            .unwrap();
    }

    #[test]
    fn rejects_single_lane_shared_memory_tasklet() {
        // Phase 48.3 — a 1-lane shared-memory tasklet has a coordinator
        // but no data workers and cannot make progress; the builder
        // must reject up front.
        let lease = cpu();
        let err = lease
            .cpu()
            .shared_memory_tasklet::<TestShared, TestScratch>()
            .cores(1)
            .workers(1)
            .shared_bytes(0)
            .scratch_bytes_per_worker(0)
            .launch(|_c| Ok(()))
            .unwrap_err();
        assert_eq!(err, TaskletError::InvalidWorkerCount);
    }

    #[test]
    fn fuel_checkpoint_succeeds_until_pool_exhausted() {
        // Phase 48.13 W2b — V2 shared-pool launch with a small max_fuel.
        // The coordinator calls fuel_checkpoint repeatedly until the
        // pool exhausts, and the test verifies the deterministic
        // transition from Ok(()) to Err(FuelExhausted).
        let lease = cpu();
        let result = lease
            .cpu()
            .shared_memory_tasklet::<TestShared, TestScratch>()
            .cores(2)
            .workers(2)
            .shared_bytes(0)
            .scratch_bytes_per_worker(0)
            // workers=2 → precharge math: slice = max(20_000/2/4, 4096)
            // = max(2500, 4096) = 4096; precharge consumes 2*4096 = 8192
            // out of 20_000, leaving 11_808 in the pool for checkpoints.
            .with_max_fuel(20_000)
            .launch(|coord| {
                // Drain the pool in 1024-unit ticks. We don't know
                // the exact starting balance (it depends on the
                // precharge math), but we know it's bounded above
                // by max_fuel and that successive checkpoints must
                // eventually return FuelExhausted.
                let mut succeeded: u32 = 0;
                let mut hit_exhaustion = false;
                for _ in 0..1000 {
                    match coord.fuel_checkpoint(1024) {
                        Ok(()) => succeeded += 1,
                        Err(FuelExhausted) => {
                            hit_exhaustion = true;
                            break;
                        }
                    }
                }
                assert!(hit_exhaustion, "pool must eventually exhaust");
                // After a failed 1024-tick the pool may still hold
                // 0..1023 units. Drain it byte-by-byte until even a
                // single-unit checkpoint fails, then assert any
                // further debit fails.
                for _ in 0..1024 {
                    if coord.fuel_checkpoint(1).is_err() {
                        break;
                    }
                }
                assert_eq!(coord.fuel_checkpoint(1), Err(FuelExhausted));
                assert_eq!(coord.fuel_checkpoint(1024), Err(FuelExhausted));
                // amount == 0 is a silent no-op returning Ok(()) even
                // post-exhaustion.
                assert_eq!(coord.fuel_checkpoint(0), Ok(()));
                coord.set_output(&succeeded.to_le_bytes());
                Ok(())
            })
            .expect("launch");
        let succeeded = u32::from_le_bytes(result.output.try_into().unwrap());
        // Sanity: at least one checkpoint must have succeeded (the pool
        // is large enough to accommodate several 1024-unit debits).
        assert!(succeeded >= 1, "at least one checkpoint must succeed");
    }

    #[test]
    fn fuel_checkpoint_returns_err_when_no_pool_installed() {
        // Phase 48.13 W2b — a launch that does NOT call with_max_fuel
        // simulates a V1 worker. fuel_checkpoint returns
        // Err(FuelExhausted) on every call (other than amount==0).
        let lease = cpu();
        lease
            .cpu()
            .shared_memory_tasklet::<TestShared, TestScratch>()
            .cores(2)
            .workers(2)
            .shared_bytes(0)
            .scratch_bytes_per_worker(0)
            // No with_max_fuel — V1 simulation.
            .launch(|coord| {
                assert_eq!(coord.fuel_checkpoint(1), Err(FuelExhausted));
                assert_eq!(coord.fuel_checkpoint(1024), Err(FuelExhausted));
                // amount == 0 is still a no-op.
                assert_eq!(coord.fuel_checkpoint(0), Ok(()));
                coord
                    .parallel_for_workers(|w| {
                        assert_eq!(w.fuel_checkpoint(1), Err(FuelExhausted));
                        assert_eq!(w.fuel_checkpoint(0), Ok(()));
                    })
                    .unwrap();
                Ok(())
            })
            .expect("launch");
    }

    #[test]
    fn cancel_propagates_to_workers() {
        let lease = cpu();
        let result = lease
            .cpu()
            .shared_memory_tasklet::<TestShared, TestScratch>()
            .cores(2)
            .workers(2)
            .shared_bytes(0)
            .scratch_bytes_per_worker(0)
            .launch(|coord| {
                coord.cancel();
                assert!(coord.cancelled());
                Ok(())
            });
        assert_eq!(result.unwrap_err(), TaskletError::Cancelled);
    }
}