RFC: Distributed Computation Mesh · Brahmastra-Labs logicaffeine · Discussion #18

Mac Terminal: largo run --id "Alpha" --mode server
Browser Tab:  Open logicaffeine.com/studio?id=Alpha

# Direct P2P connection - no relay server required
# Both nodes see the same state, can push work to each other

┌─────────────────────────────────────────────────────────────────────┐
│                    TRUE PEER-TO-PEER MESH                            │
├─────────────────────────────────────────────────────────────────────┤
│                                                                      │
│   ┌─────────────┐                              ┌────────────┐        │
│   │  Mac CLI    │◄────────────────────────────►│  Browser   │        │
│   │  (Native)   │      Direct WebRTC P2P       │  (WASM)    │        │
│   │             │      via libp2p-webrtc       │            │        │
│   └──────┬──────┘                              └─────┬──────┘        │
│          │                                           │               │
│          │         Program ID: "Alpha"               │               │
│          │         Kademlia DHT Discovery            │               │
│          │                                           │               │
│   ┌──────┴──────┐                              ┌─────┴──────┐        │
│   │ Distributed │◄────────────────────────────►│ Distributed│        │
│   │ <GCounter>  │         GossipSub            │ <GCounter> │        │
│   └─────────────┘                              └────────────┘        │
│                                                                      │
│   VFS (Native)          Anti-Entropy           VFS (OPFS)           │
│   /home/.lsf/           Merkle Search Tree     Browser OPFS         │
│                                                                      │
└─────────────────────────────────────────────────────────────────────┘

pub enum SovereigntyClass {
    /// Mac/Server - full journal, quorum voter, JIT provider, relay
    Authority,
    /// Desktop Browser - OPFS journal, work-stealer, can relay
    Citizen,
    /// Mobile/Guest - RAM-only, task consumer, leaf node
    Ephemeral,
}

## Main
# The Mac identifies as an Authority
Enable Networked Mode with ID "Alpha" as Authority.

# The Browser identifies as a Citizen
Enable Networked Mode with ID "Alpha" as Citizen.

// reliable_broadcast.rs - Pending Ack Store with Exponential Backoff
pub struct ReliableBroadcast {
    pending: HashMap<DeltaId, PendingDelta>,
    backoff_config: BackoffConfig,
}

#[derive(Clone)]
pub struct PendingDelta {
    delta: WriteDelta,
    required_acks: HashSet<PeerId>,  // Authorities that must ack
    received_acks: HashSet<PeerId>,
    attempts: u32,
    next_retry: Instant,
    created_at: Instant,
}

impl ReliableBroadcast {
    /// Track a delta that needs Authority acknowledgments
    pub fn track(&mut self, delta: WriteDelta, authorities: Vec<PeerId>) -> DeltaId {
        let id = DeltaId::new();
        self.pending.insert(id, PendingDelta {
            delta,
            required_acks: authorities.into_iter().collect(),
            received_acks: HashSet::new(),
            attempts: 0,
            next_retry: Instant::now(),
            created_at: Instant::now(),
        });
        id
    }

    /// Record an ack from a peer
    pub fn record_ack(&mut self, delta_id: &DeltaId, peer: PeerId) -> AckResult {
        if let Some(pending) = self.pending.get_mut(delta_id) {
            pending.received_acks.insert(peer);

            // Check if we have majority quorum
            let required = (pending.required_acks.len() / 2) + 1;
            if pending.received_acks.len() >= required {
                self.pending.remove(delta_id);
                return AckResult::QuorumReached;
            }
        }
        AckResult::Waiting
    }

    /// Get deltas that need retry (exponential backoff)
    pub fn due_for_retry(&mut self) -> Vec<(DeltaId, WriteDelta, Vec<PeerId>)> {
        let now = Instant::now();
        let mut retries = vec![];

        for (id, pending) in &mut self.pending {
            if now >= pending.next_retry {
                // Who hasn't acked yet?
                let missing: Vec<_> = pending.required_acks
                    .difference(&pending.received_acks)
                    .cloned()
                    .collect();

                if !missing.is_empty() {
                    retries.push((*id, pending.delta.clone(), missing));

                    // Exponential backoff: 100ms, 200ms, 400ms, 800ms, max 30s
                    pending.attempts += 1;
                    let backoff = Duration::from_millis(
                        (100 * 2u64.pow(pending.attempts.min(8))).min(30_000)
                    );
                    pending.next_retry = now + backoff;
                }
            }
        }

        retries
    }

    /// Abandon deltas older than timeout - let MST Anti-Entropy handle them
    pub fn abandon_stale(&mut self, timeout: Duration) -> Vec<DeltaId> {
        let now = Instant::now();
        let stale: Vec<_> = self.pending
            .iter()
            .filter(|(_, p)| now.duration_since(p.created_at) > timeout)
            .map(|(id, _)| *id)
            .collect();

        for id in &stale {
            self.pending.remove(id);
        }

        stale  // These will be reconciled via MST later
    }
}

pub enum AckResult {
    QuorumReached,
    Waiting,
}

// liveness.rs - Prevent "Ghost Tasks" from hanging the mesh
pub struct LivenessGuard {
    peers: HashMap<PeerId, PeerLiveness>,
    pulse_interval: Duration,      // 5 seconds
    stale_threshold: u32,          // 3 missed pulses = stale
}

#[derive(Clone)]
pub struct PeerLiveness {
    last_pulse: Instant,
    missed_pulses: u32,
    sovereignty: SovereigntyClass,
    state: PeerState,
}

#[derive(Clone, Copy, PartialEq)]
pub enum PeerState {
    Active,
    Stale,      // Missed pulses, ignored for scheduling
    Disconnected,
}

impl LivenessGuard {
    /// WASM Citizens must call this every 5 seconds
    pub fn record_pulse(&mut self, peer: PeerId) {
        if let Some(liveness) = self.peers.get_mut(&peer) {
            liveness.last_pulse = Instant::now();
            liveness.missed_pulses = 0;
            liveness.state = PeerState::Active;
        }
    }

    /// Run periodically to detect stale peers
    pub fn check_liveness(&mut self) -> Vec<PeerId> {
        let now = Instant::now();
        let mut newly_stale = vec![];

        for (peer_id, liveness) in &mut self.peers {
            // Only check non-Authority peers (Authorities don't need pulses)
            if liveness.sovereignty != SovereigntyClass::Authority {
                let elapsed = now.duration_since(liveness.last_pulse);
                let missed = (elapsed.as_secs() / self.pulse_interval.as_secs()) as u32;

                if missed > liveness.missed_pulses {
                    liveness.missed_pulses = missed;

                    if missed >= self.stale_threshold && liveness.state == PeerState::Active {
                        liveness.state = PeerState::Stale;
                        newly_stale.push(*peer_id);
                    }
                }
            }
        }

        newly_stale
    }

    /// Get only active peers for scheduling
    pub fn active_peers(&self) -> Vec<PeerId> {
        self.peers.iter()
            .filter(|(_, l)| l.state == PeerState::Active)
            .map(|(id, _)| *id)
            .collect()
    }
}

// session.rs - Identity persistence and fast reconnect
pub struct SessionManager {
    /// PeerId stored in LocalStorage for identity persistence
    keypair: Keypair,
    /// Last known MST root hash
    last_mst_root: Option<Hash>,
}

