graphrun 0.1.3

use crate::catalog::{Catalog, ExecutionKind};
use crate::cluster::{ClusterNetwork, MemberConfig};
use crate::compiler::compile_yaml;
use crate::domain::{
    Command, CommandBody, RunStatus, State, active_runs, reconstruct, run_events, run_output,
};
use crate::error::{Error, ErrorKind, Result};
use crate::ids::{CommandId, EventId, RunId};
use crate::ir::Definition;
use crate::limits;
use crate::rpc::serve_grpc;
use crate::storage::{LocalNetwork, StorageHandle, TypeConfig, load_domain_readonly};
use crate::value::Value;
use crate::write::{health_view, inspect_view, now, write_raft};
use openraft::{BasicNode, ChangeMembers, Config, Raft, ServerState, SnapshotPolicy};
use serde::{Deserialize, Serialize};
use std::collections::{BTreeMap, HashMap};
use std::net::SocketAddr;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, LazyLock, Mutex};
use std::thread::JoinHandle;
use std::time::Duration;
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::net::{UnixListener, UnixStream};
use tokio::sync::{Notify, Semaphore};

fn wake(notify: &Notify) {
    notify.notify_waiters();
    notify.notify_one();
}

pub static BLOCKING_IN_FLIGHT: AtomicU64 = AtomicU64::new(0);
pub static SLOW_RELEASE: AtomicBool = AtomicBool::new(false);

#[derive(Clone, Debug)]
pub struct LedgerEntry {
    pub physical: u32,
    pub output: Value,
}

static EFFECT_LEDGER: LazyLock<Mutex<HashMap<String, LedgerEntry>>> =
    LazyLock::new(|| Mutex::new(HashMap::new()));

pub fn ledger_apply(key: &str, compute: impl FnOnce() -> Value) -> Value {
    let mut guard = EFFECT_LEDGER.lock().unwrap();
    if let Some(row) = guard.get_mut(key) {
        row.physical = row.physical.saturating_add(1);
        return row.output.clone();
    }
    let output = compute();
    guard.insert(
        key.to_owned(),
        LedgerEntry {
            physical: 1,
            output: output.clone(),
        },
    );
    output
}

pub fn ledger_get(key: &str) -> Option<LedgerEntry> {
    EFFECT_LEDGER.lock().unwrap().get(key).cloned()
}

pub fn ledger_clear() {
    EFFECT_LEDGER.lock().unwrap().clear();
}

fn keyed_output(key: Option<&str>, kind: &str, compute: impl FnOnce() -> Value) -> Result<Value> {
    match key {
        Some(key) if !key.is_empty() => apply_keyed(key, kind, compute()),
        _ => Ok(compute()),
    }
}

fn apply_keyed(key: &str, kind: &str, output: Value) -> Result<Value> {
    if let Some(url) = crate::provider::provider_url() {
        let resp = crate::provider::apply_effect(&url, key, kind, &output)?;
        match resp.status {
            crate::provider::EffectStatus::Applied => Ok(resp.output.unwrap_or(output)),
            crate::provider::EffectStatus::Pending => Err(Error::new(
                crate::error::ErrorKind::Unavailable,
                "provider effect pending",
            )),
            crate::provider::EffectStatus::NotApplied => Err(Error::new(
                crate::error::ErrorKind::FailedPrecondition,
                "provider effect not applied",
            )),
            crate::provider::EffectStatus::Unknown => Err(Error::new(
                crate::error::ErrorKind::Unavailable,
                "provider effect unknown",
            )),
        }
    } else {
        Ok(ledger_apply(key, || output))
    }
}

/// In-process workflow runtime.
///
/// Open with [`Engine::local`] on a data directory in your Tokio process.
/// There is no in-memory shortcut: commands still go through one-member Raft
/// onto a redb file in that directory.
pub struct Engine {
    raft: Raft<TypeConfig>,
    storage: StorageHandle,
    storage_thread: Mutex<Option<JoinHandle<()>>>,
    data_dir: PathBuf,
    notify: Arc<Notify>,
    worker: Option<tokio::task::JoinHandle<()>>,
    scheduler: Option<tokio::task::JoinHandle<()>>,
    control: tokio::task::JoinHandle<()>,
    raft_server: Option<tokio::task::JoinHandle<()>>,
    snapshot_task: Option<tokio::task::JoinHandle<()>>,
    cluster_net: Option<ClusterNetwork>,
}

#[derive(Clone, Debug, Serialize, Deserialize)]
#[serde(tag = "op", rename_all = "snake_case")]
pub enum ControlRequest {
    Start {
        yaml: String,
        catalog: Catalog,
        input: Value,
        wait_ms: Option<u64>,
    },
    Signal {
        run: String,
        name: String,
        key: String,
        event_id: String,
        payload: Value,
    },
    Cancel {
        run: String,
        reason: String,
    },
    Inspect {
        run: String,
    },
    List,
    History {
        run: String,
    },
    Snapshot,
    Health,
    Acknowledge {
        reason: String,
    },
    ResolveBlocked {
        run: String,
        forward: String,
        input: Value,
    },
    Abandon {
        run: String,
        reason: String,
    },
    Join {
        node_id: u64,
        addr: String,
        ca_pem: String,
        cert_pem: String,
        key_pem: String,
        server_name: String,
    },
    Promote {
        node_id: u64,
    },
    Remove {
        node_id: u64,
    },
}

#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct ControlResponse {
    pub ok: bool,
    pub error: Option<String>,
    pub body: serde_json::Value,
}

impl Engine {
    /// Open a one-member engine on `data_dir`.
    ///
    /// Creates the directory if needed. Restarting on the same path continues
    /// from durable state. Local mode runs built-in fixture handlers for
    /// catalog activity names; unknown names echo their input.
    ///
    /// ```
    /// # tokio::runtime::Runtime::new().unwrap().block_on(async {
    /// let dir = tempfile::tempdir().unwrap();
    /// let engine = graphrun::Engine::local(dir.path()).await.unwrap();
    /// engine.shutdown().await.unwrap();
    /// # });
    /// ```
    pub async fn local(data_dir: impl AsRef<Path>) -> Result<Self> {
        let data_dir = data_dir.as_ref().to_path_buf();
        std::fs::create_dir_all(&data_dir).map_err(|err| Error::invalid(err.to_string()))?;
        write_identity(&data_dir)?;
        let db_path = data_dir.join("member.redb");
        let (storage, storage_thread) = StorageHandle::open(&db_path)?;
        let restored_path = data_dir.join("restored-domain.json");
        if restored_path.exists() {
            let bytes =
                std::fs::read(&restored_path).map_err(|err| Error::invalid(err.to_string()))?;
            let domain: State =
                serde_json::from_slice(&bytes).map_err(|err| Error::invalid(err.to_string()))?;
            storage.install_domain(domain).await?;
            let _ = std::fs::remove_file(&restored_path);
        }
        let log_store = storage.log_store();
        let state_machine = storage.state_machine();
        let config = Config {
            cluster_name: "graphrun-local".to_owned(),
            heartbeat_interval: 250,
            election_timeout_min: 1000,
            election_timeout_max: 2000,
            max_payload_entries: 4,
            snapshot_policy: SnapshotPolicy::Never,
            max_in_snapshot_log_to_keep: 1024,
            ..Config::default()
        }
        .validate()
        .map_err(|err| Error::invalid(err.to_string()))?;
        let raft =
            Raft::<TypeConfig>::new(1, Arc::new(config), LocalNetwork, log_store, state_machine)
                .await
                .map_err(|err| Error::invalid(err.to_string()))?;
        let members = BTreeMap::from([(1, BasicNode::new(""))]);
        match raft.initialize(members).await {
            Ok(()) => {}
            Err(err) => {
                let msg = err.to_string();
                if !msg.contains("NotAllowed") && !msg.contains("not allowed") {
                    return Err(Error::invalid(msg));
                }
            }
        }
        raft.wait(Some(Duration::from_secs(5)))
            .state(ServerState::Leader, "local member is leader")
            .await
            .map_err(|err| Error::invalid(err.to_string()))?;
        let notify = Arc::new(Notify::new());
        let scheduler = Some(tokio::spawn(scheduler_loop(
            raft.clone(),
            storage.clone(),
            notify.clone(),
        )));
        let worker = tokio::spawn(worker_loop(raft.clone(), storage.clone(), notify.clone()));
        let sock = data_dir.join("control.sock");
        let _ = std::fs::remove_file(&sock);
        let listener = UnixListener::bind(&sock).map_err(|err| Error::invalid(err.to_string()))?;
        let control = tokio::spawn(control_loop(
            listener,
            raft.clone(),
            storage.clone(),
            notify.clone(),
            None,
        ));
        let snapshot_task = Some(tokio::spawn(snapshot_controller(
            raft.clone(),
            storage.clone(),
        )));
        Ok(Self {
            raft,
            storage,
            storage_thread: Mutex::new(Some(storage_thread)),
            data_dir,
            notify,
            worker: Some(worker),
            scheduler,
            control,
            raft_server: None,
            snapshot_task,
            cluster_net: None,
        })
    }

