car_server_core/

sync.rs

//! Daemon-held multi-device sync + execution-lease subsystem — the `sync.*` /
//! `lease.*` WS surface's engine (slice B6 of
//! `docs/proposals/multi-device-sync.md`).
//!
//! One [`SyncSubsystem`] per daemon **is one device** in the user's sync fleet:
//! it owns a [`car_sync::SyncSession`] (the pump over an append-only oplog +
//! deterministic fold) against a [`car_sync::FsRelay`] rooted at
//! `<journal_dir>/sync/relay/` — so the single-user two-device case (two Macs
//! sharing that directory, e.g. via a synced folder, or two daemons on one host
//! in tests) converges out of the box — plus an in-process linearizable
//! [`car_sync::InMemoryLeaseCoordinator`] for the execution lease.
//!
//! The handler layer (`handler.rs`) is thin: it parses params and calls the
//! `&mut self` methods here under the subsystem's `tokio::sync::Mutex`, which is
//! why the convergence + fence + lease behaviour is unit-tested directly on
//! [`SyncSubsystem`] (two instances sharing an `FsRelay` dir + a cloned
//! coordinator) rather than only through the WS round-trip.
//!
//! ## State domains: wired vs. pending (honest boundary)
//!
//! - **Conversation — wired end-to-end.** [`SyncSubsystem::record_turn`] routes
//!   a conversation write **through the oplog** (a `Surface::Conversation` op),
//!   and [`SyncSubsystem::resume`] returns the repaired, provider-valid
//!   `Vec<Message>` from [`car_sync::SyncState::resume_messages`] — so
//!   transcript resume across devices is real, not a stub. What is **not** done:
//!   auto-teeing the daemon's *existing internal* conversation persistence
//!   (car-inference/memgine) into the oplog — a host uses `sync.record_turn`
//!   explicitly. That internal reroute is the pending B2 adoption step.
//! - **Intent ledger — wired.** [`SyncSubsystem::record_intent`] writes the
//!   leased-execution `Surface::Intent` ledger (terminal-guarded), and
//!   [`SyncSubsystem::fence_check`] runs the B6 dispatch fence over it.
//! - **Any other surface — a generic tee.** [`SyncSubsystem::append`] records
//!   an op on any [`car_sync::Surface`] (knowledge/skill/declagent/routing/…),
//!   so a host can tee those domains into the oplog today. Rerouting the
//!   daemon's own knowledge/registry write paths through it is the pending step.
//!
//! ## Encryption & distributed coordination
//!
//! Two subsystem flavors:
//!
//! - [`SyncSubsystem::open`] — the **local** default: a shared-directory
//!   `FsRelay` + an in-process `InMemoryLeaseCoordinator`, cleartext payloads.
//!   The single-host / synced-folder case.
//! - [`SyncSubsystem::open_remote`] — the **Parslee-backed** path (selected when
//!   `.car/config.toml` `[sync] backend = "parslee"`): a [`NetworkRelay`] +
//!   distributed [`NetworkLeaseCoordinator`] over a [`car_sync::SyncTransport`]
//!   (the real `car_parslee::ParsleeSyncTransport`, or a `LoopbackTransport`
//!   reference server in tests), scoped to the user's Parslee identity, with op
//!   payloads **E2E-encrypted** under a login-derived [`SyncKeyProvider`]. The
//!   session encrypts-on-append / decrypts-before-fold, so the relay holds only
//!   ciphertext, and the lease is a genuinely cross-device fencing register.
//!   This is "phone + Mac after one login" and it is exercised end-to-end
//!   (`two_remote_devices_converge_e2e_through_the_network_relay`).
//!
//! The cross-device **key** is login-derived: [`car_sync::DerivedKeyProvider`]
//! `from_passphrase` is the zero-knowledge source that works today (same
//! passphrase → same keys on every device, the server never sees it); a
//! Parslee-issued per-user master is the alternative. Config **propagation** has
//! a tested tee primitive ([`SyncSubsystem::tee_config`]/[`SyncSubsystem::config_get`],
//! partition-guarded). Under E2E, checkpoint publishing is **guarded off**
//! ([`SyncSession::publish_checkpoint`]) so no cleartext/inconsistent snapshot
//! reaches the relay.
//!
//! Remaining follow-ups: the live Parslee **server** implementing the contract
//! (cross-repo — `docs/proposals/parslee-sync-backend.md`); per-subsystem
//! **adoption** of the config tee (each config write-path calling `tee_config`
//! + applying `config_get` on pull); and per-scope **encrypted** checkpoint push
//! (restores relay-side GC — until then the server owns retention).

use std::path::Path;

use car_sync::{
    check_dispatch, frontier_of, system_clock, FsRelay, InMemoryLeaseCoordinator, Intent,
    IntentStatus, LeaseCoordinator, NetworkLeaseCoordinator, NetworkRelay, Relay, RelayConfig,
    Scope, Surface, SyncKeyProvider, SyncSession, SyncTransport, Turn, WallClock,
};
use serde_json::{json, Value};
use std::sync::Arc;

use crate::assistant::governance::{
    ActionState, AssistantCheckpoint, SupervisedActionRecord, ACTION_REGISTRY_KIND,
    CHECKPOINT_REGISTRY_KIND,
};

/// The reserved `Surface::Registry` sub-kind portable config domains tee onto
/// (LWW per domain). One sub-surface keeps all synced config in a single folded
/// registry the daemon reads back by domain.
const CONFIG_KIND: &str = "config";

/// Registry kind carrying each device's peer-reachable A2A endpoint.
///
/// This is how "find CAR on my other machine through Parslee" works without a
/// Parslee-side change: the sync roster knows which devices exist and when they
/// were last seen, but carries nothing dialable. Rather than add an address to
/// the roster — which would publish a per-device endpoint to the service — each
/// device announces its own endpoint on the **oplog**, which is end-to-end
/// encrypted and already the mechanism by which this user's devices converge.
/// Parslee relays the bytes and cannot read them.
///
/// Folds LWW per `id` (the device id), so a machine that changes address simply
/// overwrites its own entry and every other device converges on the new one.
const HOST_ENDPOINT_KIND: &str = "host_endpoint";

