car-sync

Multi-device sync core for the Common Agent Runtime — the oplog + deterministic fold + checkpoint/compaction + relay transport + execution lease/fencing + transcript resume (slices B1, B4, B3, B5, and B2 of docs/proposals/multi-device-sync.md).

The proposal's frame: sync events, not files. Every state-changing operation becomes a content-addressed, replica-tagged OpRecord in an append-only oplog; sync is "send me the ops I don't have"; and every device folds the full op-set into materialized state deterministically. Because the fold is commutative, associative, and idempotent over the op-set (CRDT properties), two devices writing simultaneously converge the moment they exchange ops.

What's here (pure, library-only)

• oplog — OpRecord { op_id, hlc, device_id, seq, prev, scope, surface, payload } exactly as the proposal specs it, plus the per-device seq/prev hash-chain that makes a log order-verifiable (verify_log). op_id is content-derived via the shipped B7 discipline (SHA-256 + 0x1f separators, the car_proto::deterministic_run_id pattern) over every field including chain position, so it is both the retransmission-dedup key and a tamper-evident cover. DeviceLog is the writer: it maintains the chain and stamps the HLC shape ({wall_ms, counter, device_id}) from HlcClock, the real hybrid logical clock (B3) — standard send/receive rules (max of local wall and everything witnessed; counter ticks on ties), monotone under clock skew, regression, and same-millisecond bursts. Wall readings are injectable (WallClock; system_clock() is the one opt-in real-time source — library logic never reads system time), and DeviceLog::new defaults to an always-0 wall under which the HLC degenerates to exactly B1's pure-Lamport order: one code path, same wire shape.

• fold — fold(ops) -> SyncState under the proposal's per-surface fold rules: grow-only surfaces (conversations, knowledge, skills, trajectories, runs) union by stable ID; registry surfaces (declagents, file registries) resolve LWW-per-record by HLC — not per file. Routing follows "sync the observations, not the result" and folds as a multiset keyed by op_id (two byte-identical observations are two events — only retransmission dedups): the fold materializes the canonically-ordered observation stream and SyncState::replay runs a caller-injected apply (the EMA stays out of the crate). state_hash is the divergence-detection invariant ("same frontier ⇒ same snapshot hash"); registry_as_lww projects a folded registry onto car_state::crdt::LwwMap, and the tests prove fold(union of ops) ≡ crdt_merge(per-device exports) where the domains overlap.

• journal — durable JSONL persistence in the car-eventlog idiom: append-only, flush-per-record, torn-line tolerant on load, tail-healing on open (a torn last line is newline-terminated before the next append so the post-crash record isn't glued onto garbage), and single-writer enforced via an exclusive advisory lock on <path>.lock (the car-registry supervisor pattern) — a second open on the same path fails with WouldBlock. B4 adds truncate_to: an atomic temp+rename rewrite to a retained tail, executed under that same lock, stamping a TruncationMarker (naming the covering checkpoint) as the new file's first line — atomic with the truncation. The marker fences the permanent-fork hazard at runtime: load refuses a marked journal (use load_with_marker + resume_anchored), and DeviceLog::resume refuses an own-chain non-zero start (ChainError::TruncatedChain).

• checkpoint (B4) — Checkpoint: a serialized fold at a frontier — per-device {seq, hlc, head} frontier entries, covered scope tags, the SyncState snapshot, state_hash (the divergence invariant), and checkpoint_hash: the whole-record content address (frontier + scopes + state; file <checkpoint_hash>.checkpoint.json). Same frontier ⇒ same file (relay dedup for free) AND same file ⇒ same checkpoint — the fold dedups cross-device, so two different frontiers can share one folded state, and a state-only address would fork chains on resume. load recomputes both hashes and cross-checks the file name; a tampered state, tampered frontier, or renamed file is rejected loudly, never folded. The recorded chain heads make a checkpoint the anchored head of a truncated log: verify_anchored(checkpoint, tail) proves the composition carries the full log's integrity guarantee, resume_anchored continues a device's chain past truncation without forking at seq 0, and fold_onto(checkpoint.state, tail) == fold(full log) — byte-identical, per surface including the routing multiset — is the tested equivalence that makes compaction safe.

• compact (B4) — the GC protocol's device-side core. Per-surface snapshot retention per the proposal's table (RetentionPolicy::proposal_default): conversations last-N by payload timestamp, runs 50/agent + 30 days (parity with RunStore::gc), trajectories last-D, knowledge/skills keep-all. Every retention-dropped id-bearing entry leaves a minimal tombstone stub ({"id", "tombstone": true}, original op identity kept, quota-neutral, idempotent), so a "supersedes" reference resolves against a tombstone even when it arrives after compaction — no time hole; id-less entries drop entirely. Event-stream surfaces reject any rule but keep-all — observation multisets replay from genesis. The age rules' reference instant defaults to as_of_from_ops (max payload timestamp among the below-frontier ops — pure over the fold's own inputs, so every device derives the same instant). AckTable is the fold-frontier bookkeeping (per-device acked HLC, monotone-only advance, temp+rename persistence); the stable frontier is min(acked) and compaction refuses to drop anything above any device's acked frontier (or to run at all for a device with no ack entry, or over a journal that already carries a truncation marker). compact_and_truncate enforces the crash-ordering invariant by construction: checkpoint durable FIRST, then truncate — a crash between the steps leaves the full journal plus a redundant (idempotently recomputable) checkpoint; acknowledged data is never lost; an empty compaction is a no-op (no empty checkpoint file).