    pub async fn member(config: MemberConfig) -> Result<Self> {
        crate::tls::install_provider();
        let data_dir = config.data_dir.clone();
        std::fs::create_dir_all(&data_dir).map_err(|err| Error::invalid(err.to_string()))?;
        write_identity(&data_dir)?;
        let db_path = data_dir.join("member.redb");
        let (storage, storage_thread) = StorageHandle::open(&db_path)?;
        let log_store = storage.log_store();
        let state_machine = storage.state_machine();
        let raft_config = Config {
            cluster_name: "graphrun".to_owned(),
            heartbeat_interval: 250,
            election_timeout_min: 1000,
            election_timeout_max: 2000,
            max_payload_entries: 4,
            snapshot_policy: SnapshotPolicy::Never,
            max_in_snapshot_log_to_keep: 1024,
            ..Config::default()
        }
        .validate()
        .map_err(|err| Error::invalid(err.to_string()))?;
        let network = ClusterNetwork::new(config.peers.clone());
        let raft = Raft::<TypeConfig>::new(
            config.node_id,
            Arc::new(raft_config),
            network.clone(),
            log_store,
            state_machine,
        )
        .await
        .map_err(|err| Error::invalid(err.to_string()))?;
        let notify = Arc::new(Notify::new());
        let scheduler = Some(tokio::spawn(scheduler_loop(
            raft.clone(),
            storage.clone(),
            notify.clone(),
        )));
        let raft_server = {
            let raft = raft.clone();
            let bind = config.bind;
            let tls = config.tls.clone();
            let storage = storage.clone();
            let notify = notify.clone();
            tokio::spawn(async move {
                let _ = serve_grpc(bind, tls, raft, storage, notify).await;
            })
        };
        if config.initialize {
            let mut members =
                BTreeMap::from([(config.node_id, BasicNode::new(config.bind.to_string()))]);
            for (id, (addr, _)) in &config.peers {
                members.insert(*id, BasicNode::new(addr.to_string()));
            }
            match raft.initialize(members).await {
                Ok(()) => {}
                Err(err) => {
                    let msg = err.to_string();
                    if !msg.contains("NotAllowed") && !msg.contains("not allowed") {
                        return Err(Error::invalid(msg));
                    }
                }
            }
            raft.wait(Some(Duration::from_secs(10)))
                .state(ServerState::Leader, "member is leader")
                .await
                .map_err(|err| Error::invalid(err.to_string()))?;
        }
        let worker = if config.host_activities {
            Some(tokio::spawn(worker_loop(
                raft.clone(),
                storage.clone(),
                notify.clone(),
            )))
        } else {
            None
        };
        let sock = data_dir.join("control.sock");
        let _ = std::fs::remove_file(&sock);
        let listener = UnixListener::bind(&sock).map_err(|err| Error::invalid(err.to_string()))?;
        let control = tokio::spawn(control_loop(
            listener,
            raft.clone(),
            storage.clone(),
            notify.clone(),
            Some(network.clone()),
        ));
        let snapshot_task = Some(tokio::spawn(snapshot_controller(
            raft.clone(),
            storage.clone(),
        )));
        Ok(Self {
            raft,
            storage,
            storage_thread: Mutex::new(Some(storage_thread)),
            data_dir,
            notify,
            worker,
            scheduler,
            control,
            raft_server: Some(raft_server),
            snapshot_task,
            cluster_net: Some(network),
        })
    }

    pub fn data_dir(&self) -> &Path {
        &self.data_dir
    }

    pub fn control_sock(&self) -> PathBuf {
        self.data_dir.join("control.sock")
    }

    /// Start a run from a compiled definition and return its id immediately.
    pub async fn start(
        &self,
        definition: Definition,
        catalog: Catalog,
        input: Value,
    ) -> Result<RunId> {
        let run = RunId::generate();
        let command = Command {
            id: CommandId::generate(),
            time: now(),
            body: CommandBody::Start {
                run,
                definition: Box::new(definition),
                input,
                catalog: Box::new(catalog),
            },
        };
        self.write(command).await?;
        wake(&self.notify);
        Ok(run)
    }

    /// Compile `yaml` against `catalog` and [`start`](Self::start) it.
    pub async fn start_yaml(&self, yaml: &str, catalog: &Catalog, input: Value) -> Result<RunId> {
        let definition = compile_yaml(yaml, catalog)?;
        self.start(definition, catalog.clone(), input).await
    }

    /// Deliver an event to a `wait_signal` on `run`.
    ///
    /// `name` and `key` must match the wait. `event_id` is 32 hex characters;
    /// a duplicate id is idempotent.
    pub async fn signal(
        &self,
        run: RunId,
        event_id: EventId,
        name: &str,
        key: &str,
        payload: Value,
    ) -> Result<()> {
        let command = Command {
            id: CommandId::generate(),
            time: now(),
            body: CommandBody::Signal {
                run,
                event_id,
                signal: name.to_owned(),
                key: key.to_owned(),
                payload,
            },
        };
        self.write(command).await?;
        wake(&self.notify);
        Ok(())
    }

    pub async fn cancel(&self, run: RunId, reason: &str) -> Result<()> {
        let command = Command {
            id: CommandId::generate(),
            time: now(),
            body: CommandBody::Cancel {
                run,
                reason: reason.to_owned(),
            },
        };
        self.write(command).await?;
        wake(&self.notify);
        Ok(())
    }

    pub async fn inspect(&self, run: RunId) -> Result<State> {
        let state = self.storage.query_state().await;
        if !state.runs.contains_key(&run) {
            return Err(Error::new(ErrorKind::NotFound, "unknown run"));
        }
        Ok(state)
    }

    pub async fn inspect_json(&self, run: RunId) -> Result<serde_json::Value> {
        let state = self.inspect(run).await?;
        Ok(inspect_view(&state, run))
    }

    pub async fn health(&self) -> serde_json::Value {
        let metrics = self.raft.metrics().borrow().clone();
        let state = self.storage.query_state().await;
        health_view(
            format!("{:?}", metrics.state),
            metrics.last_applied.map(|id| id.index),
            metrics.last_log_index,
            metrics.membership_config.voter_ids().collect(),
            &state,
        )
    }

    pub async fn list(&self) -> serde_json::Value {
        let state = self.storage.query_state().await;
        let runs: Vec<_> = state
            .runs
            .values()
            .map(|run| {
                serde_json::json!({
                    "run": run.id.to_hex(),
                    "definition": run.definition.id,
                    "status": match run.status {
                        RunStatus::Active => "active",
                        RunStatus::Succeeded { .. } => "succeeded",
                        RunStatus::Failed { .. } => "failed",
                    },
                })
            })
            .collect();
        serde_json::json!({ "runs": runs })
    }

    pub async fn history(&self, run: RunId) -> Result<Vec<crate::domain::DomainEvent>> {
        let state = self.inspect(run).await?;
        Ok(crate::domain::history_or_unavailable(&state, run, now())?.to_vec())
    }

    pub async fn run_worker(endpoint: String, tls: crate::tls::TlsMaterial) -> Result<()> {
        crate::tls::install_provider();
        let session = crate::ids::WorkerSessionId::generate();
        let channel = tonic::transport::Channel::from_shared(endpoint)
            .map_err(|err| Error::invalid(err.to_string()))?
            .tls_config(crate::rpc::client_tls(&tls)?)
            .map_err(|err| Error::invalid(err.to_string()))?
            .connect()
            .await
            .map_err(|err| Error::invalid(err.to_string()))?;
        let mut client = crate::generated::worker_client::WorkerClient::new(channel);
        let reg = client
            .register(crate::generated::RegisterRequest {
                session_id: session.to_hex(),
                activities: vec!["*".to_owned()],
                capacity: crate::limits::CLAIM_BATCH,
            })
            .await
            .map_err(|err| Error::invalid(err.to_string()))?;
        if !reg.into_inner().error.is_empty() {
            return Err(Error::invalid("worker register failed"));
        }
        loop {
            let claimed = client
                .claim(crate::generated::ClaimRequest {
                    command_id: CommandId::generate().to_hex(),
                    session_id: session.to_hex(),
                    capacity: crate::limits::CLAIM_BATCH,
                })
                .await
                .map_err(|err| Error::invalid(err.to_string()))?
                .into_inner();
            if !claimed.error.is_empty() || claimed.assignments.is_empty() {
                tokio::time::sleep(Duration::from_millis(50)).await;
                continue;
            }
            for assignment in claimed.assignments {
                let input: Value =
                    serde_json::from_slice(&assignment.input_json).unwrap_or(Value::Null);
                if assignment.role.contains("reconcil") {
                    let outcome = builtin_reconcile(
                        &assignment.activity_name,
                        &input,
                        Some(assignment.effect_key.as_str()),
                    );
                    let tag = match outcome.0 {
                        crate::domain::ReconcileOutcome::Applied => "applied",
                        crate::domain::ReconcileOutcome::NotApplied => "not_applied",
                        crate::domain::ReconcileOutcome::Unknown => "unknown",
                    };
                    let _ = client
                        .reconcile(crate::generated::ReconcileRequest {
                            command_id: CommandId::generate().to_hex(),
                            session_id: session.to_hex(),
                            run_id: assignment.run_id,
                            activation_id: assignment.activation_id,
                            outcome: tag.to_owned(),
                            output_json: outcome
                                .1
                                .map(|value| serde_json::to_vec(&value).unwrap_or_default())
                                .unwrap_or_default(),
                            generation: assignment.generation,
                            revision: assignment.revision,
                        })
                        .await;
                    continue;
                }
                let Ok(output) = dispatch_handler(
                    &assignment.activity_name,
                    &input,
                    Some(&assignment.effect_key),
                ) else {
                    continue;
                };
                let _ = client
                    .report(crate::generated::ReportRequest {
                        command_id: CommandId::generate().to_hex(),
                        session_id: session.to_hex(),
                        run_id: assignment.run_id,
                        activation_id: assignment.activation_id,
                        generation: assignment.generation,
                        revision: assignment.revision,
                        output_json: serde_json::to_vec(&output).unwrap_or_default(),
                    })
                    .await;
            }
        }
    }