/// One device's announced A2A endpoint and peer identity.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct HostEndpoint {
    pub device_id: String,
    pub name: String,
    pub url: String,
    /// Base64 ed25519 public key. Empty from a device running a build that
    /// predates peer auth — such a peer is listed but cannot be authenticated,
    /// which is the honest state rather than a silent downgrade to trusting it.
    pub pubkey: String,
}
/// Surface tag for the assistant's synced knowledge facts (matches
/// `Surface::Knowledge.tag()` and the write-half's `sync.append` surface).
const KNOWLEDGE_SURFACE: &str = "knowledge";

/// A daemon device's sync endpoint + lease coordinator. Held behind a
/// `tokio::sync::Mutex` on `ServerState`; every method is `&mut self`.
pub struct SyncSubsystem {
    device_id: String,
    session: SyncSession,
    relay: Box<dyn Relay + Send>,
    coordinator: Box<dyn LeaseCoordinator + Send>,
}

impl std::fmt::Debug for SyncSubsystem {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("SyncSubsystem")
            .field("device_id", &self.device_id)
            .finish_non_exhaustive()
    }
}

impl SyncSubsystem {
    /// Open the daemon-default subsystem rooted at `root` (`<journal_dir>/sync/`):
    /// a persistent per-device id at `root/device-id`, this device's oplog +
    /// checkpoints under `root/<device_id>/`, and a shared `FsRelay` at
    /// `root/relay/`. Uses the real system wall clock and a fresh in-process
    /// lease coordinator.
    pub fn open(root: &Path) -> Result<Self, String> {
        std::fs::create_dir_all(root)
            .map_err(|e| format!("sync: create {}: {e}", root.display()))?;
        let device_id = load_or_mint_device_id(root)?;
        let device_dir = root.join(&device_id);
        let relay_dir = root.join("relay");
        let coordinator = Box::new(InMemoryLeaseCoordinator::new(system_clock()));
        Self::open_with(
            device_id,
            &device_dir,
            &relay_dir,
            coordinator,
            system_clock(),
        )
    }

    /// Open with explicit paths, coordinator, and clock — the injection point
    /// the tests use to run two devices (distinct `device_dir`s, one shared
    /// `relay_dir`, a cloned coordinator register).
    pub fn open_with(
        device_id: String,
        device_dir: &Path,
        relay_dir: &Path,
        coordinator: Box<dyn LeaseCoordinator + Send>,
        wall: WallClock,
    ) -> Result<Self, String> {
        std::fs::create_dir_all(device_dir)
            .map_err(|e| format!("sync: create {}: {e}", device_dir.display()))?;
        let journal_path = device_dir.join("oplog.jsonl");
        let checkpoint_dir = device_dir.join("checkpoints");
        let session = SyncSession::open(
            device_id.clone(),
            &journal_path,
            &checkpoint_dir,
            wall.clone(),
        )
        .map_err(|e| format!("sync: open session: {e}"))?;
        let relay = FsRelay::open(relay_dir, RelayConfig::default(), wall)
            .map_err(|e| format!("sync: open relay {}: {e}", relay_dir.display()))?;
        Ok(Self {
            device_id,
            session,
            relay: Box::new(relay),
            coordinator,
        })
    }

    /// Open a **remote** (Parslee-backed) subsystem. This device's oplog +
    /// checkpoints stay local under `root/<device_id>/`, but the relay and lease
    /// register are the network `transport` (e.g. `ParsleeSyncTransport`, or a
    /// `LoopbackTransport` in tests), scoped to `scope` — the user's Parslee
    /// identity (`user:<id>` / `org:<id>`) — so every device on that login
    /// converges through one service. Op payloads are E2E-encrypted under
    /// `key_provider` (login-derived), so the relay only ever holds ciphertext.
    /// The daemon selects this over [`SyncSubsystem::open`] when
    /// `.car/config.toml` `[sync] backend = "parslee"`.
    pub fn open_remote(
        root: &Path,
        transport: Arc<dyn SyncTransport>,
        scope: impl Into<String>,
        key_provider: Arc<dyn SyncKeyProvider>,
    ) -> Result<Self, String> {
        std::fs::create_dir_all(root)
            .map_err(|e| format!("sync: create {}: {e}", root.display()))?;
        let device_id = load_or_mint_device_id(root)?;
        let device_dir = root.join(&device_id);
        std::fs::create_dir_all(&device_dir)
            .map_err(|e| format!("sync: create {}: {e}", device_dir.display()))?;
        let scope = scope.into();
        let journal_path = device_dir.join("oplog.jsonl");
        let checkpoint_dir = device_dir.join("checkpoints");
        let session = SyncSession::open(
            device_id.clone(),
            &journal_path,
            &checkpoint_dir,
            system_clock(),
        )
        .map_err(|e| format!("sync: open session: {e}"))?
        .with_key_provider(key_provider);
        let relay: Box<dyn Relay + Send> =
            Box::new(NetworkRelay::new(transport.clone(), scope.clone()));
        let coordinator: Box<dyn LeaseCoordinator + Send> =
            Box::new(NetworkLeaseCoordinator::new(transport, scope));
        Ok(Self {
            device_id,
            session,
            relay,
            coordinator,
        })
    }

    pub fn device_id(&self) -> &str {
        &self.device_id
    }

    // ----- sync.* -----------------------------------------------------------

    /// `sync.status` — the roster, this device's journal frontier (per-device
    /// max seq), the relay's stable frontier, and the divergence-invariant
    /// state hash.
    pub fn status(&mut self) -> Result<Value, String> {
        let stable = self
            .relay
            .stable_frontier()
            .map_err(|e| format!("sync: stable_frontier: {e}"))?;
        let roster = self
            .relay
            .roster()
            .map_err(|e| format!("sync: roster: {e}"))?;
        Ok(json!({
            "device_id": self.device_id,
            "state_hash": self.session.state_hash(),
            "journal_frontier": frontier_of(self.session.ops()),
            "stable_frontier": stable,
            "base_checkpoint": self.session.base().map(|c| c.checkpoint_hash.clone()),
            "roster": roster,
        }))
    }

    /// `sync.append` — record an op on any surface (the generic domain tee).
    pub fn append(
        &mut self,
        scope: Scope,
        surface: Surface,
        payload: Value,
    ) -> Result<Value, String> {
        let op = self
            .session
            .append(scope, surface, payload)
            .map_err(|e| format!("sync: append: {e}"))?;
        Ok(json!({ "op_id": op.op_id, "seq": op.seq, "hlc": op.hlc }))
    }

