# 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`](../../../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.
- **`fold_at`** — `fold_at(ops, &frontier) -> SyncState` folds the op-set **as
it stood at a past per-device seq frontier**: the state a branch would have
forked from, without forking anything. A filter plus `fold`, so no wire
change, no digest change, no migration.
It exists for counterfactual replay of the state suffix — assembling a
context two ways and diffing them. CLAUDE.md records why that matters:
StateBench's per-track numbers carry a **±15pp noise floor**, so a per-track
delta means nothing on its own and the only trustworthy check is to dump the
assembled context both ways and compare. Re-running reintroduces variation
unrelated to the edit; folding at a fixed frontier holds everything else
constant.
A device **absent** from the frontier contributes nothing — a frontier
describes what a reader had *seen*, so treating absent as unbounded would
fold in a whole device's history and make the replayed state larger than the
moment being reconstructed. `fold_at(ops, &frontier_of(ops)) == fold(ops)`,
and folding at a frontier equals folding that prefix; both are pinned by
tests. This is the read half of
`docs/proposals/oplog-branch-semantics.md`; the write half (sibling chains
that merge or discard) stays gated there.
- **`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.
- **`crypto`** (B6) — the E2E encryption boundary. A `PayloadCipher` trait with
a `LocalKeyCipher` ChaCha20-Poly1305 reference implementation, so the relay
holds only ciphertext while `op_id` / `seq` / `hlc` stay cleartext (the fold
and the frontier arithmetic still work on an untrusted relay).
The **live surface** (B6) is not in this crate: the `sync.*` / `lease.*` daemon
wire methods live in `car-server-core::sync`, over a daemon-held
`SyncSubsystem` — one device, holding a `SyncSession` + `FsRelay` +
`InMemoryLeaseCoordinator`. The executor dispatch fence (`check_dispatch`: a
committed-run oracle read plus a linearizable "still epoch N?" read at the point
of effect) is B5's deferred exactly-once completion, and also lives there.
Remaining B6 follow-ups: live-stream decrypt-before-fold; login-derived / org
key distribution; a distributed `LeaseCoordinator` and a network `Relay`
backend; and auto-teeing the daemon's own conversation/knowledge write paths
into the oplog.
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 — the relay here is a trait + an
in-process/filesystem loopback; the daemon wire surface (`car-server-core::sync`,
shipped) speaks the same contract. Later slices: the network relay
backend + 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, and 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
```rust
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:
```rust
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:
```rust
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
```