    pub fn is_leader(&self) -> bool {
        self.raft.metrics().borrow().state == ServerState::Leader
    }

    pub fn voter_ids(&self) -> Vec<u64> {
        self.raft
            .metrics()
            .borrow()
            .membership_config
            .voter_ids()
            .collect()
    }

    pub fn insert_peer(&self, id: u64, addr: SocketAddr, tls: crate::tls::TlsMaterial) {
        if let Some(net) = &self.cluster_net {
            net.insert_peer(id, addr, tls);
        }
    }

    pub async fn add_learner(&self, id: u64, addr: SocketAddr) -> Result<()> {
        self.raft
            .add_learner(id, BasicNode::new(addr.to_string()), true)
            .await
            .map_err(|err| Error::invalid(err.to_string()))?;
        Ok(())
    }

    pub async fn add_voter(&self, id: u64) -> Result<()> {
        self.raft
            .change_membership(
                ChangeMembers::AddVoterIds(std::iter::once(id).collect()),
                true,
            )
            .await
            .map_err(|err| Error::invalid(err.to_string()))?;
        Ok(())
    }

    pub async fn remove_voter(&self, id: u64) -> Result<()> {
        self.raft
            .change_membership(
                ChangeMembers::RemoveVoters(std::iter::once(id).collect()),
                false,
            )
            .await
            .map_err(|err| Error::invalid(err.to_string()))?;
        Ok(())
    }

    pub async fn abandon_compensation(&self, run: RunId, reason: &str) -> Result<()> {
        let command = Command {
            id: CommandId::generate(),
            time: now(),
            body: CommandBody::AbandonCompensation {
                run,
                reason: reason.to_owned(),
            },
        };
        self.write(command).await?;
        wake(&self.notify);
        Ok(())
    }

    pub async fn resolve_blocked(
        &self,
        run: RunId,
        forward: crate::ids::ActivationId,
        input: Value,
    ) -> Result<()> {
        let command = Command {
            id: CommandId::generate(),
            time: now(),
            body: CommandBody::ResolveBlocked {
                run,
                forward,
                input,
            },
        };
        self.write(command).await?;
        wake(&self.notify);
        Ok(())
    }

    pub async fn acknowledge_recovery(&self, reason: &str) -> Result<()> {
        let command = Command {
            id: CommandId::generate(),
            time: now(),
            body: CommandBody::AcknowledgeRecovery {
                reason: reason.to_owned(),
            },
        };
        self.write(command).await?;
        wake(&self.notify);
        Ok(())
    }

    pub fn backup(data_dir: impl AsRef<Path>, out: impl AsRef<Path>) -> Result<()> {
        let out = out.as_ref();
        std::fs::create_dir_all(out).map_err(|err| Error::invalid(err.to_string()))?;
        let state = load_domain_readonly(data_dir.as_ref().join("member.redb"))?;
        let body =
            serde_json::to_vec_pretty(&state).map_err(|err| Error::invalid(err.to_string()))?;
        std::fs::write(out.join("domain.json"), body)
            .map_err(|err| Error::invalid(err.to_string()))?;
        let manifest = serde_json::json!({
            "format": "graphrun.backup/v1",
            "kind": "logical-domain",
        });
        std::fs::write(
            out.join("manifest.json"),
            serde_json::to_vec_pretty(&manifest).unwrap(),
        )
        .map_err(|err| Error::invalid(err.to_string()))?;
        Ok(())
    }

    pub fn restore(from: impl AsRef<Path>, dest: impl AsRef<Path>, reason: &str) -> Result<()> {
        if reason.is_empty() {
            return Err(Error::invalid("restore requires a reason"));
        }
        let from = from.as_ref();
        let dest = dest.as_ref();
        let bytes = std::fs::read(from.join("domain.json"))
            .map_err(|err| Error::invalid(err.to_string()))?;
        let mut domain: State =
            serde_json::from_slice(&bytes).map_err(|err| Error::invalid(err.to_string()))?;
        domain.recovery = Some(crate::domain::RecoveryHold {
            reason: reason.to_owned(),
            authorized: false,
        });
        for act in domain.activations.values_mut() {
            if act.status == crate::domain::ActivationStatus::Ready {
                domain
                    .interventions
                    .insert(act.id, "restored active scope".to_owned());
            }
            act.claim = None;
        }
        std::fs::create_dir_all(dest).map_err(|err| Error::invalid(err.to_string()))?;
        let cluster = format!(
            "graphrun-restored-{}",
            crate::ids::CommandId::generate().to_hex()
        );
        let identity = serde_json::json!({
            "mode": "local",
            "node_id": 1u64,
            "cluster_name": cluster,
            "restored": true,
        });
        std::fs::write(
            dest.join("identity.json"),
            serde_json::to_vec_pretty(&identity).unwrap(),
        )
        .map_err(|err| Error::invalid(err.to_string()))?;
        std::fs::write(
            dest.join("restored-domain.json"),
            serde_json::to_vec(&domain).unwrap(),
        )
        .map_err(|err| Error::invalid(err.to_string()))?;
        Ok(())
    }

    pub async fn snapshot(&self) -> Result<()> {
        self.raft
            .trigger()
            .snapshot()
            .await
            .map_err(|err| Error::invalid(err.to_string()))
    }

    pub async fn last_applied_index(&self) -> u64 {
        self.storage.last_applied_index().await
    }

    pub fn raft_applied_index(&self) -> u64 {
        self.raft
            .metrics()
            .borrow()
            .last_applied
            .map(|id| id.index)
            .unwrap_or(0)
    }

    /// Block until `run` succeeds or fails, or `timeout` elapses.
    ///
    /// On success, returns the run output. On domain failure, returns an error
    /// with the failure code and message.
    pub async fn wait_terminal(&self, run: RunId, timeout: Duration) -> Result<Value> {
        let deadline = tokio::time::Instant::now() + timeout;
        loop {
            let state = self.storage.query_state().await;
            if let Some(output) = run_output(&state, run) {
                return Ok(output);
            }
            if let Some(run_state) = state.runs.get(&run) {
                if let RunStatus::Failed { error } = &run_state.status {
                    return Err(Error::invalid(format!(
                        "run failed: {} {}",
                        error.code, error.message
                    )));
                }
            }
            if tokio::time::Instant::now() >= deadline {
                return Err(Error::new(
                    ErrorKind::DeadlineExceeded,
                    "timed out waiting for run",
                ));
            }
            tokio::time::sleep(Duration::from_millis(20)).await;
            self.progress(run).await?;
        }
    }

    pub async fn progress(&self, run: RunId) -> Result<()> {
        self.commit_progress(run).await?;
        wake(&self.notify);
        Ok(())
    }

    pub async fn commit_progress(&self, run: RunId) -> Result<()> {
        let command = Command {
            id: CommandId::generate(),
            time: now(),
            body: CommandBody::Progress { run },
        };
        self.write(command).await
    }

    /// Stop workers, Raft, and the control socket. Does not delete `data_dir`.
    pub async fn shutdown(&self) -> Result<()> {
        if let Some(worker) = &self.worker {
            worker.abort();
        }
        if let Some(scheduler) = &self.scheduler {
            scheduler.abort();
        }
        self.control.abort();
        if let Some(server) = &self.raft_server {
            server.abort();
        }
        if let Some(task) = &self.snapshot_task {
            task.abort();
        }
        let _ = self.raft.shutdown().await;
        self.storage.shutdown();
        if let Some(thread) = self.storage_thread.lock().unwrap().take() {
            let _ = thread.join();
        }
        let _ = std::fs::remove_file(self.control_sock());
        Ok(())
    }

    async fn write(&self, command: Command) -> Result<()> {
        write_raft(&self.raft, command).await
    }
}

impl Drop for Engine {
    fn drop(&mut self) {
        if let Some(worker) = &self.worker {
            worker.abort();
        }
        if let Some(scheduler) = &self.scheduler {
            scheduler.abort();
        }
        self.control.abort();
        if let Some(server) = &self.raft_server {
            server.abort();
        }
        if let Some(task) = &self.snapshot_task {
            task.abort();
        }
        self.storage.shutdown();
        if let Some(thread) = self.storage_thread.lock().unwrap().take() {
            let _ = thread.join();
        }
        let _ = std::fs::remove_file(self.data_dir.join("control.sock"));
    }
}