    /// Replicate a **portable** config `domain` onto the oplog so every device
    /// on this login converges it (LWW per domain) — how a setting changed on
    /// the Mac reaches the phone. The **tee primitive**: the daemon's own config
    /// write-paths (agent-permissions, messaging allowlist, …) call this after a
    /// local write; that per-subsystem adoption is the remaining wiring. It
    /// **refuses** a device-local domain (`car_sync::partition`) — the guard that
    /// keeps a secret or an OS grant off the relay; `value` must already exclude
    /// any device-local sub-field (e.g. a keychain ref stays a per-device
    /// pointer). Safer than raw `sync.append` (which any surface can use).
    pub fn tee_config(&mut self, domain: &str, value: Value) -> Result<Value, String> {
        if !car_sync::is_portable(domain) {
            return Err(format!(
                "sync: refusing to tee device-local domain '{domain}' — only portable \
                 policy syncs; secrets and OS grants stay on the device"
            ));
        }
        self.append(
            Scope::Personal,
            Surface::Registry {
                kind: CONFIG_KIND.to_string(),
            },
            json!({ "id": domain, "value": value }),
        )
    }

    /// The converged (folded LWW) value of a config `domain`, or `null` if no
    /// device has teed it yet — the read side an adopting subsystem applies
    /// after a pull to reconcile a peer's change onto this device.
    pub fn config_get(&self, domain: &str) -> Value {
        self.session
            .state()
            .registries
            .get(&format!("registry:{CONFIG_KIND}"))
            .and_then(|reg| reg.get(&format!("id:{domain}")))
            .and_then(|rec| rec.payload.get("value").cloned())
            .unwrap_or(Value::Null)
    }

    /// Announce this device's peer-reachable A2A endpoint to the user's other
    /// devices.
    ///
    /// Idempotent by construction: the entry is keyed on this device's id and
    /// folds LWW, so re-announcing on every start overwrites rather than
    /// accumulates. Call it whenever the A2A listener comes up — the address can
    /// change between runs (DHCP, a new `--a2a-public-url`), and a stale entry
    /// would send peers at an address nothing answers on.
    pub fn publish_host_endpoint(
        &mut self,
        name: &str,
        url: &str,
        pubkey: &str,
    ) -> Result<Value, String> {
        let id = self.device_id().to_string();
        self.append(
            Scope::Personal,
            Surface::Registry {
                kind: HOST_ENDPOINT_KIND.to_string(),
            },
            json!({ "id": id, "name": name, "url": url, "pubkey": pubkey }),
        )
    }

    /// Every *other* device's announced endpoint.
    ///
    /// This device is excluded: it is not its own peer, and listing it would put
    /// a self-addressed row in every peer listing.
    ///
    /// The `pubkey` is what makes these peers *reachable* rather than merely
    /// visible. It arrived over the E2E-encrypted oplog, so only a device
    /// holding this login's key material could have published it — which is
    /// exactly the property that lets this host trust it without an operator
    /// comparing fingerprints by hand.
    pub fn host_endpoints(&self) -> Vec<HostEndpoint> {
        let me = self.device_id().to_string();
        // Bind the state: `session.state()` returns a temporary, and the
        // registry borrow below outlives the expression it was produced in.
        let state = self.session.state();
        let Some(reg) = state
            .registries
            .get(&format!("registry:{HOST_ENDPOINT_KIND}"))
        else {
            return Vec::new();
        };
        let mut out: Vec<HostEndpoint> = reg
            .iter()
            .filter_map(|(key, rec)| {
                let device_id = key.strip_prefix("id:")?.to_string();
                if device_id == me {
                    return None;
                }
                let url = rec.payload.get("url")?.as_str()?.trim().to_string();
                // Refuse anything that is not an absolute http(s) URL. A peer
                // entry is dialed, so a malformed or scheme-less value is a
                // request to connect somewhere unintended.
                if !(url.starts_with("http://") || url.starts_with("https://")) {
                    return None;
                }
                let name = rec
                    .payload
                    .get("name")
                    .and_then(|v| v.as_str())
                    .filter(|n| !n.trim().is_empty())
                    .unwrap_or(&device_id)
                    .to_string();
                let pubkey = rec
                    .payload
                    .get("pubkey")
                    .and_then(|v| v.as_str())
                    .unwrap_or_default()
                    .to_string();
                Some(HostEndpoint {
                    device_id,
                    name,
                    url,
                    pubkey,
                })
            })
            .collect();
        out.sort_by(|a, b| a.name.cmp(&b.name));
        out
    }

    /// The converged (folded) knowledge facts — the read side of the assistant's
    /// synced memory. Returns each fact's `{subject, body}` payload in ASCENDING
    /// `(hlc, op_id)` order (as `log_entries` yields them), so a reducer that
    /// keeps the LAST occurrence per subject gets the newest value. The assistant
    /// re-ingests these after a pull so a fact learned on one device surfaces in
    /// recall on another. Raw (unreduced) on purpose — the caller reduces
    /// newest-per-subject so the ordering contract stays in one place.
    pub fn knowledge(&self) -> Vec<Value> {
        self.session
            .state()
            .log_entries(KNOWLEDGE_SURFACE)
            .into_iter()
            .map(|rec| rec.payload.clone())
            .collect()
    }

    /// Persist the exact model-facing supervised-assistant checkpoint in the
    /// existing append-only oplog. LWW is per session id, while `revision`
    /// prevents a delayed writer from replacing a newer local checkpoint.
    pub fn assistant_checkpoint_put(
        &mut self,
        checkpoint: AssistantCheckpoint,
    ) -> Result<Value, String> {
        if checkpoint.session_id.trim().is_empty() || checkpoint.id != checkpoint.session_id {
            return Err("assistant checkpoint id must equal its non-empty session_id".into());
        }
        if let Some(current) = self.assistant_checkpoint_get(&checkpoint.session_id)? {
            if checkpoint.revision <= current.revision {
                return Err(format!(
                    "assistant checkpoint revision {} is not newer than {}",
                    checkpoint.revision, current.revision
                ));
            }
        }
        let session_id = checkpoint.session_id.clone();
        let value = serde_json::to_value(checkpoint)
            .map_err(|e| format!("sync: serialize assistant checkpoint: {e}"))?;
        self.append(
            Scope::Personal,
            Surface::Registry {
                kind: CHECKPOINT_REGISTRY_KIND.to_string(),
            },
            json!({ "id": session_id, "checkpoint": value }),
        )
    }