impl SessionManager {
    /// Browser: Store identity in LocalStorage so Mac recognizes us
    #[cfg(target_arch = "wasm32")]
    pub fn persist_identity(&self) {
        let storage = web_sys::window()
            .unwrap()
            .local_storage()
            .unwrap()
            .unwrap();

        storage.set_item(
            "fabric_keypair",
            &base64::encode(self.keypair.to_bytes())
        ).unwrap();

        if let Some(root) = &self.last_mst_root {
            storage.set_item("fabric_mst_root", &hex::encode(root)).unwrap();
        }
    }

    /// Browser: Restore identity on page load
    #[cfg(target_arch = "wasm32")]
    pub fn restore_identity() -> Option<Self> {
        let storage = web_sys::window()?.local_storage().ok()??;

        let keypair_bytes = storage.get_item("fabric_keypair").ok()??;
        let keypair = Keypair::from_bytes(&base64::decode(&keypair_bytes).ok()?).ok()?;

        let last_mst_root = storage.get_item("fabric_mst_root").ok()?
            .and_then(|s| hex::decode(&s).ok())
            .and_then(|b| Hash::from_slice(&b));

        Some(Self { keypair, last_mst_root })
    }

    /// Fast reconnect: send MST root, receive only missing entries
    pub async fn reconnect_fast(&self, fabric: &FabricHandle) -> Result<(), ReconnectError> {
        if let Some(our_root) = &self.last_mst_root {
            // Tell Authority our state
            let response = fabric.request_mst_diff(*our_root).await?;

            match response {
                MstDiffResponse::InSync => {
                    // Perfect - we missed nothing
                }
                MstDiffResponse::Behind { missing_entries } => {
                    // Apply only what we're missing
                    for entry in missing_entries {
                        self.journal.apply(entry).await?;
                    }
                }
                MstDiffResponse::Diverged { common_ancestor, their_entries, our_entries } => {
                    // Merge both sides (CRDTs make this safe)
                    for entry in their_entries {
                        self.journal.apply(entry).await?;
                    }
                    // Re-broadcast our entries they might have missed
                    for entry in our_entries {
                        fabric.publish_delta(&entry).await?;
                    }
                }
            }
        }
        Ok(())
    }
}

pub enum MstDiffResponse {
    InSync,
    Behind { missing_entries: Vec<JournalEntry> },
    Diverged {
        common_ancestor: Hash,
        their_entries: Vec<JournalEntry>,
        our_entries: Vec<JournalEntry>,
    },
}

// task_lease.rs - Lease-based ownership prevents abandoned task lockup
#[derive(Clone, Copy, PartialEq, Debug)]
pub enum TaskState {
    /// In the Global Pipe, waiting for a taker
    Pending,
    /// A Citizen is currently computing it (has a lease)
    Leased { holder: PeerId, expires: Instant },
    /// Results are back and verified
    Completed,
    /// Lease expired without a result - ready to re-queue
    Stale,
}

pub struct TaskLease {
    pub task_id: TaskId,
    pub holder: PeerId,
    pub acquired_at: Instant,
    pub expires_at: Instant,
    pub pulse_count: u32,
    pub last_pulse: Instant,
}

impl TaskLease {
    pub fn new(task_id: TaskId, holder: PeerId, ttl: Duration) -> Self {
        let now = Instant::now();
        Self {
            task_id,
            holder,
            acquired_at: now,
            expires_at: now + ttl,
            pulse_count: 0,
            last_pulse: now,
        }
    }

    pub fn is_expired(&self) -> bool {
        Instant::now() > self.expires_at
    }

    /// Extend lease on pulse (heartbeat)
    pub fn extend(&mut self, extension: Duration) {
        self.pulse_count += 1;
        self.last_pulse = Instant::now();
        self.expires_at = Instant::now() + extension;
    }
}

// task_manager.rs - Authority manages task lifecycle
pub struct TaskManager {
    /// All in-flight tasks with their leases
    in_flight: HashMap<TaskId, TaskLease>,
    /// Tasks waiting to be picked up
    pending_queue: VecDeque<WorkUnit>,
    /// Completed task results (cached briefly for late acks)
    completed: HashMap<TaskId, (TaskResult, Instant)>,
    /// Config
    lease_ttl: Duration,           // Default: 30 seconds
    pulse_interval: Duration,      // Expected: 5 seconds
    stale_threshold: u32,          // Missed pulses before revoke: 3
}

impl TaskManager {
    /// Citizen requests a task - Authority grants lease
    pub fn grant_lease(&mut self, peer: PeerId) -> Option<(WorkUnit, TaskLease)> {
        if let Some(work) = self.pending_queue.pop_front() {
            let lease = TaskLease::new(work.id, peer, self.lease_ttl);
            self.in_flight.insert(work.id, lease.clone());
            Some((work, lease))
        } else {
            None
        }
    }

    /// Citizen sends pulse while computing - extends lease
    pub fn record_pulse(&mut self, task_id: TaskId, peer: PeerId) -> Result<(), LeaseError> {
        if let Some(lease) = self.in_flight.get_mut(&task_id) {
            if lease.holder != peer {
                return Err(LeaseError::NotHolder);
            }
            lease.extend(self.lease_ttl);
            Ok(())
        } else {
            Err(LeaseError::NotFound)
        }
    }

    /// Citizen completes task - Authority releases lease
    pub fn complete_task(&mut self, task_id: TaskId, result: TaskResult, peer: PeerId) -> Result<(), LeaseError> {
        if let Some(lease) = self.in_flight.remove(&task_id) {
            if lease.holder != peer {
                // Task was re-assigned (peer was too slow)
                // But thanks to idempotency, we can still accept the result!
                // CRDTs make double-completion safe
            }
            self.completed.insert(task_id, (result, Instant::now()));
            Ok(())
        } else {
            // Task already completed by another node - that's fine (idempotent)
            Ok(())
        }
    }

    /// Run periodically by Authority to reclaim abandoned tasks
    pub fn reclaim_stale_leases(&mut self) -> Vec<WorkUnit> {
        let now = Instant::now();
        let mut reclaimed = vec![];

        let stale_ids: Vec<_> = self.in_flight.iter()
            .filter(|(_, lease)| lease.is_expired())
            .map(|(id, _)| *id)
            .collect();

        for task_id in stale_ids {
            if let Some(lease) = self.in_flight.remove(&task_id) {
                tracing::warn!(
                    "Reclaiming abandoned task {:?} from peer {:?} (missed {} pulses)",
                    task_id, lease.holder,
                    (now.duration_since(lease.last_pulse).as_secs() / self.pulse_interval.as_secs())
                );
                // Move task back to pending queue
                if let Some(work) = self.recover_work_unit(&task_id) {
                    reclaimed.push(work.clone());
                    self.pending_queue.push_back(work);
                }
            }
        }

        reclaimed
    }

    /// Citizen voluntarily releases lease (detected throttling)
    pub fn voluntary_release(&mut self, task_id: TaskId, peer: PeerId) -> Result<(), LeaseError> {
        if let Some(lease) = self.in_flight.remove(&task_id) {
            if lease.holder == peer {
                tracing::info!("Peer {:?} voluntarily released task {:?}", peer, task_id);
                if let Some(work) = self.recover_work_unit(&task_id) {
                    self.pending_queue.push_front(work); // Priority re-queue
                }
            }
        }
        Ok(())
    }
}

#[derive(Debug, thiserror::Error)]
pub enum LeaseError {
    #[error("Task not found")]
    NotFound,
    #[error("Peer is not the lease holder")]
    NotHolder,
    #[error("Lease already expired")]
    Expired,
}