pub fn replay(data_dir: impl AsRef<Path>) -> Result<State> {
    let db_path = data_dir.as_ref().join("member.redb");
    let state = load_domain_readonly(&db_path)?;
    for (run, events) in &state.history {
        let Some(run_state) = state.runs.get(run) else {
            continue;
        };
        let rebuilt = reconstruct(
            run_state.definition.clone(),
            run_state.catalog.clone(),
            events,
        )?;
        let live = run_output(&state, *run);
        let replayed = run_output(&rebuilt, *run);
        if live != replayed {
            return Err(Error::invalid(format!(
                "replay mismatch for run {}",
                run.to_hex()
            )));
        }
    }
    Ok(state)
}

pub async fn connect_control(
    sock: impl AsRef<Path>,
    req: ControlRequest,
) -> Result<ControlResponse> {
    let mut stream = UnixStream::connect(sock.as_ref())
        .await
        .map_err(|err| Error::invalid(err.to_string()))?;
    let mut line = serde_json::to_string(&req).map_err(|err| Error::invalid(err.to_string()))?;
    line.push('\n');
    stream
        .write_all(line.as_bytes())
        .await
        .map_err(|err| Error::invalid(err.to_string()))?;
    let mut reader = BufReader::new(stream);
    let mut response = String::new();
    reader
        .read_line(&mut response)
        .await
        .map_err(|err| Error::invalid(err.to_string()))?;
    serde_json::from_str(&response).map_err(|err| Error::invalid(err.to_string()))
}

fn write_identity(dir: &Path) -> Result<()> {
    let path = dir.join("identity.json");
    if path.exists() {
        return Ok(());
    }
    let body = serde_json::json!({
        "mode": "local",
        "node_id": 1u64,
        "cluster_name": "graphrun-local",
    });
    std::fs::write(&path, serde_json::to_vec_pretty(&body).unwrap())
        .map_err(|err| Error::invalid(err.to_string()))
}

async fn control_loop(
    listener: UnixListener,
    raft: Raft<TypeConfig>,
    storage: StorageHandle,
    notify: Arc<Notify>,
    cluster_net: Option<ClusterNetwork>,
) {
    loop {
        let Ok((stream, _)) = listener.accept().await else {
            continue;
        };
        let raft = raft.clone();
        let storage = storage.clone();
        let notify = notify.clone();
        let cluster_net = cluster_net.clone();
        tokio::spawn(async move {
            let _ = handle_control(stream, raft, storage, notify, cluster_net).await;
        });
    }
}

async fn handle_control(
    stream: UnixStream,
    raft: Raft<TypeConfig>,
    storage: StorageHandle,
    notify: Arc<Notify>,
    cluster_net: Option<ClusterNetwork>,
) -> Result<()> {
    let mut reader = BufReader::new(stream);
    let mut line = String::new();
    reader
        .read_line(&mut line)
        .await
        .map_err(|err| Error::invalid(err.to_string()))?;
    let req: ControlRequest =
        serde_json::from_str(&line).map_err(|err| Error::invalid(err.to_string()))?;
    let resp = dispatch_control(req, &raft, &storage, &notify, cluster_net.as_ref()).await;
    let mut out = serde_json::to_string(&resp).unwrap();
    out.push('\n');
    let mut stream = reader.into_inner();
    stream
        .write_all(out.as_bytes())
        .await
        .map_err(|err| Error::invalid(err.to_string()))?;
    Ok(())
}

async fn dispatch_control(
    req: ControlRequest,
    raft: &Raft<TypeConfig>,
    storage: &StorageHandle,
    notify: &Notify,
    cluster_net: Option<&ClusterNetwork>,
) -> ControlResponse {
    let result = match req {
        ControlRequest::Health => {
            let metrics = raft.metrics().borrow().clone();
            let state = storage.query_state().await;
            Ok(health_view(
                format!("{:?}", metrics.state),
                metrics.last_applied.map(|id| id.index),
                metrics.last_log_index,
                metrics.membership_config.voter_ids().collect(),
                &state,
            ))
        }
        ControlRequest::List => {
            let state = storage.query_state().await;
            let runs: Vec<_> = state
                .runs
                .values()
                .map(|run| {
                    serde_json::json!({
                        "run": run.id.to_hex(),
                        "definition": run.definition.id,
                        "status": match run.status {
                            RunStatus::Active => "active",
                            RunStatus::Succeeded { .. } => "succeeded",
                            RunStatus::Failed { .. } => "failed",
                        },
                    })
                })
                .collect();
            Ok(serde_json::json!({"runs": runs}))
        }
        ControlRequest::Start {
            yaml,
            catalog,
            input,
            wait_ms,
        } => match start_via_raft(raft, notify, &yaml, catalog, input).await {
            Ok(run) => {
                if let Some(ms) = wait_ms {
                    match wait_via_raft(raft, storage, notify, run, Duration::from_millis(ms)).await
                    {
                        Ok(output) => Ok(serde_json::json!({
                            "run": run.to_hex(),
                            "status": "succeeded",
                            "output": output,
                        })),
                        Err(err) => Ok(serde_json::json!({
                            "run": run.to_hex(),
                            "status": "pending",
                            "error": err.to_string(),
                        })),
                    }
                } else {
                    Ok(serde_json::json!({"run": run.to_hex(), "status": "started"}))
                }
            }
            Err(err) => Err(err),
        },
        ControlRequest::Signal {
            run,
            name,
            key,
            event_id,
            payload,
        } => {
            let parsed = parse_run_event(&run, &event_id);
            match parsed {
                Ok((run, event_id)) => write_raft(
                    raft,
                    Command {
                        id: CommandId::generate(),
                        time: now(),
                        body: CommandBody::Signal {
                            run,
                            event_id,
                            signal: name,
                            key,
                            payload,
                        },
                    },
                )
                .await
                .map(|_| {
                    wake(notify);
                    serde_json::json!({"status":"ok"})
                }),
                Err(err) => Err(err),
            }
        }
        ControlRequest::Cancel { run, reason } => match RunId::from_hex(&run) {
            Ok(run) => write_raft(
                raft,
                Command {
                    id: CommandId::generate(),
                    time: now(),
                    body: CommandBody::Cancel { run, reason },
                },
            )
            .await
            .map(|_| {
                wake(notify);
                serde_json::json!({"status":"ok"})
            }),
            Err(err) => Err(Error::invalid(err)),
        },
        ControlRequest::Inspect { run } => match RunId::from_hex(&run) {
            Ok(run) => {
                let state = storage.query_state().await;
                if state.runs.contains_key(&run) {
                    Ok(inspect_view(&state, run))
                } else {
                    Err(Error::new(ErrorKind::NotFound, "unknown run"))
                }
            }
            Err(err) => Err(Error::invalid(err)),
        },
        ControlRequest::History { run } => match RunId::from_hex(&run) {
            Ok(run) => {
                let state = storage.query_state().await;
                match serde_json::to_value(run_events(&state, run)) {
                    Ok(body) => Ok(body),
                    Err(err) => Err(Error::invalid(err.to_string())),
                }
            }
            Err(err) => Err(Error::invalid(err)),
        },
        ControlRequest::Snapshot => raft
            .trigger()
            .snapshot()
            .await
            .map(|_| serde_json::json!({"status":"ok"}))
            .map_err(|err| Error::invalid(err.to_string())),
        ControlRequest::Acknowledge { reason } => write_raft(
            raft,
            Command {
                id: CommandId::generate(),
                time: now(),
                body: CommandBody::AcknowledgeRecovery { reason },
            },
        )
        .await
        .map(|_| {
            wake(notify);
            serde_json::json!({"status":"ok"})
        }),
        ControlRequest::ResolveBlocked {
            run,
            forward,
            input,
        } => match (
            RunId::from_hex(&run),
            crate::ids::ActivationId::from_hex(&forward),
        ) {
            (Ok(run), Ok(forward)) => write_raft(
                raft,
                Command {
                    id: CommandId::generate(),
                    time: now(),
                    body: CommandBody::ResolveBlocked {
                        run,
                        forward,
                        input,
                    },
                },
            )
            .await
            .map(|_| {
                wake(notify);
                serde_json::json!({"status":"ok"})
            }),
            (Err(err), _) | (_, Err(err)) => Err(Error::invalid(err)),
        },
        ControlRequest::Abandon { run, reason } => match RunId::from_hex(&run) {
            Ok(run) => write_raft(
                raft,
                Command {
                    id: CommandId::generate(),
                    time: now(),
                    body: CommandBody::AbandonCompensation { run, reason },
                },
            )
            .await
            .map(|_| {
                wake(notify);
                serde_json::json!({"status":"ok"})
            }),
            Err(err) => Err(Error::invalid(err)),
        },
        ControlRequest::Join {
            node_id,
            addr,
            ca_pem,
            cert_pem,
            key_pem,
            server_name,
        } => match (cluster_net, addr.parse::<std::net::SocketAddr>()) {
            (None, _) => Err(Error::invalid("join requires a clustered member")),
            (_, Err(err)) => Err(Error::invalid(err.to_string())),
            (Some(net), Ok(parsed)) => {
                net.insert_peer(
                    node_id,
                    parsed,
                    crate::tls::TlsMaterial {
                        ca_pem,
                        cert_pem,
                        key_pem,
                        server_name,
                    },
                );
                raft.add_learner(node_id, BasicNode::new(parsed.to_string()), true)
                    .await
                    .map(|_| serde_json::json!({"status":"ok","node_id": node_id}))
                    .map_err(|err| Error::invalid(err.to_string()))
            }
        },
        ControlRequest::Promote { node_id } => raft
            .change_membership(
                ChangeMembers::AddVoterIds(std::iter::once(node_id).collect()),
                true,
            )
            .await
            .map(|_| serde_json::json!({"status":"ok","node_id": node_id}))
            .map_err(|err| Error::invalid(err.to_string())),
        ControlRequest::Remove { node_id } => raft
            .change_membership(
                ChangeMembers::RemoveVoters(std::iter::once(node_id).collect()),
                false,
            )
            .await
            .map(|_| serde_json::json!({"status":"ok","node_id": node_id}))
            .map_err(|err| Error::invalid(err.to_string())),
    };
    match result {
        Ok(body) => ControlResponse {
            ok: true,
            error: None,
            body,
        },
        Err(err) => ControlResponse {
            ok: false,
            error: Some(err.to_string()),
            body: serde_json::Value::Null,
        },
    }
}