• relay (B3) — the Relay trait (push / pull(since per-device seq frontier) → {ops, latest_checkpoint_ptr} / ack / checkpoint_put/get / roster) with two reference implementations: InMemoryRelay and the shared-directory FsRelay loopback (the realistic single-user two-Mac case: exclusive advisory lock, temp+rename state file, content-addressed checkpoint files re-verified on every load). The relay holds one verified chain per device — a pushed op must continue the relay-held chain (fork / gap / foreign / tampered = typed error; retransmission dedups on op_id). Pull cursors are per-device seqs, not HLCs, so a straggler's late (old-HLC) ops still reach every peer. Stable frontier = min(acked) over non-evicted roster devices; a device silent past RelayConfig::eviction_horizon_ms (the proposal's horizon H) is roster-marked Evicted — its ack no longer pins GC — and is reinstated only by an ack at/above the current stable frontier. GC drops an op only when BOTH at/below the stable frontier AND covered by a stored checkpoint; retained chains stay gap-free (prefix drops, dropped head remembered for chain continuity), and a pull into truncated space fails loudly (FrontierTruncated — the cold-bootstrap signal). Checkpoints dedup on checkpoint_hash, the whole-record content address, never state_hash (the B4 contract); the latest pointer advances only to a frontier-dominating checkpoint. Deliberate deviation (documented in the module docs, binding on B6): the proposal's "reject ops older than H" rule is not implemented — seq+checkpoint truncation makes accepted-op-in-truncated-space structurally impossible, so straggler re-entry is lossless instead of lossy.

• session (B3) — SyncSession, the device-side pump holding the binding contracts by construction: append journals (flushed) before an op is pushable (journal-durable before transmit — B1 MUST); pulled ops are verify_log/verify_anchored-checked before folding; folds are journaled before the ack, whose value is derived from journal-held ops only, so acking merely-received state has no API path (B4 MUST). Retry/crash-mid-pump idempotent (op_id dedup both directions, monotone acks). Cold bootstrap / post-eviction re-entry (bootstrap/rebase): checkpoint_get + pull(since = checkpoint frontier) + resume_anchored — never DeviceLog::resume; the rebased journal carries the truncation marker so the naive-load fence holds — with locally-held ops the checkpoint doesn't cover (a returning straggler's unpushed writes) carried across the rebase and pushed after. publish_checkpoint computes and uploads the device-side checkpoint at the relay's stable frontier (device-computed because under B6's E2E the relay can never fold).