    /// Load the latest exact checkpoint for one supervised session.
    pub fn assistant_checkpoint_get(
        &self,
        session_id: &str,
    ) -> Result<Option<AssistantCheckpoint>, String> {
        let state = self.session.state();
        let value = state
            .registries
            .get(&format!("registry:{CHECKPOINT_REGISTRY_KIND}"))
            .and_then(|reg| reg.get(&format!("id:{session_id}")))
            .and_then(|rec| rec.payload.get("checkpoint"));
        value
            .map(|v| {
                serde_json::from_value(v.clone())
                    .map_err(|e| format!("sync: decode assistant checkpoint: {e}"))
            })
            .transpose()
    }

    /// Append one monotone supervised-action lifecycle record. This ledger is
    /// separate from scheduled-run Intent because approval and indeterminate
    /// dispatch are distinct safety states, not aliases for pending/failed.
    pub fn assistant_action_put(
        &mut self,
        record: SupervisedActionRecord,
    ) -> Result<Value, String> {
        if record.id.trim().is_empty()
            || record.id != record.scope.action_id(&record.session_id, &record.call_id)
        {
            return Err("assistant action id does not match its canonical scope digest".into());
        }
        match self.assistant_action_get(&record.id)? {
            None if record.state != ActionState::Proposed => {
                return Err("a supervised action must begin in proposed state".into());
            }
            Some(current) => {
                let mut expected = current.clone();
                expected.transition(record.state, record.receipt.clone())?;
                if expected != record {
                    return Err("supervised action mutation changed immutable scope fields".into());
                }
            }
            None => {}
        }
        let id = record.id.clone();
        let value = serde_json::to_value(record)
            .map_err(|e| format!("sync: serialize supervised action: {e}"))?;
        self.append(
            Scope::Personal,
            Surface::Registry {
                kind: ACTION_REGISTRY_KIND.to_string(),
            },
            json!({ "id": id, "record": value }),
        )
    }

    pub fn assistant_action_get(
        &self,
        action_id: &str,
    ) -> Result<Option<SupervisedActionRecord>, String> {
        let state = self.session.state();
        let value = state
            .registries
            .get(&format!("registry:{ACTION_REGISTRY_KIND}"))
            .and_then(|reg| reg.get(&format!("id:{action_id}")))
            .and_then(|rec| rec.payload.get("record"));
        value
            .map(|v| {
                serde_json::from_value(v.clone())
                    .map_err(|e| format!("sync: decode supervised action: {e}"))
            })
            .transpose()
    }

    /// `sync.record_turn` — the Conversation domain, routed through the oplog so
    /// `sync.resume` is real. `role` ∈ `user|assistant|tool`.
    pub fn record_turn(
        &mut self,
        scope: Scope,
        conversation_id: &str,
        role: &str,
        content: &str,
        tool_calls: Vec<Value>,
        tool_use_id: Option<&str>,
        timestamp: u64,
    ) -> Result<Value, String> {
        // Fail fast rather than persist garbage into the durable oplog. The WS
        // handler + FFI proxy used to `.unwrap_or("")` every field, so a caller
        // that omitted/typo'd a param silently recorded a malformed turn:
        // - an empty `conversation_id` is the scoping key `resume_messages` folds
        //   on, so blank pools unrelated callers' turns into one anonymous thread;
        // - a turn with no `content` AND no `tool_calls`/`tool_use_id` is an empty
        //   record that only bloats the journal and shows up as a blank message on
        //   resume. A tool turn legitimately has empty content (identified by
        //   tool_use_id), so it's exempt from the content check.
        if conversation_id.trim().is_empty() {
            return Err("sync.record_turn requires a non-empty `conversation_id`".into());
        }
        if content.is_empty() && tool_calls.is_empty() && tool_use_id.is_none() {
            return Err(
                "sync.record_turn requires `content` (or tool_calls / tool_use_id for a tool turn)"
                    .into(),
            );
        }
        let payload = match role {
            "assistant" => Turn::assistant_payload(conversation_id, content, tool_calls, timestamp),
            "tool" | "tool_result" => Turn::tool_payload(
                conversation_id,
                tool_use_id.unwrap_or_default(),
                content,
                timestamp,
            ),
            _ => Turn::user_payload(conversation_id, content, timestamp),
        };
        self.append(scope, Surface::Conversation, payload)
    }

    /// `sync.record_intent` — write a leased-execution intent to the
    /// `Surface::Intent` ledger (terminal-guarded: a `pending`/`failed` write
    /// for an already-committed run is a no-op). This is what populates the
    /// committed-run oracle the dispatch fence reads.
    pub fn record_intent(&mut self, scope: Scope, intent: &Intent) -> Result<Value, String> {
        let recorded = self
            .session
            .record_intent(scope, intent)
            .map_err(|e| format!("sync: record_intent: {e}"))?;
        Ok(match recorded {
            Some(op) => json!({ "recorded": true, "op_id": op.op_id }),
            None => {
                json!({ "recorded": false, "reason": "run already committed (terminal guard)" })
            }
        })
    }

    /// `sync.pump` — one reconciliation round (push journal-durable own ops →
    /// pull → verify → fold → ack). This drives push/pull/ack against the relay.
    pub fn pump(&mut self) -> Result<Value, String> {
        let report = self
            .session
            .pump(self.relay.as_mut())
            .map_err(|e| format!("sync: pump: {e}"))?;
        Ok(json!({
            "pushed": report.pushed,
            "push_deduped": report.push_deduped,
            "folded": report.folded,
            "acked": report.acked,
            "state_hash": self.session.state_hash(),
        }))
    }

    /// `sync.checkpoint` — compute + publish a device-side checkpoint at the
    /// relay's stable frontier (the E2E-ready snapshot; the relay never folds).
    pub fn checkpoint(&mut self) -> Result<Value, String> {
        let published = self
            .session
            .publish_checkpoint(self.relay.as_mut())
            .map_err(|e| format!("sync: publish_checkpoint: {e}"))?;
        Ok(match published {
            Some(c) => json!({ "published": true, "checkpoint_hash": c.checkpoint_hash }),
            None => json!({ "published": false }),
        })
    }

    /// `sync.rebase` — cold bootstrap / straggler re-entry: re-anchor on the
    /// relay's latest checkpoint (carrying uncovered local ops across).
    pub fn rebase(&mut self) -> Result<Value, String> {
        let rebased = self
            .session
            .rebase(self.relay.as_mut())
            .map_err(|e| format!("sync: rebase: {e}"))?;
        Ok(json!({
            "rebased": rebased,
            "base_checkpoint": self.session.base().map(|c| c.checkpoint_hash.clone()),
        }))
    }