fn parse_run_event(run: &str, event_id: &str) -> Result<(RunId, EventId)> {
    let run = RunId::from_hex(run).map_err(Error::invalid)?;
    let event_id = EventId::from_hex(event_id).map_err(Error::invalid)?;
    Ok((run, event_id))
}

async fn start_via_raft(
    raft: &Raft<TypeConfig>,
    notify: &Notify,
    yaml: &str,
    catalog: Catalog,
    input: Value,
) -> Result<RunId> {
    let definition = compile_yaml(yaml, &catalog)?;
    let run = RunId::generate();
    write_raft(
        raft,
        Command {
            id: CommandId::generate(),
            time: now(),
            body: CommandBody::Start {
                run,
                definition: Box::new(definition),
                input,
                catalog: Box::new(catalog),
            },
        },
    )
    .await?;
    wake(notify);
    Ok(run)
}

async fn wait_via_raft(
    raft: &Raft<TypeConfig>,
    storage: &StorageHandle,
    notify: &Notify,
    run: RunId,
    timeout: Duration,
) -> Result<Value> {
    let deadline = tokio::time::Instant::now() + timeout;
    loop {
        let state = storage.query_state().await;
        if let Some(output) = run_output(&state, run) {
            return Ok(output);
        }
        if let Some(run_state) = state.runs.get(&run) {
            if let RunStatus::Failed { error } = &run_state.status {
                return Err(Error::invalid(format!(
                    "run failed: {} {}",
                    error.code, error.message
                )));
            }
        }
        if tokio::time::Instant::now() >= deadline {
            return Err(Error::new(
                ErrorKind::DeadlineExceeded,
                "timed out waiting for run",
            ));
        }
        let _ = write_raft(
            raft,
            Command {
                id: CommandId::generate(),
                time: now(),
                body: CommandBody::Progress { run },
            },
        )
        .await;
        wake(notify);
        tokio::time::sleep(Duration::from_millis(20)).await;
    }
}

pub const SNAPSHOT_ENTRY_THRESHOLD: u64 = 20_000;
pub const SNAPSHOT_INTERVAL: Duration = Duration::from_secs(30 * 60);

pub fn should_snapshot(applied: u64, last_snapshot: u64, since_snapshot: Duration) -> bool {
    applied.saturating_sub(last_snapshot) >= SNAPSHOT_ENTRY_THRESHOLD
        || (since_snapshot >= SNAPSHOT_INTERVAL && applied > last_snapshot)
}

async fn snapshot_controller(raft: Raft<TypeConfig>, _storage: StorageHandle) {
    let mut last_snapshot = 0u64;
    let mut since_snapshot = tokio::time::Instant::now();
    loop {
        tokio::time::sleep(Duration::from_secs(1)).await;
        let applied = raft
            .metrics()
            .borrow()
            .last_applied
            .map(|id| id.index)
            .unwrap_or(0);
        if should_snapshot(applied, last_snapshot, since_snapshot.elapsed()) {
            if raft.trigger().snapshot().await.is_ok() {
                last_snapshot = applied;
                since_snapshot = tokio::time::Instant::now();
            }
        }
    }
}

async fn scheduler_loop(raft: Raft<TypeConfig>, storage: StorageHandle, notify: Arc<Notify>) {
    loop {
        tokio::select! {
            _ = notify.notified() => {}
            _ = tokio::time::sleep(Duration::from_millis(50)) => {}
        }
        let state = storage.query_state().await;
        for run in active_runs(&state) {
            let _ = write_raft(
                &raft,
                Command {
                    id: CommandId::generate(),
                    time: now(),
                    body: CommandBody::Progress { run },
                },
            )
            .await;
        }
    }
}

async fn worker_loop(raft: Raft<TypeConfig>, storage: StorageHandle, notify: Arc<Notify>) {
    let session = crate::ids::WorkerSessionId::generate();
    let blocking_slots = Arc::new(Semaphore::new(limits::BLOCKING_POOL_DEFAULT as usize));
    let _ = write_raft(
        &raft,
        Command {
            id: CommandId::generate(),
            time: now(),
            body: CommandBody::RegisterSession {
                session,
                activities: vec!["*".to_owned()],
                capacity: crate::limits::CLAIM_BATCH,
            },
        },
    )
    .await;
    loop {
        tokio::select! {
            _ = notify.notified() => {}
            _ = tokio::time::sleep(Duration::from_millis(50)) => {}
        }
        let claim = Command {
            id: CommandId::generate(),
            time: now(),
            body: CommandBody::Claim {
                session,
                capacity: crate::limits::CLAIM_BATCH,
            },
        };
        if write_raft(&raft, claim.clone()).await.is_err() {
            let _ = write_raft(
                &raft,
                Command {
                    id: CommandId::generate(),
                    time: now(),
                    body: CommandBody::RegisterSession {
                        session,
                        activities: vec!["*".to_owned()],
                        capacity: crate::limits::CLAIM_BATCH,
                    },
                },
            )
            .await;
            continue;
        }
        let mut did_work = false;
        let state = storage.query_state().await;
        let events = state.commands.get(&claim.id).cloned().unwrap_or_default();
        for assignment in crate::domain::assignments_from(&state, &events) {
            let blocking = state
                .runs
                .get(&assignment.run)
                .and_then(|run_state| {
                    run_state
                        .catalog
                        .activities
                        .get(&crate::ids::ActivityKey::new(
                            &assignment.activity_name,
                            assignment.activity_version,
                        ))
                })
                .is_some_and(|contract| contract.execution == ExecutionKind::Blocking);
            if assignment.role == crate::ids::ExecutionRole::Reconciliation {
                let outcome = builtin_reconcile(
                    &assignment.activity_name,
                    &assignment.input,
                    Some(&assignment.effect_key.to_hex()),
                );
                let _ = write_raft(
                    &raft,
                    Command {
                        id: CommandId::generate(),
                        time: now(),
                        body: CommandBody::Reconcile {
                            run: assignment.run,
                            activation: assignment.activation,
                            session: assignment.session,
                            generation: assignment.generation,
                            revision: assignment.revision,
                            outcome: outcome.0,
                            output: outcome.1,
                        },
                    },
                )
                .await;
            } else {
                let output = if blocking {
                    let name = assignment.activity_name.clone();
                    let input = assignment.input.clone();
                    let Ok(permit) = blocking_slots.clone().acquire_owned().await else {
                        continue;
                    };
                    match tokio::task::spawn_blocking(move || {
                        let _permit = permit;
                        builtin_handler(&name, &input)
                    })
                    .await
                    {
                        Ok(Ok(output)) => output,
                        _ => continue,
                    }
                } else {
                    match dispatch_handler(
                        &assignment.activity_name,
                        &assignment.input,
                        Some(&assignment.effect_key.to_hex()),
                    ) {
                        Ok(output) => output,
                        Err(_) => continue,
                    }
                };
                let _ = write_raft(
                    &raft,
                    Command {
                        id: CommandId::generate(),
                        time: now(),
                        body: CommandBody::ReportAssigned {
                            run: assignment.run,
                            activation: assignment.activation,
                            output,
                            session: assignment.session,
                            generation: assignment.generation,
                            revision: assignment.revision,
                        },
                    },
                )
                .await;
            }
            did_work = true;
        }
        if did_work {
            wake(&notify);
        }
    }
}

pub fn builtin_handler(name: &str, input: &Value) -> Result<Value> {
    dispatch_handler(name, input, None)
}

