Add the cloud sync engine

Walks the local store and maps it onto the cloud object model per the
design doc: captures, generations (+ version wavs), profiles (+ samples,
avatar), and the settings singletons. Records carry the full row (paths
storage-relative so restores re-anchor under the destination data dir);
audio travels as per-asset ciphertext blobs.

Envelopes are randomized, so a new cloud_sync_state table remembers both
the plaintext fingerprint (did content actually change?) and the
ciphertext hash the server holds (re-declare unchanged blobs without
re-encrypting or re-uploading). updated_at is excluded from fingerprints
- its onupdate trigger fires as a side effect of applying pulls and
would bounce synced objects otherwise. Push runs before pull; conflicts
are last-writer-wins per object. Own pushes echoing back through the
feed are recognized by ciphertext hash and skipped.

Tests simulate two machines (own DB, data dir, keychain) syncing through
an in-process fake cloud: full backup/restore with byte-identical audio,
incremental edit (only the record blob re-uploads), delete propagation,
LWW convergence, and the blindness check that server storage only ever
holds VBX1 envelopes.

backend/database/__init__.py

@@ -12,6 +12,7 @@ from .models import (
     CaptureSettings,
     ChannelDeviceMapping,
     CloudSettings,
+    CloudSyncState,
     EffectPreset,
     Generation,
     GenerationSettings,
@@ -34,6 +35,7 @@ __all__ = [
     "CaptureSettings",
     "ChannelDeviceMapping",
     "CloudSettings",
+    "CloudSyncState",
     "EffectPreset",
     "Generation",
     "GenerationSettings",

backend/database/migrations.py

@@ -285,13 +285,16 @@ def _migrate_mcp_bindings(engine, inspector, tables: set[str]) -> None:
 
 
 def _migrate_cloud_settings(engine, inspector, tables: set[str]) -> None:
-    """Add ``sync_device_id`` — the server-assigned id from registering this
-    install as an encryption-capable sync device."""
+    """Add the cloud sync columns: ``sync_device_id`` (the server-assigned id
+    from registering this install as an encryption-capable sync device) and
+    ``sync_cursor`` (highest applied seq from the sync feed)."""
     if "cloud_settings" not in tables:
         return
     columns = _get_columns(inspector, "cloud_settings")
     if "sync_device_id" not in columns:
         _add_column(engine, "cloud_settings", "sync_device_id VARCHAR", "sync_device_id")
+    if "sync_cursor" not in columns:
+        _add_column(engine, "cloud_settings", "sync_cursor INTEGER NOT NULL DEFAULT 0", "sync_cursor")
 
 
 def _supports_drop_column(engine) -> bool:

backend/database/models.py

@@ -3,7 +3,7 @@
 from datetime import datetime
 import uuid
 
-from sqlalchemy import Column, String, Integer, Float, DateTime, Text, ForeignKey, Boolean, JSON
+from sqlalchemy import Column, String, Integer, Float, DateTime, Text, ForeignKey, Boolean, JSON, UniqueConstraint
 from sqlalchemy.ext.declarative import declarative_base
 
 from ..utils.capture_chords import (
@@ -257,9 +257,41 @@ class CloudSettings(Base):
     # this install registers as an encryption-capable device; the matching
     # X25519 private key lives in the OS keychain (services/cloud_keys.py).
     sync_device_id = Column(String, nullable=True)
+    # Highest server seq this install has applied from the sync feed.
+    sync_cursor = Column(Integer, nullable=False, default=0)
     updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
 
 
+class CloudSyncState(Base):
+    """Per-entity sync bookkeeping for cloud backup (one row per synced object).
+
+    Envelopes are randomized (fresh content key + nonce per encryption), so the
+    ciphertext hash changes on every re-encrypt even when the content didn't.
+    To keep the server's changed-hash dedup working, this table remembers what
+    was last uploaded: the plaintext fingerprints (to detect real local edits)
+    and the ciphertext hashes the server currently holds (to re-declare
+    unchanged blobs without re-encrypting or re-uploading them).
+    """
+
+    __tablename__ = "cloud_sync_state"
+    __table_args__ = (UniqueConstraint("kind", "client_id", name="uq_cloud_sync_state_kind_client"),)
+
+    id = Column(String, primary_key=True, default=lambda: str(uuid.uuid4()))
+    kind = Column(String, nullable=False)  # capture | generation | profile | settings
+    client_id = Column(String, nullable=False)  # the local entity id
+    server_object_id = Column(String, nullable=True)
+    # Client-bumped LWW version (object.version on the server).
+    version = Column(Integer, nullable=False, default=1)
+    # SHA-256 of the canonical plaintext record JSON at last push/pull.
+    record_fingerprint = Column(String, nullable=True)
+    # SHA-256 + size of the record ciphertext the server currently holds.
+    record_hash = Column(String, nullable=True)
+    record_size = Column(Integer, nullable=False, default=0)
+    # Per-asset bookkeeping, JSON: {clientAssetId: {role, fingerprint, hash, size}}
+    assets_json = Column(Text, nullable=False, default="{}")
+    last_synced_at = Column(DateTime, nullable=True)
+
+
 class MCPClientBinding(Base):
     """Per-MCP-client settings (voice profile, engine, personality default).
 

backend/services/cloud_sync.py

@@ -0,0 +1,535 @@
+"""
+The cloud sync engine: encrypted backup + multi-device restore.
+
+Walks the local store (SQLite rows + audio files), maps each entity onto the
+cloud's object model, and drives the push/pull loop against the blind server.
+Everything crosses the wire as VBX1 ciphertext (``cloud_crypto``); the server
+only ever learns kinds, ids, sizes, and hashes.
+
+Mapping (cloud repo ``docs/DESIGN.md`` §5):
+
+| local entity                          | kind       | record (encrypted JSON)   | assets            |
+| ------------------------------------- | ---------- | ------------------------- | ----------------- |
+| ``captures`` row + wav                | capture    | the row                   | the capture audio |
+| ``generations`` row + version wavs    | generation | the row + version rows    | each version wav  |
+| ``profiles`` row + samples + avatar   | profile    | the row + sample rows     | sample wavs, avatar |
+| ``capture_settings`` / ``generation_settings`` | settings | the row          | —                 |
+
+Path columns are stored storage-relative inside the (encrypted) record, so a
+restore re-anchors them under the destination machine's data dir.
+
+Change detection: envelopes are randomized, so ``CloudSyncState`` keeps the
+plaintext fingerprint (did the content actually change?) alongside the
+ciphertext hash the server holds (re-declare unchanged blobs without
+re-encrypting). Conflicts are last-writer-wins per object, matching §6 —
+push runs before pull, so local edits are declared before remote state lands.
+
+AAD binding: records are bound to ``(clientId, "record", version)`` and
+re-encrypted on every version bump. Asset blobs are bound to
+``(clientId, "asset:<clientAssetId>", 1)`` — assets are content-addressed and
+practically immutable (audio never changes in place), so their slot binding
+doesn't chase the object version.
+"""
+
+import hashlib
+import json
+import logging
+from dataclasses import dataclass, field
+from datetime import datetime
+from pathlib import Path
+
+from sqlalchemy.orm import Session
+
+from .. import config
+from ..database import (
+    Capture,
+    CaptureSettings,
+    CloudSettings as DBCloudSettings,
+    CloudSyncState,
+    Generation,
+    GenerationSettings,
+    GenerationVersion,
+    ProfileSample,
+    VoiceProfile,
+)
+from . import cloud_account, cloud_crypto
+from .cloud_api import CloudApiClient
+
+logger = logging.getLogger(__name__)
+
+_ASSET_AAD_VERSION = 1
+
+
+class CloudSyncError(Exception):
+    """Sync could not run or an object failed to round-trip."""
+
+
+# ---------------------------------------------------------------------------
+# Local object collection (push side)
+
+
+@dataclass(frozen=True)
+class LocalAsset:
+    client_asset_id: str
+    role: str  # audio | version | sample | avatar
+    path: Path
+
+
+@dataclass(frozen=True)
+class LocalObject:
+    kind: str
+    client_id: str
+    record: dict
+    assets: list[LocalAsset] = field(default_factory=list)
+
+
+_PATH_COLUMNS = {"audio_path", "avatar_path"}
+
+
+def _row_to_record(row) -> dict:
+    """All mapped columns as JSON-safe values; paths storage-relative,
+    datetimes ISO-8601."""
+    record: dict = {}
+    for column in row.__mapper__.columns:
+        value = getattr(row, column.key)
+        if value is None:
+            record[column.key] = None
+        elif column.key in _PATH_COLUMNS:
+            # Rows normally hold data-dir-relative paths already; only rebase
+            # absolute ones. (to_storage_path on a relative value would resolve
+            # it against the CWD and corrupt it.)
+            record[column.key] = config.to_storage_path(value) if Path(value).is_absolute() else value
+        elif isinstance(value, datetime):
+            record[column.key] = value.isoformat()
+        else:
+            record[column.key] = value
+    return record
+
+
+def _is_datetime_column(column) -> bool:
+    try:
+        return column.type.python_type is datetime
+    except NotImplementedError:  # e.g. JSON columns don't declare a python_type
+        return False
+
+
+def _record_to_row(model, record: dict, existing=None):
+    """Build or update a model instance from a record dict."""
+    row = existing if existing is not None else model()
+    for column in row.__mapper__.columns:
+        if column.key not in record:
+            continue
+        value = record[column.key]
+        if isinstance(value, str) and _is_datetime_column(column):
+            value = datetime.fromisoformat(value)
+        setattr(row, column.key, value)
+    return row
+
+
+def _existing_path(value: str | None) -> Path | None:
+    resolved = config.resolve_storage_path(value)
+    return resolved if resolved is not None and resolved.exists() else None
+
+
+def _collect_captures(db: Session) -> list[LocalObject]:
+    objects = []
+    for row in db.query(Capture).all():
+        assets = []
+        if (path := _existing_path(row.audio_path)) is not None:
+            assets.append(LocalAsset(client_asset_id=row.id, role="audio", path=path))
+        objects.append(LocalObject(kind="capture", client_id=row.id, record=_row_to_record(row), assets=assets))
+    return objects
+
+
+def _collect_generations(db: Session) -> list[LocalObject]:
+    objects = []
+    for row in db.query(Generation).filter(Generation.status == "completed").all():
+        record = _row_to_record(row)
+        assets = []
+        if (path := _existing_path(row.audio_path)) is not None:
+            assets.append(LocalAsset(client_asset_id=row.id, role="audio", path=path))
+        versions = db.query(GenerationVersion).filter(GenerationVersion.generation_id == row.id).all()
+        record["versions"] = [_row_to_record(v) for v in versions]
+        for version in versions:
+            if (path := _existing_path(version.audio_path)) is not None:
+                assets.append(LocalAsset(client_asset_id=version.id, role="version", path=path))
+        objects.append(LocalObject(kind="generation", client_id=row.id, record=record, assets=assets))
+    return objects
+
+
+def _collect_profiles(db: Session) -> list[LocalObject]:
+    objects = []
+    for row in db.query(VoiceProfile).all():
+        record = _row_to_record(row)
+        assets = []
+        if (path := _existing_path(row.avatar_path)) is not None:
+            assets.append(LocalAsset(client_asset_id=f"{row.id}-avatar", role="avatar", path=path))
+        samples = db.query(ProfileSample).filter(ProfileSample.profile_id == row.id).all()
+        record["samples"] = [_row_to_record(s) for s in samples]
+        for sample in samples:
+            if (path := _existing_path(sample.audio_path)) is not None:
+                assets.append(LocalAsset(client_asset_id=sample.id, role="sample", path=path))
+        objects.append(LocalObject(kind="profile", client_id=row.id, record=record, assets=assets))
+    return objects
+
+
+def _collect_settings(db: Session) -> list[LocalObject]:
+    objects = []
+    for client_id, model in (("capture_settings", CaptureSettings), ("generation_settings", GenerationSettings)):
+        row = db.query(model).first()
+        if row is not None:
+            objects.append(LocalObject(kind="settings", client_id=client_id, record=_row_to_record(row)))
+    return objects
+
+
+def collect_local_objects(db: Session) -> list[LocalObject]:
+    return _collect_captures(db) + _collect_generations(db) + _collect_profiles(db) + _collect_settings(db)
+
+
+# ---------------------------------------------------------------------------
+# Applying pulled records (pull side)
+
+
+def _write_asset(record_path_value: str | None, data: bytes) -> None:
+    resolved = config.resolve_storage_path(record_path_value)
+    if resolved is None:
+        return
+    resolved.parent.mkdir(parents=True, exist_ok=True)
+    resolved.write_bytes(data)
+
+
+def _apply_children(db: Session, model, parent_filter, child_records: list[dict], blobs: dict[str, bytes]) -> None:
+    """Upsert child rows (versions/samples) by id; drop local children the
+    record no longer contains; write any pulled audio next to them."""
+    wanted = {child["id"] for child in child_records}
+    for stale in db.query(model).filter(parent_filter).all():
+        if stale.id not in wanted:
+            db.delete(stale)
+    for child in child_records:
+        existing = db.query(model).filter(model.id == child["id"]).first()
+        row = _record_to_row(model, child, existing)
+        if existing is None:
+            db.add(row)
+        if child["id"] in blobs:
+            _write_asset(child.get("audio_path"), blobs[child["id"]])
+
+
+def _apply_capture(db: Session, client_id: str, record: dict, blobs: dict[str, bytes]) -> None:
+    existing = db.query(Capture).filter(Capture.id == client_id).first()
+    row = _record_to_row(Capture, record, existing)
+    if existing is None:
+        db.add(row)
+    if client_id in blobs:
+        _write_asset(record.get("audio_path"), blobs[client_id])
+
+
+def _apply_generation(db: Session, client_id: str, record: dict, blobs: dict[str, bytes]) -> None:
+    record = dict(record)
+    versions = record.pop("versions", [])
+    existing = db.query(Generation).filter(Generation.id == client_id).first()
+    row = _record_to_row(Generation, record, existing)
+    if existing is None:
+        db.add(row)
+    if client_id in blobs:
+        _write_asset(record.get("audio_path"), blobs[client_id])
+    _apply_children(db, GenerationVersion, GenerationVersion.generation_id == client_id, versions, blobs)
+
+
+def _apply_profile(db: Session, client_id: str, record: dict, blobs: dict[str, bytes]) -> None:
+    record = dict(record)
+    samples = record.pop("samples", [])
+    existing = db.query(VoiceProfile).filter(VoiceProfile.id == client_id).first()
+    row = _record_to_row(VoiceProfile, record, existing)
+    if existing is None:
+        db.add(row)
+    if f"{client_id}-avatar" in blobs:
+        _write_asset(record.get("avatar_path"), blobs[f"{client_id}-avatar"])
+    _apply_children(db, ProfileSample, ProfileSample.profile_id == client_id, samples, blobs)
+
+
+def _apply_settings(db: Session, client_id: str, record: dict) -> None:
+    model = CaptureSettings if client_id == "capture_settings" else GenerationSettings
+    existing = db.query(model).first()
+    row = _record_to_row(model, record, existing)
+    if existing is None:
+        db.add(row)
+
+
+def _apply_record(db: Session, kind: str, client_id: str, record: dict, blobs: dict[str, bytes]) -> None:
+    if kind == "capture":
+        _apply_capture(db, client_id, record, blobs)
+    elif kind == "generation":
+        _apply_generation(db, client_id, record, blobs)
+    elif kind == "profile":
+        _apply_profile(db, client_id, record, blobs)
+    elif kind == "settings":
+        _apply_settings(db, client_id, record)
+    else:
+        raise CloudSyncError(f"unknown object kind {kind!r}")
+
+
+def _delete_local(db: Session, kind: str, client_id: str) -> None:
+    if kind == "capture":
+        db.query(Capture).filter(Capture.id == client_id).delete()
+    elif kind == "generation":
+        db.query(GenerationVersion).filter(GenerationVersion.generation_id == client_id).delete()
+        db.query(Generation).filter(Generation.id == client_id).delete()
+    elif kind == "profile":
+        db.query(ProfileSample).filter(ProfileSample.profile_id == client_id).delete()
+        db.query(VoiceProfile).filter(VoiceProfile.id == client_id).delete()
+    # settings singletons are never deleted
+
+
+# ---------------------------------------------------------------------------
+# The engine
+
+
+@dataclass
+class SyncReport:
+    pushed: int = 0
+    pushed_deletes: int = 0
+    pulled: int = 0
+    pulled_deletes: int = 0
+    cursor: int = 0
+
+
+def _canonical(record: dict) -> bytes:
+    """Canonical bytes for change detection. ``updated_at`` is excluded (at the
+    top level and in embedded child rows): its ``onupdate`` trigger can bump it
+    as a side effect of *applying* a pulled record, and letting that feed back
+    into the fingerprint would bounce an already-synced object back and forth.
+    The field still syncs — it just doesn't count as a change by itself."""
+    stripped = {k: v for k, v in record.items() if k != "updated_at"}
+    for key, value in stripped.items():
+        if isinstance(value, list):
+            stripped[key] = [
+                {k: v for k, v in item.items() if k != "updated_at"} if isinstance(item, dict) else item
+                for item in value
+            ]
+    return json.dumps(stripped, sort_keys=True, separators=(",", ":")).encode()
+
+
+def _fingerprint(data: bytes) -> str:
+    return hashlib.sha256(data).hexdigest()
+
+
+def _get_state(db: Session, kind: str, client_id: str) -> CloudSyncState | None:
+    return db.query(CloudSyncState).filter(CloudSyncState.kind == kind, CloudSyncState.client_id == client_id).first()
+
+
+def _settings_row(db: Session) -> DBCloudSettings:
+    row = db.query(DBCloudSettings).filter(DBCloudSettings.id == 1).first()
+    if row is None:
+        raise CloudSyncError("not connected to Voicebox Cloud")
+    return row
+
+
+async def _push_object(
+    client: CloudApiClient,
+    db: Session,
+    master_key: bytes,
+    obj: LocalObject,
+    state: CloudSyncState | None,
+) -> bool:
+    """Push one object if it changed. Returns True when a push happened."""
+    record_payload = json.dumps(obj.record, sort_keys=True, separators=(",", ":")).encode()
+    record_fp = _fingerprint(_canonical(obj.record))
+    known_assets: dict = json.loads(state.assets_json) if state else {}
+
+    asset_plain: dict[str, bytes] = {}
+    asset_fps: dict[str, str] = {}
+    for asset in obj.assets:
+        data = asset.path.read_bytes()
+        asset_plain[asset.client_asset_id] = data
+        asset_fps[asset.client_asset_id] = _fingerprint(data)
+
+    unchanged = (
+        state is not None
+        and state.server_object_id is not None
+        and state.record_fingerprint == record_fp
+        and {k: v["fingerprint"] for k, v in known_assets.items()} == asset_fps
+    )
+    if unchanged:
+        return False
+
+    version = (state.version + 1) if state is not None else 1
+    record_env = cloud_crypto.encrypt_blob(
+        record_payload, master_key, object_id=obj.client_id, role="record", version=version
+    )
+
+    descriptors = []
+    envelopes: dict[str, bytes] = {}
+    next_assets: dict[str, dict] = {}
+    for asset in obj.assets:
+        caid = asset.client_asset_id
+        known = known_assets.get(caid)
+        if known and known["fingerprint"] == asset_fps[caid]:
+            # Content unchanged: re-declare the ciphertext the server holds.
+            entry = {"role": asset.role, "fingerprint": asset_fps[caid], "hash": known["hash"], "size": known["size"]}
+        else:
+            envelope = cloud_crypto.encrypt_blob(
+                asset_plain[caid],
+                master_key,
+                object_id=obj.client_id,
+                role=f"asset:{caid}",
+                version=_ASSET_AAD_VERSION,
+            )
+            envelopes[caid] = envelope
+            entry = {
+                "role": asset.role,
+                "fingerprint": asset_fps[caid],
+                "hash": _fingerprint(envelope),
+                "size": len(envelope),
+            }
+        next_assets[caid] = entry
+        descriptors.append({"role": asset.role, "clientAssetId": caid, "hash": entry["hash"], "size": entry["size"]})
+
+    pushed = await client.push_object(
+        kind=obj.kind,
+        client_id=obj.client_id,
+        version=version,
+        record={"hash": _fingerprint(record_env), "size": len(record_env)},
+        assets=descriptors,
+    )
+    uploads = {u["for"]: u["url"] for u in pushed["uploads"]}
+    if "record" in uploads:
+        await client.upload_blob(uploads["record"], record_env)
+    for caid, envelope in envelopes.items():
+        url = uploads.get(f"asset:{caid}")
+        if url:
+            await client.upload_blob(url, envelope)
+    await client.commit_object(pushed["objectId"])
+
+    if state is None:
+        state = CloudSyncState(kind=obj.kind, client_id=obj.client_id)
+        db.add(state)
+    state.server_object_id = pushed["objectId"]
+    state.version = version
+    state.record_fingerprint = record_fp
+    state.record_hash = _fingerprint(record_env)
+    state.record_size = len(record_env)
+    state.assets_json = json.dumps(next_assets)
+    state.last_synced_at = datetime.utcnow()
+    return True
+
+
+async def _push_all(client: CloudApiClient, db: Session, master_key: bytes, report: SyncReport) -> None:
+    local = collect_local_objects(db)
+    local_ids = {(o.kind, o.client_id) for o in local}
+
+    for obj in local:
+        if await _push_object(client, db, master_key, obj, _get_state(db, obj.kind, obj.client_id)):
+            report.pushed += 1
+            db.commit()
+
+    # Local deletions: state rows whose entity no longer exists → tombstone.
+    for state in db.query(CloudSyncState).all():
+        if (state.kind, state.client_id) not in local_ids and state.kind != "settings":
+            if state.server_object_id:
+                await client.delete_object(state.server_object_id)
+            db.delete(state)
+            report.pushed_deletes += 1
+            db.commit()
+
+
+async def _pull_changes(client: CloudApiClient, db: Session, master_key: bytes, report: SyncReport) -> None:
+    settings = _settings_row(db)
+    cursor = settings.sync_cursor or 0
+
+    while True:
+        page = await client.get_changes(since=cursor)
+        for change in page["changes"]:
+            kind, client_id = change["kind"], change["clientId"]
+            state = _get_state(db, kind, client_id)
+
+            if change["deleted"]:
+                if state is not None:
+                    _delete_local(db, kind, client_id)
+                    db.delete(state)
+                    report.pulled_deletes += 1
+            # Our own pushes echo back through the feed; the stored ciphertext
+            # hash identifies them as already applied.
+            elif change["record"] and (state is None or state.record_hash != change["record"]["hash"]):
+                await _apply_change(client, db, master_key, change, state)
+                report.pulled += 1
+
+            cursor = change["seq"]
+
+        settings.sync_cursor = cursor
+        db.commit()
+        report.cursor = cursor
+        if not page["hasMore"]:
+            break
+
+
+async def _apply_change(
+    client: CloudApiClient,
+    db: Session,
+    master_key: bytes,
+    change: dict,
+    state: CloudSyncState | None,
+) -> None:
+    kind, client_id = change["kind"], change["clientId"]
+    record_cipher = await client.download_blob(change["record"]["url"])
+    record = json.loads(
+        cloud_crypto.decrypt_blob(
+            record_cipher, master_key, object_id=client_id, role="record", version=change["version"]
+        )
+    )
+
+    known_assets: dict = json.loads(state.assets_json) if state else {}
+    blobs: dict[str, bytes] = {}
+    next_assets: dict[str, dict] = {}
+    for asset in change["assets"]:
+        caid = asset["clientAssetId"]
+        known = known_assets.get(caid)
+        if known and known["hash"] == asset["hash"]:
+            next_assets[caid] = known
+            continue  # ciphertext we already hold locally
+        if not asset["url"]:
+            continue  # declared but never uploaded; skip until it lands
+        cipher = await client.download_blob(asset["url"])
+        plain = cloud_crypto.decrypt_blob(
+            cipher, master_key, object_id=client_id, role=f"asset:{caid}", version=_ASSET_AAD_VERSION
+        )
+        blobs[caid] = plain
+        next_assets[caid] = {
+            "role": asset["role"],
+            "fingerprint": _fingerprint(plain),
+            "hash": asset["hash"],
+            "size": asset["size"],
+        }
+
+    _apply_record(db, kind, client_id, record, blobs)
+
+    if state is None:
+        state = CloudSyncState(kind=kind, client_id=client_id)
+        db.add(state)
+    state.server_object_id = change["id"]
+    state.version = change["version"]
+    state.record_fingerprint = _fingerprint(_canonical(record))
+    state.record_hash = change["record"]["hash"]
+    state.record_size = change["record"]["size"]
+    state.assets_json = json.dumps(next_assets)
+    state.last_synced_at = datetime.utcnow()
+
+
+async def run_sync(db: Session) -> SyncReport:
+    """One full sync: push local changes, then pull and apply remote ones."""
+    settings = _settings_row(db)
+    master_key = cloud_account.load_master_key(db)
+    report = SyncReport()
+
+    async with CloudApiClient(config.get_cloud_api_url(), settings.api_key) as client:
+        await _push_all(client, db, master_key, report)
+        await _pull_changes(client, db, master_key, report)
+
+    logger.info(
+        "cloud sync: pushed %d (+%d deletes), pulled %d (+%d deletes), cursor %d",
+        report.pushed,
+        report.pushed_deletes,
+        report.pulled,
+        report.pulled_deletes,
+        report.cursor,
+    )
+    return report

backend/tests/fake_cloud.py

@@ -0,0 +1,188 @@
+"""An in-process fake of the voicebox-cloud API for tests.
+
+Implements the surface the desktop sync client uses — devices, the account-key
+escrow, the encrypted object store, the sync feed, and blob storage — behind an
+httpx.MockTransport, mirroring apps/api in the voicebox-cloud repo. Every
+request body is kept in ``seen_bodies`` so tests can assert what the server was
+shown (never key material, never plaintext).
+"""
+
+import json
+
+import httpx
+
+STORAGE_HOST = "http://cloud.test/__storage/"
+
+
+class FakeCloud:
+    def __init__(self):
+        self.devices: dict[str, dict] = {}
+        self.account_key: dict | None = None
+        self.objects: dict[str, dict] = {}  # objectId -> row (incl. assets dict)
+        self.storage: dict[str, bytes] = {}  # key -> ciphertext
+        self.seen_bodies: list[bytes] = []
+        self._next_device = 0
+        self._next_object = 0
+        self._seq = 0
+
+    def transport(self) -> httpx.MockTransport:
+        return httpx.MockTransport(self.handle)
+
+    # -- helpers --------------------------------------------------------------
+
+    @staticmethod
+    def _ok(data, status=200):
+        return httpx.Response(status, json={"ok": True, "data": data})
+
+    def _seq_next(self) -> int:
+        self._seq += 1
+        return self._seq
+
+    def _find_object(self, kind: str, client_id: str) -> dict | None:
+        return next((o for o in self.objects.values() if o["kind"] == kind and o["clientId"] == client_id), None)
+
+    # -- request routing --------------------------------------------------------
+
+    def handle(self, request: httpx.Request) -> httpx.Response:
+        if request.content:
+            self.seen_bodies.append(request.content)
+        path, method = request.url.path, request.method
+
+        if path.startswith("/__storage/"):
+            key = path[len("/__storage/") :]
+            if method == "PUT":
+                self.storage[key] = request.content
+                return httpx.Response(200)
+            data = self.storage.get(key)
+            return httpx.Response(200, content=data) if data is not None else httpx.Response(404)
+
+        if path == "/v1/devices" and method == "POST":
+            return self._register_device(json.loads(request.content))
+        if path == "/v1/devices" and method == "GET":
+            return self._ok(list(self.devices.values()))
+        if path == "/v1/devices/account-key" and method == "PUT":
+            self.account_key = json.loads(request.content)
+            return self._ok(None)
+        if path == "/v1/devices/account-key" and method == "GET":
+            return self._ok(self.account_key)
+        if path.startswith("/v1/devices/") and path.endswith("/wrapped-key"):
+            device_id = path.split("/")[3]
+            if method == "POST":
+                self.devices[device_id]["wrappedMasterKey"] = json.loads(request.content)["wrappedMasterKey"]
+                return self._ok(None)
+            return self._ok({"wrappedMasterKey": self.devices[device_id]["wrappedMasterKey"]})
+
+        if path == "/v1/objects" and method == "POST":
+            return self._upsert_object(json.loads(request.content))
+        if path.startswith("/v1/objects/") and path.endswith("/commit"):
+            return self._commit(path.split("/")[3])
+        if path.startswith("/v1/objects/") and method == "DELETE":
+            obj = self.objects.get(path.split("/")[3])
+            if obj:
+                obj["deleted"] = True
+                obj["seq"] = self._seq_next()
+            return self._ok(None)
+
+        if path == "/v1/sync/changes" and method == "GET":
+            return self._changes(request.url.params)
+
+        return httpx.Response(404, json={"ok": False, "error": {"message": f"unhandled {method} {path}"}})
+
+    # -- endpoint implementations ----------------------------------------------
+
+    def _register_device(self, body: dict) -> httpx.Response:
+        self._next_device += 1
+        device_id = f"dev-{self._next_device}"
+        self.devices[device_id] = {
+            "id": device_id,
+            "name": body["name"],
+            "publicKey": body["publicKey"],
+            "wrappedMasterKey": None,
+            "revokedAt": None,
+        }
+        return self._ok({"deviceId": device_id, "accountHasKey": self.account_key is not None}, 201)
+
+    def _upsert_object(self, body: dict) -> httpx.Response:
+        obj = self._find_object(body["kind"], body["clientId"])
+        if obj is None:
+            self._next_object += 1
+            obj = {
+                "id": f"obj-{self._next_object}",
+                "kind": body["kind"],
+                "clientId": body["clientId"],
+                "version": 0,
+                "deleted": False,
+                "record": None,
+                "assets": {},
+            }
+            self.objects[obj["id"]] = obj
+
+        obj["version"] = max(obj["version"], body["version"])
+        obj["seq"] = self._seq_next()
+        obj["deleted"] = False
+
+        uploads = []
+        record = body.get("record")
+        if record and (obj["record"] is None or obj["record"]["hash"] != record["hash"]):
+            key = f"o/{obj['id']}/record"
+            obj["record"] = {**record, "key": key}
+            uploads.append({"for": "record", "key": key, "url": STORAGE_HOST + key})
+        for asset in body.get("assets", []):
+            caid = asset["clientAssetId"]
+            existing = obj["assets"].get(caid)
+            if existing is None or existing["hash"] != asset["hash"]:
+                key = f"o/{obj['id']}/a/{caid}"
+                obj["assets"][caid] = {**asset, "key": key}
+                uploads.append({"for": f"asset:{caid}", "key": key, "url": STORAGE_HOST + key})
+        return self._ok({"objectId": obj["id"], "seq": obj["seq"], "uploads": uploads}, 201)
+
+    def _commit(self, object_id: str) -> httpx.Response:
+        obj = self.objects.get(object_id)
+        if obj is None:
+            return httpx.Response(404, json={"ok": False, "error": {"message": "object not found"}})
+        missing = []
+        if obj["record"] and obj["record"]["key"] not in self.storage:
+            missing.append("record")
+        for caid, asset in obj["assets"].items():
+            if asset["key"] not in self.storage:
+                missing.append(f"asset:{caid}")
+        if missing:
+            return httpx.Response(409, json={"ok": False, "error": {"message": f"uploads missing: {missing}"}})
+        return self._ok({"objectId": object_id, "committed": True})
+
+    def _changes(self, params) -> httpx.Response:
+        since = int(params.get("since", 0))
+        limit = int(params.get("limit", 200))
+        rows = sorted((o for o in self.objects.values() if o["seq"] > since), key=lambda o: o["seq"])[:limit]
+        changes = [
+            {
+                "id": o["id"],
+                "kind": o["kind"],
+                "clientId": o["clientId"],
+                "version": o["version"],
+                "seq": o["seq"],
+                "deleted": o["deleted"],
+                "record": (
+                    {
+                        "hash": o["record"]["hash"],
+                        "size": o["record"]["size"],
+                        "url": STORAGE_HOST + o["record"]["key"],
+                    }
+                    if o["record"]
+                    else None
+                ),
+                "assets": [
+                    {
+                        "clientAssetId": caid,
+                        "role": a["role"],
+                        "hash": a["hash"],
+                        "size": a["size"],
+                        "url": STORAGE_HOST + a["key"] if a["key"] in self.storage else None,
+                    }
+                    for caid, a in o["assets"].items()
+                ],
+            }
+            for o in rows
+        ]
+        cursor = rows[-1]["seq"] if rows else since
+        return self._ok({"changes": changes, "cursor": cursor, "hasMore": len(rows) == limit})

backend/tests/test_cloud_account.py

@@ -6,10 +6,8 @@ the master key and recovery phrase never appear in anything sent to the server.
 """
 
 import base64
-import json
 from datetime import datetime
 
-import httpx
 import keyring
 import keyring.backend
 import pytest
@@ -20,6 +18,7 @@ from backend.database.models import Base, CloudSettings
 from backend.services import cloud_account, cloud_crypto, cloud_keys
 from backend.services.cloud_account import CloudAccountError
 from backend.services.cloud_api import CloudApiClient
+from backend.tests.fake_cloud import FakeCloud
 
 USER_A = "user-a"
 
@@ -41,57 +40,6 @@ class InMemoryKeyring(keyring.backend.KeyringBackend):
         self.store.pop((service, username), None)
 
 
-class FakeCloud:
-    """Just enough of the cloud API for the identity flows, plus a transcript
-    of every request body so tests can assert what the server was shown."""
-
-    def __init__(self):
-        self.devices: dict[str, dict] = {}
-        self.account_key: dict | None = None
-        self.seen_bodies: list[bytes] = []
-        self._next_id = 0
-
-    def transport(self) -> httpx.MockTransport:
-        return httpx.MockTransport(self.handle)
-
-    def handle(self, request: httpx.Request) -> httpx.Response:
-        if request.content:
-            self.seen_bodies.append(request.content)
-        path, method = request.url.path, request.method
-
-        if path == "/v1/devices" and method == "POST":
-            body = json.loads(request.content)
-            self._next_id += 1
-            device_id = f"dev-{self._next_id}"
-            self.devices[device_id] = {
-                "id": device_id,
-                "name": body["name"],
-                "publicKey": body["publicKey"],
-                "wrappedMasterKey": None,
-                "revokedAt": None,
-            }
-            return self._ok({"deviceId": device_id, "accountHasKey": self.account_key is not None}, 201)
-        if path == "/v1/devices" and method == "GET":
-            return self._ok(list(self.devices.values()))
-        if path == "/v1/devices/account-key" and method == "PUT":
-            self.account_key = json.loads(request.content)
-            return self._ok(None)
-        if path == "/v1/devices/account-key" and method == "GET":
-            return self._ok(self.account_key)
-        if path.endswith("/wrapped-key") and method == "POST":
-            device_id = path.split("/")[3]
-            self.devices[device_id]["wrappedMasterKey"] = json.loads(request.content)["wrappedMasterKey"]
-            return self._ok(None)
-        if path.endswith("/wrapped-key") and method == "GET":
-            device_id = path.split("/")[3]
-            return self._ok({"wrappedMasterKey": self.devices[device_id]["wrappedMasterKey"]})
-        return httpx.Response(404, json={"ok": False, "error": {"message": f"unhandled {method} {path}"}})
-
-    @staticmethod
-    def _ok(data, status=200):
-        return httpx.Response(status, json={"ok": True, "data": data})
-
-
 @pytest.fixture
 def fake_keyring(monkeypatch):
     backend = InMemoryKeyring()

backend/tests/test_cloud_sync.py

@@ -0,0 +1,278 @@
+"""Tests for the cloud sync engine (services/cloud_sync.py).
+
+Simulates two installs ("machines") of the desktop app — each with its own
+SQLite database, data directory, and keychain — syncing through the in-process
+FakeCloud. Covers the full backup → restore path, incremental pushes, deletes,
+and the blindness invariant (nothing plaintext ever lands in server storage).
+"""
+
+from datetime import datetime
+
+import keyring
+import pytest
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+from backend import config
+from backend.database.models import (
+    Base,
+    Capture,
+    CaptureSettings,
+    CloudSettings,
+    Generation,
+    GenerationVersion,
+    ProfileSample,
+    VoiceProfile,
+)
+from backend.services import cloud_account, cloud_sync
+from backend.services.cloud_api import CloudApiClient
+from backend.tests.fake_cloud import FakeCloud
+
+AUDIO_A = b"RIFFfake-capture-audio" + b"\x11" * 4000
+AUDIO_B = b"RIFFfake-generation-audio" + b"\x22" * 4000
+AUDIO_C = b"RIFFfake-version-audio" + b"\x33" * 4000
+AUDIO_D = b"RIFFfake-sample-audio" + b"\x44" * 4000
+AVATAR = b"\x89PNGfake-avatar" + b"\x55" * 500
+
+
+class Install:
+    """One simulated machine: its own DB, data dir, and keychain store."""
+
+    def __init__(self, root, name: str):
+        self.data_dir = root / name
+        self.data_dir.mkdir()
+        self.keychain: dict[tuple[str, str], str] = {}
+        engine = create_engine("sqlite://")
+        Base.metadata.create_all(engine)
+        self.db = sessionmaker(bind=engine)()
+        self.db.add(
+            CloudSettings(
+                id=1,
+                api_key=f"voicebox_{name}",
+                device_name=name,
+                account_user_id="user-1",
+                connected_at=datetime(2026, 7, 1),
+            )
+        )
+        self.db.commit()
+
+    def activate(self, monkeypatch):
+        """Point global config + keychain at this machine."""
+        config.set_data_dir(self.data_dir)
+        store = self.keychain
+        monkeypatch.setattr(keyring, "get_password", lambda s, u: store.get((s, u)))
+        monkeypatch.setattr(keyring, "set_password", lambda s, u, p: store.__setitem__((s, u), p))
+        monkeypatch.setattr(keyring, "delete_password", lambda s, u: store.pop((s, u), None))
+
+    def write_file(self, relative: str, data: bytes) -> str:
+        path = self.data_dir / relative
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_bytes(data)
+        return str(path)
+
+
+@pytest.fixture
+def cloud(monkeypatch):
+    fake = FakeCloud()
+    monkeypatch.setattr(
+        cloud_sync,
+        "CloudApiClient",
+        lambda url, key: CloudApiClient(url, key, transport=fake.transport()),
+    )
+    monkeypatch.setattr(
+        cloud_account,
+        "_client",
+        lambda row: CloudApiClient("http://cloud.test", row.api_key, transport=fake.transport()),
+    )
+    return fake
+
+
+def seed_content(install: Install) -> None:
+    db = install.db
+    profile = VoiceProfile(
+        id="prof-1",
+        name="Morgan",
+        description="test voice",
+        language="en",
+        avatar_path=install.write_file("profiles/prof-1/avatar.png", AVATAR),
+        personality="dry wit",
+    )
+    db.add(profile)
+    db.add(
+        ProfileSample(
+            id="samp-1",
+            profile_id="prof-1",
+            audio_path=install.write_file("profiles/prof-1/samples/samp-1.wav", AUDIO_D),
+            reference_text="hello there",
+        )
+    )
+    db.add(
+        Capture(
+            id="cap-1",
+            audio_path=install.write_file("captures/cap-1.wav", AUDIO_A),
+            source="dictation",
+            language="en",
+            transcript_raw="the raw transcript",
+            transcript_refined="the refined transcript",
+        )
+    )
+    db.add(
+        Generation(
+            id="gen-1",
+            profile_id="prof-1",
+            text="hello world",
+            audio_path=install.write_file("generations/gen-1.wav", AUDIO_B),
+            status="completed",
+            source="manual",
+        )
+    )
+    db.add(
+        GenerationVersion(
+            id="ver-1",
+            generation_id="gen-1",
+            label="Take 2",
+            audio_path=install.write_file("generations/ver-1.wav", AUDIO_C),
+        )
+    )
+    db.add(CaptureSettings(id=1, stt_model="turbo", language="auto"))
+    db.commit()
+
+
+async def connect(install: Install, monkeypatch, phrase: str | None = None) -> str | None:
+    install.activate(monkeypatch)
+    result = await cloud_account.setup_device(install.db)
+    if phrase is not None:
+        await cloud_account.restore_with_phrase(install.db, phrase)
+    return result
+
+
+class TestSyncEngine:
+    async def test_backup_then_restore_on_second_machine(self, cloud, tmp_path, monkeypatch):
+        a = Install(tmp_path, "machine-a")
+        phrase = await connect(a, monkeypatch)
+        seed_content(a)
+
+        report = await cloud_sync.run_sync(a.db)
+        assert report.pushed == 4  # capture, generation, profile, capture_settings
+        assert report.pulled == 0  # own echoes are recognized by ciphertext hash
+
+        # Server blindness: every stored blob is a VBX1 envelope, no plaintext.
+        assert cloud.storage
+        for blob in cloud.storage.values():
+            assert blob[:4] == b"VBX1"
+            assert b"transcript" not in blob
+            assert AUDIO_A not in blob
+        for body in cloud.seen_bodies:
+            assert b"the raw transcript" not in body
+            assert b"Morgan" not in body
+
+        # Fresh machine: restore identity from phrase, then pull everything.
+        b = Install(tmp_path, "machine-b")
+        await connect(b, monkeypatch, phrase=phrase)
+        report_b = await cloud_sync.run_sync(b.db)
+        assert report_b.pulled == report.pushed
+
+        cap = b.db.query(Capture).one()
+        assert cap.id == "cap-1"
+        assert cap.transcript_raw == "the raw transcript"
+        assert (b.data_dir / "captures/cap-1.wav").read_bytes() == AUDIO_A
+
+        prof = b.db.query(VoiceProfile).one()
+        assert prof.name == "Morgan"
+        assert prof.personality == "dry wit"
+        assert (b.data_dir / "profiles/prof-1/avatar.png").read_bytes() == AVATAR
+        samp = b.db.query(ProfileSample).one()
+        assert samp.reference_text == "hello there"
+        assert (b.data_dir / "profiles/prof-1/samples/samp-1.wav").read_bytes() == AUDIO_D
+
+        gen = b.db.query(Generation).one()
+        assert gen.text == "hello world"
+        assert (b.data_dir / "generations/gen-1.wav").read_bytes() == AUDIO_B
+        ver = b.db.query(GenerationVersion).one()
+        assert ver.label == "Take 2"
+        assert (b.data_dir / "generations/ver-1.wav").read_bytes() == AUDIO_C
+
+        settings = b.db.query(CaptureSettings).one()
+        assert settings.stt_model == "turbo"
+
+        # Second sync on B is a no-op in both directions.
+        report_b2 = await cloud_sync.run_sync(b.db)
+        assert (report_b2.pushed, report_b2.pulled) == (0, 0)
+
+    async def test_incremental_edit_propagates(self, cloud, tmp_path, monkeypatch):
+        a = Install(tmp_path, "machine-a")
+        phrase = await connect(a, monkeypatch)
+        seed_content(a)
+        await cloud_sync.run_sync(a.db)
+
+        b = Install(tmp_path, "machine-b")
+        await connect(b, monkeypatch, phrase=phrase)
+        await cloud_sync.run_sync(b.db)
+
+        # Edit on A: only the capture should push, and only its record blob
+        # should re-upload (the audio is unchanged).
+        a.activate(monkeypatch)
+        blobs_before = dict(cloud.storage)
+        cap = a.db.query(Capture).one()
+        cap.transcript_refined = "edited on machine A"
+        a.db.commit()
+        report_a = await cloud_sync.run_sync(a.db)
+        assert report_a.pushed == 1
+        changed_keys = [k for k, v in cloud.storage.items() if blobs_before.get(k) != v]
+        assert changed_keys == [k for k in changed_keys if k.endswith("/record")]
+
+        b.activate(monkeypatch)
+        report_b = await cloud_sync.run_sync(b.db)
+        assert report_b.pulled == 1
+        assert b.db.query(Capture).one().transcript_refined == "edited on machine A"
+
+    async def test_delete_propagates(self, cloud, tmp_path, monkeypatch):
+        a = Install(tmp_path, "machine-a")
+        phrase = await connect(a, monkeypatch)
+        seed_content(a)
+        await cloud_sync.run_sync(a.db)
+
+        b = Install(tmp_path, "machine-b")
+        await connect(b, monkeypatch, phrase=phrase)
+        await cloud_sync.run_sync(b.db)
+
+        a.activate(monkeypatch)
+        cap = a.db.query(Capture).one()
+        a.db.delete(cap)
+        a.db.commit()
+        report_a = await cloud_sync.run_sync(a.db)
+        assert report_a.pushed_deletes == 1
+
+        b.activate(monkeypatch)
+        report_b = await cloud_sync.run_sync(b.db)
+        assert report_b.pulled_deletes == 1
+        assert b.db.query(Capture).count() == 0
+
+    async def test_last_writer_wins_on_conflict(self, cloud, tmp_path, monkeypatch):
+        a = Install(tmp_path, "machine-a")
+        phrase = await connect(a, monkeypatch)
+        seed_content(a)
+        await cloud_sync.run_sync(a.db)
+
+        b = Install(tmp_path, "machine-b")
+        await connect(b, monkeypatch, phrase=phrase)
+        await cloud_sync.run_sync(b.db)
+
+        # Concurrent edits to the same capture on both machines.
+        a.activate(monkeypatch)
+        a.db.query(Capture).one().transcript_refined = "A's edit"
+        a.db.commit()
+        await cloud_sync.run_sync(a.db)
+
+        b.activate(monkeypatch)
+        b.db.query(Capture).one().transcript_refined = "B's edit"
+        b.db.commit()
+        await cloud_sync.run_sync(b.db)  # B pushes after A: B is the last writer
+
+        a.activate(monkeypatch)
+        await cloud_sync.run_sync(a.db)
+        assert a.db.query(Capture).one().transcript_refined == "B's edit"
+
+        b.activate(monkeypatch)
+        await cloud_sync.run_sync(b.db)
+        assert b.db.query(Capture).one().transcript_refined == "B's edit"