    /// `sync.transcript` — the ordered, role-threaded raw transcript projection.
    pub fn transcript(&self, conversation_id: &str) -> Value {
        json!(self.session.state().transcript(conversation_id))
    }

    /// `sync.resume` — the repaired, provider-valid `Vec<Message>` a host
    /// replays to continue the conversation (the verbatim resume path).
    pub fn resume(&self, conversation_id: &str) -> Result<Value, String> {
        serde_json::to_value(self.session.state().resume_messages(conversation_id))
            .map_err(|e| format!("sync: serialize resume messages: {e}"))
    }

    /// `sync.fence_check` — the B6 executor dispatch fence at the point of
    /// effect: the durable committed-run oracle read + the linearizable "am I
    /// still epoch N?" read. Only `may_dispatch == true` authorizes the effect.
    pub fn fence_check(
        &mut self,
        agent_id: &str,
        run_id: &str,
        epoch: u64,
    ) -> Result<Value, String> {
        let state = self.session.state();
        let decision = check_dispatch(
            self.coordinator.as_mut(),
            &state,
            agent_id,
            run_id,
            &self.device_id,
            epoch,
        )
        .map_err(|e| format!("sync: fence_check: {e}"))?;
        let may = decision.may_dispatch();
        Ok(json!({ "decision": decision, "may_dispatch": may }))
    }

    // ----- lease.* ----------------------------------------------------------

    /// `lease.acquire` — CAS-acquire the per-agent execution lease (this device
    /// is the holder); on grant the monotone fencing `epoch` bumps.
    pub fn lease_acquire(&mut self, agent_id: &str, ttl_ms: u64) -> Result<Value, String> {
        let lease = self
            .coordinator
            .acquire(agent_id, &self.device_id, ttl_ms)
            .map_err(|e| format!("lease: acquire: {e}"))?;
        serde_json::to_value(lease).map_err(|e| e.to_string())
    }

    /// `lease.renew` — heartbeat the lease (no epoch bump), iff still the holder.
    pub fn lease_renew(
        &mut self,
        agent_id: &str,
        epoch: u64,
        ttl_ms: u64,
    ) -> Result<Value, String> {
        let lease = self
            .coordinator
            .renew(agent_id, &self.device_id, epoch, ttl_ms)
            .map_err(|e| format!("lease: renew: {e}"))?;
        serde_json::to_value(lease).map_err(|e| e.to_string())
    }

    /// `lease.release` — clean handoff (next acquire skips the TTL wait).
    pub fn lease_release(&mut self, agent_id: &str, epoch: u64) -> Result<Value, String> {
        self.coordinator
            .release(agent_id, &self.device_id, epoch)
            .map_err(|e| format!("lease: release: {e}"))?;
        Ok(json!({ "released": true }))
    }

    /// `lease.status` — the linearizable read of the current lease (or `null`).
    pub fn lease_status(&mut self, agent_id: &str) -> Result<Value, String> {
        let current = self
            .coordinator
            .current(agent_id)
            .map_err(|e| format!("lease: status: {e}"))?;
        Ok(json!({ "lease": current }))
    }
}

/// Load the daemon's persistent device id, or mint + persist a fresh uuid.
fn load_or_mint_device_id(root: &Path) -> Result<String, String> {
    let path = root.join("device-id");
    if path.exists() {
        let id = std::fs::read_to_string(&path)
            .map_err(|e| format!("sync: read device-id: {e}"))?
            .trim()
            .to_string();
        if !id.is_empty() {
            return Ok(id);
        }
    }
    let id = format!("device-{}", uuid::Uuid::new_v4());
    std::fs::write(&path, &id).map_err(|e| format!("sync: write device-id: {e}"))?;
    Ok(id)
}

/// Parse an optional `scope` param: `{scope: "personal"}` (default) or
/// `{scope: {org: "acme"}}` / `{org: "acme"}` → `Shared`.
pub fn parse_scope(params: &Value) -> Scope {
    if let Some(org) = params
        .get("scope")
        .and_then(|s| s.get("org"))
        .or_else(|| params.get("org"))
        .and_then(Value::as_str)
    {
        return Scope::Shared {
            org: org.to_string(),
        };
    }
    Scope::Personal
}

/// Parse a `surface` param string into a [`Surface`]. `registry:<kind>` maps to
/// `Registry { kind }`. Unknown surfaces error (never silently defaulted).
pub fn parse_surface(s: &str) -> Result<Surface, String> {
    Ok(match s {
        "routing" => Surface::Routing,
        "declagent" => Surface::Declagent,
        "conversation" => Surface::Conversation,
        "knowledge" => Surface::Knowledge,
        "skill" => Surface::Skill,
        "trajectory" => Surface::Trajectory,
        "run" => Surface::Run,
        "intent" => Surface::Intent,
        other => {
            if let Some(kind) = other.strip_prefix("registry:") {
                Surface::Registry {
                    kind: kind.to_string(),
                }
            } else {
                return Err(format!("unknown sync surface '{other}'"));
            }
        }
    })
}