pub(crate) fn dispatch_handler(
    name: &str,
    input: &Value,
    effect_key: Option<&str>,
) -> Result<Value> {
    match name {
        "counter.increment" => {
            let Value::Object(fields) = input else {
                return Err(Error::invalid("counter input"));
            };
            let value = fields
                .get("value")
                .and_then(Value::as_i64)
                .ok_or_else(|| Error::invalid("counter.value"))?
                + 1;
            Ok(Value::Object(BTreeMap::from([(
                "value".to_owned(),
                Value::Int(value),
            )])))
        }
        "inventory.reserve" => {
            let Value::Object(fields) = input else {
                return Err(Error::invalid("order input"));
            };
            let order_id = fields
                .get("order_id")
                .cloned()
                .unwrap_or(Value::String(String::new()));
            let amount = fields.get("amount").cloned().unwrap_or(Value::Int(0));
            keyed_output(effect_key, "forward", || {
                Value::Object(BTreeMap::from([
                    ("order_id".to_owned(), order_id.clone()),
                    ("amount".to_owned(), amount.clone()),
                    (
                        "reservation_id".to_owned(),
                        Value::String("res-1".to_owned()),
                    ),
                ]))
            })
        }
        "payment.charge" => {
            let Value::Object(fields) = input else {
                return Err(Error::invalid("reserved input"));
            };
            let order_id = fields
                .get("order_id")
                .cloned()
                .unwrap_or(Value::String(String::new()));
            let amount = fields.get("amount").cloned().unwrap_or(Value::Int(0));
            keyed_output(effect_key, "forward", || {
                Value::Object(BTreeMap::from([
                    ("order_id".to_owned(), order_id.clone()),
                    ("amount".to_owned(), amount.clone()),
                    ("payment_id".to_owned(), Value::String("pay-1".to_owned())),
                ]))
            })
        }
        "tax.quote" => {
            let Value::Object(fields) = input else {
                return Err(Error::invalid("order input"));
            };
            let amount = fields.get("amount").and_then(Value::as_i64).unwrap_or(0);
            Ok(Value::Object(BTreeMap::from([(
                "cents".to_owned(),
                Value::Int(amount / 10),
            )])))
        }
        "shipping.quote" => Ok(Value::Object(BTreeMap::from([(
            "cents".to_owned(),
            Value::Int(500),
        )]))),
        "test.block" => {
            BLOCKING_IN_FLIGHT.fetch_add(1, Ordering::SeqCst);
            std::thread::sleep(Duration::from_millis(400));
            BLOCKING_IN_FLIGHT.fetch_sub(1, Ordering::SeqCst);
            Ok(input.clone())
        }
        "remote.echo" | "test.gate" => Ok(input.clone()),
        "inventory.release" | "payment.refund" => {
            if name == "inventory.release" && SLOW_RELEASE.load(Ordering::SeqCst) {
                std::thread::sleep(Duration::from_millis(800));
            }
            if let Some(key) = effect_key {
                let undo = format!("{key}:undo");
                apply_keyed(&undo, "compensate", Value::Null)?;
            }
            Ok(Value::Null)
        }
        "test.manual" => Ok(input.clone()),
        _ => Ok(input.clone()),
    }
}