// browser_worker.rs - WASM-side throttle detection
#[cfg(target_arch = "wasm32")]
pub struct BrowserWorker {
    current_task: Option<(TaskId, Instant)>,
    expected_duration: Duration,
    pulse_interval: Duration,
}

#[cfg(target_arch = "wasm32")]
impl BrowserWorker {
    /// Check if we're being throttled (task taking 10x longer than expected)
    pub fn is_throttled(&self) -> bool {
        if let Some((_, started)) = &self.current_task {
            let elapsed = Instant::now().duration_since(*started);
            elapsed > self.expected_duration * 10
        } else {
            false
        }
    }

    /// Called periodically during long computations
    pub async fn maybe_yield(&mut self, fabric: &FabricHandle) {
        if self.is_throttled() {
            if let Some((task_id, _)) = self.current_task.take() {
                // Voluntarily release - let a faster node take over
                fabric.voluntary_release(task_id).await;
                tracing::info!("Voluntarily released throttled task {:?}", task_id);
            }
        } else {
            // Send pulse to keep lease alive
            if let Some((task_id, _)) = &self.current_task {
                fabric.send_pulse(*task_id).await;
            }
        }
    }

    /// Use scheduler.yield() to cooperatively yield to browser
    pub async fn cooperative_compute<F, R>(&mut self, task_id: TaskId, f: F) -> Option<R>
    where
        F: FnOnce() -> R,
    {
        self.current_task = Some((task_id, Instant::now()));

        // Yield control periodically to browser event loop
        let result = f();

        self.current_task = None;
        Some(result)
    }
}

impl LivenessGuard {
    /// Enhanced: Also tracks per-task pulses, not just peer liveness
    pub fn record_task_pulse(&mut self, peer: PeerId, task_id: TaskId) {
        // Update peer liveness
        self.record_pulse(peer);

        // Update task lease
        if let Some(task_manager) = &mut self.task_manager {
            let _ = task_manager.record_pulse(task_id, peer);
        }
    }
}

// snapshot_catchup.rs - Switch from deltas to snapshots for slow consumers
pub struct SnapshotCatchup {
    /// Threshold before switching to snapshot mode
    delta_lag_threshold: usize,  // Default: 500 unapplied deltas
    /// Per-peer delta lag tracking
    peer_lag: HashMap<PeerId, DeltaLagState>,
}

#[derive(Clone)]
pub struct DeltaLagState {
    /// Number of deltas sent but not acked
    pending_deltas: usize,
    /// Last known applied delta sequence number
    last_applied_seq: u64,
    /// Current sync mode
    mode: SyncMode,
}

#[derive(Clone, Copy, PartialEq)]
pub enum SyncMode {
    /// Normal: Stream deltas incrementally
    Incremental,
    /// Paused: Stop streaming, wait for stabilization
    Paused,
    /// Snapshot: Send full state as compressed payload
    SnapshotPending,
}

impl SnapshotCatchup {
    /// Called when we receive a delta ack from a peer
    pub fn record_ack(&mut self, peer: PeerId, acked_seq: u64) {
        if let Some(state) = self.peer_lag.get_mut(&peer) {
            state.pending_deltas = state.pending_deltas.saturating_sub(1);
            state.last_applied_seq = acked_seq;

            // Peer caught up - resume incremental mode
            if state.pending_deltas < self.delta_lag_threshold / 2 {
                state.mode = SyncMode::Incremental;
            }
        }
    }

    /// Called when we send a delta to a peer
    pub fn record_delta_sent(&mut self, peer: PeerId) {
        if let Some(state) = self.peer_lag.get_mut(&peer) {
            state.pending_deltas += 1;

            // Peer is falling behind - pause deltas
            if state.pending_deltas >= self.delta_lag_threshold {
                state.mode = SyncMode::Paused;
                tracing::warn!("Peer {:?} lagging ({} pending), switching to snapshot mode",
                    peer, state.pending_deltas);
            }
        }
    }

    /// Check if peer needs a snapshot instead of deltas
    pub fn needs_snapshot(&self, peer: &PeerId) -> bool {
        self.peer_lag.get(peer)
            .map(|s| s.mode == SyncMode::Paused || s.mode == SyncMode::SnapshotPending)
            .unwrap_or(false)
    }

    /// Generate compressed snapshot for slow consumer
    pub async fn generate_snapshot<T: Serialize>(&self, state: &T) -> CompressedSnapshot {
        let serialized = bincode::serialize(state).unwrap();
        let compressed = zstd::encode_all(&serialized[..], 3).unwrap();

        CompressedSnapshot {
            data: compressed,
            uncompressed_size: serialized.len(),
            timestamp: Instant::now(),
        }
    }
}

#[derive(Serialize, Deserialize)]
pub struct CompressedSnapshot {
    pub data: Vec<u8>,
    pub uncompressed_size: usize,
    pub timestamp: Instant,
}

// vclock_causality.rs - Logical time replaces wall-clock time
use crate::crdt::VClock;

/// Every distributed mutation carries a VClock for causal ordering
/// SystemTime is metadata for display only, NOT for conflict resolution
#[derive(Serialize, Deserialize, Clone)]
pub struct CausalMutation<T> {
    /// The actual data change
    pub payload: T,
    /// Logical timestamp for ordering (SOURCE OF TRUTH)
    pub vclock: VClock,
    /// Wall-clock time (DISPLAY ONLY - never used for ordering)
    pub wall_time: SystemTime,
    /// Origin node
    pub origin: ReplicaId,
}

impl<T> CausalMutation<T> {
    pub fn new(payload: T, vclock: VClock, origin: ReplicaId) -> Self {
        Self {
            payload,
            vclock,
            wall_time: SystemTime::now(),  // For human display only
            origin,
        }
    }

    /// Compare causality using VClock, NOT wall time
    pub fn happened_before(&self, other: &Self) -> bool {
        self.vclock.partial_cmp(&other.vclock) == Some(std::cmp::Ordering::Less)
    }

    /// Check if mutations are concurrent (neither happened-before the other)
    pub fn is_concurrent_with(&self, other: &Self) -> bool {
        self.vclock.partial_cmp(&other.vclock).is_none()
    }
}

/// Enhanced LWW that uses VClock instead of SystemTime
pub struct VClockLWW<T> {
    value: T,
    vclock: VClock,
    /// Wall time kept for display only
    last_modified_display: SystemTime,
}

impl<T: Clone> VClockLWW<T> {
    pub fn set(&mut self, value: T, mutation: &CausalMutation<()>) {
        // Compare VClocks, not timestamps
        if mutation.vclock > self.vclock {
            self.value = value;
            self.vclock = mutation.vclock.clone();
            self.last_modified_display = mutation.wall_time;
        }
    }

    pub fn merge(&mut self, other: &Self) {
        // VClock comparison for conflict resolution
        if other.vclock > self.vclock {
            self.value = other.value.clone();
            self.vclock = other.vclock.clone();
            self.last_modified_display = other.last_modified_display;
        }
    }
}

/// Node-local clock manager - increments on every local mutation
pub struct LocalClock {
    replica_id: ReplicaId,
    vclock: VClock,
}

impl LocalClock {
    /// Increment and return new VClock for a mutation
    pub fn tick(&mut self) -> VClock {
        self.vclock.increment(self.replica_id);
        self.vclock.clone()
    }

    /// Merge received VClock and increment (for receiving remote mutations)
    pub fn receive(&mut self, remote: &VClock) -> VClock {
        self.vclock.merge(remote);
        self.tick()
    }
}