/// Parse an [`IntentStatus`] string.
pub fn parse_intent_status(s: &str) -> Result<IntentStatus, String> {
    Ok(match s {
        "pending" => IntentStatus::Pending,
        "committed" => IntentStatus::Committed,
        "failed" => IntentStatus::Failed,
        other => return Err(format!("unknown intent status '{other}'")),
    })
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::atomic::{AtomicU64, Ordering};
    use std::sync::Arc;

    fn manual_clock() -> (Arc<AtomicU64>, WallClock) {
        let t = Arc::new(AtomicU64::new(0));
        let reader = t.clone();
        (t, Arc::new(move || reader.load(Ordering::SeqCst)))
    }

    /// Two devices (distinct oplog dirs) syncing through one shared FsRelay dir
    /// and one shared lease register — the realistic single-user case, driven
    /// entirely through the `SyncSubsystem` methods the WS handlers call.
    fn two_devices(relay_dir: &Path, a_dir: &Path, b_dir: &Path) -> (SyncSubsystem, SyncSubsystem) {
        let coord = InMemoryLeaseCoordinator::new({
            let (_t, w) = manual_clock();
            w
        });
        let (_ta, wa) = manual_clock();
        let (_tb, wb) = manual_clock();
        let a = SyncSubsystem::open_with(
            "mac-a".into(),
            a_dir,
            relay_dir,
            Box::new(coord.clone()),
            wa,
        )
        .unwrap();
        let b = SyncSubsystem::open_with("mac-b".into(), b_dir, relay_dir, Box::new(coord), wb)
            .unwrap();
        (a, b)
    }

    #[test]
    fn record_turn_rejects_empty_conversation_id_and_empty_turns() {
        let tmp = tempfile::tempdir().unwrap();
        let relay = tmp.path().join("relay");
        let (mut a, _b) = two_devices(&relay, &tmp.path().join("a"), &tmp.path().join("b"));

        // Empty conversation_id → error (would otherwise pool into an anonymous thread).
        assert!(a
            .record_turn(Scope::Personal, "", "user", "hi", vec![], None, 1)
            .unwrap_err()
            .contains("conversation_id"));
        assert!(a
            .record_turn(Scope::Personal, "   ", "user", "hi", vec![], None, 1)
            .is_err());

        // Empty content with no tool_calls/tool_use_id → error (empty record).
        assert!(a
            .record_turn(Scope::Personal, "c1", "user", "", vec![], None, 1)
            .unwrap_err()
            .contains("content"));

        // A real turn, and a tool turn with empty content but a tool_use_id, are accepted.
        a.record_turn(Scope::Personal, "c1", "user", "hello", vec![], None, 1)
            .unwrap();
        a.record_turn(Scope::Personal, "c1", "tool", "", vec![], Some("call_1"), 2)
            .unwrap();
    }

    #[test]
    fn two_devices_converge_a_conversation_through_the_shared_relay() {
        let tmp = tempfile::tempdir().unwrap();
        let relay = tmp.path().join("relay");
        let (mut a, mut b) = two_devices(&relay, &tmp.path().join("a"), &tmp.path().join("b"));

        // a records a user turn + a knowledge fact; b records an assistant turn.
        a.record_turn(Scope::Personal, "c1", "user", "hello", vec![], None, 1)
            .unwrap();
        a.append(
            Scope::Personal,
            Surface::Knowledge,
            json!({"id": "f1", "body": "sky is blue"}),
        )
        .unwrap();
        a.pump().unwrap();
        b.pump().unwrap();
        b.record_turn(
            Scope::Personal,
            "c1",
            "assistant",
            "hi there",
            vec![],
            None,
            2,
        )
        .unwrap();
        b.pump().unwrap();
        a.pump().unwrap();

        // Divergence invariant: identical state hash after exchange.
        let sa = a.status().unwrap();
        let sb = b.status().unwrap();
        assert_eq!(sa["state_hash"], sb["state_hash"], "two devices converge");

        // Transcript resume is real on BOTH devices: the ordered user→assistant
        // turns come back as provider-valid messages.
        let resume_a = a.resume("c1").unwrap();
        let resume_b = b.resume("c1").unwrap();
        assert_eq!(resume_a, resume_b);
        let msgs = resume_a.as_array().unwrap();
        assert_eq!(msgs.len(), 2, "user + assistant");
        assert_eq!(msgs[0]["role"], json!("user"));
        assert_eq!(msgs[1]["role"], json!("assistant"));

        // The knowledge fact folded on the other device too.
        let transcript = a.transcript("c1");
        assert_eq!(transcript.as_array().unwrap().len(), 2);
    }

    #[test]
    fn two_devices_discover_each_others_a2a_endpoints_through_the_relay() {
        // "Find CAR on my other machine through Parslee", at the data layer.
        // Two daemons on one login announce their A2A endpoints; after a pump
        // each can dial the other. The relay carries only ciphertext, so the
        // endpoints converge without Parslee being able to read them — which is
        // why this rides the oplog instead of an address field on the roster.
        use car_sync::{DerivedKeyProvider, LoopbackTransport};
        let tmp = tempfile::tempdir().unwrap();
        let transport: Arc<dyn SyncTransport> = Arc::new(LoopbackTransport::new().unwrap());
        let provider: Arc<dyn SyncKeyProvider> =
            Arc::new(DerivedKeyProvider::new(b"parslee-login-master".to_vec()));

        let mut mac = SyncSubsystem::open_remote(
            &tmp.path().join("mac"),
            transport.clone(),
            "user:matt",
            provider.clone(),
        )
        .unwrap();
        let mut desktop = SyncSubsystem::open_remote(
            &tmp.path().join("desktop"),
            transport.clone(),
            "user:matt",
            provider.clone(),
        )
        .unwrap();

        // Neither knows the other before anything is announced.
        assert!(mac.host_endpoints().is_empty());
        assert!(desktop.host_endpoints().is_empty());

        mac.publish_host_endpoint("mac-studio", "http://192.168.1.10:8731", "MAC-PUBKEY")
            .unwrap();
        desktop
            .publish_host_endpoint("desktop", "http://192.168.1.20:8731", "DESKTOP-PUBKEY")
            .unwrap();
        mac.pump().unwrap();
        desktop.pump().unwrap();
        mac.pump().unwrap();

        let seen_by_mac = mac.host_endpoints();
        assert_eq!(seen_by_mac.len(), 1, "mac should see exactly the desktop");
        assert_eq!(seen_by_mac[0].name, "desktop");
        assert_eq!(seen_by_mac[0].url, "http://192.168.1.20:8731");
        // The key travels with the endpoint; without it the peer would be
        // visible but unauthenticatable.
        assert_eq!(seen_by_mac[0].pubkey, "DESKTOP-PUBKEY");

        let seen_by_desktop = desktop.host_endpoints();
        assert_eq!(seen_by_desktop.len(), 1);
        assert_eq!(seen_by_desktop[0].name, "mac-studio");
        assert_eq!(seen_by_desktop[0].url, "http://192.168.1.10:8731");
        assert_eq!(seen_by_desktop[0].pubkey, "MAC-PUBKEY");

        // A device that moves overwrites its own entry rather than adding one:
        // the surface folds LWW per device id, so a stale address cannot linger
        // and send peers somewhere nothing answers.
        desktop
            .publish_host_endpoint("desktop", "http://10.0.0.5:8731", "DESKTOP-PUBKEY")
            .unwrap();
        desktop.pump().unwrap();
        mac.pump().unwrap();
        let after_move = mac.host_endpoints();
        assert_eq!(after_move.len(), 1, "re-announcing must not accumulate");
        assert_eq!(after_move[0].url, "http://10.0.0.5:8731");
    }

    #[test]
    fn a_malformed_or_scheme_less_endpoint_is_not_offered_as_a_peer() {
        // An entry is dialed, so anything that is not an absolute http(s) URL is
        // a request to connect somewhere unintended and must not reach a listing.
        use car_sync::{DerivedKeyProvider, LoopbackTransport};
        let tmp = tempfile::tempdir().unwrap();
        let transport: Arc<dyn SyncTransport> = Arc::new(LoopbackTransport::new().unwrap());
        let provider: Arc<dyn SyncKeyProvider> = Arc::new(DerivedKeyProvider::new(b"k".to_vec()));
        let mut a = SyncSubsystem::open_remote(
            &tmp.path().join("a"),
            transport.clone(),
            "user:matt",
            provider.clone(),
        )
        .unwrap();
        let mut b = SyncSubsystem::open_remote(
            &tmp.path().join("b"),
            transport.clone(),
            "user:matt",
            provider.clone(),
        )
        .unwrap();
        b.publish_host_endpoint("hostile", "file:///etc/passwd", "K")
            .unwrap();
        b.pump().unwrap();
        a.pump().unwrap();
        assert!(
            a.host_endpoints().is_empty(),
            "a non-http(s) endpoint must never be offered as a dialable peer"
        );
    }

    #[test]
    fn two_remote_devices_converge_e2e_through_the_network_relay() {
        // The capstone: two daemon subsystems built via `open_remote` — the
        // Parslee-backed path — share ONE network service (a LoopbackTransport
        // reference server) under the same login scope + login-derived key.
        // They converge a conversation + knowledge fact end-to-end, with the
        // relay carrying only ciphertext. This is "phone + Mac after one login".
        use car_sync::{DerivedKeyProvider, LoopbackTransport};
        let tmp = tempfile::tempdir().unwrap();
        let transport: Arc<dyn SyncTransport> = Arc::new(LoopbackTransport::new().unwrap());
        let provider: Arc<dyn SyncKeyProvider> =
            Arc::new(DerivedKeyProvider::new(b"parslee-login-master".to_vec()));

        let mut mac = SyncSubsystem::open_remote(
            &tmp.path().join("mac"),
            transport.clone(),
            "user:matt",
            provider.clone(),
        )
        .unwrap();
        let mut phone = SyncSubsystem::open_remote(
            &tmp.path().join("phone"),
            transport.clone(),
            "user:matt",
            provider.clone(),
        )
        .unwrap();

        mac.record_turn(
            Scope::Personal,
            "c1",
            "user",
            "hello from mac",
            vec![],
            None,
            1,
        )
        .unwrap();
        mac.append(
            Scope::Personal,
            Surface::Knowledge,
            json!({"id": "f1", "body": "sensitive note"}),
        )
        .unwrap();
        mac.pump().unwrap();
        phone.pump().unwrap();
        phone
            .record_turn(
                Scope::Personal,
                "c1",
                "assistant",
                "hi from phone",
                vec![],
                None,
                2,
            )
            .unwrap();
        phone.pump().unwrap();
        mac.pump().unwrap();

        // Converge: identical state hash across the two devices.
        assert_eq!(
            mac.status().unwrap()["state_hash"],
            phone.status().unwrap()["state_hash"],
            "remote-backed devices converge through the network relay"
        );

        // The E2E round-trip works on both: the ordered transcript resumes as
        // provider-valid messages (which is only possible if the fold decrypted).
        let resume = phone.resume("c1").unwrap();
        let msgs = resume.as_array().unwrap();
        assert_eq!(msgs.len(), 2, "user + assistant, decrypted on the phone");
        assert_eq!(msgs[0]["role"], json!("user"));
        assert_eq!(msgs[1]["role"], json!("assistant"));

        // A different login (wrong master) shares the relay but cannot read it —
        // it folds the same ciphertext to a different (undecrypted) state.
        let intruder: Arc<dyn SyncKeyProvider> =
            Arc::new(DerivedKeyProvider::new(b"someone-elses-login".to_vec()));
        let mut evil = SyncSubsystem::open_remote(
            &tmp.path().join("evil"),
            transport.clone(),
            "user:matt",
            intruder,
        )
        .unwrap();
        evil.pump().unwrap();
        assert_ne!(
            evil.status().unwrap()["state_hash"],
            mac.status().unwrap()["state_hash"],
            "a wrong-key reader cannot reconstruct the plaintext state"
        );
    }

    #[test]
    fn config_tees_across_devices_and_refuses_device_local_domains() {
        // A setting changed on the Mac reaches the phone (LWW per domain), and a
        // device-local domain (a secret / OS grant) is refused — the partition
        // guard that keeps secrets off the relay.
        let tmp = tempfile::tempdir().unwrap();
        let relay = tmp.path().join("relay");
        let (mut mac, mut phone) =
            two_devices(&relay, &tmp.path().join("a"), &tmp.path().join("b"));

        mac.tee_config(
            "agent_permissions",
            json!({"milo": {"full_access": "require_approval"}}),
        )
        .unwrap();
        mac.pump().unwrap();
        phone.pump().unwrap();

        assert_eq!(
            phone.config_get("agent_permissions"),
            json!({"milo": {"full_access": "require_approval"}}),
            "the phone converges the Mac's config change"
        );

        // Last-writer-wins: the phone updates the same domain, the Mac converges.
        phone
            .tee_config(
                "agent_permissions",
                json!({"milo": {"full_access": "deny"}}),
            )
            .unwrap();
        phone.pump().unwrap();
        mac.pump().unwrap();
        assert_eq!(
            mac.config_get("agent_permissions")["milo"]["full_access"],
            json!("deny")
        );

        // A device-local domain (secret / OS grant) must be refused.
        assert!(mac
            .tee_config("keychain_secrets", json!({"slack": "xoxb-…"}))
            .is_err());
        assert!(mac.tee_config("parslee_tokens", json!("tok")).is_err());
        assert_eq!(mac.config_get("keychain_secrets"), json!(null));
    }

    #[test]
    fn dispatch_fence_refuses_stale_epoch_and_already_committed() {
        let tmp = tempfile::tempdir().unwrap();
        let relay = tmp.path().join("relay");
        // Shared coordinator (frozen clock → a's lease never expires here).
        let (_tc, wc) = manual_clock();
        let coord = InMemoryLeaseCoordinator::new(wc);
        let (_ta, wa) = manual_clock();
        let (_tb, wb) = manual_clock();
        let mut a = SyncSubsystem::open_with(
            "mac-a".into(),
            &tmp.path().join("a"),
            &relay,
            Box::new(coord.clone()),
            wa,
        )
        .unwrap();
        let mut b = SyncSubsystem::open_with(
            "mac-b".into(),
            &tmp.path().join("b"),
            &relay,
            Box::new(coord),
            wb,
        )
        .unwrap();

        // a acquires epoch 1 and may dispatch.
        let lease = a.lease_acquire("milo", 100).unwrap();
        assert_eq!(lease["epoch"], json!(1));
        let f = a.fence_check("milo", "run-1", 1).unwrap();
        assert_eq!(f["may_dispatch"], json!(true));

        // a records the run committed and syncs it to b.
        a.record_intent(
            Scope::Personal,
            &Intent::new("milo", "run-1", 1, IntentStatus::Committed),
        )
        .unwrap();
        a.pump().unwrap();
        b.pump().unwrap();

        // The oracle now refuses a re-dispatch of run-1 on BOTH devices — even
        // a, the legitimate current holder (idempotency beats liveness).
        let f = a.fence_check("milo", "run-1", 1).unwrap();
        assert_eq!(f["decision"]["decision"], json!("already_committed"));
        assert_eq!(f["may_dispatch"], json!(false));
        let fb = b.fence_check("milo", "run-1", 1).unwrap();
        assert_eq!(fb["decision"]["decision"], json!("already_committed"));

        // b is not the holder at epoch 1 → a fresh (uncommitted) run on b is
        // fenced as not-held; after b steals it becomes the holder.
        let stale = b.fence_check("milo", "run-2", 1).unwrap();
        assert_eq!(stale["decision"]["decision"], json!("stale_epoch"));
        assert_eq!(stale["may_dispatch"], json!(false));
    }

    #[test]
    fn lease_is_visible_across_two_devices_sharing_the_register() {
        let tmp = tempfile::tempdir().unwrap();
        let relay = tmp.path().join("relay");
        let (mut a, mut b) = two_devices(&relay, &tmp.path().join("a"), &tmp.path().join("b"));

        // a acquires; b sees it held by mac-a (shared linearizable register).
        a.lease_acquire("milo", 1_000_000).unwrap();
        let status = b.lease_status("milo").unwrap();
        assert_eq!(status["lease"]["holder"], json!("mac-a"));

        // b cannot acquire an unexpired lease.
        assert!(b.lease_acquire("milo", 100).is_err());

        // a releases; b now acquires epoch 2 (monotone, never reused).
        a.lease_release("milo", 1).unwrap();
        let lease = b.lease_acquire("milo", 100).unwrap();
        assert_eq!(
            (lease["epoch"].as_u64(), lease["holder"].as_str()),
            (Some(2), Some("mac-b"))
        );
    }

    #[test]
    fn assistant_checkpoint_and_action_survive_process_reopen_exactly() {
        use crate::assistant::governance::{
            ActionScope, AssistantCheckpoint, CompletionMatrix, CredentialCapability,
            ResumeDirective, SupervisedActionRecord,
        };
        use car_inference::tasks::generate::{
            ContentBlock, Message, Provenance, ThinkingBlock, ToolCall,
        };

        let tmp = tempfile::tempdir().unwrap();
        let device = tmp.path().join("device");
        let relay = tmp.path().join("relay");
        let repo = tmp.path().join("repo");
        std::fs::create_dir_all(repo.join(".git")).unwrap();
        let (_t, wall) = manual_clock();
        let coordinator = InMemoryLeaseCoordinator::new(wall.clone());
        let checkpoint = AssistantCheckpoint {
            id: "task-1".into(),
            session_id: "task-1".into(),
            revision: 1,
            repository_root: repo.clone(),
            messages: vec![
                Message::System {
                    content: "system exact".into(),
                },
                Message::User {
                    content: "diagnose it".into(),
                },
                Message::UserMultimodal {
                    content: vec![ContentBlock::Text {
                        text: "screenshot attached".into(),
                    }],
                },
                Message::Assistant {
                    content: "checking".into(),
                    tool_calls: vec![serde_json::from_value::<ToolCall>(json!({
                        "id": "call-1",
                        "name": "read_file",
                        "arguments": {"path": "src/app.rs"}
                    }))
                    .unwrap()],
                    thinking: vec![ThinkingBlock {
                        text: "opaque reasoning".into(),
                        signature: Some("provider-signature".into()),
                        redacted_data: None,
                    }],
                    model_id: None,
                    local_last_resort: false,
                },
                Message::ToolResult {
                    tool_use_id: "call-1".into(),
                    content: "source".into(),
                    provenance: Provenance::Internal,
                },
                Message::ProviderOutputItems {
                    protocol: "openai-responses".into(),
                    items: vec![json!({"type": "reasoning", "encrypted_content": "opaque"})],
                },
            ],
            goal: Some(json!({"check": "./verify"})),
            compaction: Some(json!({"generation": 2, "supersedes": 1})),
            completion: CompletionMatrix::default(),
        };
        let scope = ActionScope {
            tool: "shell".into(),
            parameters: json!({"command": "git push origin HEAD:main"}),
            repository_root: repo,
            target: "origin/main".into(),
            environment: "fixture".into(),
            credential_capabilities: vec![CredentialCapability("git:origin".into())],
        };
        let mut action = SupervisedActionRecord::propose("task-1", "call-7", scope);

        {
            let mut first = SyncSubsystem::open_with(
                "mac-a".into(),
                &device,
                &relay,
                Box::new(coordinator.clone()),
                wall.clone(),
            )
            .unwrap();
            first.assistant_checkpoint_put(checkpoint.clone()).unwrap();
            first.assistant_action_put(action.clone()).unwrap();
            action.transition(ActionState::Approved, None).unwrap();
            first.assistant_action_put(action.clone()).unwrap();
            action.transition(ActionState::Dispatched, None).unwrap();
            first.assistant_action_put(action.clone()).unwrap();
        }

        let reopened =
            SyncSubsystem::open_with("mac-a".into(), &device, &relay, Box::new(coordinator), wall)
                .unwrap();
        assert_eq!(
            reopened.assistant_checkpoint_get("task-1").unwrap(),
            Some(checkpoint)
        );
        let recovered = reopened
            .assistant_action_get(&action.id)
            .unwrap()
            .expect("durable action");
        assert_eq!(recovered, action);
        assert_eq!(
            recovered.resume_directive(),
            ResumeDirective::MarkIndeterminate,
            "a crash after dispatch must not replay the external effect"
        );
    }
}