car-server-core 0.32.1

//! 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 (documented boundary)
//!
//! The E2E payload-encryption boundary ships as a tested library primitive in
//! [`car_sync::crypto`] ([`car_sync::LocalKeyCipher`] over ChaCha20-Poly1305).
//! It is **not** applied to this daemon's live oplog stream yet: the fold groups
//! on cleartext `payload["id"]`, so a live E2E device must decrypt-before-fold —
//! that fold-path integration, plus login-derived key distribution, is the
//! documented follow-up. Likewise the `InMemoryLeaseCoordinator` is linearizable
//! **within one daemon**; a cross-daemon distributed coordinator (Cosmos
//! `if_match` / single-writer daemon / Postgres advisory lock) is the B6 backend
//! follow-up — it implements the same [`car_sync::LeaseCoordinator`] trait.

use std::path::Path;

use car_sync::{
    check_dispatch, frontier_of, system_clock, FsRelay, InMemoryLeaseCoordinator, Intent,
    IntentStatus, LeaseCoordinator, Relay, RelayConfig, Scope, Surface, SyncSession, Turn,
    WallClock,
};
use serde_json::{json, Value};

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

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

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