diff --git a/CLAUDE.md b/CLAUDE.md
index 4965910..bf107a3 100644
--- a/CLAUDE.md
+++ b/CLAUDE.md
@@ -36,12 +36,13 @@ CI runs the full suite only on Linux (service container); the Windows leg is a c
 ## Architecture & conventions
 
 - **Library/binary split.** `src/main.rs` is only runtime plumbing (load `.env`, init tracing, connect pool, run migrations, bind, serve with graceful shutdown). The router is built by `waveflow_server::app(config, state)` in `src/lib.rs` so integration tests spawn the *same* app in-process (`tests/support.rs::spawn_app`). Put logic behind `app()`, not in `main`.
-- **`AppState`** (`src/lib.rs`) holds the shared singletons threaded through every handler — currently just the `PgPool` (cheap to clone, `Arc`-backed). Add new singletons here.
+- **`AppState`** (`src/lib.rs`) holds the shared singletons threaded through every handler. Cheap to clone — every field is `Arc`-backed. Fields today: `db: PgPool`, `jwt_verifier: Arc<JwtVerifier>` (Phase 1.d.1), `stream_ctx: Option<Arc<StreamCtx>>` (Phase 1.e — `None` disables streaming), `sync: SyncHub` (Phase 1.f — broadcast `Sender` + `DashMap<(user_id, device_id), AckEntry>` + flush/compaction tasks; tests construct via `SyncHub::for_tests(pool)` which skips the background loops so `flush_acks` / `compact_once` can be driven by hand for determinism). Add new singletons here.
 - **API is one file per resource** under `src/api/`, each exposing a `router()` merged in `src/api/mod.rs`. `/health` (liveness, no DB) and `/ready` (DB-aware readiness) are unversioned infra probes; every real resource mounts under `/api/v1/`.
 - **No SQL in handlers.** SQL lives in the DB layer (`src/db.rs`) or in a `waveflow-core::repository::postgres::*` method; handlers stay pure HTTP orchestration. `db::ping` (`/ready`'s `SELECT 1`) and `db::users::create` are the in-tree pattern; everything tenant-scoped goes through `PostgresProfileRepository::*_for_user`. This mirrors the desktop's Tauri-command ↔ `waveflow-core` boundary.
 - **Tenancy is enforced at the storage layer.** Server handlers under `/api/v1/*` extract `UserId` from the `middleware::authenticate` middleware and call `*_for_user` methods only — never the single-tenant trait surface. `PostgresProfileRepository` deliberately does NOT implement `ProfileRepository`, so the compiler stops a careless `list_all()` from leaking another tenant's rows. Apply the same pattern when adding library / track / playlist repositories.
 - **Auth: JWT-only (Phase 1.d.2).** `middleware::authenticate` requires a Bearer JWT signed by the upstream Better Auth issuer. The middleware verifies the token via [`AppState::jwt_verifier`], then `db::users::find_or_provision_by_external_id(state.db, &sub, now_ms)` lazy-onboards the user on first request — a valid signature is the authoritative onboarding signal, so no separate `POST /api/v1/users` exists. Boot requires the full `WAVEFLOW_JWT_*` triple (`_JWKS_URL` / `_ISSUER` / `_AUDIENCE`); the legacy `X-User-Id` dev shim retired alongside `WAVEFLOW_DEV_AUTH`.
 - **Streaming (Phase 1.e).** Two endpoints: `POST /api/v1/profiles/{p}/libraries/{l}/tracks/{t}/stream-url` (JWT-authed) verifies tenant ownership and signs a short-lived (≤ 60 s) URL via [`stream_token::mint`]; `GET /api/v1/stream/{token}` is mounted OUTSIDE the JWT layer because browsers can't attach a Bearer to `<audio src>` — the HMAC in the token IS the auth. The stream handler canonicalises `<music_root>/<file_path>` and refuses anything resolving outside `WAVEFLOW_MUSIC_ROOT` (path-traversal guard via `std::fs::canonicalize` + prefix check). Range requests handled in-process: `Accept-Ranges: bytes`, 206 + `Content-Range` on partial, 416 on unsatisfiable. Both endpoints answer 503 when streaming is disabled at boot (`WAVEFLOW_MUSIC_ROOT` + `WAVEFLOW_STREAM_SECRET` unset). Until 1.f's sync ships, files must be placed manually under the music root.
+- **Sync (Phase 1.f).** Append-only `sync_op` log keyed on `BIGSERIAL id`; per-`(user, device)` UNIQUEs on `operation_id` (idempotency) and `lamport_ts` (monotonicity). Three REST routes + one WebSocket under `/api/v1/sync/*`: `POST /ops` push (idempotent replay via `ON CONFLICT operation_id DO NOTHING`, 409 + `stored_max` on lamport regression), `GET /ops?since=N` pull (410 + `compacted_up_to` when `0 < since < watermark`; `since=0` is the bootstrap path and always skips the guard so fresh devices can converge from an empty cursor — read-only, never advances the cursor), `POST /ack` (the **only** path that writes `device_sync_cursor`, buffered in memory + flushed every 5 s), `GET /ws?device_id=…` WebSocket fan-out (one broadcast channel, per-frame `user_id` filter so cross-tenant ops never leak). Live state in `sync::SyncHub`: tokio `broadcast::Sender` + `DashMap<(user_id, device_id), AckEntry>` + daily compaction task that flushes ACKs first then collapses superseded ops in the same Postgres transaction as the watermark UPSERT. Watermark monotonic by UPSERT `WHERE` clause. Stale devices (`last_seen_at < now - 90 d`) skipped from the compaction MIN. Tests drive `SyncHub::for_tests` (no background tasks) and call `flush_acks` / `compact_once` directly for determinism.
 - **Don't leak DB errors to unauthenticated probes.** `/ready` logs the sqlx error via `tracing::warn!` but returns a fixed sentinel body (`{status, db}`) so a load balancer never sees the connection-URL host or credentials. Apply the same discipline to any other unauthenticated endpoint.
 - **Migrations are immutable once merged.** They're embedded at compile time via `sqlx::migrate!("./migrations")` (`db::MIGRATOR`); the `_sqlx_migrations` table stores each file's checksum, so editing an applied migration makes the server refuse to start. Schema changes = a new dated migration file (`YYYYMMDDHHMMSS_name.sql`). Boot applies pending migrations *before* opening the listener, which is what makes `/ready` trustworthy.
 - **Schema parity with the desktop SQLite migrations.** Postgres tables mirror the shapes in the desktop repo's `src-tauri/migrations/app/` so `PostgresProfileRepository` and `SqliteProfileRepository` (in `waveflow-core`) satisfy the same trait against identical rows. Keep types compatible (e.g. `BIGSERIAL` ↔ SQLite `INTEGER PK`, epoch-millis `BIGINT` for timestamps).
diff --git a/Cargo.lock b/Cargo.lock
index d6b313f..d7d96ef 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -89,6 +89,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "31b698c5f9a010f6573133b09e0de5408834d0c82f8d7475a89fc1867a71cd90"
 dependencies = [
  "axum-core",
+ "base64",
  "bytes",
  "form_urlencoded",
  "futures-util",
@@ -107,8 +108,10 @@ dependencies = [
  "serde_json",
  "serde_path_to_error",
  "serde_urlencoded",
+ "sha1 0.10.6",
  "sync_wrapper",
  "tokio",
+ "tokio-tungstenite 0.29.0",
  "tower",
  "tower-layer",
  "tower-service",
@@ -445,6 +448,20 @@ dependencies = [
  "syn",
 ]
 
+[[package]]
+name = "dashmap"
+version = "6.2.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e6361d5c062261c78a176addb82d4c821ae42bed6089de0e12603cd25de2059c"
+dependencies = [
+ "cfg-if",
+ "crossbeam-utils",
+ "hashbrown 0.14.5",
+ "lock_api",
+ "once_cell",
+ "parking_lot_core",
+]
+
 [[package]]
 name = "data-encoding"
 version = "2.11.0"
@@ -905,6 +922,12 @@ dependencies = [
  "zerocopy",
 ]
 
+[[package]]
+name = "hashbrown"
+version = "0.14.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e5274423e17b7c9fc20b6e7e208532f9b19825d82dfd615708b70edd83df41f1"
+
 [[package]]
 name = "hashbrown"
 version = "0.15.5"
@@ -1076,7 +1099,7 @@ dependencies = [
  "tokio",
  "tokio-rustls",
  "tower-service",
- "webpki-roots",
+ "webpki-roots 1.0.7",
 ]
 
 [[package]]
@@ -2058,7 +2081,7 @@ dependencies = [
  "wasm-bindgen",
  "wasm-bindgen-futures",
  "web-sys",
- "webpki-roots",
+ "webpki-roots 1.0.7",
 ]
 
 [[package]]
@@ -2304,6 +2327,17 @@ dependencies = [
  "serde",
 ]
 
+[[package]]
+name = "sha1"
+version = "0.10.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e3bf829a2d51ab4a5ddf1352d8470c140cadc8301b2ae1789db023f01cedd6ba"
+dependencies = [
+ "cfg-if",
+ "cpufeatures 0.2.17",
+ "digest 0.10.7",
+]
+
 [[package]]
 name = "sha1"
 version = "0.11.0"
@@ -2496,7 +2530,8 @@ dependencies = [
  "tokio-stream",
  "tracing",
  "url",
- "webpki-roots",
+ "uuid",
+ "webpki-roots 1.0.7",
 ]
 
 [[package]]
@@ -2558,11 +2593,12 @@ dependencies = [
  "log",
  "percent-encoding",
  "serde",
- "sha1",
+ "sha1 0.11.0",
  "sha2 0.11.0",
  "sqlx-core",
  "thiserror",
  "tracing",
+ "uuid",
 ]
 
 [[package]]
@@ -2598,6 +2634,7 @@ dependencies = [
  "stringprep",
  "thiserror",
  "tracing",
+ "uuid",
  "whoami",
 ]
 
@@ -2624,6 +2661,7 @@ dependencies = [
  "thiserror",
  "tracing",
  "url",
+ "uuid",
 ]
 
 [[package]]
@@ -3046,6 +3084,34 @@ dependencies = [
  "tokio",
 ]
 
+[[package]]
+name = "tokio-tungstenite"
+version = "0.27.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "489a59b6730eda1b0171fcfda8b121f4bee2b35cba8645ca35c5f7ba3eb736c1"
+dependencies = [
+ "futures-util",
+ "log",
+ "rustls",
+ "rustls-pki-types",
+ "tokio",
+ "tokio-rustls",
+ "tungstenite 0.27.0",
+ "webpki-roots 0.26.11",
+]
+
+[[package]]
+name = "tokio-tungstenite"
+version = "0.29.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8f72a05e828585856dacd553fba484c242c46e391fb0e58917c942ee9202915c"
+dependencies = [
+ "futures-util",
+ "log",
+ "tokio",
+ "tungstenite 0.29.0",
+]
+
 [[package]]
 name = "tokio-util"
 version = "0.7.18"
@@ -3189,6 +3255,41 @@ version = "0.2.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b"
 
+[[package]]
+name = "tungstenite"
+version = "0.27.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "eadc29d668c91fcc564941132e17b28a7ceb2f3ebf0b9dae3e03fd7a6748eb0d"
+dependencies = [
+ "bytes",
+ "data-encoding",
+ "http",
+ "httparse",
+ "log",
+ "rand 0.9.4",
+ "rustls",
+ "rustls-pki-types",
+ "sha1 0.10.6",
+ "thiserror",
+ "utf-8",
+]
+
+[[package]]
+name = "tungstenite"
+version = "0.29.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6c01152af293afb9c7c2a57e4b559c5620b421f6d133261c60dd2d0cdb38e6b8"
+dependencies = [
+ "bytes",
+ "data-encoding",
+ "http",
+ "httparse",
+ "log",
+ "rand 0.9.4",
+ "sha1 0.10.6",
+ "thiserror",
+]
+
 [[package]]
 name = "typenum"
 version = "1.20.1"
@@ -3268,6 +3369,12 @@ dependencies = [
  "xmlwriter",
 ]
 
+[[package]]
+name = "utf-8"
+version = "0.7.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "09cc8ee72d2a9becf2f2febe0205bbed8fc6615b7cb429ad062dc7b7ddd036a9"
+
 [[package]]
 name = "utf8_iter"
 version = "1.0.4"
@@ -3309,6 +3416,7 @@ dependencies = [
  "quote",
  "regex",
  "syn",
+ "uuid",
 ]
 
 [[package]]
@@ -3331,6 +3439,7 @@ checksum = "ddd74a9687298c6858e9b88ec8935ec45d22e8fd5e6394fa1bd4e99a87789c76"
 dependencies = [
  "getrandom 0.4.2",
  "js-sys",
+ "serde_core",
  "wasm-bindgen",
 ]
 
@@ -3510,7 +3619,9 @@ dependencies = [
  "axum",
  "base64",
  "chrono",
+ "dashmap",
  "dotenvy",
+ "futures-util",
  "hmac 0.12.1",
  "jsonwebtoken",
  "p256",
@@ -3524,6 +3635,7 @@ dependencies = [
  "tempfile",
  "thiserror",
  "tokio",
+ "tokio-tungstenite 0.27.0",
  "tokio-util",
  "tower",
  "tower-http",
@@ -3532,6 +3644,7 @@ dependencies = [
  "utoipa",
  "utoipa-axum",
  "utoipa-scalar",
+ "uuid",
  "waveflow-core",
 ]
 
@@ -3555,6 +3668,15 @@ dependencies = [
  "wasm-bindgen",
 ]
 
+[[package]]
+name = "webpki-roots"
+version = "0.26.11"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "521bc38abb08001b01866da9f51eb7c5d647a19260e00054a8c7fd5f9e57f7a9"
+dependencies = [
+ "webpki-roots 1.0.7",
+]
+
 [[package]]
 name = "webpki-roots"
 version = "1.0.7"
diff --git a/Cargo.toml b/Cargo.toml
index 37cdffd..89ea97b 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -32,7 +32,7 @@ waveflow-core = { git = "https://github.com/InstaZDLL/WaveFlow", rev = "25b9ada6
 # mandate `sslmode=require` (Neon, Supabase, Prisma Accelerate, RDS).
 # Pure-Rust rustls matches the convention reqwest uses elsewhere in
 # the project — no OpenSSL system dep to ship in the container image.
-sqlx = { version = "0.9", default-features = false, features = ["runtime-tokio", "postgres", "macros", "migrate", "chrono", "tls-rustls"] }
+sqlx = { version = "0.9", default-features = false, features = ["runtime-tokio", "postgres", "macros", "migrate", "chrono", "uuid", "tls-rustls"] }
 
 # Wall-clock timestamps for `created_at` / `last_used_at`. Same
 # crate the desktop uses so a shared timestamp helper stays cheap.
@@ -44,16 +44,29 @@ chrono = "0.4"
 # without a parallel paths(...) list to maintain. `utoipa-scalar`
 # serves the Scalar API reference (Stripe-style — modern, ~500 KB
 # bundle, dark mode, integrated search) from `/reference`.
-utoipa = { version = "5", features = ["axum_extras"] }
+utoipa = { version = "5", features = ["axum_extras", "uuid"] }
 utoipa-axum = "0.2"
 utoipa-scalar = { version = "0.3", features = ["axum"] }
 
 # HTTP layer. axum 0.8 is the line current for Tokio 1.45+ — the same
 # generation used by the desktop's reqwest 0.12, so we're not pulling
 # two competing hyper trees once `waveflow-core` lands as a dependency
-# in 1.b.2.
-axum = "0.8"
-tokio = { version = "1", features = ["macros", "rt-multi-thread", "signal"] }
+# in 1.b.2. The `ws` feature gates the WebSocket extractor the
+# sync-broadcast endpoint (Phase 1.f) needs.
+axum = { version = "0.8", features = ["ws"] }
+tokio = { version = "1", features = ["macros", "rt-multi-thread", "signal", "sync", "time"] }
+# Stream combinators for the WebSocket handler — `SinkExt::send` +
+# `StreamExt::next` on the split halves of an `axum::extract::ws::WebSocket`.
+futures-util = { version = "0.3", default-features = false, features = ["sink", "std"] }
+# UUIDv4 for `sync_op.operation_id` — the idempotency key the client
+# generates and the server enforces uniqueness on per `(user_id,
+# device_id, operation_id)`. `v4` for the random generator, `serde`
+# for axum's `Json<…>` round-trip.
+uuid = { version = "1", features = ["v4", "serde"] }
+# Lock-free concurrent map for the in-memory ACK buffer in
+# `sync::SyncHub`. Per-key shard locks mean handler-side `record_ack`
+# never blocks behind the 5 s flusher.
+dashmap = "6"
 tower = "0.5"
 tower-http = { version = "0.6", features = ["trace", "request-id", "timeout", "util"] }
 
@@ -155,6 +168,21 @@ p256 = { version = "0.13", features = ["ecdsa", "pkcs8"] }
 # no cleanup boilerplate.
 tempfile = "3"
 
+# WebSocket client for the sync fan-out integration tests. The
+# tokio-tungstenite version that pairs with tokio 1.x ships only
+# rustls feature flags — `rustls-tls-webpki-roots` matches the
+# pure-Rust convention used elsewhere in the project.
+tokio-tungstenite = { version = "0.27", default-features = false, features = ["connect", "rustls-tls-webpki-roots"] }
+
+# `record_ack` takes a `uuid::Uuid`; the sync tests construct one per
+# op. Use the `serde` feature so the same value round-trips through
+# the JSON request body.
+uuid = { version = "1", features = ["v4", "serde"] }
+
+# Streaming tests already pulled `futures-util` transitively; the
+# sync WS test needs `StreamExt::next` + `SinkExt::send` directly.
+futures-util = { version = "0.3", default-features = false, features = ["sink", "std"] }
+
 [profile.release]
 # Match the desktop app's release profile so behaviour parity holds at
 # the binary level. Single-codegen-unit + LTO trade compile time for
diff --git a/migrations/20260601000000_sync_ops.sql b/migrations/20260601000000_sync_ops.sql
new file mode 100644
index 0000000..0143e35
--- /dev/null
+++ b/migrations/20260601000000_sync_ops.sql
@@ -0,0 +1,123 @@
+-- Multi-device sync log. Phase 1.f schema per RFC-001 §6.6.
+--
+-- The contract:
+--   - `sync_op` is an append-only log keyed on `BIGSERIAL id`, the
+--     single watermark every device pulls against. A per-(user, device)
+--     `operation_id` UUID is the idempotency key so a client retry of
+--     a successfully-pushed batch yields the same `id` instead of a
+--     duplicate row, and a per-(user, device) `lamport_ts` enforces
+--     ordering so a delayed-then-late replay is rejected as a regression
+--     instead of silently re-ordering history.
+--
+--   - `device_sync_cursor` tracks how far each device has confirmed
+--     reading. Only the explicit `POST /sync/ack` path writes here —
+--     the `GET /sync/ops` pull is read-only on purpose, so a client
+--     can re-pull without committing to having processed the rows.
+--
+--   - `sync_compaction_watermark` is the "below this id, history has
+--     been collapsed" floor. The compaction job updates it in the same
+--     transaction as the row deletes; a client requesting `since < N`
+--     where N = compacted_up_to gets 410 Gone (resurrected-device case).
+--
+-- All three tables ON DELETE CASCADE via `user_id` so a user delete
+-- fan-outs cleanly. No cross-table FKs beyond `users(id)` — the log
+-- is intentionally decoupled from the entity tables (`playlist`,
+-- `library`, etc.) so server-side schema evolution doesn't require
+-- migrating historic ops.
+
+CREATE TABLE sync_op (
+    id              BIGSERIAL   PRIMARY KEY,
+    user_id         BIGINT      NOT NULL
+                    REFERENCES users(id) ON DELETE CASCADE,
+
+    -- Stable identifier of the originating client. Free-form TEXT so
+    -- the desktop can use a UUID generated at first launch, the web
+    -- can use a per-tab UUID, and the test harness can use friendly
+    -- names like "device-a" without round-tripping through a uuid
+    -- parser. The (user_id, device_id) pair is what makes a Lamport
+    -- clock meaningful — two unrelated devices can use the same
+    -- lamport sequence without colliding.
+    device_id       TEXT        NOT NULL,
+
+    -- Client-generated UUID, the idempotency key. A POST replaying a
+    -- previously-accepted op with the same `operation_id` short-
+    -- circuits via `ON CONFLICT (user_id, device_id, operation_id)
+    -- DO NOTHING` and returns the existing row, so a network blip
+    -- mid-push never inflates the log.
+    operation_id    UUID        NOT NULL,
+
+    -- Per-(user, device) Lamport clock. The unique constraint below
+    -- makes monotonicity inviolable at the storage layer — a stale
+    -- replay attempting to land at an already-used lamport position
+    -- surfaces as a 23505 unique violation, which the handler maps
+    -- to 409 + the stored max so the client can resync its clock.
+    lamport_ts      BIGINT      NOT NULL,
+
+    -- Op shape. `entity` names the type ("playlist", "library", …);
+    -- `entity_id` is a TEXT to accommodate cross-type id schemes (the
+    -- desktop uses BIGINT ids for some tables and UUID-ish strings
+    -- for others). `field` is NULL on whole-entity ops (insert /
+    -- delete) and named on partial updates ("set name", "set color").
+    entity          TEXT        NOT NULL,
+    entity_id       TEXT        NOT NULL,
+    field           TEXT,
+    op              TEXT        NOT NULL,
+    payload         JSONB,
+
+    -- ms-epoch wall-clock the server recorded the op at. Used by
+    -- operators to correlate the log with external timelines; the
+    -- compaction job uses `sync_op.id` (not `created_at`) as the
+    -- horizon so a backwards-running wall clock never breaks the
+    -- monotonic-by-id invariant.
+    created_at      BIGINT      NOT NULL,
+
+    UNIQUE (user_id, device_id, operation_id),
+    UNIQUE (user_id, device_id, lamport_ts)
+);
+
+-- The pull path is `WHERE user_id = $1 AND id > $since ORDER BY id`.
+-- The (user_id, id) composite covers both the equality + range; a
+-- user_id-only index would still work but would force a heap scan
+-- across the matching range for the id sort.
+CREATE INDEX sync_op_user_id_idx ON sync_op (user_id, id);
+
+CREATE TABLE device_sync_cursor (
+    user_id         BIGINT      NOT NULL
+                    REFERENCES users(id) ON DELETE CASCADE,
+    device_id       TEXT        NOT NULL,
+
+    -- Highest `sync_op.id` this device has confirmed processing.
+    -- Compaction MIN reads across every active device for the user;
+    -- a single lagging device pins the floor for the whole tenant,
+    -- which is intentional — we'd rather keep history than drop ops
+    -- a device still needs.
+    last_seen_id    BIGINT      NOT NULL,
+
+    -- Wall-clock of the last ACK. Drives the stale-device cutoff so
+    -- a device that hasn't talked to the server in > 90 days stops
+    -- pinning the compaction floor (resurrected-device guard then
+    -- handles the rejoin via 410 Gone).
+    last_seen_at    BIGINT      NOT NULL,
+
+    PRIMARY KEY (user_id, device_id)
+);
+
+-- Index for the compaction MIN query — `MIN(last_seen_id)` over
+-- `WHERE user_id = $1 AND last_seen_at >= $stale_threshold`. The
+-- column order matches the predicate shape (eq → range → projected
+-- column) so the index can answer the query without a heap visit.
+CREATE INDEX device_sync_cursor_user_recent_idx
+    ON device_sync_cursor (user_id, last_seen_at, last_seen_id);
+
+CREATE TABLE sync_compaction_watermark (
+    user_id         BIGINT      PRIMARY KEY
+                    REFERENCES users(id) ON DELETE CASCADE,
+
+    -- All ops with `id <= compacted_up_to` have been collapsed; pulls
+    -- starting `since < compacted_up_to` MUST 410 Gone instead of
+    -- returning a partial history that would let the client converge
+    -- on a state inconsistent with peers. Monotonically increasing;
+    -- the compaction UPSERT refuses to lower it.
+    compacted_up_to BIGINT      NOT NULL DEFAULT 0,
+    updated_at      BIGINT      NOT NULL
+);
diff --git a/src/api/mod.rs b/src/api/mod.rs
index a7d306d..63a5124 100644
--- a/src/api/mod.rs
+++ b/src/api/mod.rs
@@ -35,6 +35,7 @@ mod playlists;
 mod profiles;
 mod ready;
 mod stream;
+mod sync;
 mod tracks;
 
 /// Combined router for every API module. Mounted at the root by
@@ -53,6 +54,7 @@ pub fn router(state: AppState) -> OpenApiRouter {
     let libraries_router = libraries::router(state.clone()).layer(auth_layer.clone());
     let tracks_router = tracks::router(state.clone()).layer(auth_layer.clone());
     let playlists_router = playlists::router(state.clone()).layer(auth_layer.clone());
+    let sync_router = sync::router(state.clone()).layer(auth_layer.clone());
     // Mint stays JWT-authed (verifies tenant ownership before signing).
     let stream_mint_router = stream::auth_router(state.clone()).layer(auth_layer);
     // The stream endpoint itself is HMAC-authed by the token in the
@@ -68,6 +70,7 @@ pub fn router(state: AppState) -> OpenApiRouter {
         .merge(libraries_router)
         .merge(tracks_router)
         .merge(playlists_router)
+        .merge(sync_router)
         .merge(stream_mint_router)
         .merge(stream_public_router)
 }
diff --git a/src/api/sync.rs b/src/api/sync.rs
new file mode 100644
index 0000000..97a200e
--- /dev/null
+++ b/src/api/sync.rs
@@ -0,0 +1,518 @@
+//! `/api/v1/sync/*` — multi-device sync surface per RFC-001 §6.6.
+//!
+//! Three REST routes + one WebSocket:
+//!
+//! - `POST /api/v1/sync/ops` — push a batch. Per-op idempotency on
+//!   `operation_id`; per-device monotonicity on `lamport_ts`. New rows
+//!   broadcast to every WebSocket subscriber owned by the same user.
+//! - `GET  /api/v1/sync/ops?since=N` — pull rows with `id > N`.
+//!   Resurrected-device guard: a `since` below the compaction
+//!   watermark returns `410 Gone` so the client knows to resync from
+//!   scratch instead of converging on a half-collapsed view.
+//! - `POST /api/v1/sync/ack` — the **only** path that advances the
+//!   per-device cursor. Body is buffered in memory and UPSERTed every
+//!   five seconds (or synchronously by the compaction job before the
+//!   MIN read).
+//! - `GET  /api/v1/sync/ws?device_id=…` — WebSocket. Server pushes
+//!   `{"type":"op","op":{…}}` envelopes for the authenticated user;
+//!   client sends `{"ack": N}` frames to advance its cursor. Disconnect
+//!   triggers a synchronous flush of that device's buffered ACK so a
+//!   browser tab close doesn't lose the position.
+
+use axum::{
+    extract::{
+        ws::{Message, WebSocket, WebSocketUpgrade},
+        Extension, Query, State,
+    },
+    http::StatusCode,
+    response::IntoResponse,
+    Json,
+};
+use chrono::Utc;
+use futures_util::{SinkExt, StreamExt};
+use serde::{Deserialize, Serialize};
+use sqlx::{postgres::PgRow, Row};
+use tokio::sync::broadcast::error::RecvError;
+use utoipa::ToSchema;
+use utoipa_axum::{router::OpenApiRouter, routes};
+use uuid::Uuid;
+
+use crate::{
+    db,
+    middleware::UserId,
+    sync::{build_broadcast, SyncOp, SyncOpIn},
+    AppState,
+};
+
+/// Max ops per batch. A hard cap on the request body keeps a buggy or
+/// hostile client from holding a transaction open across a million-row
+/// insert. The desktop's scanner pipeline commits every 200 rows; 1024
+/// is comfortable headroom for a coalesced flush.
+const MAX_BATCH_SIZE: usize = 1024;
+
+/// Max rows per pull response. 1024 matches the push cap; clients
+/// loop on `since = last_id` until they receive a short page.
+const PULL_PAGE_SIZE: i64 = 1024;
+
+#[derive(Debug, Deserialize, ToSchema)]
+pub struct PushBatchRequest {
+    /// Originating device. Same value the WS handshake uses on
+    /// `?device_id=…`.
+    pub device_id: String,
+    pub ops: Vec<SyncOpIn>,
+}
+
+#[derive(Debug, Serialize, ToSchema)]
+pub struct PushBatchResponse {
+    /// Every op the server now considers durably stored — both fresh
+    /// inserts AND duplicate replays of previously-accepted rows. The
+    /// client uses this to confirm what landed; a missing op means
+    /// the batch was aborted (409 / 500).
+    pub accepted: Vec<SyncOp>,
+}
+
+#[derive(Debug, Serialize, ToSchema)]
+pub struct LamportRegression {
+    pub error: &'static str,
+    pub device_id: String,
+    pub stored_max: i64,
+    pub offending_lamport_ts: i64,
+}
+
+#[derive(Debug, Deserialize, ToSchema, utoipa::IntoParams)]
+pub struct PullQuery {
+    /// Last `sync_op.id` the client has confirmed seeing. `0` (or
+    /// omitted) means "send me everything from the start".
+    #[serde(default)]
+    pub since: i64,
+}
+
+#[derive(Debug, Serialize, ToSchema)]
+pub struct PullResponse {
+    pub ops: Vec<SyncOp>,
+    /// Highest `id` in this batch — convenience so the client doesn't
+    /// have to scan `ops` to compute the next `since`. Equals `since`
+    /// when the page is empty.
+    pub last_id: i64,
+}
+
+#[derive(Debug, Serialize, ToSchema)]
+pub struct ResurrectedDeviceGone {
+    pub error: &'static str,
+    pub compacted_up_to: i64,
+}
+
+#[derive(Debug, Deserialize, ToSchema)]
+pub struct AckRequest {
+    pub device_id: String,
+    pub last_seen_id: i64,
+}
+
+#[derive(Debug, Deserialize)]
+struct WsQuery {
+    device_id: String,
+}
+
+pub fn router(state: AppState) -> OpenApiRouter {
+    OpenApiRouter::new()
+        .routes(routes!(push_ops, pull_ops))
+        .routes(routes!(ack_ops))
+        .route("/api/v1/sync/ws", axum::routing::get(ws_upgrade))
+        .with_state(state)
+}
+
+/// Push a batch of ops. Per-op semantics:
+///
+/// - `ON CONFLICT (user_id, device_id, operation_id) DO NOTHING`
+///   absorbs idempotent replays.
+/// - The `(user_id, device_id, lamport_ts)` UNIQUE bubbles up as
+///   sqlx `23505` on a regression — caught + reported as 409.
+///
+/// All ops land in a single transaction; broadcast fires after the
+/// commit so a subscriber never sees an op that gets rolled back.
+#[utoipa::path(
+    post,
+    path = "/api/v1/sync/ops",
+    tag = "sync",
+    params(
+        ("authorization" = String, Header, description = "Bearer JWT issued by Better Auth"),
+    ),
+    request_body = PushBatchRequest,
+    responses(
+        (status = 200, description = "Batch accepted (one entry per op, fresh or dup)", body = PushBatchResponse),
+        (status = 400, description = "Empty `device_id`, oversized batch, or malformed op"),
+        (status = 401, description = "Missing or invalid bearer token"),
+        (status = 409, description = "Lamport regression — stored max returned in body", body = LamportRegression),
+        (status = 500, description = "Database or internal failure"),
+    ),
+)]
+async fn push_ops(
+    State(state): State<AppState>,
+    Extension(UserId(user_id)): Extension<UserId>,
+    Json(req): Json<PushBatchRequest>,
+) -> impl IntoResponse {
+    let device_id = req.device_id.trim();
+    if device_id.is_empty() {
+        return (StatusCode::BAD_REQUEST, "device_id is required").into_response();
+    }
+    if req.ops.is_empty() {
+        return (StatusCode::OK, Json(PushBatchResponse { accepted: vec![] })).into_response();
+    }
+    if req.ops.len() > MAX_BATCH_SIZE {
+        return (
+            StatusCode::BAD_REQUEST,
+            format!("batch exceeds {MAX_BATCH_SIZE} ops"),
+        )
+            .into_response();
+    }
+
+    let now = Utc::now().timestamp_millis();
+    let pool = state.sync.pool();
+
+    let mut tx = match pool.begin().await {
+        Ok(tx) => tx,
+        Err(err) => {
+            tracing::error!(error = %err, user_id, "sync push begin tx failed");
+            return (StatusCode::INTERNAL_SERVER_ERROR, "tx begin failed").into_response();
+        }
+    };
+
+    let mut accepted: Vec<SyncOp> = Vec::with_capacity(req.ops.len());
+    let mut freshly_inserted: Vec<SyncOp> = Vec::new();
+
+    for op_in in &req.ops {
+        let entity = op_in.entity.trim();
+        let entity_id = op_in.entity_id.trim();
+        let op_kind = op_in.op.trim();
+        if entity.is_empty() || entity_id.is_empty() || op_kind.is_empty() {
+            tx.rollback().await.ok();
+            return (
+                StatusCode::BAD_REQUEST,
+                "entity / entity_id / op are required",
+            )
+                .into_response();
+        }
+
+        let insert_res = db::sync::insert_op_returning(
+            &mut tx,
+            user_id,
+            device_id,
+            op_in.operation_id,
+            op_in.lamport_ts,
+            entity,
+            entity_id,
+            op_in.field.as_deref(),
+            op_kind,
+            op_in.payload.as_ref(),
+            now,
+        )
+        .await;
+
+        match insert_res {
+            Ok(Some(row)) => {
+                let op = row_to_op(&row);
+                accepted.push(op.clone());
+                freshly_inserted.push(op);
+            }
+            Ok(None) => {
+                // Duplicate `operation_id` — fold in the previously-
+                // stored row so the client's `accepted` list is whole.
+                match db::sync::fetch_op_by_operation_id(
+                    &mut tx,
+                    user_id,
+                    device_id,
+                    op_in.operation_id,
+                )
+                .await
+                {
+                    Ok(row) => accepted.push(row_to_op(&row)),
+                    Err(err) => {
+                        tracing::error!(error = %err, user_id, device_id, "sync dup lookup failed");
+                        tx.rollback().await.ok();
+                        return (StatusCode::INTERNAL_SERVER_ERROR, "dup lookup failed")
+                            .into_response();
+                    }
+                }
+            }
+            Err(sqlx::Error::Database(db_err)) if db_err.code().as_deref() == Some("23505") => {
+                // Unique violation that wasn't on `operation_id`
+                // (that branch is absorbed by the ON CONFLICT) — so it
+                // can only be the `(user_id, device_id, lamport_ts)`
+                // constraint, i.e. a regression. Read the current max
+                // and return it so the client can resync its clock.
+                tx.rollback().await.ok();
+                let stored_max = match db::sync::lamport_max(pool, user_id, device_id).await {
+                    Ok(n) => n,
+                    Err(err) => {
+                        // Surface the read failure rather than masking
+                        // it behind a `0` that would tell the client
+                        // "your clock is fine, retry" when it isn't.
+                        tracing::error!(error = %err, user_id, device_id, "lamport_max read failed");
+                        return (StatusCode::INTERNAL_SERVER_ERROR, "lamport_max read failed")
+                            .into_response();
+                    }
+                };
+                return (
+                    StatusCode::CONFLICT,
+                    Json(LamportRegression {
+                        error: "lamport_regression",
+                        device_id: device_id.to_string(),
+                        stored_max,
+                        offending_lamport_ts: op_in.lamport_ts,
+                    }),
+                )
+                    .into_response();
+            }
+            Err(err) => {
+                tracing::error!(error = %err, user_id, device_id, "sync insert failed");
+                tx.rollback().await.ok();
+                return (StatusCode::INTERNAL_SERVER_ERROR, "insert failed").into_response();
+            }
+        }
+    }
+
+    if let Err(err) = tx.commit().await {
+        tracing::error!(error = %err, user_id, "sync push commit failed");
+        return (StatusCode::INTERNAL_SERVER_ERROR, "commit failed").into_response();
+    }
+
+    // Broadcast outside the transaction — subscribers must never
+    // observe an op that rolls back. The pre-serialised `Arc<String>`
+    // means each subscriber gets a cheap clone.
+    for op in &freshly_inserted {
+        state.sync.broadcast(build_broadcast(user_id, op));
+    }
+
+    (StatusCode::OK, Json(PushBatchResponse { accepted })).into_response()
+}
+
+/// Pull ops with `id > since`. Read-only — never advances the device
+/// cursor (that's `POST /sync/ack`'s sole job). Resurrected-device
+/// guard: any `since` below the compaction watermark returns 410.
+#[utoipa::path(
+    get,
+    path = "/api/v1/sync/ops",
+    tag = "sync",
+    params(
+        ("authorization" = String, Header, description = "Bearer JWT issued by Better Auth"),
+        PullQuery,
+    ),
+    responses(
+        (status = 200, description = "Page of ops, capped at 1024", body = PullResponse),
+        (status = 401, description = "Missing or invalid bearer token"),
+        (status = 410, description = "since < compacted_up_to — client must full-resync", body = ResurrectedDeviceGone),
+        (status = 500, description = "Database or internal failure"),
+    ),
+)]
+async fn pull_ops(
+    State(state): State<AppState>,
+    Extension(UserId(user_id)): Extension<UserId>,
+    Query(query): Query<PullQuery>,
+) -> impl IntoResponse {
+    if query.since < 0 {
+        return (StatusCode::BAD_REQUEST, "since must be >= 0").into_response();
+    }
+    let pool = state.sync.pool();
+
+    // Resurrected-device guard. `since == 0` ("send me everything")
+    // is always allowed — the compaction watermark only matters when
+    // the client is claiming a position partway up the log. The
+    // `compacted_up_to == 0` row (never compacted) is also a no-op
+    // guard, so the condition is symmetric: `since > 0 AND since <
+    // compacted_up_to`.
+    if query.since > 0 {
+        // Propagate a watermark read failure as 500 instead of
+        // silently dropping it into `None` — masking a transient
+        // pool / Postgres hiccup as "no compaction floor" could let a
+        // resurrected device skip the 410 path and replay a partial
+        // log into a state inconsistent with its peers.
+        let compacted = match db::sync::fetch_compacted_up_to(pool, user_id).await {
+            Ok(c) => c,
+            Err(err) => {
+                tracing::error!(error = %err, user_id, "watermark read failed");
+                return (StatusCode::INTERNAL_SERVER_ERROR, "watermark read failed")
+                    .into_response();
+            }
+        };
+        if let Some(watermark) = compacted {
+            if query.since < watermark {
+                return (
+                    StatusCode::GONE,
+                    Json(ResurrectedDeviceGone {
+                        error: "resync_required",
+                        compacted_up_to: watermark,
+                    }),
+                )
+                    .into_response();
+            }
+        }
+    }
+
+    let rows = db::sync::pull_ops_since(pool, user_id, query.since, PULL_PAGE_SIZE).await;
+
+    match rows {
+        Ok(rows) => {
+            let ops: Vec<SyncOp> = rows.iter().map(row_to_op).collect();
+            let last_id = ops.last().map(|o| o.id).unwrap_or(query.since);
+            (StatusCode::OK, Json(PullResponse { ops, last_id })).into_response()
+        }
+        Err(err) => {
+            tracing::error!(error = %err, user_id, "sync pull failed");
+            (StatusCode::INTERNAL_SERVER_ERROR, "pull failed").into_response()
+        }
+    }
+}
+
+/// Record an ACK. Goes into the in-memory buffer; the flush task
+/// UPSERTs the row within `flush_interval` (default 5 s). The
+/// compaction job synchronously flushes the buffer before reading the
+/// MIN, so a fresh ACK never gets stranded in memory at compaction
+/// time.
+#[utoipa::path(
+    post,
+    path = "/api/v1/sync/ack",
+    tag = "sync",
+    params(
+        ("authorization" = String, Header, description = "Bearer JWT issued by Better Auth"),
+    ),
+    request_body = AckRequest,
+    responses(
+        (status = 204, description = "ACK buffered"),
+        (status = 400, description = "Empty `device_id` or negative `last_seen_id`"),
+        (status = 401, description = "Missing or invalid bearer token"),
+    ),
+)]
+async fn ack_ops(
+    State(state): State<AppState>,
+    Extension(UserId(user_id)): Extension<UserId>,
+    Json(req): Json<AckRequest>,
+) -> impl IntoResponse {
+    let device_id = req.device_id.trim();
+    if device_id.is_empty() {
+        return (StatusCode::BAD_REQUEST, "device_id is required").into_response();
+    }
+    if req.last_seen_id < 0 {
+        return (StatusCode::BAD_REQUEST, "last_seen_id must be >= 0").into_response();
+    }
+    state.sync.record_ack(
+        user_id,
+        device_id,
+        req.last_seen_id,
+        Utc::now().timestamp_millis(),
+    );
+    StatusCode::NO_CONTENT.into_response()
+}
+
+/// WebSocket upgrade. The `device_id` rides in the query string so the
+/// socket task knows who's talking from the first byte — an in-band
+/// "hello" frame would race with the first broadcast.
+async fn ws_upgrade(
+    ws: WebSocketUpgrade,
+    State(state): State<AppState>,
+    Extension(UserId(user_id)): Extension<UserId>,
+    Query(params): Query<WsQuery>,
+) -> impl IntoResponse {
+    let device_id = params.device_id.trim().to_string();
+    if device_id.is_empty() {
+        return (StatusCode::BAD_REQUEST, "device_id is required").into_response();
+    }
+    ws.on_upgrade(move |socket| handle_socket(socket, state, user_id, device_id))
+        .into_response()
+}
+
+#[derive(Debug, Deserialize)]
+struct InboundAck {
+    ack: i64,
+}
+
+async fn handle_socket(socket: WebSocket, state: AppState, user_id: i64, device_id: String) {
+    let (mut sender, mut receiver) = socket.split();
+    let mut broadcast_rx = state.sync.subscribe();
+
+    loop {
+        tokio::select! {
+            // Inbound frame from the client. We only understand
+            // `{"ack": N}` today; anything else is logged and dropped
+            // so a future protocol extension doesn't break old
+            // clients.
+            inbound = receiver.next() => {
+                match inbound {
+                    Some(Ok(Message::Text(text))) => {
+                        if let Ok(ack) = serde_json::from_str::<InboundAck>(&text) {
+                            if ack.ack >= 0 {
+                                state.sync.record_ack(
+                                    user_id,
+                                    &device_id,
+                                    ack.ack,
+                                    Utc::now().timestamp_millis(),
+                                );
+                            }
+                        }
+                    }
+                    Some(Ok(Message::Close(_))) | None => break,
+                    Some(Ok(_)) => { /* ignore non-text frames */ }
+                    Some(Err(err)) => {
+                        tracing::debug!(error = %err, user_id, device_id, "ws inbound error");
+                        break;
+                    }
+                }
+            }
+            // Outbound broadcast. Filter to this user before sending —
+            // a cross-tenant frame must never leave the socket.
+            recv = broadcast_rx.recv() => {
+                match recv {
+                    Ok(payload) => {
+                        if payload.user_id == user_id {
+                            // Send the pre-serialised frame. A failure
+                            // here means the socket died; bail and let
+                            // the on-disconnect flush run.
+                            if sender
+                                .send(Message::Text((*payload.frame).clone().into()))
+                                .await
+                                .is_err()
+                            {
+                                break;
+                            }
+                        }
+                    }
+                    Err(RecvError::Closed) => break,
+                    Err(RecvError::Lagged(skipped)) => {
+                        // The subscriber fell behind the broadcast
+                        // buffer. The cleanest recovery is to drop the
+                        // socket and let the client pull via REST +
+                        // reconnect — they have `since = last_seen_id`
+                        // to resume from.
+                        tracing::warn!(
+                            user_id,
+                            device_id,
+                            skipped,
+                            "ws broadcast lagged, closing socket",
+                        );
+                        break;
+                    }
+                }
+            }
+        }
+    }
+
+    // On disconnect, flush this device's ACK synchronously so a
+    // crash on the next compaction tick doesn't over-retain.
+    if let Err(err) = state.sync.flush_acks().await {
+        tracing::warn!(error = %err, user_id, device_id, "ws teardown flush failed");
+    }
+}
+
+fn row_to_op(row: &PgRow) -> SyncOp {
+    SyncOp {
+        id: row.get("id"),
+        operation_id: row.get::<Uuid, _>("operation_id"),
+        device_id: row.get("device_id"),
+        lamport_ts: row.get("lamport_ts"),
+        entity: row.get("entity"),
+        entity_id: row.get("entity_id"),
+        field: row.get("field"),
+        op: row.get("op"),
+        payload: row.get::<Option<serde_json::Value>, _>("payload"),
+        created_at: row.get("created_at"),
+    }
+}
diff --git a/src/db.rs b/src/db.rs
index 47b4bba..26d31d1 100644
--- a/src/db.rs
+++ b/src/db.rs
@@ -74,6 +74,142 @@ pub async fn ping(pool: &PgPool) -> Result<(), sqlx::Error> {
         .map(|_| ())
 }
 
+/// Sync log helpers. Lives here for the same reason as [`users`]: the
+/// handlers in `api/sync.rs` should stay pure HTTP orchestration, so
+/// every `INSERT` / `SELECT` against `sync_op` /
+/// `sync_compaction_watermark` lands on a function in this module.
+///
+/// Tx-aware functions accept `&mut sqlx::PgConnection` (= `&mut *tx`
+/// for an open transaction) so the batch handler can keep its
+/// transaction across N inserts. Single-statement helpers take
+/// `&PgPool` since they don't compose with a transaction.
+pub mod sync {
+    use sqlx::{postgres::PgRow, PgConnection, PgPool};
+    use uuid::Uuid;
+
+    /// Append one op. Returns the inserted row, or `None` when the
+    /// `(user_id, device_id, operation_id)` UNIQUE absorbed an
+    /// idempotent replay. The `(user_id, device_id, lamport_ts)`
+    /// UNIQUE is *not* covered by `ON CONFLICT` — a violation there
+    /// bubbles up as a `sqlx::Error::Database` with SQLSTATE 23505
+    /// for the caller to map to its 409 path.
+    #[allow(clippy::too_many_arguments)]
+    pub async fn insert_op_returning(
+        conn: &mut PgConnection,
+        user_id: i64,
+        device_id: &str,
+        operation_id: Uuid,
+        lamport_ts: i64,
+        entity: &str,
+        entity_id: &str,
+        field: Option<&str>,
+        op: &str,
+        payload: Option<&serde_json::Value>,
+        created_at: i64,
+    ) -> Result<Option<PgRow>, sqlx::Error> {
+        sqlx::query(
+            "INSERT INTO sync_op \
+                (user_id, device_id, operation_id, lamport_ts, entity, entity_id, field, op, payload, created_at) \
+             VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10) \
+             ON CONFLICT (user_id, device_id, operation_id) DO NOTHING \
+             RETURNING id, operation_id, device_id, lamport_ts, entity, entity_id, field, op, payload, created_at",
+        )
+        .bind(user_id)
+        .bind(device_id)
+        .bind(operation_id)
+        .bind(lamport_ts)
+        .bind(entity)
+        .bind(entity_id)
+        .bind(field)
+        .bind(op)
+        .bind(payload)
+        .bind(created_at)
+        .fetch_optional(conn)
+        .await
+    }
+
+    /// Fetch the row matching a previously-accepted `operation_id`.
+    /// Caller has already confirmed the row exists via the
+    /// `ON CONFLICT DO NOTHING` returning `None`, so this is a plain
+    /// `fetch_one`.
+    pub async fn fetch_op_by_operation_id(
+        conn: &mut PgConnection,
+        user_id: i64,
+        device_id: &str,
+        operation_id: Uuid,
+    ) -> Result<PgRow, sqlx::Error> {
+        sqlx::query(
+            "SELECT id, operation_id, device_id, lamport_ts, entity, entity_id, field, op, payload, created_at \
+             FROM sync_op \
+             WHERE user_id = $1 AND device_id = $2 AND operation_id = $3",
+        )
+        .bind(user_id)
+        .bind(device_id)
+        .bind(operation_id)
+        .fetch_one(conn)
+        .await
+    }
+
+    /// Current `MAX(lamport_ts)` for a device. Returns `0` when the
+    /// device has no rows yet. Used after a lamport-regression 23505
+    /// to tell the client how far ahead the server is.
+    pub async fn lamport_max(
+        pool: &PgPool,
+        user_id: i64,
+        device_id: &str,
+    ) -> Result<i64, sqlx::Error> {
+        sqlx::query_scalar(
+            "SELECT COALESCE(MAX(lamport_ts), 0) FROM sync_op \
+             WHERE user_id = $1 AND device_id = $2",
+        )
+        .bind(user_id)
+        .bind(device_id)
+        .fetch_one(pool)
+        .await
+    }
+
+    /// Read the compaction watermark for a user. `None` means the
+    /// compaction job hasn't touched this tenant yet (no row), which
+    /// the pull guard treats as "no floor". A transport / pool error
+    /// is propagated — silently treating it as `None` would let a
+    /// resurrected-device case slip through during a DB hiccup.
+    pub async fn fetch_compacted_up_to(
+        pool: &PgPool,
+        user_id: i64,
+    ) -> Result<Option<i64>, sqlx::Error> {
+        sqlx::query_scalar(
+            "SELECT compacted_up_to FROM sync_compaction_watermark \
+             WHERE user_id = $1",
+        )
+        .bind(user_id)
+        .fetch_optional(pool)
+        .await
+    }
+
+    /// Fetch the next page of ops with `id > since`, capped at
+    /// `limit`, ordered ascending so the client can stream straight
+    /// into its local replay.
+    pub async fn pull_ops_since(
+        pool: &PgPool,
+        user_id: i64,
+        since: i64,
+        limit: i64,
+    ) -> Result<Vec<PgRow>, sqlx::Error> {
+        sqlx::query(
+            "SELECT id, operation_id, device_id, lamport_ts, entity, entity_id, field, op, payload, created_at \
+             FROM sync_op \
+             WHERE user_id = $1 AND id > $2 \
+             ORDER BY id ASC \
+             LIMIT $3",
+        )
+        .bind(user_id)
+        .bind(since)
+        .bind(limit)
+        .fetch_all(pool)
+        .await
+    }
+}
+
 /// User-table helpers. Keeps the raw SQL out of handlers — same
 /// boundary the project's no-SQL-in-handlers rule enforces for the
 /// `/ready` probe.
diff --git a/src/lib.rs b/src/lib.rs
index 2a875e9..471982a 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -28,6 +28,7 @@ pub mod config;
 pub mod db;
 pub mod middleware;
 pub mod stream_token;
+pub mod sync;
 
 pub use config::Config;
 
@@ -75,6 +76,11 @@ pub struct AppState {
     /// boot; `None` disables both the mint and the stream endpoints
     /// (they answer 503).
     pub stream_ctx: Option<std::sync::Arc<StreamCtx>>,
+    /// Multi-device sync hub (Phase 1.f). Owns the broadcast channel
+    /// every WebSocket subscribes to, the in-memory ACK buffer the
+    /// 5 s flusher drains, and the daily compaction task. Cheap to
+    /// clone — every inner field is `Arc`-backed.
+    pub sync: sync::SyncHub,
 }
 
 /// Per-process streaming context built once at boot. Wraps the HMAC
diff --git a/src/main.rs b/src/main.rs
index 7257ab2..7222045 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -18,7 +18,9 @@ use waveflow_server::{
     app,
     auth::{JwtVerifier, JwtVerifierConfig},
     config::Config,
-    db, AppState, StreamCtx,
+    db,
+    sync::{SyncHub, DEFAULT_COMPACTION_INTERVAL, DEFAULT_FLUSH_INTERVAL},
+    AppState, StreamCtx,
 };
 
 #[tokio::main]
@@ -78,10 +80,29 @@ async fn main() -> anyhow::Result<()> {
     let local = listener.local_addr()?;
     info!(addr = %local, "waveflow-server listening");
 
+    // Sync hub — broadcast channel + ACK debouncer + nightly
+    // compaction. The two `JoinHandle`s are intentionally retained as
+    // `_flush_task` / `_compaction_task` so the underscored bindings
+    // keep ownership through `main`'s scope — dropping a handle
+    // doesn't cancel the task itself, but rebinding-by-shadowing or
+    // explicit `drop` would, and a future refactor that loses the
+    // bindings should hear from the borrow checker on the next read.
+    let (sync_hub, _flush_task, _compaction_task) = SyncHub::spawn(
+        db.clone(),
+        DEFAULT_FLUSH_INTERVAL,
+        DEFAULT_COMPACTION_INTERVAL,
+    );
+    info!(
+        flush_interval = ?DEFAULT_FLUSH_INTERVAL,
+        compaction_interval = ?DEFAULT_COMPACTION_INTERVAL,
+        "sync hub started",
+    );
+
     let state = AppState {
         db,
         jwt_verifier,
         stream_ctx,
+        sync: sync_hub,
     };
     axum::serve(
         listener,
diff --git a/src/sync.rs b/src/sync.rs
new file mode 100644
index 0000000..2055192
--- /dev/null
+++ b/src/sync.rs
@@ -0,0 +1,515 @@
+//! Multi-device sync hub. Phase 1.f per RFC-001 §6.6.
+//!
+//! Three responsibilities live here:
+//!
+//! 1. **Broadcast fan-out** — every successful `POST /api/v1/sync/ops`
+//!    pushes its newly-assigned rows onto a single tokio
+//!    `broadcast::Sender`. WebSocket handlers subscribe at connect and
+//!    filter on `user_id` so cross-tenant ops never reach a foreign
+//!    socket. One channel rather than per-user channels keeps the
+//!    state structure trivial — the per-receive `user_id` equality
+//!    check is cheap compared to the lock + map lookup a per-user
+//!    `DashMap<i64, Sender>` would need on every emit.
+//!
+//! 2. **ACK debouncing** — `POST /api/v1/sync/ack` (and WebSocket
+//!    `{"ack": N}` frames) write into an in-memory `AckBuffer` keyed
+//!    on `(user_id, device_id)`. A background task wakes every
+//!    `flush_interval` (default 5 s) and flushes every dirty entry to
+//!    `device_sync_cursor` in a single transaction. The compaction job
+//!    flushes the buffer synchronously before it reads — so the MIN
+//!    it computes always reflects the latest ACK every device sent,
+//!    closing the "compaction MIN reads a stale Postgres row" race
+//!    the RFC calls out as the dangerous failure mode.
+//!
+//! 3. **Compaction** — a daily tokio task collapses superseded ops
+//!    `WHERE id <= MIN(last_seen_id)` (excluding stale devices). The
+//!    delete and the `sync_compaction_watermark` update happen in the
+//!    same Postgres transaction so a kill mid-compaction leaves either
+//!    the old state OR the new state, never one without the other.
+//!
+//! Boot is in [`SyncHub::spawn`], which returns the hub itself plus the
+//! `JoinHandle`s the binary needs to keep alive. Tests can build a
+//! hub via [`SyncHub::for_tests`] which skips the background tasks so
+//! the test stays deterministic.
+
+use std::{
+    sync::{
+        atomic::{AtomicBool, AtomicI64, Ordering},
+        Arc,
+    },
+    time::Duration,
+};
+
+use chrono::Utc;
+use dashmap::DashMap;
+use serde::{Deserialize, Serialize};
+use sqlx::PgPool;
+use tokio::{
+    sync::broadcast,
+    task::JoinHandle,
+    time::{interval, MissedTickBehavior},
+};
+use utoipa::ToSchema;
+use uuid::Uuid;
+
+/// Capacity of the global broadcast channel. Each `POST /sync/ops`
+/// pushes one row per accepted op; with thousands of concurrently-
+/// connected WebSocket subscribers, a slow consumer could lag behind
+/// — `broadcast` then returns `RecvError::Lagged` instead of buffering
+/// without bound. 4096 is enough headroom for a bursty batch from a
+/// freshly-rescanned library without forcing an unbounded queue.
+const BROADCAST_CAPACITY: usize = 4096;
+
+/// Default ACK flush cadence. Five seconds matches the RFC and gives
+/// a sensible knob: short enough that a crash loses at most one
+/// window of cursors (the compaction job rejoins the lost ground from
+/// the in-memory buffer anyway), long enough that a chatty WebSocket
+/// session doesn't write-amplify the cursor row.
+pub const DEFAULT_FLUSH_INTERVAL: Duration = Duration::from_secs(5);
+
+/// Default compaction cadence. Daily is the spec's call.
+pub const DEFAULT_COMPACTION_INTERVAL: Duration = Duration::from_secs(24 * 60 * 60);
+
+/// A device that hasn't ACKed for this long stops pinning the
+/// compaction floor. 90 days is the spec's call — long enough that a
+/// laptop-on-shelf-for-summer comes back fine, short enough that a
+/// permanently-lost device doesn't drag the log forever.
+pub const STALE_DEVICE_MS: i64 = 90 * 24 * 60 * 60 * 1000;
+
+/// Wire format for a single op the client pushes. `payload` stays
+/// opaque to the server — it's stored as JSONB and replayed verbatim
+/// to other devices, so the schema can evolve client-side without a
+/// server release.
+#[derive(Debug, Clone, Deserialize, Serialize, ToSchema)]
+pub struct SyncOpIn {
+    /// Client-generated UUIDv4. Idempotency key against
+    /// `(user_id, device_id, operation_id)`.
+    pub operation_id: Uuid,
+    /// Per-(user, device) monotonic counter. Strictly increasing.
+    pub lamport_ts: i64,
+    /// Entity type, e.g. `"playlist"`, `"library"`.
+    pub entity: String,
+    /// Free-form entity id (TEXT to span the desktop's mixed id schemes).
+    pub entity_id: String,
+    /// `None` for whole-entity ops (insert / delete), `Some` for
+    /// partial updates ("set name", "set color").
+    #[serde(default)]
+    pub field: Option<String>,
+    /// `"set" | "delete" | "insert" | "noop"`. Kept as a free-form
+    /// string so the protocol can grow without a server-side enum
+    /// expansion + migration.
+    pub op: String,
+    /// Op-specific JSON payload.
+    #[serde(default)]
+    pub payload: Option<serde_json::Value>,
+}
+
+/// Wire format for an accepted op. Mirrors the row shape so the
+/// pull (`GET /sync/ops`) and broadcast paths emit the same JSON
+/// straight from the `SELECT … RETURNING …` rowset.
+#[derive(Debug, Clone, Deserialize, Serialize, ToSchema)]
+pub struct SyncOp {
+    /// Server-assigned BIGSERIAL — the monotonic watermark every
+    /// device tracks via `last_seen_id`.
+    pub id: i64,
+    /// Echoed back so the client can correlate a POST response with
+    /// the original request even when `accepted` reorders.
+    pub operation_id: Uuid,
+    /// Echoed back so a pulling device can keep its own clock in sync.
+    pub device_id: String,
+    pub lamport_ts: i64,
+    pub entity: String,
+    pub entity_id: String,
+    pub field: Option<String>,
+    pub op: String,
+    pub payload: Option<serde_json::Value>,
+    pub created_at: i64,
+}
+
+/// Broadcast payload. We pre-serialise to JSON once at emit time so
+/// every receiver (potentially hundreds of WebSocket subscribers)
+/// avoids re-serialising the same row.
+#[derive(Debug, Clone)]
+pub struct SyncBroadcast {
+    /// Owner of the op. WS subscribers compare against the
+    /// connection's authenticated `user_id`; mismatches are dropped
+    /// before the frame ever leaves the socket task.
+    pub user_id: i64,
+    /// Pre-serialised `{"type":"op","op":{…}}` envelope, ready to
+    /// hand to `WebSocket::send(Message::Text(...))`.
+    pub frame: Arc<String>,
+}
+
+/// Per-(user, device) ACK accumulator. Atomics let the
+/// REST + WS handlers update without taking a lock; the periodic
+/// flusher reads + clears `dirty` to decide what to UPSERT.
+#[derive(Debug)]
+struct AckEntry {
+    last_seen_id: AtomicI64,
+    last_seen_at_ms: AtomicI64,
+    dirty: AtomicBool,
+}
+
+/// Shared sync state. Cheap to clone — every field is `Arc`-backed
+/// (the broadcast `Sender` clones into a new handle, the `DashMap`
+/// is `Arc`-internal, the pool is `Arc`-backed).
+#[derive(Clone)]
+pub struct SyncHub {
+    inner: Arc<SyncHubInner>,
+}
+
+struct SyncHubInner {
+    broadcast_tx: broadcast::Sender<SyncBroadcast>,
+    ack_buffer: DashMap<(i64, String), Arc<AckEntry>>,
+    db: PgPool,
+}
+
+impl SyncHub {
+    /// Build a hub against the live pool. Spawns the flush + compaction
+    /// tasks and returns their join handles so the binary can keep them
+    /// alive for the lifetime of the process.
+    pub fn spawn(
+        db: PgPool,
+        flush_interval: Duration,
+        compaction_interval: Duration,
+    ) -> (Self, JoinHandle<()>, JoinHandle<()>) {
+        let (broadcast_tx, _) = broadcast::channel(BROADCAST_CAPACITY);
+        let hub = Self {
+            inner: Arc::new(SyncHubInner {
+                broadcast_tx,
+                ack_buffer: DashMap::new(),
+                db,
+            }),
+        };
+
+        let flush_task = {
+            let hub = hub.clone();
+            tokio::spawn(async move { hub.run_flush_loop(flush_interval).await })
+        };
+        let compaction_task = {
+            let hub = hub.clone();
+            tokio::spawn(async move { hub.run_compaction_loop(compaction_interval).await })
+        };
+
+        (hub, flush_task, compaction_task)
+    }
+
+    /// Test helper — no background tasks. Tests drive [`flush_acks`]
+    /// and [`compact_once`] directly so each step is observable.
+    pub fn for_tests(db: PgPool) -> Self {
+        let (broadcast_tx, _) = broadcast::channel(BROADCAST_CAPACITY);
+        Self {
+            inner: Arc::new(SyncHubInner {
+                broadcast_tx,
+                ack_buffer: DashMap::new(),
+                db,
+            }),
+        }
+    }
+
+    /// Subscribe a new WebSocket session to the global broadcast.
+    /// The receiver itself filters per-user — see the WS handler.
+    pub fn subscribe(&self) -> broadcast::Receiver<SyncBroadcast> {
+        self.inner.broadcast_tx.subscribe()
+    }
+
+    /// Emit a fan-out frame. Returns the receiver count for the metric
+    /// the handler logs; the underlying `send` returns `Err` when no
+    /// subscribers are attached, which is fine and means "nothing to
+    /// do".
+    pub fn broadcast(&self, payload: SyncBroadcast) {
+        let _ = self.inner.broadcast_tx.send(payload);
+    }
+
+    /// Record an ACK. Idempotent + monotonic — a lower `last_seen_id`
+    /// is ignored. Marks the entry dirty so the next flush picks it up.
+    pub fn record_ack(&self, user_id: i64, device_id: &str, last_seen_id: i64, now_ms: i64) {
+        let entry = self
+            .inner
+            .ack_buffer
+            .entry((user_id, device_id.to_string()))
+            .or_insert_with(|| {
+                Arc::new(AckEntry {
+                    last_seen_id: AtomicI64::new(0),
+                    last_seen_at_ms: AtomicI64::new(0),
+                    dirty: AtomicBool::new(false),
+                })
+            })
+            .clone();
+        // CAS-loop bumps `last_seen_id` only if the incoming value is
+        // strictly higher. A no-op CAS short-circuits without writing
+        // (and crucially without marking the row dirty) so a chatty
+        // client re-acknowledging the same id never amplifies writes.
+        let mut current = entry.last_seen_id.load(Ordering::Relaxed);
+        loop {
+            if last_seen_id <= current {
+                return;
+            }
+            match entry.last_seen_id.compare_exchange_weak(
+                current,
+                last_seen_id,
+                Ordering::AcqRel,
+                Ordering::Relaxed,
+            ) {
+                Ok(_) => break,
+                Err(seen) => current = seen,
+            }
+        }
+        entry.last_seen_at_ms.store(now_ms, Ordering::Relaxed);
+        entry.dirty.store(true, Ordering::Release);
+    }
+
+    /// Flush every dirty ACK to `device_sync_cursor`. Public so the
+    /// compaction job can call it synchronously before reading the MIN
+    /// (otherwise the MIN reflects stale ACKs and compaction would
+    /// over-retain history at best, miscalculate the floor at worst).
+    pub async fn flush_acks(&self) -> Result<usize, sqlx::Error> {
+        // Snapshot the dirty rows under a single pass over the map.
+        // We `.clone()` the `Arc<AckEntry>` so the iteration releases
+        // each shard lock quickly and the actual SQL runs lock-free.
+        let mut staged: Vec<(i64, String, i64, i64)> = Vec::new();
+        for entry in self.inner.ack_buffer.iter() {
+            let ((user_id, device_id), state) = (entry.key(), entry.value());
+            // Clear dirty optimistically — if the SQL fails we re-set
+            // it from the error path so the next flush retries. The
+            // ordering matters: `swap(false, AcqRel)` returns the old
+            // value, so we only stage entries that were actually dirty
+            // since the last flush.
+            if state.dirty.swap(false, Ordering::AcqRel) {
+                staged.push((
+                    *user_id,
+                    device_id.clone(),
+                    state.last_seen_id.load(Ordering::Acquire),
+                    state.last_seen_at_ms.load(Ordering::Acquire),
+                ));
+            }
+        }
+        if staged.is_empty() {
+            return Ok(0);
+        }
+
+        let mut tx = self.inner.db.begin().await?;
+        let mut applied = 0usize;
+        for (user_id, device_id, last_seen_id, last_seen_at) in &staged {
+            // Monotonic UPSERT — refuse to lower an existing
+            // last_seen_id. A pull-without-process scenario where the
+            // client acked N + then crashed mid-process would otherwise
+            // let a "behind" cursor reach the cursor and pin compaction
+            // too high.
+            let res = sqlx::query(
+                "INSERT INTO device_sync_cursor \
+                    (user_id, device_id, last_seen_id, last_seen_at) \
+                 VALUES ($1, $2, $3, $4) \
+                 ON CONFLICT (user_id, device_id) DO UPDATE \
+                 SET last_seen_id = EXCLUDED.last_seen_id, \
+                     last_seen_at = EXCLUDED.last_seen_at \
+                 WHERE device_sync_cursor.last_seen_id < EXCLUDED.last_seen_id",
+            )
+            .bind(user_id)
+            .bind(device_id)
+            .bind(last_seen_id)
+            .bind(last_seen_at)
+            .execute(&mut *tx)
+            .await;
+            match res {
+                Ok(result) => {
+                    // The UPSERT's `WHERE last_seen_id <
+                    // EXCLUDED.last_seen_id` guard can match zero
+                    // rows when the persisted cursor is already at
+                    // or beyond the staged value (e.g. a previous
+                    // flush + a re-ACK of the same id). Count rows
+                    // actually written rather than statements run so
+                    // the returned tally matches the observable DB
+                    // change.
+                    applied += result.rows_affected() as usize;
+                }
+                Err(err) => {
+                    // Re-mark the staged entries dirty so the next
+                    // flush retries them — losing an ACK to a
+                    // transient DB hiccup would let the cursor drift
+                    // and over-retain forever.
+                    self.requeue_dirty(&staged);
+                    tx.rollback().await.ok();
+                    return Err(err);
+                }
+            }
+        }
+        tx.commit().await?;
+        Ok(applied)
+    }
+
+    fn requeue_dirty(&self, staged: &[(i64, String, i64, i64)]) {
+        for (user_id, device_id, _, _) in staged {
+            if let Some(entry) = self.inner.ack_buffer.get(&(*user_id, device_id.clone())) {
+                entry.dirty.store(true, Ordering::Release);
+            }
+        }
+    }
+
+    async fn run_flush_loop(self, period: Duration) {
+        let mut ticker = interval(period);
+        ticker.set_missed_tick_behavior(MissedTickBehavior::Skip);
+        // First tick fires immediately — burn it so we don't flush an
+        // empty buffer on boot.
+        ticker.tick().await;
+        loop {
+            ticker.tick().await;
+            if let Err(err) = self.flush_acks().await {
+                tracing::error!(error = %err, "sync ack flush failed");
+            }
+        }
+    }
+
+    async fn run_compaction_loop(self, period: Duration) {
+        let mut ticker = interval(period);
+        ticker.set_missed_tick_behavior(MissedTickBehavior::Skip);
+        ticker.tick().await; // burn the immediate first tick
+        loop {
+            ticker.tick().await;
+            if let Err(err) = self.compact_once().await {
+                tracing::error!(error = %err, "sync compaction failed");
+            }
+        }
+    }
+
+    /// Run one compaction pass. Public so tests can step through.
+    /// Algorithm:
+    ///
+    /// 1. Flush the in-memory ACK buffer to `device_sync_cursor` so the
+    ///    MIN below sees every device's true position.
+    /// 2. For each user with at least one fresh cursor (ACKed within
+    ///    [`STALE_DEVICE_MS`]):
+    ///    - Compute `MIN(last_seen_id)` across the fresh cursors.
+    ///    - In the same transaction, delete superseded ops
+    ///      (`id <= min`, keeping only the latest per
+    ///      `(entity, entity_id, COALESCE(field, ''))`) and UPSERT the
+    ///      watermark to `min`.
+    ///
+    /// Invariants:
+    ///
+    /// - Phase 1 flush is the "ACK MIN includes unflushed" guarantee.
+    /// - The delete + watermark UPSERT happen in the same Postgres
+    ///   transaction so a crash mid-compaction is atomic.
+    /// - Watermark is monotonic — the UPSERT `WHERE` clause refuses
+    ///   to lower an existing value.
+    pub async fn compact_once(&self) -> Result<CompactionReport, sqlx::Error> {
+        // Phase 1: flush pending ACKs so the MIN read below sees the
+        // freshest device positions.
+        self.flush_acks().await?;
+
+        let now = Utc::now().timestamp_millis();
+        let stale_threshold = now - STALE_DEVICE_MS;
+        let mut report = CompactionReport::default();
+
+        let users: Vec<i64> = sqlx::query_scalar(
+            "SELECT DISTINCT user_id FROM device_sync_cursor \
+             WHERE last_seen_at >= $1",
+        )
+        .bind(stale_threshold)
+        .fetch_all(&self.inner.db)
+        .await?;
+
+        for user_id in users {
+            let mut tx = self.inner.db.begin().await?;
+
+            let min_seen: Option<i64> = sqlx::query_scalar(
+                "SELECT MIN(last_seen_id) FROM device_sync_cursor \
+                 WHERE user_id = $1 AND last_seen_at >= $2",
+            )
+            .bind(user_id)
+            .bind(stale_threshold)
+            .fetch_one(&mut *tx)
+            .await?;
+
+            let Some(min) = min_seen else {
+                tx.commit().await?;
+                continue;
+            };
+
+            // Collapse: keep only the most-recent op per
+            // `(entity, entity_id, COALESCE(field, ''))` at or below
+            // `min`. Older partial updates to the same field are
+            // dead weight by definition — no device behind the
+            // watermark needs them.
+            let deleted = sqlx::query(
+                "DELETE FROM sync_op \
+                 WHERE id IN ( \
+                    SELECT id FROM ( \
+                        SELECT id, ROW_NUMBER() OVER ( \
+                            PARTITION BY entity, entity_id, COALESCE(field, '') \
+                            ORDER BY lamport_ts DESC, id DESC \
+                        ) AS rn \
+                        FROM sync_op \
+                        WHERE user_id = $1 AND id <= $2 \
+                    ) ranked \
+                    WHERE rn > 1 \
+                 )",
+            )
+            .bind(user_id)
+            .bind(min)
+            .execute(&mut *tx)
+            .await?
+            .rows_affected();
+
+            // Monotonic UPSERT — refuse to lower an existing
+            // watermark. A simultaneous compaction (we don't expect
+            // one, but the invariant is cheap to preserve) wouldn't
+            // be able to roll history back.
+            sqlx::query(
+                "INSERT INTO sync_compaction_watermark \
+                    (user_id, compacted_up_to, updated_at) \
+                 VALUES ($1, $2, $3) \
+                 ON CONFLICT (user_id) DO UPDATE \
+                 SET compacted_up_to = EXCLUDED.compacted_up_to, \
+                     updated_at = EXCLUDED.updated_at \
+                 WHERE sync_compaction_watermark.compacted_up_to \
+                       < EXCLUDED.compacted_up_to",
+            )
+            .bind(user_id)
+            .bind(min)
+            .bind(now)
+            .execute(&mut *tx)
+            .await?;
+
+            tx.commit().await?;
+
+            report.users_compacted += 1;
+            report.rows_deleted += deleted;
+        }
+
+        Ok(report)
+    }
+
+    /// Direct access to the underlying pool — handlers reuse it for
+    /// the per-request reads they own (push, pull, watermark check).
+    pub fn pool(&self) -> &PgPool {
+        &self.inner.db
+    }
+}
+
+/// Summary of one compaction pass. Returned from [`SyncHub::compact_once`]
+/// so tests can assert on observable progress without re-querying the
+/// schema.
+#[derive(Debug, Default, Clone, Copy)]
+pub struct CompactionReport {
+    pub users_compacted: usize,
+    pub rows_deleted: u64,
+}
+
+/// Build a `SyncBroadcast` envelope for an accepted op. Pre-serialises
+/// the frame so every receiver hands the cached `Arc<String>` to
+/// `WebSocket::send` without re-serialising.
+pub fn build_broadcast(user_id: i64, op: &SyncOp) -> SyncBroadcast {
+    // The frame schema is `{"type":"op","op":{…}}`. Pre-serialise once
+    // here so every subscriber dispatches an `Arc<String>` clone
+    // rather than re-serialising the same row N times.
+    let frame = serde_json::to_string(&serde_json::json!({
+        "type": "op",
+        "op": op,
+    }))
+    .expect("SyncOp serialises");
+    SyncBroadcast {
+        user_id,
+        frame: Arc::new(frame),
+    }
+}
diff --git a/tests/openapi.rs b/tests/openapi.rs
index 56b6f0a..4157abd 100644
--- a/tests/openapi.rs
+++ b/tests/openapi.rs
@@ -73,6 +73,17 @@ async fn openapi_doc_lists_every_handler(pool: PgPool) {
         "missing stream path in spec"
     );
 
+    // Phase 1.f sync surface — drop the `#[utoipa::path]` macro from
+    // any of these and the desktop / web stop seeing it in the docs.
+    assert!(
+        paths.contains_key("/api/v1/sync/ops"),
+        "missing /api/v1/sync/ops in spec"
+    );
+    assert!(
+        paths.contains_key("/api/v1/sync/ack"),
+        "missing /api/v1/sync/ack in spec"
+    );
+
     // `POST /api/v1/users` retired in Phase 1.d.2 — the JWT path's
     // lazy auto-provisioning is the only way to land a `users` row,
     // so the spec must NOT advertise the legacy bootstrap endpoint.
diff --git a/tests/support.rs b/tests/support.rs
index 8d5caa4..f03e10c 100644
--- a/tests/support.rs
+++ b/tests/support.rs
@@ -25,8 +25,10 @@ mod jwks_harness;
 
 use std::{net::SocketAddr, sync::Arc};
 
+use std::time::Duration;
+
 use sqlx::PgPool;
-use waveflow_server::{app, auth::JwtVerifier, AppState, Config, StreamCtx};
+use waveflow_server::{app, auth::JwtVerifier, sync::SyncHub, AppState, Config, StreamCtx};
 
 pub use jwks_harness::{good_claims, header_with_kid, JwksHarness, TEST_KID};
 
@@ -69,12 +71,61 @@ pub async fn spawn_app_with_jwt(pool: PgPool, verifier: Arc<JwtVerifier>) -> Str
     };
 
     let state = AppState {
-        db: pool,
+        db: pool.clone(),
         jwt_verifier: verifier,
         // Streaming is off by default in the shared helper. Tests
         // that exercise the stream endpoints use
         // `spawn_app_with_jwt_and_stream` instead.
         stream_ctx: None,
+        // Sync hub — the standard helper spawns the live flush +
+        // compaction tasks against a quick cadence so any test that
+        // exercises the REST surface gets the real wiring. Sync
+        // tests that want deterministic control use the variant
+        // below.
+        sync: SyncHub::spawn(pool, Duration::from_millis(50), Duration::from_secs(3600)).0,
+    };
+    tokio::spawn(async move {
+        axum::serve(
+            listener,
+            app(config, state).into_make_service_with_connect_info::<SocketAddr>(),
+        )
+        .await
+        .unwrap();
+    });
+
+    format!("http://{addr}")
+}
+
+/// Spawn the app with a caller-controlled `SyncHub`. Sync tests pass
+/// [`SyncHub::for_tests`] so the flush + compaction loops stay off and
+/// the test drives them step-by-step via [`SyncHub::flush_acks`] /
+/// [`SyncHub::compact_once`]. Returns the base URL alongside the hub
+/// so the test can read its state directly.
+pub async fn spawn_app_with_sync(
+    pool: PgPool,
+    verifier: Arc<JwtVerifier>,
+    sync: SyncHub,
+) -> String {
+    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = listener.local_addr().unwrap();
+
+    let config = Config {
+        bind_addr: addr,
+        request_timeout_secs: 5,
+        database_url: "<test>".into(),
+        db_max_connections: 1,
+        jwt_jwks_url: String::new(),
+        jwt_issuer: String::new(),
+        jwt_audience: String::new(),
+        music_root: None,
+        stream_secret: None,
+    };
+
+    let state = AppState {
+        db: pool,
+        jwt_verifier: verifier,
+        stream_ctx: None,
+        sync,
     };
     tokio::spawn(async move {
         axum::serve(
@@ -204,9 +255,10 @@ pub async fn spawn_app_with_jwt_and_stream(
     };
 
     let state = AppState {
-        db: pool,
+        db: pool.clone(),
         jwt_verifier: verifier,
         stream_ctx: Some(Arc::new(StreamCtx { music_root, secret })),
+        sync: SyncHub::spawn(pool, Duration::from_millis(50), Duration::from_secs(3600)).0,
     };
     tokio::spawn(async move {
         axum::serve(
diff --git a/tests/sync.rs b/tests/sync.rs
new file mode 100644
index 0000000..b5bcaf6
--- /dev/null
+++ b/tests/sync.rs
@@ -0,0 +1,748 @@
+//! End-to-end tests for `/api/v1/sync/*` (Phase 1.f).
+//!
+//! Coverage map:
+//!
+//! - REST contract: push → pull round-trip, idempotent replay, lamport
+//!   regression 409, oversized batch 400, empty `device_id` 400, pull
+//!   never advances the cursor.
+//! - ACK pipeline: `POST /sync/ack` lands the row only after the
+//!   in-memory buffer is flushed; ack semantics are monotonic.
+//! - Resurrected device: a pull with `since` below the compaction
+//!   watermark returns 410 + the watermark.
+//! - Compaction: collapse keeps only the latest op per (entity,
+//!   entity_id, field), stale devices are skipped, watermark advances.
+//! - Tenant isolation: user A's ops never reach user B's pull / WS.
+//! - WebSocket fan-out: a push by one device reaches the other's
+//!   socket in real time, scoped to the same user.
+
+mod support;
+
+use std::time::Duration;
+
+use futures_util::{SinkExt, StreamExt};
+use reqwest::StatusCode;
+use serde_json::{json, Value};
+use sqlx::PgPool;
+use support::{spawn_app_with_sync, spawn_authenticated, spawn_two_authenticated, JwksHarness};
+use tokio_tungstenite::tungstenite::Message;
+use uuid::Uuid;
+use waveflow_server::sync::SyncHub;
+
+fn op_payload(operation_id: Uuid, lamport_ts: i64, entity_id: &str, value: &str) -> Value {
+    json!({
+        "operation_id": operation_id,
+        "lamport_ts": lamport_ts,
+        "entity": "playlist",
+        "entity_id": entity_id,
+        "field": "name",
+        "op": "set",
+        "payload": { "value": value },
+    })
+}
+
+#[sqlx::test(migrator = "waveflow_server::db::MIGRATOR")]
+async fn push_then_pull_round_trip(pool: PgPool) {
+    let auth = spawn_authenticated(pool, "round-trip").await;
+    let op_id = Uuid::new_v4();
+
+    let push: Value = reqwest::Client::new()
+        .post(format!("{}/api/v1/sync/ops", auth.base))
+        .bearer_auth(&auth.token)
+        .json(&json!({
+            "device_id": "device-a",
+            "ops": [op_payload(op_id, 1, "pl-1", "Soirée")],
+        }))
+        .send()
+        .await
+        .unwrap()
+        .error_for_status()
+        .unwrap()
+        .json()
+        .await
+        .unwrap();
+    assert_eq!(push["accepted"].as_array().unwrap().len(), 1);
+    let row = &push["accepted"][0];
+    assert_eq!(row["operation_id"], json!(op_id));
+    assert_eq!(row["device_id"], "device-a");
+    let assigned_id = row["id"].as_i64().unwrap();
+    assert!(assigned_id > 0);
+
+    let pull: Value = reqwest::Client::new()
+        .get(format!("{}/api/v1/sync/ops?since=0", auth.base))
+        .bearer_auth(&auth.token)
+        .send()
+        .await
+        .unwrap()
+        .error_for_status()
+        .unwrap()
+        .json()
+        .await
+        .unwrap();
+    assert_eq!(pull["ops"].as_array().unwrap().len(), 1);
+    assert_eq!(pull["ops"][0]["id"], assigned_id);
+    assert_eq!(pull["last_id"], assigned_id);
+
+    // Empty pull when `since` is already at the head.
+    let empty: Value = reqwest::Client::new()
+        .get(format!("{}/api/v1/sync/ops?since={assigned_id}", auth.base))
+        .bearer_auth(&auth.token)
+        .send()
+        .await
+        .unwrap()
+        .error_for_status()
+        .unwrap()
+        .json()
+        .await
+        .unwrap();
+    assert!(empty["ops"].as_array().unwrap().is_empty());
+    assert_eq!(empty["last_id"], assigned_id);
+}
+
+#[sqlx::test(migrator = "waveflow_server::db::MIGRATOR")]
+async fn idempotent_replay_returns_same_id(pool: PgPool) {
+    let auth = spawn_authenticated(pool, "idem-user").await;
+    let op_id = Uuid::new_v4();
+    let body = json!({
+        "device_id": "device-a",
+        "ops": [op_payload(op_id, 1, "pl-1", "First")],
+    });
+
+    let first: Value = reqwest::Client::new()
+        .post(format!("{}/api/v1/sync/ops", auth.base))
+        .bearer_auth(&auth.token)
+        .json(&body)
+        .send()
+        .await
+        .unwrap()
+        .error_for_status()
+        .unwrap()
+        .json()
+        .await
+        .unwrap();
+    let first_id = first["accepted"][0]["id"].as_i64().unwrap();
+
+    let second: Value = reqwest::Client::new()
+        .post(format!("{}/api/v1/sync/ops", auth.base))
+        .bearer_auth(&auth.token)
+        .json(&body)
+        .send()
+        .await
+        .unwrap()
+        .error_for_status()
+        .unwrap()
+        .json()
+        .await
+        .unwrap();
+    let second_id = second["accepted"][0]["id"].as_i64().unwrap();
+    assert_eq!(
+        first_id, second_id,
+        "idempotent replay must return the original row id"
+    );
+
+    // And only one row physically exists.
+    let count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM sync_op")
+        .fetch_one(&auth.pool)
+        .await
+        .unwrap();
+    assert_eq!(count, 1);
+}
+
+#[sqlx::test(migrator = "waveflow_server::db::MIGRATOR")]
+async fn lamport_regression_returns_409_with_stored_max(pool: PgPool) {
+    let auth = spawn_authenticated(pool, "lamport").await;
+
+    let _: Value = reqwest::Client::new()
+        .post(format!("{}/api/v1/sync/ops", auth.base))
+        .bearer_auth(&auth.token)
+        .json(&json!({
+            "device_id": "device-a",
+            "ops": [
+                op_payload(Uuid::new_v4(), 10, "pl-1", "a"),
+                op_payload(Uuid::new_v4(), 11, "pl-1", "b"),
+            ],
+        }))
+        .send()
+        .await
+        .unwrap()
+        .error_for_status()
+        .unwrap()
+        .json()
+        .await
+        .unwrap();
+
+    // Now replay an op at lamport_ts=10, which is already taken by a
+    // DIFFERENT operation_id — the unique constraint on (user_id,
+    // device_id, lamport_ts) fires and we expect 409 + stored_max=11.
+    let resp = reqwest::Client::new()
+        .post(format!("{}/api/v1/sync/ops", auth.base))
+        .bearer_auth(&auth.token)
+        .json(&json!({
+            "device_id": "device-a",
+            "ops": [op_payload(Uuid::new_v4(), 10, "pl-2", "regression")],
+        }))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(resp.status(), StatusCode::CONFLICT);
+    let body: Value = resp.json().await.unwrap();
+    assert_eq!(body["error"], "lamport_regression");
+    assert_eq!(body["device_id"], "device-a");
+    assert_eq!(body["stored_max"], 11);
+    assert_eq!(body["offending_lamport_ts"], 10);
+}
+
+#[sqlx::test(migrator = "waveflow_server::db::MIGRATOR")]
+async fn pull_does_not_advance_cursor(pool: PgPool) {
+    // Deterministic harness — `SyncHub::for_tests` skips the 5 s
+    // flusher, so the assertion below isn't racing a background task.
+    let harness = std::sync::Arc::new(JwksHarness::spawn().await);
+    let sync = SyncHub::for_tests(pool.clone());
+    let base = spawn_app_with_sync(pool.clone(), harness.verifier_arc(), sync.clone()).await;
+    let token = harness.mint(
+        &support::good_claims("pull-readonly"),
+        &support::header_with_kid(support::TEST_KID),
+    );
+    // Warm-up request lazy-provisions the user row.
+    reqwest::Client::new()
+        .get(format!("{base}/api/v1/profiles"))
+        .bearer_auth(&token)
+        .send()
+        .await
+        .unwrap()
+        .error_for_status()
+        .unwrap();
+
+    let _: Value = reqwest::Client::new()
+        .post(format!("{base}/api/v1/sync/ops"))
+        .bearer_auth(&token)
+        .json(&json!({
+            "device_id": "device-a",
+            "ops": [op_payload(Uuid::new_v4(), 1, "pl-1", "x")],
+        }))
+        .send()
+        .await
+        .unwrap()
+        .error_for_status()
+        .unwrap()
+        .json()
+        .await
+        .unwrap();
+
+    // A pull. Read-only by contract — must not touch the cursor.
+    let _: Value = reqwest::Client::new()
+        .get(format!("{base}/api/v1/sync/ops?since=0"))
+        .bearer_auth(&token)
+        .send()
+        .await
+        .unwrap()
+        .error_for_status()
+        .unwrap()
+        .json()
+        .await
+        .unwrap();
+
+    // Force-drain any pending ACK buffer so a stray write would show
+    // up here. The pull above never recorded one, so the flush is a
+    // no-op; the assertion is the real signal.
+    sync.flush_acks().await.unwrap();
+    let count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM device_sync_cursor")
+        .fetch_one(&pool)
+        .await
+        .unwrap();
+    assert_eq!(count, 0, "GET /sync/ops must never advance the cursor");
+}
+
+#[sqlx::test(migrator = "waveflow_server::db::MIGRATOR")]
+async fn ack_writes_cursor_after_flush(pool: PgPool) {
+    // Deterministic harness — drive `flush_acks` by hand so the
+    // assertion isn't racing the 5 s loop.
+    let harness = std::sync::Arc::new(JwksHarness::spawn().await);
+    let sync = SyncHub::for_tests(pool.clone());
+    let base = spawn_app_with_sync(pool.clone(), harness.verifier_arc(), sync.clone()).await;
+    let token = harness.mint(
+        &support::good_claims("ack-flush"),
+        &support::header_with_kid(support::TEST_KID),
+    );
+    reqwest::Client::new()
+        .get(format!("{base}/api/v1/profiles"))
+        .bearer_auth(&token)
+        .send()
+        .await
+        .unwrap()
+        .error_for_status()
+        .unwrap();
+    let user_id: i64 = sqlx::query_scalar("SELECT id FROM users WHERE external_id = 'ack-flush'")
+        .fetch_one(&pool)
+        .await
+        .unwrap();
+
+    // Land an op so there's an `id` to ACK at.
+    let push: Value = reqwest::Client::new()
+        .post(format!("{base}/api/v1/sync/ops"))
+        .bearer_auth(&token)
+        .json(&json!({
+            "device_id": "device-a",
+            "ops": [op_payload(Uuid::new_v4(), 1, "pl-1", "x")],
+        }))
+        .send()
+        .await
+        .unwrap()
+        .error_for_status()
+        .unwrap()
+        .json()
+        .await
+        .unwrap();
+    let head_id = push["accepted"][0]["id"].as_i64().unwrap();
+
+    let ack = reqwest::Client::new()
+        .post(format!("{base}/api/v1/sync/ack"))
+        .bearer_auth(&token)
+        .json(&json!({ "device_id": "device-a", "last_seen_id": head_id }))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(ack.status(), StatusCode::NO_CONTENT);
+
+    // Force the flush — no sleeps, no polling.
+    let flushed = sync.flush_acks().await.unwrap();
+    assert_eq!(flushed, 1, "first flush must UPSERT the new cursor row");
+    let last_seen: i64 = sqlx::query_scalar(
+        "SELECT last_seen_id FROM device_sync_cursor \
+         WHERE user_id = $1 AND device_id = 'device-a'",
+    )
+    .bind(user_id)
+    .fetch_one(&pool)
+    .await
+    .unwrap();
+    assert_eq!(last_seen, head_id);
+
+    // Re-ACK at a lower id must NOT regress — the in-memory CAS
+    // short-circuits without even marking the entry dirty, so the
+    // second flush has nothing to write AND the cursor row is
+    // unchanged.
+    reqwest::Client::new()
+        .post(format!("{base}/api/v1/sync/ack"))
+        .bearer_auth(&token)
+        .json(&json!({ "device_id": "device-a", "last_seen_id": 0 }))
+        .send()
+        .await
+        .unwrap()
+        .error_for_status()
+        .unwrap();
+    let flushed_after_regress = sync.flush_acks().await.unwrap();
+    assert_eq!(
+        flushed_after_regress, 0,
+        "regressing ACK must not produce a flush write",
+    );
+    let still: i64 = sqlx::query_scalar(
+        "SELECT last_seen_id FROM device_sync_cursor \
+         WHERE user_id = $1 AND device_id = 'device-a'",
+    )
+    .bind(user_id)
+    .fetch_one(&pool)
+    .await
+    .unwrap();
+    assert_eq!(still, head_id, "ACK must be monotonic");
+}
+
+#[sqlx::test(migrator = "waveflow_server::db::MIGRATOR")]
+async fn resurrected_device_returns_410(pool: PgPool) {
+    let auth = spawn_authenticated(pool, "resurrected").await;
+
+    // Manually plant a high watermark to simulate a compacted log
+    // without having to actually compact. The whole point of this
+    // test is the *guard*, not the compaction job.
+    sqlx::query(
+        "INSERT INTO sync_compaction_watermark (user_id, compacted_up_to, updated_at) \
+         VALUES ($1, 100, 0)",
+    )
+    .bind(auth.user_id)
+    .execute(&auth.pool)
+    .await
+    .unwrap();
+
+    let resp = reqwest::Client::new()
+        .get(format!("{}/api/v1/sync/ops?since=50", auth.base))
+        .bearer_auth(&auth.token)
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(resp.status(), StatusCode::GONE);
+    let body: Value = resp.json().await.unwrap();
+    assert_eq!(body["error"], "resync_required");
+    assert_eq!(body["compacted_up_to"], 100);
+
+    // `since=0` is the special "send me everything" case and must
+    // *not* trip the guard — otherwise a fresh device would never
+    // bootstrap.
+    let ok = reqwest::Client::new()
+        .get(format!("{}/api/v1/sync/ops?since=0", auth.base))
+        .bearer_auth(&auth.token)
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(ok.status(), StatusCode::OK);
+}
+
+#[sqlx::test(migrator = "waveflow_server::db::MIGRATOR")]
+async fn tenant_isolation_pull(pool: PgPool) {
+    let two = spawn_two_authenticated(pool, "tenant-a", "tenant-b").await;
+
+    let _: Value = reqwest::Client::new()
+        .post(format!("{}/api/v1/sync/ops", two.base))
+        .bearer_auth(&two.a.token)
+        .json(&json!({
+            "device_id": "device-a",
+            "ops": [op_payload(Uuid::new_v4(), 1, "pl-a", "for A")],
+        }))
+        .send()
+        .await
+        .unwrap()
+        .error_for_status()
+        .unwrap()
+        .json()
+        .await
+        .unwrap();
+
+    // User B's pull must NOT see user A's ops.
+    let pull_b: Value = reqwest::Client::new()
+        .get(format!("{}/api/v1/sync/ops?since=0", two.base))
+        .bearer_auth(&two.b.token)
+        .send()
+        .await
+        .unwrap()
+        .error_for_status()
+        .unwrap()
+        .json()
+        .await
+        .unwrap();
+    assert!(
+        pull_b["ops"].as_array().unwrap().is_empty(),
+        "user B must not see user A's ops, got: {pull_b}"
+    );
+}
+
+#[sqlx::test(migrator = "waveflow_server::db::MIGRATOR")]
+async fn oversized_batch_is_rejected(pool: PgPool) {
+    let auth = spawn_authenticated(pool, "oversized").await;
+    let ops: Vec<Value> = (0..1025)
+        .map(|i| op_payload(Uuid::new_v4(), i as i64 + 1, "pl-1", "x"))
+        .collect();
+    let resp = reqwest::Client::new()
+        .post(format!("{}/api/v1/sync/ops", auth.base))
+        .bearer_auth(&auth.token)
+        .json(&json!({ "device_id": "device-a", "ops": ops }))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
+}
+
+#[sqlx::test(migrator = "waveflow_server::db::MIGRATOR")]
+async fn compaction_collapses_old_ops_and_advances_watermark(pool: PgPool) {
+    // Manual control over both loops — push three SET ops over the
+    // same (entity, entity_id, field), ACK to the head, run
+    // `compact_once` exactly once, and observe only the latest op
+    // survives at or below the watermark.
+    let harness = std::sync::Arc::new(JwksHarness::spawn().await);
+    let sync = SyncHub::for_tests(pool.clone());
+    let base = spawn_app_with_sync(pool.clone(), harness.verifier_arc(), sync.clone()).await;
+    let token = harness.mint(
+        &support::good_claims("compaction"),
+        &support::header_with_kid(support::TEST_KID),
+    );
+
+    // Provision the user via the warm-up request.
+    let _ = reqwest::Client::new()
+        .get(format!("{base}/api/v1/profiles"))
+        .bearer_auth(&token)
+        .send()
+        .await
+        .unwrap()
+        .error_for_status()
+        .unwrap();
+    let user_id: i64 = sqlx::query_scalar("SELECT id FROM users WHERE external_id = 'compaction'")
+        .fetch_one(&pool)
+        .await
+        .unwrap();
+
+    // Push three superseding SETs on the same field, plus an
+    // unrelated entity op that compaction must leave alone.
+    for lamport in 1..=3 {
+        let _: Value = reqwest::Client::new()
+            .post(format!("{base}/api/v1/sync/ops"))
+            .bearer_auth(&token)
+            .json(&json!({
+                "device_id": "device-a",
+                "ops": [op_payload(Uuid::new_v4(), lamport, "pl-keep", &format!("v{lamport}"))],
+            }))
+            .send()
+            .await
+            .unwrap()
+            .error_for_status()
+            .unwrap()
+            .json()
+            .await
+            .unwrap();
+    }
+    let _: Value = reqwest::Client::new()
+        .post(format!("{base}/api/v1/sync/ops"))
+        .bearer_auth(&token)
+        .json(&json!({
+            "device_id": "device-a",
+            "ops": [op_payload(Uuid::new_v4(), 4, "pl-other", "untouched")],
+        }))
+        .send()
+        .await
+        .unwrap()
+        .error_for_status()
+        .unwrap()
+        .json()
+        .await
+        .unwrap();
+
+    let head_id: i64 = sqlx::query_scalar("SELECT MAX(id) FROM sync_op")
+        .fetch_one(&pool)
+        .await
+        .unwrap();
+
+    // ACK to the head. Buffered — flush by running `compact_once`,
+    // which calls `flush_acks` internally before reading the MIN.
+    let ack = reqwest::Client::new()
+        .post(format!("{base}/api/v1/sync/ack"))
+        .bearer_auth(&token)
+        .json(&json!({ "device_id": "device-a", "last_seen_id": head_id }))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(ack.status(), StatusCode::NO_CONTENT);
+
+    let report = sync.compact_once().await.unwrap();
+    assert_eq!(report.users_compacted, 1);
+    assert_eq!(report.rows_deleted, 2, "two superseded SETs collapse away");
+
+    // The latest `pl-keep` SET + the `pl-other` op remain.
+    let remaining: Vec<(i64, String, String)> =
+        sqlx::query_as("SELECT id, entity_id, payload->>'value' FROM sync_op ORDER BY id ASC")
+            .fetch_all(&pool)
+            .await
+            .unwrap();
+    assert_eq!(remaining.len(), 2, "got {remaining:?}");
+    assert_eq!(remaining[0].1, "pl-keep");
+    assert_eq!(remaining[0].2, "v3", "only the latest SET survives");
+    assert_eq!(remaining[1].1, "pl-other");
+
+    // Watermark equals the MIN we computed (= head_id, the only
+    // device is fully caught up).
+    let watermark: i64 = sqlx::query_scalar(
+        "SELECT compacted_up_to FROM sync_compaction_watermark WHERE user_id = $1",
+    )
+    .bind(user_id)
+    .fetch_one(&pool)
+    .await
+    .unwrap();
+    assert_eq!(watermark, head_id);
+
+    // Second compaction with no new ACKs is a no-op (no rows
+    // deleted, watermark unchanged).
+    let report2 = sync.compact_once().await.unwrap();
+    assert_eq!(report2.rows_deleted, 0);
+    let watermark2: i64 = sqlx::query_scalar(
+        "SELECT compacted_up_to FROM sync_compaction_watermark WHERE user_id = $1",
+    )
+    .bind(user_id)
+    .fetch_one(&pool)
+    .await
+    .unwrap();
+    assert_eq!(watermark2, head_id, "watermark must be monotonic");
+}
+
+#[sqlx::test(migrator = "waveflow_server::db::MIGRATOR")]
+async fn compaction_skips_stale_devices(pool: PgPool) {
+    let harness = std::sync::Arc::new(JwksHarness::spawn().await);
+    let sync = SyncHub::for_tests(pool.clone());
+    let base = spawn_app_with_sync(pool.clone(), harness.verifier_arc(), sync.clone()).await;
+    let token = harness.mint(
+        &support::good_claims("stale"),
+        &support::header_with_kid(support::TEST_KID),
+    );
+    let _ = reqwest::Client::new()
+        .get(format!("{base}/api/v1/profiles"))
+        .bearer_auth(&token)
+        .send()
+        .await
+        .unwrap()
+        .error_for_status()
+        .unwrap();
+    let user_id: i64 = sqlx::query_scalar("SELECT id FROM users WHERE external_id = 'stale'")
+        .fetch_one(&pool)
+        .await
+        .unwrap();
+
+    // Push three superseding ops as one device.
+    for lamport in 1..=3 {
+        let _: Value = reqwest::Client::new()
+            .post(format!("{base}/api/v1/sync/ops"))
+            .bearer_auth(&token)
+            .json(&json!({
+                "device_id": "device-a",
+                "ops": [op_payload(Uuid::new_v4(), lamport, "pl-1", &format!("v{lamport}"))],
+            }))
+            .send()
+            .await
+            .unwrap()
+            .error_for_status()
+            .unwrap()
+            .json()
+            .await
+            .unwrap();
+    }
+    let head_id: i64 = sqlx::query_scalar("SELECT MAX(id) FROM sync_op")
+        .fetch_one(&pool)
+        .await
+        .unwrap();
+
+    // Device A is up to date.
+    let _ = reqwest::Client::new()
+        .post(format!("{base}/api/v1/sync/ack"))
+        .bearer_auth(&token)
+        .json(&json!({ "device_id": "device-a", "last_seen_id": head_id }))
+        .send()
+        .await
+        .unwrap()
+        .error_for_status()
+        .unwrap();
+    // Plant a STALE cursor for device-stale (200 days ago) at id 1 —
+    // if the cutoff weren't applied, the MIN would be 1 and nothing
+    // would compact. We expect the compactor to skip this row.
+    let stale_ts = chrono::Utc::now().timestamp_millis() - 200 * 24 * 60 * 60 * 1000;
+    sqlx::query(
+        "INSERT INTO device_sync_cursor (user_id, device_id, last_seen_id, last_seen_at) \
+         VALUES ($1, 'device-stale', 1, $2)",
+    )
+    .bind(user_id)
+    .bind(stale_ts)
+    .execute(&pool)
+    .await
+    .unwrap();
+
+    let report = sync.compact_once().await.unwrap();
+    assert_eq!(
+        report.rows_deleted, 2,
+        "compaction must ignore the stale device and collapse using device-a's MIN",
+    );
+    let watermark: i64 = sqlx::query_scalar(
+        "SELECT compacted_up_to FROM sync_compaction_watermark WHERE user_id = $1",
+    )
+    .bind(user_id)
+    .fetch_one(&pool)
+    .await
+    .unwrap();
+    assert_eq!(watermark, head_id);
+}
+
+#[sqlx::test(migrator = "waveflow_server::db::MIGRATOR")]
+async fn websocket_fans_out_to_same_tenant(pool: PgPool) {
+    use tokio_tungstenite::tungstenite::client::IntoClientRequest;
+
+    let auth = spawn_authenticated(pool, "ws-user").await;
+
+    // Reach the WS endpoint over plain `ws://`. The bearer + query
+    // string identify the device.
+    let ws_url = format!(
+        "ws://{}/api/v1/sync/ws?device_id=ws-device",
+        auth.base.trim_start_matches("http://"),
+    );
+    let mut req = ws_url.into_client_request().unwrap();
+    req.headers_mut().insert(
+        "authorization",
+        format!("Bearer {}", auth.token).parse().unwrap(),
+    );
+    let (mut socket, _resp) = tokio_tungstenite::connect_async(req)
+        .await
+        .expect("ws upgrade failed");
+
+    // Push from a different device under the same user — the WS
+    // session should receive the op envelope.
+    let op_id = Uuid::new_v4();
+    let _: Value = reqwest::Client::new()
+        .post(format!("{}/api/v1/sync/ops", auth.base))
+        .bearer_auth(&auth.token)
+        .json(&json!({
+            "device_id": "pusher",
+            "ops": [op_payload(op_id, 1, "pl-1", "from-ws-test")],
+        }))
+        .send()
+        .await
+        .unwrap()
+        .error_for_status()
+        .unwrap()
+        .json()
+        .await
+        .unwrap();
+
+    // Receive — bail on a 1 s timeout so a regression doesn't hang CI.
+    let msg = tokio::time::timeout(Duration::from_secs(1), socket.next())
+        .await
+        .expect("ws frame timed out")
+        .expect("ws stream ended")
+        .expect("ws frame error");
+    let text = match msg {
+        Message::Text(t) => t.to_string(),
+        other => panic!("expected text frame, got {other:?}"),
+    };
+    let payload: Value = serde_json::from_str(&text).unwrap();
+    assert_eq!(payload["type"], "op");
+    assert_eq!(payload["op"]["operation_id"], json!(op_id));
+    assert_eq!(payload["op"]["device_id"], "pusher");
+
+    // Close cleanly so the on-disconnect flush runs.
+    socket.send(Message::Close(None)).await.ok();
+}
+
+#[sqlx::test(migrator = "waveflow_server::db::MIGRATOR")]
+async fn websocket_isolates_other_tenants(pool: PgPool) {
+    use tokio_tungstenite::tungstenite::client::IntoClientRequest;
+
+    let two = spawn_two_authenticated(pool, "ws-a", "ws-b").await;
+
+    // User B subscribes; user A pushes — B must not receive the frame.
+    let ws_url = format!(
+        "ws://{}/api/v1/sync/ws?device_id=b-device",
+        two.base.trim_start_matches("http://"),
+    );
+    let mut req = ws_url.into_client_request().unwrap();
+    req.headers_mut().insert(
+        "authorization",
+        format!("Bearer {}", two.b.token).parse().unwrap(),
+    );
+    let (mut socket, _resp) = tokio_tungstenite::connect_async(req)
+        .await
+        .expect("ws upgrade failed");
+
+    // Push as user A.
+    let _: Value = reqwest::Client::new()
+        .post(format!("{}/api/v1/sync/ops", two.base))
+        .bearer_auth(&two.a.token)
+        .json(&json!({
+            "device_id": "a-pusher",
+            "ops": [op_payload(Uuid::new_v4(), 1, "pl-leak", "should-not-reach-b")],
+        }))
+        .send()
+        .await
+        .unwrap()
+        .error_for_status()
+        .unwrap()
+        .json()
+        .await
+        .unwrap();
+
+    // Wait 500 ms for the broadcast to (not) arrive. A successful
+    // recv would fail the test.
+    match tokio::time::timeout(Duration::from_millis(500), socket.next()).await {
+        Err(_timeout) => { /* expected — nothing crossed */ }
+        Ok(Some(Ok(Message::Close(_)))) | Ok(None) => { /* socket died, also fine */ }
+        Ok(other) => panic!("tenant B's WS received a frame from tenant A's push: {other:?}"),
+    }
+    socket.send(Message::Close(None)).await.ok();
+}