// task_generation.rs - Prevent zombie task execution
#[derive(Clone, Copy, PartialEq, Eq, Hash, Debug, Serialize, Deserialize)]
pub struct GenerationId(u64);

impl GenerationId {
    pub fn initial() -> Self {
        Self(1)
    }

    pub fn increment(&self) -> Self {
        Self(self.0 + 1)
    }
}

/// Enhanced TaskLease with generation tracking
pub struct GenerationalTaskLease {
    pub task_id: TaskId,
    pub holder: PeerId,
    pub generation: GenerationId,
    pub expires_at: Instant,
    pub pulse_count: u32,
}

/// Enhanced TaskManager with generation tracking
impl TaskManager {
    /// Reclaim task and INCREMENT generation
    pub fn reclaim_with_generation(&mut self, task_id: TaskId) -> Option<(WorkUnit, GenerationId)> {
        if let Some(lease) = self.in_flight.remove(&task_id) {
            let new_generation = lease.generation.increment();

            // Store the new generation
            self.task_generations.insert(task_id, new_generation);

            // Re-queue with new generation
            if let Some(mut work) = self.recover_work_unit(&task_id) {
                work.generation = new_generation;
                self.pending_queue.push_back(work.clone());
                return Some((work, new_generation));
            }
        }
        None
    }

    /// Handle result submission with generation check
    pub fn submit_result(
        &mut self,
        task_id: TaskId,
        generation: GenerationId,
        result: TaskResult,
        peer: PeerId,
    ) -> SubmitOutcome {
        let current_generation = self.task_generations.get(&task_id)
            .copied()
            .unwrap_or(GenerationId::initial());

        if generation < current_generation {
            // Stale generation - accept result (idempotent) but kill zombie
            tracing::warn!(
                "Stale result from {:?} for task {:?} (gen {} < current {})",
                peer, task_id, generation.0, current_generation.0
            );

            // Accept the result anyway (might be valid data)
            if !self.completed.contains_key(&task_id) {
                self.completed.insert(task_id, (result, Instant::now()));
            }

            // Signal zombie to stop
            return SubmitOutcome::StaleGeneration {
                stop_signal: StopWorkSignal { task_id, peer }
            };
        }

        // Current generation - normal completion
        self.in_flight.remove(&task_id);
        self.completed.insert(task_id, (result, Instant::now()));
        SubmitOutcome::Accepted
    }
}

pub enum SubmitOutcome {
    /// Result accepted, task completed
    Accepted,
    /// Result accepted but submitter should stop (stale generation)
    StaleGeneration { stop_signal: StopWorkSignal },
    /// Task not found (already completed)
    NotFound,
}

/// Signal sent to zombie workers to stop execution
#[derive(Serialize, Deserialize, Clone)]
pub struct StopWorkSignal {
    pub task_id: TaskId,
    pub peer: PeerId,
}

/// Browser-side handling of stop signals
#[cfg(target_arch = "wasm32")]
impl BrowserWorker {
    /// Handle incoming stop signal - abort current work
    pub fn handle_stop_signal(&mut self, signal: StopWorkSignal) {
        if let Some((current_task, _)) = &self.current_task {
            if *current_task == signal.task_id {
                tracing::info!("Received stop signal for {:?}, aborting", signal.task_id);
                self.current_task = None;
                // Cancel any pending computation
                self.abort_handle.take().map(|h| h.abort());
            }
        }
    }
}

// adaptive_compaction.rs - Storage-aware journal management
#[cfg(target_arch = "wasm32")]
pub struct AdaptiveCompactor {
    /// Threshold to trigger aggressive compaction (0.0-1.0)
    warning_threshold: f64,   // Default: 0.7 (70%)
    critical_threshold: f64,  // Default: 0.85 (85%)
    /// Current compaction policy
    policy: CompactionPolicy,
}

#[derive(Clone, Copy, PartialEq)]
pub enum CompactionPolicy {
    /// Normal: Keep recent WAL entries for replay
    Normal { wal_retention_days: u32 },
    /// Aggressive: Keep only snapshot + MST root
    Aggressive,
    /// Critical: Emergency - snapshot only, no history
    Critical,
}

#[cfg(target_arch = "wasm32")]
impl AdaptiveCompactor {
    /// Check storage quota using navigator.storage.estimate()
    pub async fn check_storage_pressure(&mut self) -> StoragePressure {
        let window = web_sys::window().unwrap();
        let navigator = window.navigator();
        let storage = navigator.storage();

        let estimate = JsFuture::from(storage.estimate()).await.unwrap();
        let estimate: web_sys::StorageEstimate = estimate.unchecked_into();

        let usage = estimate.usage().unwrap_or(0) as f64;
        let quota = estimate.quota().unwrap_or(u64::MAX) as f64;
        let ratio = usage / quota;

        let pressure = if ratio >= self.critical_threshold {
            StoragePressure::Critical { used_ratio: ratio }
        } else if ratio >= self.warning_threshold {
            StoragePressure::Warning { used_ratio: ratio }
        } else {
            StoragePressure::Normal { used_ratio: ratio }
        };

        // Update policy based on pressure
        self.policy = match pressure {
            StoragePressure::Critical { .. } => CompactionPolicy::Critical,
            StoragePressure::Warning { .. } => CompactionPolicy::Aggressive,
            StoragePressure::Normal { .. } => CompactionPolicy::Normal { wal_retention_days: 7 },
        };

        pressure
    }

    /// Run compaction based on current policy
    pub async fn compact(&self, journal: &mut Journal, mst: &MerkleSearchTree) -> CompactionResult {
        match self.policy {
            CompactionPolicy::Normal { wal_retention_days } => {
                // Keep entries from last N days
                let cutoff = SystemTime::now() - Duration::from_secs(wal_retention_days as u64 * 86400);
                let removed = journal.remove_entries_before(cutoff).await?;
                CompactionResult::Normal { entries_removed: removed }
            }
            CompactionPolicy::Aggressive => {
                // Generate snapshot, keep only recent entries
                let snapshot = self.create_snapshot(journal).await?;
                journal.truncate_to_snapshot(&snapshot).await?;

                // Keep MST root for fast reconnect
                let mst_root = mst.root_hash();

                CompactionResult::Aggressive {
                    snapshot_size: snapshot.data.len(),
                    mst_root,
                }
            }
            CompactionPolicy::Critical => {
                // Emergency: snapshot only, discard ALL history
                let snapshot = self.create_snapshot(journal).await?;
                journal.replace_with_snapshot(&snapshot).await?;

                tracing::warn!("Critical compaction: all WAL history discarded");

                CompactionResult::Critical {
                    snapshot_size: snapshot.data.len(),
                    history_discarded: true,
                }
            }
        }
    }

    /// Estimate space savings before compaction
    pub async fn estimate_savings(&self, journal: &Journal) -> StorageSavings {
        let current_size = journal.total_size().await;
        let snapshot_size = journal.estimate_snapshot_size().await;

        StorageSavings {
            current_bytes: current_size,
            after_compaction_bytes: snapshot_size,
            savings_bytes: current_size.saturating_sub(snapshot_size),
            savings_percent: ((current_size - snapshot_size) as f64 / current_size as f64) * 100.0,
        }
    }
}

#[derive(Debug)]
pub enum StoragePressure {
    Normal { used_ratio: f64 },
    Warning { used_ratio: f64 },
    Critical { used_ratio: f64 },
}

#[derive(Debug)]
pub enum CompactionResult {
    Normal { entries_removed: usize },
    Aggressive { snapshot_size: usize, mst_root: Option<Hash> },
    Critical { snapshot_size: usize, history_discarded: bool },
}

