car-server-core 0.34.0

//! 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;

/// 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";

/// 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)
    }

    /// `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> {
        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 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_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"))
        );
    }
}