pub fn builtin_reconcile(
    name: &str,
    input: &Value,
    effect_key: Option<&str>,
) -> (crate::domain::ReconcileOutcome, Option<Value>) {
    if let (Some(url), Some(key)) = (crate::provider::provider_url(), effect_key) {
        match crate::provider::probe_effect(&url, key) {
            Ok(resp) => {
                return match resp.status {
                    crate::provider::EffectStatus::Applied => (
                        crate::domain::ReconcileOutcome::Applied,
                        resp.output.or_else(|| Some(input.clone())),
                    ),
                    crate::provider::EffectStatus::NotApplied => {
                        (crate::domain::ReconcileOutcome::NotApplied, None)
                    }
                    crate::provider::EffectStatus::Pending
                    | crate::provider::EffectStatus::Unknown => {
                        (crate::domain::ReconcileOutcome::Unknown, None)
                    }
                };
            }
            Err(_) => return (crate::domain::ReconcileOutcome::Unknown, None),
        }
    }
    match name {
        "test.lookup" | "inventory.lookup" | "payment.lookup" => (
            crate::domain::ReconcileOutcome::Applied,
            Some(input.clone()),
        ),
        _ => (crate::domain::ReconcileOutcome::Unknown, None),
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    static COMPENSATION_TEST: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(());

    fn catalog() -> Catalog {
        Catalog::from_json(include_bytes!(
            "../../docs/specs/v1/examples/activity-catalog.json"
        ))
        .unwrap()
    }

    fn order() -> Value {
        Value::Object(BTreeMap::from([
            ("order_id".to_owned(), Value::String("o1".to_owned())),
            ("amount".to_owned(), Value::Int(1000)),
        ]))
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn local_sequence_survives_restart() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = include_str!("../../docs/specs/v1/examples/sequence.yaml");
        let run = {
            let engine = Engine::local(dir.path()).await.unwrap();
            let run = engine.start_yaml(yaml, &catalog(), order()).await.unwrap();
            let output = engine
                .wait_terminal(run, Duration::from_secs(10))
                .await
                .unwrap();
            let Value::Object(fields) = output else {
                panic!("receipt");
            };
            assert_eq!(fields.get("payment_id").unwrap().as_str(), Some("pay-1"));
            engine.shutdown().await.unwrap();
            run
        };
        let engine = Engine::local(dir.path()).await.unwrap();
        let state = engine.inspect(run).await.unwrap();
        let output = run_output(&state, run).expect("persisted run");
        let Value::Object(fields) = output else {
            panic!("receipt");
        };
        assert_eq!(fields.get("payment_id").unwrap().as_str(), Some("pay-1"));
        engine.shutdown().await.unwrap();
        let replayed = replay(dir.path()).unwrap();
        assert!(run_output(&replayed, run).is_some());
        let before = std::fs::metadata(dir.path().join("member.redb"))
            .unwrap()
            .modified()
            .unwrap();
        let _ = replay(dir.path()).unwrap();
        let after = std::fs::metadata(dir.path().join("member.redb"))
            .unwrap()
            .modified()
            .unwrap();
        assert_eq!(before, after, "readonly replay must not write the store");
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn local_engine_survives_past_session_lease() {
        let dir = tempfile::tempdir().unwrap();
        let engine = Engine::local(dir.path()).await.unwrap();
        let yaml = r#"
dsl: graphrun/v1
id: bump
version: 1
input_schema: counter/v1
output_schema: counter/v1
start: inc
nodes:
  inc:
    kind: activity
    activity: {name: counter.increment, version: 1}
    input: {from: workflow.input}
    next: done
  done:
    kind: complete
    output: {from: nodes.inc.output}
"#;
        let input = Value::Object(BTreeMap::from([("value".to_owned(), Value::Int(0))]));
        for i in 0..80 {
            if i == 10 {
                tokio::time::sleep(crate::policy::SESSION_LEASE + Duration::from_millis(200)).await;
            }
            let run = engine
                .start_yaml(yaml, &catalog(), input.clone())
                .await
                .unwrap();
            let output = engine
                .wait_terminal(run, Duration::from_secs(5))
                .await
                .unwrap_or_else(|err| panic!("run {i}: {err}"));
            assert_eq!(
                output,
                Value::Object(BTreeMap::from([("value".to_owned(), Value::Int(1))]))
            );
        }
        engine.shutdown().await.unwrap();
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn second_local_owner_is_rejected() {
        let dir = tempfile::tempdir().unwrap();
        let engine = Engine::local(dir.path()).await.unwrap();
        let err = match Engine::local(dir.path()).await {
            Ok(_) => panic!("second owner must be rejected"),
            Err(err) => err,
        };
        assert_eq!(err.kind, ErrorKind::FailedPrecondition);
        engine.shutdown().await.unwrap();
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn event_wait_survives_restart() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = include_str!("../../docs/specs/v1/examples/events.yaml");
        let input = Value::Object(BTreeMap::from([(
            "key".to_owned(),
            Value::String("k1".to_owned()),
        )]));
        let run = {
            let engine = Engine::local(dir.path()).await.unwrap();
            let run = engine.start_yaml(yaml, &catalog(), input).await.unwrap();
            for _ in 0..50 {
                engine.progress(run).await.unwrap();
                let state = engine.inspect(run).await.unwrap();
                if state.waits.values().any(|wait| wait.pending) {
                    break;
                }
                tokio::time::sleep(Duration::from_millis(20)).await;
            }
            engine.shutdown().await.unwrap();
            run
        };
        let engine = Engine::local(dir.path()).await.unwrap();
        engine
            .signal(
                run,
                EventId::generate(),
                "approval",
                "k1",
                Value::Object(BTreeMap::from([("approved".to_owned(), Value::Bool(true))])),
            )
            .await
            .unwrap();
        let output = engine
            .wait_terminal(run, Duration::from_secs(10))
            .await
            .unwrap();
        assert_eq!(output.pointer("/approved").unwrap(), &Value::Bool(true));
        engine.shutdown().await.unwrap();
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn snapshot_writes_file() {
        let dir = tempfile::tempdir().unwrap();
        let engine = Engine::local(dir.path()).await.unwrap();
        let run = engine
            .start_yaml(
                include_str!("../../docs/specs/v1/examples/sequence.yaml"),
                &catalog(),
                order(),
            )
            .await
            .unwrap();
        let _ = engine.wait_terminal(run, Duration::from_secs(10)).await;
        engine.snapshot().await.unwrap();
        tokio::time::sleep(Duration::from_millis(200)).await;
        engine.shutdown().await.unwrap();
        let snaps = dir.path().join("snapshots");
        let found = snaps
            .read_dir()
            .map(|entries| entries.filter_map(|e| e.ok()).count())
            .unwrap_or(0);
        assert!(found > 0, "expected snapshot file in {}", snaps.display());
        let replayed = replay(dir.path()).unwrap();
        assert!(run_output(&replayed, run).is_some());
    }

    #[test]
    fn snapshot_controller_threshold() {
        assert!(!should_snapshot(10, 0, Duration::from_secs(1)));
        assert!(should_snapshot(
            SNAPSHOT_ENTRY_THRESHOLD,
            0,
            Duration::from_secs(1)
        ));
        assert!(should_snapshot(5, 0, SNAPSHOT_INTERVAL));
        assert!(!should_snapshot(0, 0, SNAPSHOT_INTERVAL));
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    #[ignore = "applies 20_000 raft entries (~4 min)"]
    async fn snapshot_controller_fires_at_20000_entries() {
        let dir = tempfile::tempdir().unwrap();
        let engine = Engine::local(dir.path()).await.unwrap();
        let run = engine
            .start_yaml(
                include_str!("../../docs/specs/v1/examples/events.yaml"),
                &catalog(),
                Value::Object(BTreeMap::from([(
                    "key".to_owned(),
                    Value::String("k1".to_owned()),
                )])),
            )
            .await
            .unwrap();
        let mut writes = 0u64;
        while engine.raft_applied_index() < SNAPSHOT_ENTRY_THRESHOLD {
            engine.commit_progress(run).await.unwrap();
            writes += 1;
            if writes == 32 {
                assert!(
                    engine.raft_applied_index() > 0,
                    "raft applied index stayed 0 after {writes} writes"
                );
            }
            if writes.is_multiple_of(2_000) {
                eprintln!(
                    "snapshot fill writes={writes} applied={}",
                    engine.raft_applied_index()
                );
            }
            if writes > SNAPSHOT_ENTRY_THRESHOLD + 256 {
                break;
            }
        }
        let applied = engine.raft_applied_index();
        assert!(
            applied >= SNAPSHOT_ENTRY_THRESHOLD,
            "applied {applied} below threshold"
        );
        let snaps = dir.path().join("snapshots");
        let mut found = 0;
        for _ in 0..8 {
            tokio::time::sleep(Duration::from_millis(500)).await;
            found = snaps
                .read_dir()
                .map(|entries| {
                    entries
                        .flatten()
                        .filter(|entry| {
                            entry.path().extension().and_then(|ext| ext.to_str()) == Some("snap")
                        })
                        .count()
                })
                .unwrap_or(0);
            if found > 0 {
                break;
            }
        }
        engine.shutdown().await.unwrap();
        assert!(
            found > 0,
            "controller did not write a snapshot after {applied} applied entries in {}",
            snaps.display()
        );
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn local_saga_failure_compensates() {
        let _guard = COMPENSATION_TEST.lock().await;
        let dir = tempfile::tempdir().unwrap();
        let engine = Engine::local(dir.path()).await.unwrap();
        let run = engine
            .start_yaml(
                include_str!("../../docs/specs/v1/examples/saga.yaml"),
                &catalog(),
                Value::Object(BTreeMap::from([
                    ("order_id".to_owned(), Value::String("o1".to_owned())),
                    ("amount".to_owned(), Value::Int(1000)),
                    ("fail_after_payment".to_owned(), Value::Bool(true)),
                ])),
            )
            .await
            .unwrap();
        let deadline = tokio::time::Instant::now() + Duration::from_secs(20);
        loop {
            let state = engine.inspect(run).await.unwrap();
            let failed = matches!(
                state.runs.get(&run).unwrap().status,
                RunStatus::Failed { .. }
            );
            let compensated =
                state.obligations.iter().all(|item| {
                    matches!(item.status, crate::domain::ObligationStatus::Compensated)
                }) && !state.obligations.is_empty();
            if failed && compensated {
                break;
            }
            if tokio::time::Instant::now() >= deadline {
                panic!(
                    "saga did not compensate; failed={failed} obligations={}",
                    state.obligations.len()
                );
            }
            let _ = engine.progress(run).await;
            tokio::time::sleep(Duration::from_millis(20)).await;
        }
        let events = engine.history(run).await.unwrap();
        let handlers: Vec<_> = events
            .iter()
            .filter_map(|event| match event {
                crate::domain::DomainEvent::CompensationStarted { handler, .. } => {
                    Some(handler.as_str())
                }
                _ => None,
            })
            .collect();
        assert_eq!(handlers, ["payment.refund", "inventory.release"]);
        engine.shutdown().await.unwrap();
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn local_nested_saga_transfers() {
        let _guard = COMPENSATION_TEST.lock().await;
        let dir = tempfile::tempdir().unwrap();
        let engine = Engine::local(dir.path()).await.unwrap();
        let run = engine
            .start_yaml(
                include_str!("../../docs/specs/v1/examples/nested-saga.yaml"),
                &catalog(),
                order(),
            )
            .await
            .unwrap();
        let deadline = tokio::time::Instant::now() + Duration::from_secs(20);
        loop {
            let state = engine.inspect(run).await.unwrap();
            let failed = matches!(
                state.runs.get(&run).unwrap().status,
                RunStatus::Failed { .. }
            );
            let compensated = state.obligations.iter().all(|item| {
                matches!(
                    item.status,
                    crate::domain::ObligationStatus::Compensated
                        | crate::domain::ObligationStatus::Released
                )
            }) && !state.obligations.is_empty();
            if failed && compensated {
                break;
            }
            if tokio::time::Instant::now() >= deadline {
                panic!("nested saga did not settle");
            }
            let _ = engine.progress(run).await;
            tokio::time::sleep(Duration::from_millis(20)).await;
        }
        let events = engine.history(run).await.unwrap();
        let handlers: Vec<_> = events
            .iter()
            .filter_map(|event| match event {
                crate::domain::DomainEvent::CompensationStarted { handler, .. } => {
                    Some(handler.as_str())
                }
                _ => None,
            })
            .collect();
        assert_eq!(handlers, ["payment.refund", "inventory.release"]);
        engine.shutdown().await.unwrap();
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn blocking_cancel_does_not_prove_termination() {
        let dir = tempfile::tempdir().unwrap();
        let engine = Engine::local(dir.path()).await.unwrap();
        let yaml = r#"
dsl: graphrun/v1
id: block_once
version: 1
input_schema: counter/v1
output_schema: counter/v1
start: block
nodes:
  block:
    kind: activity
    activity: {name: test.block, version: 1}
    input: {from: workflow.input}
    next: done
  done:
    kind: complete
    output: {from: nodes.block.output}
"#;
        let run = engine
            .start_yaml(
                yaml,
                &catalog(),
                Value::Object(BTreeMap::from([("value".to_owned(), Value::Int(1))])),
            )
            .await
            .unwrap();
        let started = tokio::time::Instant::now();
        loop {
            if BLOCKING_IN_FLIGHT.load(Ordering::SeqCst) == 1 {
                break;
            }
            if started.elapsed() > Duration::from_secs(5) {
                panic!("blocking handler never entered");
            }
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
        engine.cancel(run, "stop").await.unwrap();
        assert_eq!(
            BLOCKING_IN_FLIGHT.load(Ordering::SeqCst),
            1,
            "cancel must not terminate the blocking thread"
        );
        let _ = engine.inspect(run).await.unwrap();
        let _ = engine.progress(run).await;
        let inflight_deadline = tokio::time::Instant::now() + Duration::from_secs(2);
        while BLOCKING_IN_FLIGHT.load(Ordering::SeqCst) != 0 {
            if tokio::time::Instant::now() >= inflight_deadline {
                panic!("blocking handler never returned");
            }
            tokio::time::sleep(Duration::from_millis(20)).await;
        }
        let state = engine.inspect(run).await.unwrap();
        assert!(matches!(
            state.runs.get(&run).unwrap().status,
            RunStatus::Failed { .. }
        ));
        engine.shutdown().await.unwrap();
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn snapshot_unfinished_parallel_branches() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = r#"
dsl: graphrun/v1
id: unfinished_parallel
version: 1
input_schema: unit/v1
output_schema: {tuple: [approval/v1, approval/v1]}
signals:
  approval: {schema: approval/v1}
start: both
nodes:
  both:
    kind: parallel
    branches:
      - name: left
        input: {literal: null}
        body:
          input_schema: unit/v1
          output_schema: approval/v1
          start: wait
          nodes:
            wait:
              kind: wait_signal
              signal: approval
              key: {literal: "k1"}
              consume_from: buffered
              timeout: null
              next: done
            done:
              kind: complete
              output: {from: nodes.wait.output}
      - name: right
        input: {literal: null}
        body:
          input_schema: unit/v1
          output_schema: approval/v1
          start: wait
          nodes:
            wait:
              kind: wait_signal
              signal: approval
              key: {literal: "k2"}
              consume_from: buffered
              timeout: null
              next: done
            done:
              kind: complete
              output: {from: nodes.wait.output}
    next: finish
  finish:
    kind: complete
    output: {from: nodes.both.output}
"#;
        let run = {
            let engine = Engine::local(dir.path()).await.unwrap();
            let run = engine
                .start_yaml(yaml, &catalog(), Value::Null)
                .await
                .unwrap();
            let mut opened = false;
            for _ in 0..50 {
                engine.progress(run).await.unwrap();
                let state = engine.inspect(run).await.unwrap();
                let pending = state.waits.values().filter(|wait| wait.pending).count();
                let branches: Vec<_> = state
                    .scopes
                    .values()
                    .filter_map(|scope| match &scope.role {
                        crate::domain::ScopeRole::ParallelBranch { name, .. } => Some(name.clone()),
                        _ => None,
                    })
                    .collect();
                if pending == 2 && branches.len() == 2 {
                    assert!(branches.contains(&"left".to_owned()));
                    assert!(branches.contains(&"right".to_owned()));
                    engine.snapshot().await.unwrap();
                    tokio::time::sleep(Duration::from_millis(200)).await;
                    opened = true;
                    break;
                }
                tokio::time::sleep(Duration::from_millis(20)).await;
            }
            assert!(opened, "parallel waits did not open");
            engine.shutdown().await.unwrap();
            run
        };
        let engine = Engine::local(dir.path()).await.unwrap();
        let state = engine.inspect(run).await.unwrap();
        let branches: Vec<_> = state
            .scopes
            .values()
            .filter_map(|scope| match &scope.role {
                crate::domain::ScopeRole::ParallelBranch { name, .. } => Some(name.clone()),
                _ => None,
            })
            .collect();
        assert_eq!(branches.len(), 2);
        assert!(state.waits.values().filter(|wait| wait.pending).count() >= 1);
        engine
            .signal(
                run,
                EventId::generate(),
                "approval",
                "k1",
                Value::Object(BTreeMap::from([("approved".to_owned(), Value::Bool(true))])),
            )
            .await
            .unwrap();
        engine
            .signal(
                run,
                EventId::generate(),
                "approval",
                "k2",
                Value::Object(BTreeMap::from([("approved".to_owned(), Value::Bool(true))])),
            )
            .await
            .unwrap();
        let output = engine
            .wait_terminal(run, Duration::from_secs(10))
            .await
            .unwrap();
        let Value::Array(items) = output else {
            panic!("tuple");
        };
        assert_eq!(items.len(), 2);
        let state = engine.inspect(run).await.unwrap();
        let branch_count = state
            .scopes
            .values()
            .filter(|scope| matches!(scope.role, crate::domain::ScopeRole::ParallelBranch { .. }))
            .count();
        assert_eq!(
            branch_count, 2,
            "snapshot restore must not duplicate branches"
        );
        engine.shutdown().await.unwrap();
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn cancel_during_saga_compensates() {
        let _guard = COMPENSATION_TEST.lock().await;
        let dir = tempfile::tempdir().unwrap();
        let engine = Engine::local(dir.path()).await.unwrap();
        let mut fail_order = order();
        if let Value::Object(fields) = &mut fail_order {
            fields.insert("fail_after_payment".to_owned(), Value::Bool(true));
        }
        let run = engine
            .start_yaml(
                include_str!("../../docs/specs/v1/examples/saga.yaml"),
                &catalog(),
                fail_order,
            )
            .await
            .unwrap();
        let deadline = tokio::time::Instant::now() + Duration::from_secs(10);
        loop {
            let state = engine.inspect(run).await.unwrap();
            if state
                .obligations
                .iter()
                .any(|item| item.handler == "inventory.release")
            {
                break;
            }
            if tokio::time::Instant::now() >= deadline {
                panic!("reserve obligation never registered");
            }
            tokio::time::sleep(Duration::from_millis(20)).await;
        }
        if let Err(err) = engine.cancel(run, "stop").await {
            let state = engine.inspect(run).await.unwrap();
            let inactive = !matches!(state.runs.get(&run).unwrap().status, RunStatus::Active);
            assert!(
                inactive || err.to_string().contains("not active"),
                "cancel failed: {err}"
            );
        }
        let deadline = tokio::time::Instant::now() + Duration::from_secs(15);
        loop {
            let state = engine.inspect(run).await.unwrap();
            let failed = matches!(
                &state.runs.get(&run).unwrap().status,
                RunStatus::Failed { .. }
            );
            let compensated = !state.obligations.is_empty()
                && state.obligations.iter().all(|item| {
                    matches!(item.status, crate::domain::ObligationStatus::Compensated)
                });
            if failed && compensated {
                break;
            }
            if tokio::time::Instant::now() >= deadline {
                panic!("cancel did not compensate");
            }
            tokio::time::sleep(Duration::from_millis(20)).await;
        }
        engine.shutdown().await.unwrap();
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn compensation_resumes_after_restart() {
        let _guard = COMPENSATION_TEST.lock().await;
        SLOW_RELEASE.store(true, Ordering::SeqCst);
        let dir = tempfile::tempdir().unwrap();
        let run = {
            let engine = Engine::local(dir.path()).await.unwrap();
            let run = engine
                .start_yaml(
                    include_str!("../../docs/specs/v1/examples/saga.yaml"),
                    &catalog(),
                    Value::Object(BTreeMap::from([
                        ("order_id".to_owned(), Value::String("o1".to_owned())),
                        ("amount".to_owned(), Value::Int(1000)),
                        ("fail_after_payment".to_owned(), Value::Bool(true)),
                    ])),
                )
                .await
                .unwrap();
            let deadline = tokio::time::Instant::now() + Duration::from_secs(15);
            loop {
                let state = engine.inspect(run).await.unwrap();
                let refunded = state.obligations.iter().any(|item| {
                    item.handler == "payment.refund"
                        && matches!(item.status, crate::domain::ObligationStatus::Compensated)
                });
                let release_open = state.obligations.iter().any(|item| {
                    item.handler == "inventory.release"
                        && !matches!(item.status, crate::domain::ObligationStatus::Compensated)
                });
                if refunded && release_open {
                    break;
                }
                if tokio::time::Instant::now() >= deadline {
                    SLOW_RELEASE.store(false, Ordering::SeqCst);
                    panic!("did not catch mid-compensation");
                }
                tokio::time::sleep(Duration::from_millis(20)).await;
            }
            engine.shutdown().await.unwrap();
            run
        };
        SLOW_RELEASE.store(false, Ordering::SeqCst);
        let engine = Engine::local(dir.path()).await.unwrap();
        let deadline = tokio::time::Instant::now() + Duration::from_secs(15);
        loop {
            let state = engine.inspect(run).await.unwrap();
            let done = !state.obligations.is_empty()
                && state.obligations.iter().all(|item| {
                    matches!(item.status, crate::domain::ObligationStatus::Compensated)
                });
            if done {
                break;
            }
            if tokio::time::Instant::now() >= deadline {
                panic!("compensation did not resume");
            }
            tokio::time::sleep(Duration::from_millis(20)).await;
        }
        let events = engine.history(run).await.unwrap();
        let releases = events
            .iter()
            .filter(|event| {
                matches!(
                    event,
                    crate::domain::DomainEvent::CompensationStarted { handler, .. } if handler == "inventory.release"
                )
            })
            .count();
        assert_eq!(releases, 1);
        engine.shutdown().await.unwrap();
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn logical_restore_suspends_until_authorized() {
        let src = tempfile::tempdir().unwrap();
        let backup = tempfile::tempdir().unwrap();
        let dest = tempfile::tempdir().unwrap();
        let run = {
            let engine = Engine::local(src.path()).await.unwrap();
            let run = engine
                .start_yaml(
                    include_str!("../../docs/specs/v1/examples/events.yaml"),
                    &catalog(),
                    Value::Object(BTreeMap::from([(
                        "key".to_owned(),
                        Value::String("k1".to_owned()),
                    )])),
                )
                .await
                .unwrap();
            let deadline = tokio::time::Instant::now() + Duration::from_secs(10);
            loop {
                let state = engine.inspect(run).await.unwrap();
                if state.waits.values().any(|wait| wait.pending) {
                    break;
                }
                if tokio::time::Instant::now() >= deadline {
                    panic!("wait never opened");
                }
                tokio::time::sleep(Duration::from_millis(20)).await;
            }
            engine.shutdown().await.unwrap();
            Engine::backup(src.path(), backup.path()).unwrap();
            run
        };
        Engine::restore(backup.path(), dest.path(), "disaster").unwrap();
        let identity = std::fs::read_to_string(dest.path().join("identity.json")).unwrap();
        assert!(identity.contains("graphrun-restored-"));
        assert!(!dest.path().join("member.redb").exists());
        let engine = Engine::local(dest.path()).await.unwrap();
        let state = engine.inspect(run).await.unwrap();
        assert!(matches!(
            state.runs.get(&run).unwrap().status,
            RunStatus::Active
        ));
        assert!(state.recovery.as_ref().is_some_and(|hold| !hold.authorized));
        let err = engine
            .start_yaml(
                include_str!("../../docs/specs/v1/examples/sequence.yaml"),
                &catalog(),
                order(),
            )
            .await
            .expect_err("start must stay suspended");
        assert!(err.to_string().contains("suspended"));
        engine
            .acknowledge_recovery("operator authorized restore")
            .await
            .unwrap();
        engine
            .signal(
                run,
                EventId::generate(),
                "approval",
                "k1",
                Value::Object(BTreeMap::from([("approved".to_owned(), Value::Bool(true))])),
            )
            .await
            .unwrap();
        let output = engine
            .wait_terminal(run, Duration::from_secs(10))
            .await
            .unwrap();
        assert_eq!(output.pointer("/approved").unwrap(), &Value::Bool(true));
        engine.shutdown().await.unwrap();
    }
}