#[derive(Debug)]
pub struct StorageSavings {
    pub current_bytes: usize,
    pub after_compaction_bytes: usize,
    pub savings_bytes: usize,
    pub savings_percent: f64,
}

impl Journal {
    /// Periodic storage check (runs every 5 minutes in browser)
    #[cfg(target_arch = "wasm32")]
    pub async fn storage_maintenance(&mut self) {
        let pressure = self.compactor.check_storage_pressure().await;

        match pressure {
            StoragePressure::Warning { used_ratio } => {
                tracing::warn!("Storage at {:.1}%, triggering aggressive compaction", used_ratio * 100.0);
                self.compactor.compact(&mut self.inner, &self.mst).await;
            }
            StoragePressure::Critical { used_ratio } => {
                tracing::error!("Storage critical at {:.1}%, emergency compaction", used_ratio * 100.0);
                self.compactor.compact(&mut self.inner, &self.mst).await;

                // Notify user they may need to free space
                self.emit_storage_warning(used_ratio).await;
            }
            StoragePressure::Normal { .. } => {
                // Normal compaction on schedule
            }
        }
    }
}

// quorum.rs - Split-brain prevention via majority quorum
pub struct QuorumConfig {
    /// Total known Authority nodes in the mesh
    authority_count: usize,
}

impl QuorumConfig {
    /// Calculate required acks for write to succeed
    /// Uses floor(n/2) + 1 to ensure majority
    pub fn required_acks(&self) -> usize {
        (self.authority_count / 2) + 1
    }

    /// Check if we can even attempt a write
    pub fn can_attempt_write(&self, reachable_authorities: usize) -> bool {
        reachable_authorities >= self.required_acks()
    }
}

// Examples:
// - 1 Mac (local dev): required = 1, quorum = 1
// - 2 Macs: required = 2, BOTH must ack (prevents split-brain)
// - 3 Macs: required = 2, any 2 of 3 must ack
// - 50 browsers: They are WITNESSES, not voters. Don't affect quorum math.

// lease.rs - Prevent concurrent leaders during partition
pub struct SovereignLease {
    holder: Option<PeerId>,
    expires: Instant,
    term: u64,
}

impl SovereignLease {
    /// Request leadership (only Authorities can hold leases)
    pub async fn acquire(&mut self, fabric: &FabricHandle) -> Result<LeaseGuard, LeaseError> {
        if !fabric.local_capabilities().is_authority() {
            return Err(LeaseError::NotAuthority);
        }

        // Must get majority agreement to become leader
        let authorities = fabric.authority_peers().await;
        let required = (authorities.len() / 2) + 1;

        let votes = fabric.request_leadership_votes(self.term + 1).await;

        if votes.len() >= required {
            self.holder = Some(fabric.local_peer_id());
            self.term += 1;
            self.expires = Instant::now() + Duration::from_secs(30);
            Ok(LeaseGuard { lease: self, fabric: fabric.clone() })
        } else {
            Err(LeaseError::InsufficientVotes)
        }
    }
}

// auth.rs - Only nodes knowing the Program ID can join
use hmac::{Hmac, Mac};
use sha2::Sha256;

pub struct MeshAuthenticator {
    program_id: ProgramId,
    /// Derived from program_id - acts as shared secret
    mesh_key: [u8; 32],
}

impl MeshAuthenticator {
    pub fn new(program_id: ProgramId) -> Self {
        // Derive mesh key from program ID
        let mut mac = Hmac::<Sha256>::new_from_slice(b"fabric/mesh/auth").unwrap();
        mac.update(program_id.as_bytes());
        let mesh_key: [u8; 32] = mac.finalize().into_bytes().into();

        Self { program_id, mesh_key }
    }

    /// Generate challenge for incoming peer
    pub fn generate_challenge(&self) -> Challenge {
        let nonce: [u8; 32] = rand::random();
        Challenge { nonce }
    }

    /// Verify peer knows the program ID
    pub fn verify_response(&self, challenge: &Challenge, response: &[u8]) -> bool {
        let mut mac = Hmac::<Sha256>::new_from_slice(&self.mesh_key).unwrap();
        mac.update(&challenge.nonce);
        let expected = mac.finalize().into_bytes();

        // Constant-time comparison
        expected.as_slice() == response
    }

    /// Compute response to prove we know the program ID
    pub fn compute_response(&self, challenge: &Challenge) -> Vec<u8> {
        let mut mac = Hmac::<Sha256>::new_from_slice(&self.mesh_key).unwrap();
        mac.update(&challenge.nonce);
        mac.finalize().into_bytes().to_vec()
    }
}

pub struct Challenge {
    nonce: [u8; 32],
}

// Integration with libp2p handshake
impl FabricBehaviour {
    /// After noise handshake, verify peer knows program ID
    pub async fn authenticate_peer(&self, peer: PeerId) -> Result<(), AuthError> {
        let challenge = self.authenticator.generate_challenge();
        self.send_challenge(peer, &challenge).await?;

        let response = self.await_response(peer).await?;

        if self.authenticator.verify_response(&challenge, &response) {
            Ok(())
        } else {
            self.disconnect(peer).await;
            Err(AuthError::InvalidResponse)
        }
    }
}

┌─────────────────────────────────────────────────────────────────────┐
│                    libp2p Transport Stack                            │
│                                                                      │
│   Native (Mac/Linux)              WASM (Browser)                     │
│   ┌──────────────────┐            ┌──────────────────┐              │
│   │ QUIC + TCP       │            │ WebRTC           │              │
│   │ WebRTC (server)  │◄──────────►│ WebSocket        │              │
│   │ Circuit Relay v2 │            │ (fallback)       │              │
│   └────────┬─────────┘            └────────┬─────────┘              │
│            │                               │                         │
│            └───────────┬───────────────────┘                         │
│                        │                                             │
│              ┌─────────┴─────────┐                                   │
│              │    GossipSub      │                                   │
│              │    Kademlia DHT   │                                   │
│              │    Identify       │                                   │
│              └───────────────────┘                                   │
└─────────────────────────────────────────────────────────────────────┘

// webrtc.rs - WebRTC transport configuration for both native and WASM

/// Configure libp2p for the current platform
pub fn build_transport() -> libp2p::core::transport::Boxed<(PeerId, StreamMuxerBox)> {
    #[cfg(not(target_arch = "wasm32"))]
    {
        // Native: QUIC + TCP + WebRTC server
        let quic = libp2p::quic::tokio::Transport::new(quic_config);
        let tcp = libp2p::tcp::tokio::Transport::new(tcp_config);
        let webrtc = libp2p::webrtc::tokio::Transport::new(
            keypair.clone(),
            webrtc_certificate,
        );

        OrTransport::new(quic, OrTransport::new(tcp, webrtc))
            .upgrade(Version::V1)
            .authenticate(noise::Config::new(&keypair)?)
            .multiplex(yamux::Config::default())
            .boxed()
    }

    #[cfg(target_arch = "wasm32")]
    {
        // WASM: WebRTC client + WebSocket fallback
        let webrtc = libp2p::webrtc::Transport::new(keypair.clone());
        let websocket = libp2p::websocket::WsConfig::new(
            libp2p::wasm_ext::ExtTransport::new(libp2p::wasm_ext::ffi::websocket_transport())
        );

        OrTransport::new(webrtc, websocket)
            .upgrade(Version::V1)
            .authenticate(noise::Config::new(&keypair)?)
            .multiplex(yamux::Config::default())
            .boxed()
    }
}