• lease (B5) — execution lease + fencing: single-leader execution over the leaderless replication above. It delivers deterministic ledger convergence + a durable idempotency oracle — NOT exactly-once execution (that is B6's dispatch fence). A LeaseCoordinator trait (acquire/renew/release/current) provides a linearizable CAS register per agent — deliberately separate from Relay, which is eventually consistent and cannot host a lease. acquire grants only if unheld/expired and bumps the monotone, never-reused epoch (the fencing token); InMemoryLeaseCoordinator is the honest in-process reference (Arc<Mutex> CAS, like InMemoryRelay). Fencing is a fold property with two views on the leased Surface::Intent surface (FoldTier::Leased): committed_runs — a fence-independent, keep-all idempotency oracle (SyncState::committed_run), the correct "did this run already execute?" lookup, surviving epoch bumps AND compaction; and runs — the "who holds now" view, where pending intents are per-agent fenced (a zombie's stale pending loses deterministically, without a wall-clock race — fencing beats HLC) while committed/failed records are terminal-immune. fold_onto(checkpoint, tail) == fold(full) holds across an epoch bump. Idempotency keys on the B7 deterministic run_id. SyncSession::record_intent (ungated, journal-durable, with a terminal guard that no-ops a pending/failed write for an already-committed run) + record_intent_if_current (best-effort local lease gate) + committed_run (the oracle a B6 dispatch fence reads). The register holds only non-sensitive metadata, so it never breaches E2E. Intents are keep-all in compaction (CompactError::IntentRetention rejects any trim). Fencing is per-agent (the spec — it fences a zombie's unique post-failover pending too).

• conversation (B2) — transcript resume: the conversation surface as an ordered, role-threaded projection of the oplog. SyncState::transcript( conversation_id) folds the Surface::Conversation entries into a causally (hlc, op_id)-ordered Vec<Turn> (two devices talking to the same agent concurrently interleave deterministically by (wall, counter, device_id)), and resume_messages(conversation_id) returns the repaired, provider-valid Vec<Message> (the real car-inference-types::Message) car-inference's multi-turn path replays to continue the conversation — the verbatim conversation-resume path the 0.25 removal (docs/solutions/conversation-persistence-removed-in-0.25.md) deferred to sync, now shipped and oplog-native (not the removed ConversationStore). A conversation turn is an event stream keyed by op_id (kernel-review correction — op identity IS turn identity; content-keying silently dropped two genuine same-timestamp turns): a resent op dedups, two distinct authorings never collapse. It reuses the routing multiset machinery but is an independent multiset (no path-dependent replay — is_replay_stream == false), so it tolerates LastN retention. Because HLC order is deterministic but says nothing about concurrent turns, resume_messages runs a repair (coalesce adjacent same-role turns, drop an orphan tool_result from a LastN cut mid-exchange, strip a dangling assistant tool_call) so the sequence is never provider-invalid — the "runtime validates" thesis on the projection. The 0.25 compaction-vs-store incoherence cannot recur — the oplog is the one source of truth and the transcript is a projection of the same folded state B4's checkpoint serializes, so a resume after LastN compaction is exactly the retained window + live tail in order, and local compaction is byte- and hash-identical to global retention. car-sync builds the real Message from the shared pure-serde car-inference-types crate (no Candle/MLX weight), so a shape change is a compile error, not a runtime from_value break. The daemon/memgine adoption (feeding resume_messages into the engine's multi-turn path + the semantic-summary boundary) is B6.

Determinism discipline: all folded state is BTreeMap-backed and nothing in the fold/retention path reads a clock (age rules take an explicit as_of_ms) — the two non-determinism leaks the proposal explicitly warns about. The fold stays order-independent even on invalid input (a forged colliding op_id tiebreaks on content, not arrival order), but ops from a remote/untrusted source MUST pass verify_log (or verify_anchored, when composing with a checkpoint) before folding.

Contracts callers must hold:

• Journal-durable before transmit. DeviceLog::resume re-derives next_seq from the journal; a crash between "op transmitted" and "op journaled" re-mints that seq for a different op — a permanent DuplicateSeq fork on union. Always OplogJournal::append first.
• Device identity is asserted, not authenticated. The hash chain proves internal consistency; a forger who recomputes hashes passes verify_log, and Checkpoint::verify proves integrity, not origin. Cryptographic device identity (op/checkpoint signing) is B6 — until then, trust in a log's origin comes from the transport that delivered it.
• Checkpoint durable before truncation. compact_and_truncate holds this by construction; anyone driving Checkpoint::save + OplogJournal::truncate_to by hand MUST keep that order — truncation makes the dropped ops unrecoverable from the journal, and the checkpoint named by the truncation marker is what still accounts for them.
• An ack asserts durably-folded state (MUST, binding on B3). Report ack(frontier) only after the ops at/below it are durably persisted and folded — an ack sent from memory ahead of the fsync lets compaction drop ops the acking device then loses in a crash. The mirror of journal-durable-before-transmit.

Scope note: B1 stamps ONE seq/prev chain per device across scopes, so B4 checkpoints whole-chain and records the covered scope tags — per-scope frontiers land when B3/B6 split the relay streams (and with them the chains) by scope.

What's deliberately NOT here

No network daemon wiring, no encryption — the relay here is a trait + an in-process/filesystem loopback; the wire surface speaks the same contract later. Later slices: the sync.* WS/FFI surface + the network relay backend + E2E encryption + checkpoint/op signing + per-scope streams (B6, signing the whole-record content address and honoring scopes-as-encryption-audiences), recompaction over a checkpoint base (re-planning a journal that already carries a truncation marker — a base-anchored session skips publish_checkpoint for the same reason), device-side relay-driven local compaction (folding relay roster acks into a local AckTable for compact_and_truncate), the distributed LeaseCoordinator backend + the tier-3 executor fence call-site + the lease.*/sync.* wire surface (B6, on top of B5's core here), and transcript resume + files-are-projections write-path rerouting (B2). B7 (deterministic run ids, car_proto::deterministic_run_id) shipped earlier and is the id discipline reused here (and by B5's leased-intent idempotency key).

Quick example

use car_sync::{fold, state_hash, verify_log, DeviceLog, OplogJournal, Scope, Surface};
use serde_json::json;

let mut laptop_a = DeviceLog::new("laptop-a");
let mut laptop_b = DeviceLog::new("laptop-b");

let op1 = laptop_a.append(Scope::Personal, Surface::Knowledge,
    json!({"id": "fact-1", "body": "prefers dark mode"}));
laptop_b.observe(&op1.hlc); // lamport receive rule
let op2 = laptop_b.append(Scope::Personal, Surface::Declagent,
    json!({"id": "milo", "model": "qwen3"}));

// Any delivery order, any duplication — same state, same hash.
let ops = vec![op1, op2];
verify_log(&ops).unwrap();
let state = fold(&ops);
assert_eq!(state_hash(&state), state_hash(&fold(&ops)));

// Durable journal: append-only JSONL, torn-tail tolerant on load.
let mut journal = OplogJournal::open(std::path::Path::new("/tmp/oplog.jsonl")).unwrap();
for op in &ops { journal.append(op).unwrap(); }

Compaction (B4): once every device has acked past a frontier, checkpoint and truncate — checkpoint durable first, equivalence guaranteed:

use car_sync::{compact_and_truncate, fold, fold_onto, AckTable, OplogJournal, RetentionPolicy};

let mut acks = AckTable::new();
acks.ack("laptop-a", frontier_hlc_a); // monotone-only; min(acked) is the frontier
acks.ack("laptop-b", frontier_hlc_b);

let outcome = compact_and_truncate(
    &mut journal,
    std::path::Path::new("/tmp/checkpoints"),
    &acks,
    &RetentionPolicy::proposal_default(200, 30 * 24 * 60 * 60 * 1000),
    None, // as_of derived from the below-frontier ops — deterministic, no clock
).unwrap();

// The invariant that makes it safe (pinned per surface by the tests):
// fold_onto(checkpoint.state, retained tail) == fold(full log)
// (a truncated journal is marked: load() now errors — read the tail +
// marker explicitly, and resume via resume_anchored, never resume())
let (marker, tail) = OplogJournal::load_with_marker(journal.path()).unwrap();
assert_eq!(marker.unwrap().checkpoint_hash, outcome.plan.checkpoint.checkpoint_hash);
let reconstructed = fold_onto(&outcome.plan.checkpoint.state, &tail);
assert_eq!(reconstructed, fold(&full_log_from_before_compaction));

Sync (B3): two Macs converging through a shared-directory relay — each side pumps (push journal-durable ops → pull → verify → journal the folds → ack) until quiescent:

use car_sync::{FsRelay, Relay, RelayConfig, Scope, Surface, SyncSession, system_clock};
use serde_json::json;
use std::path::Path;

let relay_dir = Path::new("/Volumes/Shared/car-relay"); // any shared dir
let mut relay = FsRelay::open(relay_dir, RelayConfig::default(), system_clock()).unwrap();

let mut a = SyncSession::open(
    "mac-a",
    Path::new("/Users/me/.car/sync/oplog.jsonl"),
    Path::new("/Users/me/.car/sync/checkpoints"),
    system_clock(), // injectable — tests pass a controlled closure
).unwrap();

// Local write: stamped by the hybrid clock, journal-durable BEFORE it can
// ever be transmitted.
a.append(Scope::Personal, Surface::Knowledge,
    json!({"id": "fact-1", "body": "prefers dark mode"})).unwrap();

// One reconciliation round; retry-safe at every crash point.
let report = a.pump(&mut relay).unwrap();
assert_eq!(report.pushed, 1);

// A brand-new (or evicted-and-returning) device cold-bootstraps:
// checkpoint_get + pull(since = checkpoint frontier) + resume_anchored —
// locally-held unpushed writes survive the rebase and push on the next pump.
let mut b = SyncSession::bootstrap(
    "mac-b",
    Path::new("/Users/me2/.car/sync/oplog.jsonl"),
    Path::new("/Users/me2/.car/sync/checkpoints"),
    &mut relay,
    system_clock(),
).unwrap();
b.pump(&mut relay).unwrap();
assert_eq!(a.state_hash(), b.state_hash()); // the divergence invariant