/// Native node configuration - acts as Circuit Relay for browsers
#[cfg(not(target_arch = "wasm32"))]
pub struct NativeNodeConfig {
    /// Listen on WebRTC for browser connections
    pub webrtc_listen: Multiaddr,  // e.g., /ip4/0.0.0.0/udp/9000/webrtc-direct
    /// Listen on QUIC for native peer connections
    pub quic_listen: Multiaddr,    // e.g., /ip4/0.0.0.0/udp/9001/quic-v1
    /// Enable Circuit Relay v2 server
    pub relay_enabled: bool,
}

/// Browser node configuration - connects via WebRTC
#[cfg(target_arch = "wasm32")]
pub struct BrowserNodeConfig {
    /// Bootstrap peers (native nodes with WebRTC listeners)
    pub bootstrap_peers: Vec<Multiaddr>,
}

#[cfg(not(target_arch = "wasm32"))]
impl NativeNodeConfig {
    /// Enable Circuit Relay v2 server for browser-to-browser communication
    /// Authority nodes act as relay pivots for Citizens that can't directly connect
    pub fn with_relay(mut self) -> Self {
        self.relay_config = Some(relay::Config::default()
            .max_reservations(128)           // Max browser connections
            .max_circuits(64)                // Max active relays
            .max_circuit_duration(Duration::from_secs(3600))
        );
        self
    }
}

┌──────────────┐                    ┌──────────────┐
│  Browser A   │                    │  Browser B   │
│  (Citizen)   │                    │  (Citizen)   │
└──────┬───────┘                    └───────┬──────┘
       │                                    │
       │ WebRTC                      WebRTC │
       │                                    │
       ▼                                    ▼
┌──────────────────────────────────────────────────┐
│                 Mac (Authority)                   │
│                                                   │
│  1. Browser A and Browser B connect via WebRTC   │
│  2. Mac attempts dcutr hole punch                │
│  3. If direct fails → Circuit Relay v2           │
│  4. GossipSub flows through relay at LAN speed   │
└──────────────────────────────────────────────────┘

// In build_swarm: Authority includes relay server
#[cfg(not(target_arch = "wasm32"))]
pub fn build_swarm_with_relay(keypair: &Keypair, config: &NativeNodeConfig) -> Swarm<FabricBehaviour> {
    let transport = build_transport(keypair);

    let behaviour = FabricBehaviour {
        gossipsub: gossipsub::Behaviour::new(...),
        kad: kad::Behaviour::new(...),
        identify: identify::Behaviour::new(...),
        autonat: autonat::Behaviour::new(local_peer_id, autonat::Config::default()),
        dcutr: dcutr::Behaviour::new(local_peer_id),
        mdns: mdns::tokio::Behaviour::new(mdns::Config::default(), local_peer_id)?,

        // Relay v2: Authority acts as pivot for browser-to-browser
        relay: relay::Behaviour::new(local_peer_id, relay::Config::default()),
        relay_client: relay::client::Behaviour::new(local_peer_id, &keypair),
    };

    SwarmBuilder::with_tokio_executor(transport, behaviour, local_peer_id).build()
}

1. Mac starts with WebRTC listener on /ip4/192.168.1.100/udp/9000/webrtc-direct
2. Browser discovers Mac via Kademlia DHT (Program ID "Alpha" → DHT key)
3. Browser initiates WebRTC connection directly to Mac's IP
4. ICE negotiation happens via libp2p signaling (no external STUN/TURN needed on LAN)
5. GossipSub messages flow over the WebRTC DataChannel

// Native node includes Auto-NAT, Hole Punching, and mDNS
#[cfg(not(target_arch = "wasm32"))]
pub fn build_swarm(keypair: &Keypair) -> Swarm<FabricBehaviour> {
    let transport = build_transport(keypair);

    let behaviour = FabricBehaviour {
        gossipsub: gossipsub::Behaviour::new(...),
        kad: kad::Behaviour::new(...),
        identify: identify::Behaviour::new(...),

        // Auto-NAT: Detect if we're behind NAT
        autonat: autonat::Behaviour::new(local_peer_id, autonat::Config::default()),

        // Hole Punching: Traverse NAT when possible
        dcutr: dcutr::Behaviour::new(local_peer_id),

        // mDNS: Instant local network discovery (Mac finds Browser on same WiFi)
        mdns: mdns::tokio::Behaviour::new(mdns::Config::default(), local_peer_id)?,
    };

    SwarmBuilder::with_tokio_executor(transport, behaviour, local_peer_id).build()
}

#[tokio::test]
async fn test_native_webrtc_listener() {
    // Native node starts WebRTC listener
    // Verify multiaddr is /ip4/.../udp/.../webrtc-direct
}

#[tokio::test]
async fn test_browser_connects_to_native_via_webrtc() {
    // Native node listening on WebRTC
    // Browser node (simulated) connects
    // Both see each other as peers
}

#[tokio::test]
async fn test_gossipsub_over_webrtc() {
    // Native publishes to topic "test"
    // Browser subscribed to "test"
    // Browser receives message via WebRTC DataChannel
}

#[tokio::test]
async fn test_distributed_wasm_syncs_with_native() {
    // Create Distributed<GCounter> on Mac with topic "test"
    // Create Distributed<GCounter> on Browser with topic "test"
    // Mutate on Mac → Browser sees change
    // Mutate on Browser → Mac sees change
}

#[tokio::test]
async fn test_direct_p2p_no_relay_required() {
    // Mac and Browser on same network
    // No external relay server running
    // Connection succeeds via direct WebRTC
}

#[tokio::test]
async fn test_mdns_instant_discovery() {
    // Mac and Browser on same WiFi
    // Browser discovers Mac via mDNS < 100ms
    // No DHT warmup required
}

#[tokio::test]
async fn test_nat_hole_punching() {
    // Mac behind NAT
    // Auto-NAT detects NAT type
    // dcutr (Direct Connection Upgrade) punches hole
    // Browser connects via punched hole
}

// program_id.rs
use libp2p::kad::{Kademlia, KademliaEvent, QueryId, Record, RecordKey};
use sha2::{Sha256, Digest};

#[derive(Clone, Debug, Hash, Eq, PartialEq, Serialize, Deserialize)]
pub struct ProgramId(String);

impl ProgramId {
    pub fn new(id: impl Into<String>) -> Self {
        Self(id.into())
    }

    /// Derive DHT key from Program ID
    /// Hash ensures even distribution across DHT keyspace
    pub fn dht_key(&self) -> RecordKey {
        let mut hasher = Sha256::new();
        hasher.update(b"fabric/program/");
        hasher.update(self.0.as_bytes());
        RecordKey::new(&hasher.finalize())
    }

    /// Derive GossipSub topic from ID
    pub fn gossip_topic(&self) -> IdentTopic {
        IdentTopic::new(format!("fabric/{}/sync", self.0))
    }
}

// discovery.rs - DHT-based peer discovery
pub struct FabricDiscovery {
    kad: Kademlia<MemoryStore>,
    program_id: ProgramId,
    known_peers: HashSet<PeerId>,
}

impl FabricDiscovery {
    /// Announce ourselves to the DHT under the Program ID key
    pub async fn announce(&mut self) -> Result<(), DiscoveryError> {
        let record = Record {
            key: self.program_id.dht_key(),
            value: self.local_peer_info().serialize(),
            publisher: Some(self.local_peer_id),
            expires: Some(Instant::now() + Duration::from_secs(3600)),
        };
        self.kad.put_record(record, Quorum::One)?;
        Ok(())
    }

    /// Find peers sharing our Program ID
    pub async fn discover_peers(&mut self) -> Result<Vec<PeerInfo>, DiscoveryError> {
        self.kad.get_record(self.program_id.dht_key());
        // Returns via KademliaEvent::OutboundQueryProgressed
    }

    /// Bootstrap into the DHT network
    pub async fn bootstrap(&mut self, bootstrap_peers: &[Multiaddr]) -> Result<(), DiscoveryError> {
        for addr in bootstrap_peers {
            self.kad.add_address(&peer_id_from_addr(addr)?, addr.clone());
        }
        self.kad.bootstrap()?;
        Ok(())
    }
}

#[derive(Serialize, Deserialize)]
pub struct PeerInfo {
    pub peer_id: PeerId,
    pub addresses: Vec<Multiaddr>,
    pub capabilities: PeerCapabilities,
    pub timestamp: u64,
}

#[derive(Serialize, Deserialize)]
pub struct PeerCapabilities {
    pub platform: Platform,      // Native, Wasm
    pub arch: Arch,              // X86_64, Aarch64, Wasm32
    pub features: Vec<String>,   // "compute", "storage", "relay"
    pub tier: CapabilityTier,    // LowPower, WebStandard, NativeStandard, HighPerformance
    pub sovereignty: SovereigntyClass,  // Authority, Citizen, Ephemeral
}

impl PeerCapabilities {
    /// Check if this peer is an Authority (quorum voter)
    pub fn is_authority(&self) -> bool {
        matches!(self.sovereignty, SovereigntyClass::Authority)
    }

    /// Check if this peer can vote in write quorum
    pub fn can_vote_in_quorum(&self) -> bool {
        self.is_authority()
    }

    /// Check if this peer can provide JIT-compiled binaries
    pub fn can_provide_jit(&self) -> bool {
        self.is_authority() && self.tier >= CapabilityTier::NativeStandard
    }

    /// Check if this peer can act as a relay
    pub fn can_relay(&self) -> bool {
        matches!(self.sovereignty, SovereigntyClass::Authority | SovereigntyClass::Citizen)
            && self.tier >= CapabilityTier::WebStandard
    }
}

#[derive(Serialize, Deserialize, Clone, Copy)]
pub enum Platform { Native, Wasm }

#[derive(Serialize, Deserialize, Clone, Copy)]
pub enum Arch { X86_64, Aarch64, Wasm32 }

1. Mac joins with Program ID "Alpha"
   - Computes DHT key: SHA256("fabric/program/Alpha")
   - Announces itself: PUT(key, {peer_id, multiaddrs, capabilities})
   - Subscribes to GossipSub topic "fabric/Alpha/sync"

2. Browser joins with Program ID "Alpha"
   - Computes same DHT key
   - Queries DHT: GET(key) → finds Mac's peer info
   - Connects to Mac via WebRTC multiaddr
   - Subscribes to same GossipSub topic

3. Both are now in the same mesh, sharing state via GossipSub

// bootstrap.rs - Generate and parse invite links
pub struct InviteLink {
    pub program_id: ProgramId,
    pub peer_id: PeerId,
    pub addrs: Vec<Multiaddr>,
}

impl InviteLink {
    /// Generate shareable URL
    /// Example: logicaffeine.com/studio?join=Alpha&peer=12D3Koo...&addr=/ip4/192.168.1.100/udp/9000/webrtc-direct
    pub fn to_url(&self) -> String {
        let addrs_encoded = self.addrs.iter()
            .map(|a| urlencoding::encode(&a.to_string()))
            .collect::<Vec<_>>()
            .join(",");

        format!(
            "https://logicaffeine.com/studio?join={}&peer={}&addr={}",
            urlencoding::encode(&self.program_id.0),
            self.peer_id,
            addrs_encoded
        )
    }

    /// Generate QR Code (for mobile/tablet)
    pub fn to_qr(&self) -> QrCode {
        QrCode::new(self.to_url()).unwrap()
    }

    /// Parse from URL query params
    pub fn from_url(url: &str) -> Result<Self, ParseError> { ... }
}

// CLI integration
// When running: largo run --id "Alpha"
// Terminal outputs:
//
// ╭─────────────────────────────────────────────────╮
// │  Fabric Mesh: Alpha                              │
// │  Your peers can join at:                         │
// │                                                  │
// │  logicaffeine.com/studio?join=Alpha&peer=12D... │
// │                                                  │
// │  [QR CODE HERE]                                  │
// │                                                  │
// │  Or scan the QR code above                       │
// ╰─────────────────────────────────────────────────╯

#[tokio::test]
async fn test_program_id_dht_key_deterministic() {
    let id1 = ProgramId::new("Alpha");
    let id2 = ProgramId::new("Alpha");
    assert_eq!(id1.dht_key(), id2.dht_key());
}

#[tokio::test]
async fn test_different_ids_different_keys() {
    let alpha = ProgramId::new("Alpha");
    let beta = ProgramId::new("Beta");
    assert_ne!(alpha.dht_key(), beta.dht_key());
}

#[tokio::test]
async fn test_dht_discovery_finds_peers() {
    // Node A announces under "Alpha"
    // Node B queries for "Alpha"
    // Node B finds Node A's peer info
}

#[tokio::test]
async fn test_dht_discovery_isolates_programs() {
    // Node A announces under "Alpha"
    // Node B queries for "Beta"
    // Node B does NOT find Node A
}

#[tokio::test]
async fn test_browser_discovers_native_via_dht() {
    // Mac announces under "Alpha" with WebRTC multiaddr
    // Browser queries DHT for "Alpha"
    // Browser receives Mac's WebRTC address
    // Browser connects directly
}

// crc.rs - Hardware-accelerated CRC32C
use crc32c::crc32c;  // Uses SSE4.2 on x86, CRC instructions on ARM

/// Compute CRC32C checksum (hardware-accelerated on M-series Mac + modern WASM)
pub fn checksum(data: &[u8]) -> u32 {
    crc32c(data)
}

// WASM: crc32c crate auto-detects WebAssembly SIMD support
// M-series Mac: Uses ARM CRC32C instructions (3x faster than software)
// Intel Mac: Uses SSE4.2 CRC32 instructions

// memory_zone.rs - Zero-copy shared memory
#[cfg(not(target_arch = "wasm32"))]
pub struct MemoryZone {
    mmap: memmap2::MmapMut,
    dirty_pages: BitSet,
}

#[cfg(target_arch = "wasm32")]
pub struct MemoryZone {
    sab: SharedArrayBuffer,  // Requires COOP/COEP headers
    dirty_pages: BitSet,
}

impl MemoryZone {
    /// Sync dirty pages directly to network (zero-copy)
    pub async fn sync_to_fabric(&self, fabric: &FabricHandle) {
        for page_idx in self.dirty_pages.iter() {
            let page_data = self.get_page(page_idx);
            // Send page directly from memory - no intermediate buffer
            fabric.publish_page(page_idx, page_data).await;
        }
        self.dirty_pages.clear();
    }
}

/// First-class durability intent - programmer chooses trade-off
pub enum DurabilityPolicy {
    /// Return immediately after local write (Fastest)
    /// Use for: game state, caches, ephemeral data
    Local,

    /// Wait for N Authority nodes to acknowledge (Safest)
    /// Use for: financial data, user documents
    Quorum(usize),

    /// Wait for the primary Authority to acknowledge
    /// Use for: ordered operations, leader-based workflows
    Leader,
}

# High safety for financial data
Mount wallet at "wallet.journal" with Quorum(2).

# High speed for game state (voxels)
Mount world_data at "voxels.journal" with Local.

// unified.rs
pub struct UnifiedVfs {
    inner: Arc<dyn Vfs + Send + Sync>,
    fabric: Arc<FabricHandle>,
    locks: Arc<Mutex<HashMap<PathBuf, LockState>>>,
    durability_policy: DurabilityPolicy,  // Controls quorum behavior
}

impl UnifiedVfs {
    /// Create with distributed durability
    pub fn new(
        vfs: Arc<dyn Vfs + Send + Sync>,
        fabric: Arc<FabricHandle>,
        durability_policy: DurabilityPolicy,
    ) -> Self {
        Self {
            inner: vfs,
            fabric,
            locks: Arc::new(Mutex::new(HashMap::new())),
            durability_policy,
        }
    }

    /// Distributed durable write
    /// Behavior depends on DurabilityPolicy:
    /// - Local: Return immediately after local write
    /// - Quorum(N): Wait for N Authority nodes to ack
    /// - Leader: Wait for primary Authority to ack
    pub async fn write_durable(&self, path: &Path, data: &[u8]) -> VfsResult<()> {
        // 1. Acquire exclusive lock
        let _lock = self.lock_exclusive(path).await?;

        // 2. Write to local VFS
        self.inner.write(path, data).await?;

        // 3. Compute delta for replication
        let delta = WriteDelta {
            path: path.to_path_buf(),
            content_hash: blake3::hash(data),
            timestamp: SystemTime::now(),
        };

        // 4. Broadcast delta via GossipSub (always, for eventual consistency)
        self.fabric.publish_delta(&delta).await?;

        // 5. Wait for Authority acknowledgments ONLY (not Citizens/Ephemeral)
        // This is the key change: 50 browser tabs don't slow down quorum
        match &self.durability_policy {
            DurabilityPolicy::Local => {
                // Return immediately - local write is enough
                // Use for: game state, caches, ephemeral data
            }
            DurabilityPolicy::Quorum(n) => {
                // Wait for N Authority nodes to ack
                // Citizens and Ephemerals are witnesses, not voters
                self.fabric.await_authority_acks(&delta, *n).await?;
            }
            DurabilityPolicy::Leader => {
                // Wait for primary Authority to ack
                self.fabric.await_leader_ack(&delta).await?;
            }
        }

        Ok(())
    }

    /// Acquire exclusive lock (mandatory, software-level)
    pub async fn lock_exclusive(&self, path: &Path) -> VfsResult<LockGuard> {
        let normalized = Self::normalize_path(path);

        loop {
            let mut locks = self.locks.lock().await;
            match locks.get(&normalized) {
                None | Some(LockState::Unlocked) => {
                    locks.insert(normalized.clone(), LockState::Exclusive {
                        holder: self.fabric.local_replica_id(),
                    });
                    return Ok(LockGuard::new(self.locks.clone(), normalized));
                }
                Some(LockState::Exclusive { .. }) | Some(LockState::Shared { .. }) => {
                    drop(locks);
                    // Wait and retry
                    tokio::time::sleep(Duration::from_millis(10)).await;
                }
            }
        }
    }

    /// Normalize path: case-sensitive, forward slashes only
    fn normalize_path(path: &Path) -> PathBuf {
        let s = path.to_string_lossy();
        PathBuf::from(s.replace('\\', "/"))
        // Note: We do NOT lowercase - paths are case-sensitive
    }
}

// distributed_write.rs
#[derive(Serialize, Deserialize, Clone)]
pub struct WriteDelta {
    pub path: PathBuf,
    pub content_hash: [u8; 32],
    pub timestamp: SystemTime,
}

#[derive(Debug)]
pub enum LockState {
    Exclusive { holder: ReplicaId },
    Shared { readers: HashSet<ReplicaId> },
    Unlocked,
}

pub struct LockGuard {
    locks: Arc<Mutex<HashMap<PathBuf, LockState>>>,
    path: PathBuf,
}

impl Drop for LockGuard {
    fn drop(&mut self) {
        if let Ok(mut locks) = self.locks.try_lock() {
            locks.insert(self.path.clone(), LockState::Unlocked);
        }
    }
}

// fabric_handle.rs - Authority-only quorum methods
impl FabricHandle {
    /// Wait for N Authority nodes to acknowledge (not all peers)
    /// This is the key to preventing quorum bloat from browser tabs
    pub async fn await_authority_acks(&self, delta: &WriteDelta, n: usize) -> Result<(), FabricError> {
        let authorities: Vec<_> = self.connected_peers().await
            .into_iter()
            .filter(|p| p.capabilities.is_authority())
            .collect();

        if authorities.len() < n {
            return Err(FabricError::InsufficientAuthorities {
                required: n,
                available: authorities.len(),
            });
        }

        // Broadcast delta and collect acks from authorities only
        // Citizens/Ephemeral receive the delta but don't block the write
        let acks = self.broadcast_and_collect(&delta, &authorities).await;

        if acks.len() >= n {
            Ok(())
        } else {
            Err(FabricError::QuorumTimeout)
        }
    }

    /// Wait for the primary Authority (leader) to acknowledge
    pub async fn await_leader_ack(&self, delta: &WriteDelta) -> Result<(), FabricError> {
        let leader = self.current_leader().await
            .ok_or(FabricError::NoLeaderElected)?;

        let ack = self.send_and_await_ack(&delta, &leader).await?;
        Ok(())
    }

    /// Get all connected Authority peers
    pub async fn authority_peers(&self) -> Vec<PeerInfo> {
        self.connected_peers().await
            .into_iter()
            .filter(|p| p.capabilities.is_authority())
            .collect()
    }
}

#[derive(Debug, thiserror::Error)]
pub enum FabricError {
    #[error("Insufficient authorities: required {required}, available {available}")]
    InsufficientAuthorities { required: usize, available: usize },

    #[error("Quorum timeout: not enough authorities acknowledged within deadline")]
    QuorumTimeout,

    #[error("No leader elected")]
    NoLeaderElected,
}

#[tokio::test]
async fn test_case_sensitivity_enforced() {
    let vfs = UnifiedVfs::new(...);
    vfs.write_durable(Path::new("File.txt"), b"data").await.unwrap();

    // Different case = different file
    let result = vfs.read(Path::new("file.txt")).await;
    assert!(result.is_err()); // Not found
}

#[tokio::test]
async fn test_mandatory_locking() {
    let vfs = Arc::new(UnifiedVfs::new(...));

    // Process A locks
    let lock_a = vfs.lock_exclusive(Path::new("data.json")).await.unwrap();

    // Process B tries to lock - should block
    let vfs_clone = vfs.clone();
    let handle = tokio::spawn(async move {
        vfs_clone.lock_exclusive(Path::new("data.json")).await
    });

    // Give B time to attempt
    tokio::time::sleep(Duration::from_millis(50)).await;
    assert!(!handle.is_finished()); // Still waiting

    // A releases
    drop(lock_a);

    // B should now succeed
    let lock_b = handle.await.unwrap().unwrap();
}

#[tokio::test]
async fn test_distributed_durable_write() {
    // Browser writes to OPFS
    // Write blocks until Mac acknowledges
    // Only then does write return Ok
}

#[tokio::test]
async fn test_write_fails_without_quorum() {
    // Browser writes with quorum=1
    // No other peers connected
    // Write times out / returns error
}