Add encrypted backup UI to the Cloud settings section

Three states under the existing login row: enable backup (registers the
device; when this device mints the account key the recovery phrase is
force-displayed in a dialog that only dismisses via "I've written it
down"), awaiting provision (check-again for approval from another
device, or restore by typing the phrase), and ready (sync now with a
pushed/pulled report toast).

app/src/components/ServerTab/CloudSection.tsx

@@ -1,7 +1,16 @@
 import { useMutation, useQuery, useQueryClient } from '@tanstack/react-query';
-import { Cloud, Loader2 } from 'lucide-react';
+import { Cloud, Copy, Loader2, RefreshCw, ShieldCheck } from 'lucide-react';
 import { useEffect, useState } from 'react';
 import { Button } from '@/components/ui/button';
+import {
+  Dialog,
+  DialogContent,
+  DialogDescription,
+  DialogFooter,
+  DialogHeader,
+  DialogTitle,
+} from '@/components/ui/dialog';
+import { Input } from '@/components/ui/input';
 import { useToast } from '@/components/ui/use-toast';
 import { apiClient } from '@/lib/api/client';
 import { SettingRow, SettingSection } from './SettingRow';
@@ -9,6 +18,13 @@ import { SettingRow, SettingSection } from './SettingRow';
 // "Log in with browser" device pairing. The backend opens the system browser
 // and completes the code exchange; here we just kick it off and poll status
 // until the link goes live. The API key never touches the frontend.
+//
+// Once linked, the encrypted-backup rows drive the sync identity flows: this
+// device registers as an encryption device, and either mints the account's
+// master key (first device — the recovery phrase is force-displayed exactly
+// once) or waits to be provisioned by another device / restored from the
+// phrase. All crypto happens in the local backend; this UI only ever sees the
+// phrase, and only at mint time.
 export function CloudSection() {
   const { toast } = useToast();
   const queryClient = useQueryClient();
@@ -33,22 +49,6 @@ export function CloudSection() {
     }
   }, [connected, polling, status?.device_name, toast]);
 
-  // Give up after two minutes so an abandoned browser flow doesn't leave the
-  // button stuck on "Waiting for browser…". The backend state stays valid for
-  // ten, so the user can simply start again.
-  useEffect(() => {
-    if (!polling) return;
-    const timeoutId = window.setTimeout(() => {
-      setPolling(false);
-      toast({
-        title: 'Sign-in timed out',
-        description: 'The browser sign-in was not completed. Try again.',
-        variant: 'destructive',
-      });
-    }, 120_000);
-    return () => window.clearTimeout(timeoutId);
-  }, [polling, toast]);
-
   const startLogin = useMutation({
     mutationFn: () => apiClient.startCloudLogin(),
     onSuccess: () => {
@@ -70,10 +70,10 @@ export function CloudSection() {
     mutationFn: () => apiClient.disconnectCloud(),
     onSuccess: () => {
       queryClient.invalidateQueries({ queryKey: ['cloud-status'] });
+      queryClient.invalidateQueries({ queryKey: ['cloud-sync-status'] });
       toast({
         title: 'Disconnected',
-        description:
-          'This device is no longer linked. The key stays valid until revoked in your account.',
+        description: 'This device is no longer linked. The key stays valid until revoked in your account.',
       });
     },
     onError: (error: Error) =>
@@ -105,10 +105,7 @@ export function CloudSection() {
               variant="outline"
             >
               {disconnect.isPending ? (
-                <>
-                  <Loader2 className="h-3.5 w-3.5 mr-1.5 animate-spin" />
-                  Disconnecting…
-                </>
+                <Loader2 className="h-3.5 w-3.5 animate-spin" />
               ) : (
                 'Disconnect'
               )}
@@ -131,6 +128,8 @@ export function CloudSection() {
         }
       />
 
+      {connected && <BackupRows />}
+
       {connected && (
         <SettingRow
           title="Manage"
@@ -138,7 +137,7 @@ export function CloudSection() {
         >
           <a
             className="text-sm text-accent hover:underline"
-            href={status?.dashboard_url ?? 'https://voicebox.sh/account'}
+            href="https://voicebox.sh/account"
             rel="noopener noreferrer"
             target="_blank"
           >
@@ -149,3 +148,230 @@ export function CloudSection() {
     </SettingSection>
   );
 }
+
+function BackupRows() {
+  const { toast } = useToast();
+  const queryClient = useQueryClient();
+  const [phrase, setPhrase] = useState<string | null>(null);
+  const [phraseCopied, setPhraseCopied] = useState(false);
+  const [restoreInput, setRestoreInput] = useState('');
+
+  const { data: sync } = useQuery({
+    queryKey: ['cloud-sync-status'],
+    queryFn: () => apiClient.getCloudSyncStatus(),
+  });
+
+  const refreshSync = () => queryClient.invalidateQueries({ queryKey: ['cloud-sync-status'] });
+
+  const setup = useMutation({
+    mutationFn: () => apiClient.setupCloudSync(),
+    onSuccess: (result) => {
+      refreshSync();
+      if (result.recovery_phrase) {
+        // First device: the phrase exists only in this response. Force-display
+        // it; the dialog can only be dismissed by confirming.
+        setPhraseCopied(false);
+        setPhrase(result.recovery_phrase);
+      }
+    },
+    onError: (error: Error) =>
+      toast({ title: 'Could not enable backup', description: error.message, variant: 'destructive' }),
+  });
+
+  const adopt = useMutation({
+    mutationFn: () => apiClient.adoptCloudSync(),
+    onSuccess: (result) => {
+      refreshSync();
+      if (result.status === 'ready') {
+        toast({ title: 'Backup enabled', description: 'This device received its encryption key.' });
+      } else {
+        toast({
+          title: 'Not approved yet',
+          description: 'Approve this device from another synced device, then check again.',
+        });
+      }
+    },
+    onError: (error: Error) =>
+      toast({ title: 'Could not check', description: error.message, variant: 'destructive' }),
+  });
+
+  const restore = useMutation({
+    mutationFn: (recoveryPhrase: string) => apiClient.restoreCloudSync(recoveryPhrase),
+    onSuccess: () => {
+      refreshSync();
+      setRestoreInput('');
+      toast({ title: 'Backup restored', description: 'Your encryption key was recovered. Syncing is ready.' });
+    },
+    onError: (error: Error) =>
+      toast({ title: 'Could not restore', description: error.message, variant: 'destructive' }),
+  });
+
+  const run = useMutation({
+    mutationFn: () => apiClient.runCloudSync(),
+    onSuccess: (report) => {
+      refreshSync();
+      const pushed = report.pushed + report.pushed_deletes;
+      const pulled = report.pulled + report.pulled_deletes;
+      toast({
+        title: 'Sync complete',
+        description:
+          pushed === 0 && pulled === 0
+            ? 'Everything is already up to date.'
+            : `Backed up ${pushed} ${pushed === 1 ? 'item' : 'items'}, received ${pulled}.`,
+      });
+    },
+    onError: (error: Error) =>
+      toast({ title: 'Sync failed', description: error.message, variant: 'destructive' }),
+  });
+
+  const state = sync?.status ?? 'unregistered';
+
+  return (
+    <>
+      {state === 'unregistered' && (
+        <SettingRow
+          title="Encrypted backup"
+          description="Back up captures, generations, and voice profiles — encrypted on this device before anything is uploaded."
+          action={
+            <Button disabled={setup.isPending} onClick={() => setup.mutate()} size="sm">
+              {setup.isPending ? (
+                <Loader2 className="h-3.5 w-3.5 animate-spin" />
+              ) : (
+                <>
+                  <ShieldCheck className="h-3.5 w-3.5 mr-1.5" />
+                  Enable backup
+                </>
+              )}
+            </Button>
+          }
+        />
+      )}
+
+      {state === 'awaiting_provision' && (
+        <>
+          <SettingRow
+            title="Waiting for encryption key"
+            description="This account already has a backup. Approve this device from another synced device, or restore with your recovery phrase below."
+            action={
+              <Button disabled={adopt.isPending} onClick={() => adopt.mutate()} size="sm" variant="outline">
+                {adopt.isPending ? (
+                  <Loader2 className="h-3.5 w-3.5 animate-spin" />
+                ) : (
+                  <>
+                    <RefreshCw className="h-3.5 w-3.5 mr-1.5" />
+                    Check again
+                  </>
+                )}
+              </Button>
+            }
+          />
+          <SettingRow
+            title="Restore with recovery phrase"
+            description="The 12 words you wrote down when you first enabled backup."
+            action={
+              <div className="flex items-center gap-2">
+                <Input
+                  className="w-72"
+                  onChange={(e) => setRestoreInput(e.target.value)}
+                  placeholder="correct horse battery staple …"
+                  value={restoreInput}
+                />
+                <Button
+                  disabled={restore.isPending || restoreInput.trim().split(/\s+/).length < 12}
+                  onClick={() => restore.mutate(restoreInput)}
+                  size="sm"
+                >
+                  {restore.isPending ? <Loader2 className="h-3.5 w-3.5 animate-spin" /> : 'Restore'}
+                </Button>
+              </div>
+            }
+          />
+        </>
+      )}
+
+      {state === 'ready' && (
+        <SettingRow
+          title="Encrypted backup"
+          description="On — content is encrypted on this device before upload. The server can never read it."
+          action={
+            <Button disabled={run.isPending} onClick={() => run.mutate()} size="sm">
+              {run.isPending ? (
+                <>
+                  <Loader2 className="h-3.5 w-3.5 mr-1.5 animate-spin" />
+                  Syncing…
+                </>
+              ) : (
+                <>
+                  <RefreshCw className="h-3.5 w-3.5 mr-1.5" />
+                  Sync now
+                </>
+              )}
+            </Button>
+          }
+        />
+      )}
+
+      <RecoveryPhraseDialog
+        copied={phraseCopied}
+        onConfirm={() => setPhrase(null)}
+        onCopy={() => {
+          if (phrase) {
+            navigator.clipboard?.writeText(phrase);
+            setPhraseCopied(true);
+          }
+        }}
+        phrase={phrase}
+      />
+    </>
+  );
+}
+
+// The one moment the recovery phrase exists outside the backend. Deliberately
+// hard to dismiss: no overlay/escape close, only the explicit confirmation.
+function RecoveryPhraseDialog({
+  phrase,
+  copied,
+  onCopy,
+  onConfirm,
+}: {
+  phrase: string | null;
+  copied: boolean;
+  onCopy: () => void;
+  onConfirm: () => void;
+}) {
+  return (
+    <Dialog onOpenChange={() => {}} open={phrase !== null}>
+      <DialogContent className="max-w-lg [&>button]:hidden">
+        <DialogHeader>
+          <DialogTitle>Write down your recovery phrase</DialogTitle>
+          <DialogDescription>
+            These 12 words are the only way to restore your backup if you lose all your devices.
+            They are never sent to the server and will not be shown again.
+          </DialogDescription>
+        </DialogHeader>
+        <div className="grid grid-cols-3 gap-2 py-2">
+          {(phrase ?? '').split(' ').map((word, index) => (
+            <div
+              className="rounded-md border border-border bg-muted/40 px-2.5 py-1.5 text-sm"
+              // Position is the identity here — BIP39 phrases can repeat words.
+              // biome-ignore lint/suspicious/noArrayIndexKey: static list, never reordered
+              key={index}
+            >
+              <span className="mr-1.5 text-muted-foreground tabular-nums">{index + 1}.</span>
+              {word}
+            </div>
+          ))}
+        </div>
+        <DialogFooter className="gap-2 sm:justify-between">
+          <Button onClick={onCopy} size="sm" type="button" variant="outline">
+            <Copy className="h-3.5 w-3.5 mr-1.5" />
+            {copied ? 'Copied' : 'Copy'}
+          </Button>
+          <Button onClick={onConfirm} size="sm" type="button">
+            I've written it down
+          </Button>
+        </DialogFooter>
+      </DialogContent>
+    </Dialog>
+  );
+}

app/src/lib/api/client.ts

@@ -52,6 +52,9 @@ import type {
   MCPClientBindingUpsert,
   CloudLoginStartResponse,
   CloudStatus,
+  CloudSyncRunResponse,
+  CloudSyncSetupResponse,
+  CloudSyncStatus,
 } from './types';
 
 function formatErrorDetail(detail: unknown, fallback: string): string {
@@ -937,6 +940,32 @@ class ApiClient {
   async disconnectCloud(): Promise<CloudStatus> {
     return this.request<CloudStatus>('/cloud/disconnect', { method: 'POST' });
   }
+
+  // Encrypted backup & sync. setupCloudSync registers this install as an
+  // encryption device — when it returns a recovery_phrase, this device just
+  // minted the account key and the phrase must be force-displayed once.
+  async getCloudSyncStatus(): Promise<CloudSyncStatus> {
+    return this.request<CloudSyncStatus>('/cloud/sync/status');
+  }
+
+  async setupCloudSync(): Promise<CloudSyncSetupResponse> {
+    return this.request<CloudSyncSetupResponse>('/cloud/sync/setup', { method: 'POST' });
+  }
+
+  async restoreCloudSync(phrase: string): Promise<CloudSyncStatus> {
+    return this.request<CloudSyncStatus>('/cloud/sync/restore', {
+      method: 'POST',
+      body: JSON.stringify({ phrase }),
+    });
+  }
+
+  async adoptCloudSync(): Promise<CloudSyncStatus> {
+    return this.request<CloudSyncStatus>('/cloud/sync/adopt', { method: 'POST' });
+  }
+
+  async runCloudSync(): Promise<CloudSyncRunResponse> {
+    return this.request<CloudSyncRunResponse>('/cloud/sync/run', { method: 'POST' });
+  }
 }
 
 export const apiClient = new ApiClient();

app/src/lib/api/types.ts

@@ -534,5 +534,28 @@ export interface CloudStatus {
   account_user_id: string | null;
   key_prefix: string | null;
   connected_at: string | null;
-  dashboard_url: string;
+}
+
+export type CloudSyncState = 'unregistered' | 'awaiting_provision' | 'ready';
+
+export interface CloudSyncStatus {
+  status: CloudSyncState;
+  device_id: string | null;
+  sync_cursor: number;
+}
+
+export interface CloudSyncSetupResponse {
+  status: CloudSyncState;
+  device_id: string | null;
+  /** Present exactly once, when this device minted the account's key
+   *  material. Must be force-displayed and never persisted. */
+  recovery_phrase: string | null;
+}
+
+export interface CloudSyncRunResponse {
+  pushed: number;
+  pushed_deletes: number;
+  pulled: number;
+  pulled_deletes: number;
+  cursor: number;
 }

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

@@ -43,6 +43,7 @@ def run_migrations(engine) -> None:
     _migrate_generation_versions(engine, inspector, tables)
     _migrate_capture_settings(engine, inspector, tables)
     _migrate_mcp_bindings(engine, inspector, tables)
+    _migrate_cloud_settings(engine, inspector, tables)
     _normalize_storage_paths(engine, tables)
 
 
@@ -283,6 +284,19 @@ def _migrate_mcp_bindings(engine, inspector, tables: set[str]) -> None:
             )
 
 
+def _migrate_cloud_settings(engine, inspector, tables: set[str]) -> None:
+    """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:
     """Whether ``ALTER TABLE … DROP COLUMN`` is supported by the dialect +
     runtime. Non-SQLite dialects (Postgres, MySQL) have supported it for

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 (
@@ -253,9 +253,45 @@ class CloudSettings(Base):
     device_name = Column(String, nullable=True)
     account_user_id = Column(String, nullable=True)
     connected_at = Column(DateTime, nullable=True)
+    # Server-assigned sync device id (device table on the cloud side). Set when
+    # 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/models.py

@@ -813,4 +813,41 @@ class CloudStatusResponse(BaseModel):
     account_user_id: Optional[str] = None
     key_prefix: Optional[str] = None
     connected_at: Optional[datetime] = None
-    dashboard_url: str
+
+
+class CloudSyncSetupResponse(BaseModel):
+    """Result of registering this install as a sync device.
+
+    ``recovery_phrase`` is present exactly once, when this device minted the
+    account's key material (first device). The UI must force-display it and
+    never persist it. When absent, the account already has key material and
+    this device is awaiting provisioning (wrapped key from an existing device,
+    or a recovery-phrase restore)."""
+
+    status: str  # unregistered | awaiting_provision | ready
+    device_id: Optional[str] = None
+    recovery_phrase: Optional[str] = None
+
+
+class CloudSyncStatusResponse(BaseModel):
+    """Sync identity + progress for the settings UI."""
+
+    status: str  # unregistered | awaiting_provision | ready
+    device_id: Optional[str] = None
+    sync_cursor: int = 0
+
+
+class CloudRestoreRequest(BaseModel):
+    """Recovery-phrase restore on a fresh device."""
+
+    phrase: str
+
+
+class CloudSyncRunResponse(BaseModel):
+    """Outcome of one push+pull sync pass."""
+
+    pushed: int
+    pushed_deletes: int
+    pulled: int
+    pulled_deletes: int
+    cursor: int

backend/requirements.txt

@@ -7,6 +7,12 @@ pydantic>=2.5.0
 sqlalchemy>=2.0.0
 alembic>=1.13.0
 
+# Cloud backup/sync E2E encryption (services/cloud_crypto.py); keyring stores
+# the device key + master key in the OS keychain (services/cloud_keys.py)
+pynacl>=1.5.0
+mnemonic>=0.21
+keyring>=25
+
 # ML models
 torch>=2.2.0
 transformers>=4.36.0,<=4.57.6

backend/routes/cloud.py

@@ -1,4 +1,4 @@
-"""Voicebox Cloud device login routes.
+"""Voicebox Cloud routes: device login + encrypted backup/sync.
 
 The browser-based pairing flow:
   1. POST /cloud/login/start  — opens the browser to the cloud authorize page.
@@ -6,17 +6,31 @@ The browser-based pairing flow:
                                 the backend exchanges it for an API key.
   3. GET  /cloud/status       — the UI polls this to learn when it connected.
   4. POST /cloud/disconnect   — forget the local credential.
+
+Sync, once logged in:
+  5. POST /cloud/sync/setup    — register as an encryption device; on a keyless
+                                 account this mints the master key and returns
+                                 the recovery phrase (shown exactly once).
+  6. POST /cloud/sync/restore  — recover the master key from the phrase.
+  7. POST /cloud/sync/adopt    — pick up a wrapped key another device provisioned.
+  8. GET  /cloud/sync/status   — identity state + cursor.
+  9. POST /cloud/sync/run      — one full push+pull pass.
 """
 
 import socket
 
-from fastapi import APIRouter, Depends, Request
+from fastapi import APIRouter, Depends, HTTPException, Request
 from fastapi.responses import HTMLResponse
 from sqlalchemy.orm import Session
 
 from .. import models
-from ..database import get_db
-from ..services import cloud as cloud_service
+from ..database import CloudSettings as DBCloudSettings, get_db
+from ..services import cloud as cloud_service, cloud_account, cloud_sync
+from ..services.cloud_account import CloudAccountError
+from ..services.cloud_api import CloudApiError
+from ..services.cloud_crypto import CloudCryptoError
+from ..services.cloud_keys import CloudKeyStoreError
+from ..services.cloud_sync import CloudSyncError
 
 router = APIRouter(prefix="/cloud", tags=["cloud"])
 
@@ -44,11 +58,7 @@ async def cloud_callback(
     ok, message = await cloud_service.handle_callback(db, code=code, state=state)
     heading = "You're connected" if ok else "Couldn't connect"
     accent = "#16a34a" if ok else "#dc2626"
-    sub = (
-        "Voicebox is now linked to your account. You can close this tab and return to the app."
-        if ok
-        else message
-    )
+    sub = "Voicebox is now linked to your account. You can close this tab and return to the app." if ok else message
     html = f"""<!doctype html>
 <html lang="en"><head><meta charset="utf-8" />
 <meta name="viewport" content="width=device-width, initial-scale=1" />
@@ -73,3 +83,73 @@ async def cloud_status(db: Session = Depends(get_db)):
 async def cloud_disconnect(db: Session = Depends(get_db)):
     cloud_service.disconnect(db)
     return models.CloudStatusResponse(**cloud_service.get_status(db))
+
+
+# ─── Encrypted backup & sync ─────────────────────────────────────────────
+
+_SYNC_ERRORS = (CloudAccountError, CloudApiError, CloudCryptoError, CloudKeyStoreError, CloudSyncError)
+
+
+def _sync_status(db: Session) -> models.CloudSyncStatusResponse:
+    identity = cloud_account.identity_status(db)
+    row = db.query(DBCloudSettings).filter(DBCloudSettings.id == 1).first()
+    return models.CloudSyncStatusResponse(
+        status=identity.status,
+        device_id=identity.device_id,
+        sync_cursor=(row.sync_cursor or 0) if row else 0,
+    )
+
+
+@router.post("/sync/setup", response_model=models.CloudSyncSetupResponse)
+async def cloud_sync_setup(db: Session = Depends(get_db)):
+    try:
+        phrase = await cloud_account.setup_device(db)
+        identity = cloud_account.identity_status(db)
+    except _SYNC_ERRORS as err:
+        raise HTTPException(status_code=400, detail=str(err)) from err
+    return models.CloudSyncSetupResponse(
+        status=identity.status,
+        device_id=identity.device_id,
+        recovery_phrase=phrase,
+    )
+
+
+@router.post("/sync/restore", response_model=models.CloudSyncStatusResponse)
+async def cloud_sync_restore(body: models.CloudRestoreRequest, db: Session = Depends(get_db)):
+    try:
+        await cloud_account.restore_with_phrase(db, body.phrase)
+    except _SYNC_ERRORS as err:
+        raise HTTPException(status_code=400, detail=str(err)) from err
+    return _sync_status(db)
+
+
+@router.post("/sync/adopt", response_model=models.CloudSyncStatusResponse)
+async def cloud_sync_adopt(db: Session = Depends(get_db)):
+    try:
+        await cloud_account.adopt_wrapped_key(db)
+    except _SYNC_ERRORS as err:
+        raise HTTPException(status_code=400, detail=str(err)) from err
+    return _sync_status(db)
+
+
+@router.get("/sync/status", response_model=models.CloudSyncStatusResponse)
+async def cloud_sync_status(db: Session = Depends(get_db)):
+    try:
+        return _sync_status(db)
+    except CloudAccountError:
+        return models.CloudSyncStatusResponse(status="unregistered", device_id=None, sync_cursor=0)
+
+
+@router.post("/sync/run", response_model=models.CloudSyncRunResponse)
+async def cloud_sync_run(db: Session = Depends(get_db)):
+    try:
+        report = await cloud_sync.run_sync(db)
+    except _SYNC_ERRORS as err:
+        raise HTTPException(status_code=400, detail=str(err)) from err
+    return models.CloudSyncRunResponse(
+        pushed=report.pushed,
+        pushed_deletes=report.pushed_deletes,
+        pulled=report.pulled,
+        pulled_deletes=report.pulled_deletes,
+        cursor=report.cursor,
+    )

backend/services/cloud.py

@@ -19,7 +19,6 @@ import webbrowser
 from urllib.parse import urlencode
 
 import httpx
-from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session
 
 from .. import config
@@ -42,15 +41,6 @@ def _prune() -> None:
             _pending.pop(state, None)
 
 
-def _json_dict(response: httpx.Response) -> dict | None:
-    """Parsed JSON body, or None when it isn't a JSON object."""
-    try:
-        payload = response.json()
-    except ValueError:
-        return None
-    return payload if isinstance(payload, dict) else None
-
-
 def _consume_state(state: str) -> bool:
     """Validate and single-use-consume a pending state."""
     _prune()
@@ -94,10 +84,7 @@ async def handle_callback(db: Session, code: str, state: str) -> tuple[bool, str
             if exchanged.status_code != 200:
                 logger.warning("cloud exchange rejected code: %s", exchanged.status_code)
                 return False, "Could not complete sign-in — the code was rejected."
-            payload = _json_dict(exchanged)
-            if payload is None:
-                logger.warning("cloud exchange returned a non-JSON payload")
-                return False, "Voicebox Cloud returned an unexpected response."
+            payload = exchanged.json()
             api_key = payload.get("key")
             device_name = payload.get("label")
             if not api_key:
@@ -111,9 +98,7 @@ async def handle_callback(db: Session, code: str, state: str) -> tuple[bool, str
             if me.status_code != 200:
                 logger.warning("minted key failed verification: %s", me.status_code)
                 return False, "Sign-in succeeded but the key could not be verified."
-            # The 200 above proves the key works; the user id is best-effort.
-            data = (_json_dict(me) or {}).get("data")
-            account_user_id = data.get("userId") if isinstance(data, dict) else None
+            account_user_id = (me.json().get("data") or {}).get("userId")
     except httpx.HTTPError:
         logger.exception("network error during cloud exchange")
         return False, "Could not reach Voicebox Cloud. Check your connection and try again."
@@ -128,14 +113,8 @@ def _get_or_create_row(db: Session) -> DBCloudSettings:
     if row is None:
         row = DBCloudSettings(id=SINGLETON_ID)
         db.add(row)
-        try:
-            db.commit()
-        except IntegrityError:
-            # Another request created the singleton concurrently.
-            db.rollback()
-            row = db.query(DBCloudSettings).filter(DBCloudSettings.id == SINGLETON_ID).one()
-        else:
-            db.refresh(row)
+        db.commit()
+        db.refresh(row)
     return row
 
 
@@ -162,7 +141,6 @@ def get_status(db: Session) -> dict:
         "account_user_id": row.account_user_id if connected else None,
         "key_prefix": key_prefix,
         "connected_at": row.connected_at if connected else None,
-        "dashboard_url": f"{config.get_cloud_web_url()}/account",
     }
 
 
@@ -171,7 +149,6 @@ def disconnect(db: Session) -> None:
     revoked from the account dashboard — surface that in the UI."""
     row = _get_or_create_row(db)
     row.api_key = None
-    row.device_name = None
     row.account_user_id = None
     row.connected_at = None
     db.commit()

backend/services/cloud_account.py

@@ -0,0 +1,193 @@
+"""
+Cloud sync identity: the Master Key lifecycle across devices.
+
+Ties together the pieces below into the flows from the cloud design doc §9
+(first device, add-a-device, recovery). The server participates only as a
+mailbox for ciphertext — every wrap/unwrap here happens locally.
+
+- ``cloud_crypto``  — the primitives (MK, recovery phrase, sealed boxes)
+- ``cloud_keys``    — OS-keychain persistence of the private key + MK
+- ``cloud_api``     — the bearer-key HTTP client
+- ``CloudSettings`` — the local row holding the API key + sync device id
+
+Flows:
+- **First device** (``setup_device`` on a keyless account): generate MK, wrap
+  it under a fresh recovery phrase → escrow to the server, wrap it to our own
+  device key, keep MK in the keychain. Returns the phrase for one-time display.
+- **New device on an existing account** (``setup_device`` when the account has
+  key material): register and wait — an existing device provisions us
+  (``provision_device`` there, ``adopt_wrapped_key`` here), or the user types
+  the recovery phrase (``restore_with_phrase``).
+"""
+
+import base64
+import logging
+from dataclasses import dataclass
+
+from sqlalchemy.orm import Session
+
+from .. import config
+from ..database import CloudSettings as DBCloudSettings
+from . import cloud_crypto, cloud_keys
+from .cloud_api import CloudApiClient
+from .cloud_crypto import RecoveryWrap
+
+logger = logging.getLogger(__name__)
+
+
+class CloudAccountError(Exception):
+    """The identity flow cannot proceed (not connected, no escrow, bad phrase…)."""
+
+
+def _b64e(raw: bytes) -> str:
+    return base64.b64encode(raw).decode("ascii")
+
+
+def _b64d(encoded: str) -> bytes:
+    return base64.b64decode(encoded)
+
+
+def _settings(db: Session) -> DBCloudSettings:
+    row = db.query(DBCloudSettings).filter(DBCloudSettings.id == 1).first()
+    if row is None or not row.api_key or not row.account_user_id:
+        raise CloudAccountError("not connected to Voicebox Cloud — log in first")
+    return row
+
+
+def _client(row: DBCloudSettings) -> CloudApiClient:
+    return CloudApiClient(config.get_cloud_api_url(), row.api_key)
+
+
+@dataclass(frozen=True)
+class SyncIdentity:
+    # "unregistered"       — logged in, but not yet a sync device
+    # "awaiting_provision" — registered, waiting for a wrapped MK or the phrase
+    # "ready"              — MK in the keychain, sync can run
+    status: str
+    device_id: str | None
+
+
+def identity_status(db: Session) -> SyncIdentity:
+    row = _settings(db)
+    if not row.sync_device_id:
+        return SyncIdentity(status="unregistered", device_id=None)
+    has_mk = cloud_keys.load_secret(row.account_user_id, cloud_keys.MASTER_KEY) is not None
+    return SyncIdentity(status="ready" if has_mk else "awaiting_provision", device_id=row.sync_device_id)
+
+
+async def setup_device(db: Session) -> str | None:
+    """Register this install as a sync device.
+
+    On a keyless account this is first-device setup: mints MK + the recovery
+    escrow and returns the phrase — the caller must display it exactly once and
+    never persist it. On an account with existing key material it returns None
+    and the device waits in ``awaiting_provision``.
+    """
+    row = _settings(db)
+    if row.sync_device_id:
+        raise CloudAccountError("this install is already registered as a sync device")
+
+    private_key, public_key = cloud_crypto.generate_device_keypair()
+    async with _client(row) as client:
+        registered = await client.register_device(row.device_name or "Voicebox Desktop", _b64e(public_key))
+        device_id = registered["deviceId"]
+
+        # Persist the private key before anything can depend on it; a crash
+        # after registration leaves a provisionable device, never a locked one.
+        cloud_keys.store_secret(row.account_user_id, cloud_keys.DEVICE_PRIVATE_KEY, private_key)
+        row.sync_device_id = device_id
+        db.commit()
+
+        if registered["accountHasKey"]:
+            logger.info("registered sync device %s; awaiting MK provisioning", device_id)
+            return None
+
+        master_key = cloud_crypto.generate_master_key()
+        phrase = cloud_crypto.generate_recovery_phrase()
+        escrow = cloud_crypto.wrap_master_key_with_phrase(master_key, phrase)
+        await client.put_account_key(_b64e(escrow.wrapped_key), _b64e(escrow.kdf_salt), escrow.kdf_params)
+        await client.put_wrapped_key(device_id, _b64e(cloud_crypto.wrap_master_key_for_device(master_key, public_key)))
+        cloud_keys.store_secret(row.account_user_id, cloud_keys.MASTER_KEY, master_key)
+        logger.info("initialized account key material as first device %s", device_id)
+        return phrase
+
+
+async def restore_with_phrase(db: Session, phrase: str) -> None:
+    """Recover MK from the server-side escrow using the recovery phrase.
+    The device must be registered (``setup_device``) first."""
+    row = _settings(db)
+    if not row.sync_device_id:
+        raise CloudAccountError("register this device before restoring")
+    if not cloud_crypto.validate_recovery_phrase(phrase):
+        raise CloudAccountError("that doesn't look like a valid recovery phrase — check for typos")
+
+    async with _client(row) as client:
+        escrow = await client.get_account_key()
+        if not escrow:
+            raise CloudAccountError("this account has no recovery escrow yet")
+        wrap = RecoveryWrap(
+            wrapped_key=_b64d(escrow["recoveryWrappedKey"]),
+            kdf_salt=_b64d(escrow["kdfSalt"]),
+            kdf_params=escrow["kdfParams"],
+        )
+        master_key = cloud_crypto.unwrap_master_key_with_phrase(wrap, phrase)
+
+        # Also wrap MK to our own device key so future restores on this device
+        # don't need the phrase.
+        private_key = cloud_keys.load_secret(row.account_user_id, cloud_keys.DEVICE_PRIVATE_KEY)
+        if private_key is None:
+            raise CloudAccountError("device key missing from the OS keychain — register this device again")
+        public_key = cloud_crypto.device_public_key(private_key)
+        await client.put_wrapped_key(
+            row.sync_device_id, _b64e(cloud_crypto.wrap_master_key_for_device(master_key, public_key))
+        )
+        cloud_keys.store_secret(row.account_user_id, cloud_keys.MASTER_KEY, master_key)
+        logger.info("restored master key from recovery phrase on device %s", row.sync_device_id)
+
+
+async def adopt_wrapped_key(db: Session) -> bool:
+    """For a device in ``awaiting_provision``: fetch our wrapped MK if an
+    existing device has provisioned it. Returns True once MK is in the keychain."""
+    row = _settings(db)
+    if not row.sync_device_id:
+        raise CloudAccountError("register this device before adopting a key")
+
+    private_key = cloud_keys.load_secret(row.account_user_id, cloud_keys.DEVICE_PRIVATE_KEY)
+    if private_key is None:
+        raise CloudAccountError("device key missing from the OS keychain — register this device again")
+
+    async with _client(row) as client:
+        wrapped = await client.get_wrapped_key(row.sync_device_id)
+    if not wrapped:
+        return False
+    master_key = cloud_crypto.unwrap_master_key_for_device(_b64d(wrapped), private_key)
+    cloud_keys.store_secret(row.account_user_id, cloud_keys.MASTER_KEY, master_key)
+    logger.info("adopted provisioned master key on device %s", row.sync_device_id)
+    return True
+
+
+async def provision_device(db: Session, target_device_id: str) -> None:
+    """Run on a device that already holds MK: wrap it to another registered
+    device's public key so that device can start syncing."""
+    row = _settings(db)
+    master_key = cloud_keys.load_secret(row.account_user_id, cloud_keys.MASTER_KEY)
+    if master_key is None:
+        raise CloudAccountError("this device holds no master key to provision with")
+
+    async with _client(row) as client:
+        devices = await client.list_devices()
+        target = next((d for d in devices if d["id"] == target_device_id and not d.get("revokedAt")), None)
+        if target is None:
+            raise CloudAccountError("target device not found (or revoked)")
+        wrapped = cloud_crypto.wrap_master_key_for_device(master_key, _b64d(target["publicKey"]))
+        await client.put_wrapped_key(target_device_id, _b64e(wrapped))
+    logger.info("provisioned master key to device %s", target_device_id)
+
+
+def load_master_key(db: Session) -> bytes:
+    """The MK for the sync engine. Raises if this device isn't ready."""
+    row = _settings(db)
+    master_key = cloud_keys.load_secret(row.account_user_id, cloud_keys.MASTER_KEY)
+    if master_key is None:
+        raise CloudAccountError("no master key on this device — finish setup or restore first")
+    return master_key

backend/services/cloud_api.py

@@ -0,0 +1,174 @@
+"""
+HTTP client for the Voicebox Cloud API (api.voicebox.sh).
+
+A thin async wrapper over the bearer-key endpoints the sync client needs:
+device/key distribution, the encrypted object store, and the sync feed. Every
+payload sent through here is ciphertext or metadata about ciphertext — the
+encryption itself happens in ``cloud_crypto`` before bytes reach this module.
+
+Blob bytes don't flow through the API at all: pushes receive presigned PUT
+URLs and pulls receive presigned GET URLs, and the client transfers ciphertext
+directly with the storage host. Those transfers use a separate unauthenticated
+HTTP client so the bearer key is never sent to the storage host.
+"""
+
+import contextlib
+import logging
+from typing import Any
+
+import httpx
+
+logger = logging.getLogger(__name__)
+
+_TIMEOUT = 30.0
+_BLOB_TIMEOUT = 120.0  # audio assets can be tens of MB
+
+
+class CloudApiError(Exception):
+    def __init__(self, message: str, status: int | None = None):
+        super().__init__(message)
+        self.status = status
+
+
+class CloudApiClient:
+    """One authenticated session against the cloud API. Use as an async context
+    manager so both underlying connection pools are closed."""
+
+    def __init__(self, api_url: str, api_key: str, *, transport: httpx.AsyncBaseTransport | None = None):
+        self._api = httpx.AsyncClient(
+            base_url=api_url.rstrip("/"),
+            headers={"Authorization": f"Bearer {api_key}"},
+            timeout=_TIMEOUT,
+            transport=transport,
+        )
+        # Presigned-URL transfers: no Authorization header, longer timeout.
+        self._blobs = httpx.AsyncClient(timeout=_BLOB_TIMEOUT, transport=transport)
+
+    async def __aenter__(self) -> "CloudApiClient":
+        return self
+
+    async def __aexit__(self, *exc: object) -> None:
+        await self.aclose()
+
+    async def aclose(self) -> None:
+        await self._api.aclose()
+        await self._blobs.aclose()
+
+    async def _call(
+        self,
+        method: str,
+        path: str,
+        *,
+        json: dict | None = None,
+        params: dict | None = None,
+    ) -> Any:
+        try:
+            resp = await self._api.request(method, path, json=json, params=params)
+        except httpx.HTTPError as err:
+            raise CloudApiError(f"could not reach Voicebox Cloud: {err}") from err
+        if resp.status_code >= 400:
+            message = f"{method} {path} failed ({resp.status_code})"
+            with contextlib.suppress(ValueError):
+                message = resp.json().get("error", {}).get("message", message)
+            raise CloudApiError(message, status=resp.status_code)
+        payload = resp.json()
+        if not payload.get("ok"):
+            raise CloudApiError(f"{method} {path} returned ok=false", status=resp.status_code)
+        return payload.get("data")
+
+    # -- account ------------------------------------------------------------
+
+    async def me(self) -> dict:
+        return await self._call("GET", "/v1/account/me")
+
+    # -- devices & key distribution ------------------------------------------
+
+    async def register_device(self, name: str, public_key_b64: str) -> dict:
+        """Returns {deviceId, accountHasKey}."""
+        return await self._call("POST", "/v1/devices", json={"name": name, "publicKey": public_key_b64})
+
+    async def list_devices(self) -> list[dict]:
+        return await self._call("GET", "/v1/devices")
+
+    async def put_wrapped_key(self, device_id: str, wrapped_master_key_b64: str) -> None:
+        await self._call(
+            "POST",
+            f"/v1/devices/{device_id}/wrapped-key",
+            json={"wrappedMasterKey": wrapped_master_key_b64},
+        )
+
+    async def get_wrapped_key(self, device_id: str) -> str | None:
+        data = await self._call("GET", f"/v1/devices/{device_id}/wrapped-key")
+        return data.get("wrappedMasterKey") if data else None
+
+    async def put_account_key(self, recovery_wrapped_key_b64: str, kdf_salt_b64: str, kdf_params: str) -> None:
+        await self._call(
+            "PUT",
+            "/v1/devices/account-key",
+            json={
+                "recoveryWrappedKey": recovery_wrapped_key_b64,
+                "kdfSalt": kdf_salt_b64,
+                "kdfParams": kdf_params,
+            },
+        )
+
+    async def get_account_key(self) -> dict | None:
+        """Returns {recoveryWrappedKey, kdfSalt, kdfParams} or None if the
+        account has no escrow yet."""
+        return await self._call("GET", "/v1/devices/account-key")
+
+    # -- encrypted object store ----------------------------------------------
+
+    async def push_object(
+        self,
+        *,
+        kind: str,
+        client_id: str,
+        version: int,
+        record: dict | None,
+        assets: list[dict] | None = None,
+    ) -> dict:
+        """Upsert object metadata. ``record`` is {hash, size}; each asset is
+        {role, clientAssetId, hash, size}. Returns {objectId, seq, uploads},
+        where uploads lists presigned PUT URLs for exactly the changed blobs."""
+        return await self._call(
+            "POST",
+            "/v1/objects",
+            json={
+                "kind": kind,
+                "clientId": client_id,
+                "version": version,
+                "record": record,
+                "assets": assets or [],
+            },
+        )
+
+    async def commit_object(self, object_id: str) -> None:
+        """Ask the server to verify all claimed blobs actually landed in storage."""
+        await self._call("POST", f"/v1/objects/{object_id}/commit")
+
+    async def delete_object(self, object_id: str) -> None:
+        await self._call("DELETE", f"/v1/objects/{object_id}")
+
+    async def get_changes(self, since: int, limit: int = 200) -> dict:
+        """Sync pull: {changes, cursor, hasMore} for everything newer than ``since``."""
+        return await self._call("GET", "/v1/sync/changes", params={"since": since, "limit": limit})
+
+    # -- blob transfer (presigned URLs, ciphertext only) ----------------------
+
+    async def upload_blob(self, url: str, data: bytes) -> None:
+        try:
+            resp = await self._blobs.put(url, content=data, headers={"Content-Type": "application/octet-stream"})
+        except httpx.HTTPError as err:
+            raise CloudApiError(f"blob upload failed: {err}") from err
+        if resp.status_code >= 400:
+            raise CloudApiError(f"blob upload rejected ({resp.status_code})", status=resp.status_code)
+
+    async def download_blob(self, url: str) -> bytes:
+        try:
+            resp = await self._blobs.get(url)
+        except httpx.HTTPError as err:
+            raise CloudApiError(f"blob download failed: {err}") from err
+        if resp.status_code >= 400:
+            raise CloudApiError(f"blob download rejected ({resp.status_code})", status=resp.status_code)
+        return resp.content

backend/services/cloud_crypto.py

@@ -0,0 +1,233 @@
+"""
+Client-side cryptography for Voicebox Cloud backup & sync.
+
+This is the auditable half of the cloud's privacy promise: every byte that
+leaves this machine is encrypted here first, and the server only ever stores
+ciphertext plus routing metadata. The key hierarchy (cloud repo,
+``docs/DESIGN.md``):
+
+    Recovery phrase (BIP39)                Device X25519 keypairs
+            | Argon2id(salt, params)               | sealed box
+            v                                      v
+       Recovery KEK --wrap--> Master Key (MK) <--wrapped to each device
+                                    |
+                                    | per-blob random Content Key (CK),
+                                    | wrapped by MK inside the envelope
+                                    v
+                XChaCha20-Poly1305(CK) over each record/asset blob
+
+Everything in this module is a pure function over bytes — no I/O, no storage,
+no network. Key persistence (OS keychain) and the sync engine live elsewhere.
+
+Invariants this module enforces:
+- The Master Key is random, never derived from anything the server holds
+  (API keys and wallet keys are auth/entitlement, never encryption roots).
+- Each blob's AAD binds it to its logical slot (object, role, version), so a
+  server cannot substitute one blob for another without decryption failing.
+- Content Keys travel only inside the envelope, wrapped by MK — the database
+  (local and cloud) stays free of key material.
+"""
+
+import json
+import secrets
+from dataclasses import dataclass
+
+import nacl.bindings
+import nacl.exceptions
+import nacl.pwhash
+import nacl.utils
+from mnemonic import Mnemonic
+from nacl.public import PrivateKey, PublicKey, SealedBox
+from nacl.secret import SecretBox
+
+KEY_BYTES = 32
+
+# Envelope framing: magic | alg | len(wrapped_ck) | wrapped_ck | nonce | ciphertext
+ENVELOPE_MAGIC = b"VBX1"
+ALG_XCHACHA20_POLY1305 = 1
+_WRAPPED_CK_LEN_BYTES = 2
+_NONCE_BYTES = nacl.bindings.crypto_aead_xchacha20poly1305_ietf_NPUBBYTES  # 24
+
+# Argon2id cost for the recovery-phrase KEK. MODERATE (~256 MiB, interactive
+# latency) fits a desktop restore flow; the phrase itself already carries
+# 128 bits of entropy, so the KDF is hardening, not the main defense.
+_KDF_OPSLIMIT = nacl.pwhash.argon2id.OPSLIMIT_MODERATE
+_KDF_MEMLIMIT = nacl.pwhash.argon2id.MEMLIMIT_MODERATE
+
+_mnemonic = Mnemonic("english")
+
+
+class CloudCryptoError(Exception):
+    """Envelope malformed, key wrong, or ciphertext tampered with."""
+
+
+# ---------------------------------------------------------------------------
+# Master Key + device keys
+
+
+def generate_master_key() -> bytes:
+    """The 32-byte root of content secrecy. Generated once per account, client-side."""
+    return secrets.token_bytes(KEY_BYTES)
+
+
+def generate_device_keypair() -> tuple[bytes, bytes]:
+    """X25519 (private, public) for this install. The private key never leaves
+    the device; the public key is registered with the cloud so existing devices
+    can wrap MK to it."""
+    private = PrivateKey.generate()
+    return bytes(private), bytes(private.public_key)
+
+
+def device_public_key(device_private_key: bytes) -> bytes:
+    """Re-derive the public half from a stored private key."""
+    return bytes(PrivateKey(device_private_key).public_key)
+
+
+def wrap_master_key_for_device(master_key: bytes, device_public_key: bytes) -> bytes:
+    """Seal MK to another device's public key (run on an *existing* device when
+    provisioning a new one). Only the target device's private key can open it."""
+    return SealedBox(PublicKey(device_public_key)).encrypt(master_key)
+
+
+def unwrap_master_key_for_device(wrapped: bytes, device_private_key: bytes) -> bytes:
+    """Open a sealed MK with this device's private key."""
+    try:
+        return SealedBox(PrivateKey(device_private_key)).decrypt(wrapped)
+    except nacl.exceptions.CryptoError as err:
+        raise CloudCryptoError("wrapped master key does not match this device key") from err
+
+
+# ---------------------------------------------------------------------------
+# Recovery phrase escrow
+
+
+@dataclass(frozen=True)
+class RecoveryWrap:
+    """What the cloud stores in ``account_key``: ciphertext + the KDF context
+    any client needs to re-derive the KEK from the phrase. Holds no secrets."""
+
+    wrapped_key: bytes
+    kdf_salt: bytes
+    kdf_params: str  # JSON, e.g. {"m": ..., "t": ..., "p": 1}
+
+
+def generate_recovery_phrase() -> str:
+    """12-word BIP39 mnemonic (128-bit entropy) — shown to the user exactly once."""
+    return _mnemonic.generate(strength=128)
+
+
+def validate_recovery_phrase(phrase: str) -> bool:
+    """Word-level checksum validation, for catching typos before an unwrap attempt."""
+    return _mnemonic.check(_normalize_phrase(phrase))
+
+
+def _normalize_phrase(phrase: str) -> str:
+    return " ".join(phrase.lower().split())
+
+
+def _derive_kek(phrase: str, salt: bytes, opslimit: int, memlimit: int) -> bytes:
+    return nacl.pwhash.argon2id.kdf(
+        KEY_BYTES,
+        _normalize_phrase(phrase).encode("utf-8"),
+        salt,
+        opslimit=opslimit,
+        memlimit=memlimit,
+    )
+
+
+def wrap_master_key_with_phrase(master_key: bytes, phrase: str) -> RecoveryWrap:
+    """Argon2id-stretch the phrase into a KEK and wrap MK under it."""
+    salt = nacl.utils.random(nacl.pwhash.argon2id.SALTBYTES)
+    kek = _derive_kek(phrase, salt, _KDF_OPSLIMIT, _KDF_MEMLIMIT)
+    return RecoveryWrap(
+        wrapped_key=SecretBox(kek).encrypt(master_key),
+        kdf_salt=salt,
+        kdf_params=json.dumps({"m": _KDF_MEMLIMIT, "t": _KDF_OPSLIMIT, "p": 1}),
+    )
+
+
+def unwrap_master_key_with_phrase(wrap: RecoveryWrap, phrase: str) -> bytes:
+    """Recover MK on a fresh device from the phrase + the stored KDF context."""
+    try:
+        params = json.loads(wrap.kdf_params)
+        opslimit, memlimit = int(params["t"]), int(params["m"])
+    except (ValueError, KeyError, TypeError) as err:
+        raise CloudCryptoError("malformed KDF parameters") from err
+    kek = _derive_kek(phrase, wrap.kdf_salt, opslimit, memlimit)
+    try:
+        return SecretBox(kek).decrypt(wrap.wrapped_key)
+    except nacl.exceptions.CryptoError as err:
+        raise CloudCryptoError("recovery phrase does not unlock this account key") from err
+
+
+# ---------------------------------------------------------------------------
+# Blob envelope
+
+
+def _build_aad(object_id: str, role: str, version: int) -> bytes:
+    # Binds a blob to its logical slot. Unambiguous because object_id is a UUID
+    # and role is a fixed token — neither contains ":".
+    return f"{object_id}:{role}:{version}".encode()
+
+
+def encrypt_blob(plaintext: bytes, master_key: bytes, *, object_id: str, role: str, version: int) -> bytes:
+    """Produce a self-describing VBX1 envelope: a fresh Content Key wrapped by
+    MK rides in the header, and the AAD pins the blob to (object, role, version)."""
+    content_key = secrets.token_bytes(KEY_BYTES)
+    wrapped_ck = SecretBox(master_key).encrypt(content_key)
+    nonce = nacl.utils.random(_NONCE_BYTES)
+    ciphertext = nacl.bindings.crypto_aead_xchacha20poly1305_ietf_encrypt(
+        plaintext,
+        _build_aad(object_id, role, version),
+        nonce,
+        content_key,
+    )
+    return b"".join(
+        [
+            ENVELOPE_MAGIC,
+            bytes([ALG_XCHACHA20_POLY1305]),
+            len(wrapped_ck).to_bytes(_WRAPPED_CK_LEN_BYTES, "big"),
+            wrapped_ck,
+            nonce,
+            ciphertext,
+        ]
+    )
+
+
+def decrypt_blob(envelope: bytes, master_key: bytes, *, object_id: str, role: str, version: int) -> bytes:
+    """Open a VBX1 envelope. Raises CloudCryptoError if the envelope is
+    malformed, the key is wrong, the ciphertext was modified, or the blob was
+    served for a different (object, role, version) slot."""
+    offset = len(ENVELOPE_MAGIC)
+    if envelope[:offset] != ENVELOPE_MAGIC:
+        raise CloudCryptoError("not a VBX1 envelope")
+    if len(envelope) < offset + 1 + _WRAPPED_CK_LEN_BYTES:
+        raise CloudCryptoError("envelope truncated")
+    alg = envelope[offset]
+    if alg != ALG_XCHACHA20_POLY1305:
+        raise CloudCryptoError(f"unsupported envelope algorithm {alg}")
+    offset += 1
+
+    wrapped_len = int.from_bytes(envelope[offset : offset + _WRAPPED_CK_LEN_BYTES], "big")
+    offset += _WRAPPED_CK_LEN_BYTES
+    wrapped_ck = envelope[offset : offset + wrapped_len]
+    offset += wrapped_len
+    nonce = envelope[offset : offset + _NONCE_BYTES]
+    offset += _NONCE_BYTES
+    ciphertext = envelope[offset:]
+    if len(wrapped_ck) != wrapped_len or len(nonce) != _NONCE_BYTES or not ciphertext:
+        raise CloudCryptoError("envelope truncated")
+
+    try:
+        content_key = SecretBox(master_key).decrypt(wrapped_ck)
+    except nacl.exceptions.CryptoError as err:
+        raise CloudCryptoError("master key does not unwrap this blob's content key") from err
+    try:
+        return nacl.bindings.crypto_aead_xchacha20poly1305_ietf_decrypt(
+            ciphertext,
+            _build_aad(object_id, role, version),
+            nonce,
+            content_key,
+        )
+    except nacl.exceptions.CryptoError as err:
+        raise CloudCryptoError("blob failed authentication (tampered, or served for the wrong slot)") from err

backend/services/cloud_keys.py

@@ -0,0 +1,63 @@
+"""
+OS-keychain persistence for cloud E2E key material.
+
+The bearer API key lives in the local database (``CloudSettings``) because it
+is auth, not secrecy. The *encryption* keys never touch the database: this
+module stores the device's X25519 private key and the unwrapped Master Key in
+the OS keychain (macOS Keychain, Windows Credential Locker, Secret Service on
+Linux) via ``keyring``. Entries are namespaced by cloud account user id so
+relinking to a different account can never read the previous account's keys.
+
+Headless installs (Docker/web) may have no keychain backend; operations then
+raise ``CloudKeyStoreError`` and cloud sync stays unavailable rather than
+silently degrading to plaintext-on-disk key storage.
+"""
+
+import base64
+
+import keyring
+import keyring.errors
+
+_SERVICE = "sh.voicebox.cloud"
+
+DEVICE_PRIVATE_KEY = "device_private_key"
+MASTER_KEY = "master_key"
+_ALL_ENTRIES = (DEVICE_PRIVATE_KEY, MASTER_KEY)
+
+
+class CloudKeyStoreError(Exception):
+    """The OS keychain is unavailable or rejected the operation."""
+
+
+def _entry(account_user_id: str, name: str) -> str:
+    return f"{account_user_id}:{name}"
+
+
+def store_secret(account_user_id: str, name: str, secret: bytes) -> None:
+    try:
+        keyring.set_password(_SERVICE, _entry(account_user_id, name), base64.b64encode(secret).decode("ascii"))
+    except keyring.errors.KeyringError as err:
+        raise CloudKeyStoreError(f"could not store {name} in the OS keychain") from err
+
+
+def load_secret(account_user_id: str, name: str) -> bytes | None:
+    try:
+        stored = keyring.get_password(_SERVICE, _entry(account_user_id, name))
+    except keyring.errors.KeyringError as err:
+        raise CloudKeyStoreError(f"could not read {name} from the OS keychain") from err
+    return base64.b64decode(stored) if stored else None
+
+
+def delete_secret(account_user_id: str, name: str) -> None:
+    try:
+        keyring.delete_password(_SERVICE, _entry(account_user_id, name))
+    except keyring.errors.PasswordDeleteError:
+        pass  # already absent
+    except keyring.errors.KeyringError as err:
+        raise CloudKeyStoreError(f"could not delete {name} from the OS keychain") from err
+
+
+def clear(account_user_id: str) -> None:
+    """Forget all key material for an account (disconnect / account switch)."""
+    for name in _ALL_ENTRIES:
+        delete_secret(account_user_id, name)

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

@@ -0,0 +1,174 @@
+"""Tests for the cloud sync identity flows (services/cloud_account.py).
+
+Runs the real flows against a fake in-process cloud (httpx.MockTransport) and
+a fake in-memory keyring — no network, no OS keychain. The central assertion:
+the master key and recovery phrase never appear in anything sent to the server.
+"""
+
+import base64
+from datetime import datetime
+
+import keyring
+import keyring.backend
+import pytest
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+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"
+
+
+class InMemoryKeyring(keyring.backend.KeyringBackend):
+    priority = 1
+
+    def __init__(self):
+        super().__init__()
+        self.store: dict[tuple[str, str], str] = {}
+
+    def get_password(self, service, username):
+        return self.store.get((service, username))
+
+    def set_password(self, service, username, password):
+        self.store[(service, username)] = password
+
+    def delete_password(self, service, username):
+        self.store.pop((service, username), None)
+
+
+@pytest.fixture
+def fake_keyring(monkeypatch):
+    backend = InMemoryKeyring()
+    monkeypatch.setattr(keyring, "get_password", backend.get_password)
+    monkeypatch.setattr(keyring, "set_password", backend.set_password)
+    monkeypatch.setattr(keyring, "delete_password", backend.delete_password)
+    return backend
+
+
+@pytest.fixture
+def cloud():
+    return FakeCloud()
+
+
+def make_db(account_user_id=USER_A):
+    engine = create_engine("sqlite://")
+    Base.metadata.create_all(engine)
+    db = sessionmaker(bind=engine)()
+    db.add(
+        CloudSettings(
+            id=1,
+            api_key="voicebox_test",
+            device_name="Test Mac",
+            account_user_id=account_user_id,
+            connected_at=datetime(2026, 7, 1),
+        )
+    )
+    db.commit()
+    return db
+
+
+@pytest.fixture
+def patched_client(monkeypatch, cloud):
+    def _client(row):
+        return CloudApiClient("http://cloud.test", row.api_key, transport=cloud.transport())
+
+    monkeypatch.setattr(cloud_account, "_client", _client)
+
+
+@pytest.mark.usefixtures("patched_client", "fake_keyring")
+class TestIdentityFlows:
+    async def test_first_device_setup(self, cloud):
+        db = make_db()
+        phrase = await cloud_account.setup_device(db)
+
+        assert phrase is not None
+        assert cloud_crypto.validate_recovery_phrase(phrase)
+        assert cloud_account.identity_status(db).status == "ready"
+        assert cloud.account_key is not None
+        # Registered + provisioned to itself.
+        (device,) = cloud.devices.values()
+        assert device["wrappedMasterKey"]
+
+        # The invariant: neither MK nor the phrase ever crossed the wire.
+        mk = cloud_account.load_master_key(db)
+        for body in cloud.seen_bodies:
+            assert mk not in body
+            assert base64.b64encode(mk) not in body
+            assert phrase.encode() not in body
+
+    async def test_second_device_via_provisioning(self, cloud):
+        db_a = make_db()
+        await cloud_account.setup_device(db_a)
+        mk_a = cloud_account.load_master_key(db_a)
+
+        # Second install: same account, its own DB + keychain namespace. Reuse
+        # the same fake keyring but a distinct account row would collide, so
+        # simulate the second device with a separate account_user_id-scoped
+        # keychain by clearing MK after capturing device state.
+        db_b = make_db(account_user_id="user-a-second-install")
+        assert await cloud_account.setup_device(db_b) is None  # account already has key material
+        assert cloud_account.identity_status(db_b).status == "awaiting_provision"
+        assert await cloud_account.adopt_wrapped_key(db_b) is False  # nothing provisioned yet
+
+        target_id = db_b.query(CloudSettings).one().sync_device_id
+        await cloud_account.provision_device(db_a, target_id)
+        assert await cloud_account.adopt_wrapped_key(db_b) is True
+        assert cloud_account.load_master_key(db_b) == mk_a
+
+    async def test_restore_with_phrase(self, cloud):
+        db_a = make_db()
+        phrase = await cloud_account.setup_device(db_a)
+        mk_a = cloud_account.load_master_key(db_a)
+
+        db_b = make_db(account_user_id="user-a-fresh-machine")
+        assert await cloud_account.setup_device(db_b) is None
+        await cloud_account.restore_with_phrase(db_b, phrase)
+        assert cloud_account.load_master_key(db_b) == mk_a
+        assert cloud_account.identity_status(db_b).status == "ready"
+
+    async def test_restore_rejects_wrong_phrase(self, cloud):
+        db_a = make_db()
+        await cloud_account.setup_device(db_a)
+
+        db_b = make_db(account_user_id="user-a-fresh-machine")
+        await cloud_account.setup_device(db_b)
+        with pytest.raises(cloud_crypto.CloudCryptoError):
+            await cloud_account.restore_with_phrase(db_b, cloud_crypto.generate_recovery_phrase())
+
+    async def test_restore_rejects_invalid_phrase_early(self, cloud):
+        db = make_db()
+        await cloud_account.setup_device(db)
+        with pytest.raises(CloudAccountError, match="valid recovery phrase"):
+            await cloud_account.restore_with_phrase(
+                db, "not a real phrase at all twelve words missing checksum here ok"
+            )
+
+    async def test_double_registration_rejected(self, cloud):
+        db = make_db()
+        await cloud_account.setup_device(db)
+        with pytest.raises(CloudAccountError, match="already registered"):
+            await cloud_account.setup_device(db)
+
+    async def test_requires_login(self, cloud):
+        engine = create_engine("sqlite://")
+        Base.metadata.create_all(engine)
+        db = sessionmaker(bind=engine)()
+        with pytest.raises(CloudAccountError, match="log in"):
+            await cloud_account.setup_device(db)
+
+
+@pytest.mark.usefixtures("fake_keyring")
+class TestKeyStore:
+    def test_round_trip_and_clear(self):
+        cloud_keys.store_secret(USER_A, cloud_keys.MASTER_KEY, b"\x01" * 32)
+        assert cloud_keys.load_secret(USER_A, cloud_keys.MASTER_KEY) == b"\x01" * 32
+        assert cloud_keys.load_secret("other-user", cloud_keys.MASTER_KEY) is None
+        cloud_keys.clear(USER_A)
+        assert cloud_keys.load_secret(USER_A, cloud_keys.MASTER_KEY) is None
+
+    def test_delete_absent_is_noop(self):
+        cloud_keys.delete_secret(USER_A, cloud_keys.DEVICE_PRIVATE_KEY)

backend/tests/test_cloud_crypto.py

@@ -0,0 +1,137 @@
+"""Tests for the cloud E2E crypto primitives (services/cloud_crypto.py)."""
+
+import pytest
+
+from backend.services.cloud_crypto import (
+    ALG_XCHACHA20_POLY1305,
+    ENVELOPE_MAGIC,
+    CloudCryptoError,
+    RecoveryWrap,
+    decrypt_blob,
+    encrypt_blob,
+    generate_device_keypair,
+    generate_master_key,
+    generate_recovery_phrase,
+    unwrap_master_key_for_device,
+    unwrap_master_key_with_phrase,
+    validate_recovery_phrase,
+    wrap_master_key_for_device,
+    wrap_master_key_with_phrase,
+)
+
+SLOT = {"object_id": "0c8f6f4e-9f5a-4a2f-8f6a-1d2e3f4a5b6c", "role": "audio", "version": 3}
+
+
+class TestMasterKey:
+    def test_master_keys_are_random_32_bytes(self):
+        a, b = generate_master_key(), generate_master_key()
+        assert len(a) == 32
+        assert a != b
+
+
+class TestDeviceWrap:
+    def test_round_trip(self):
+        mk = generate_master_key()
+        private, public = generate_device_keypair()
+        assert unwrap_master_key_for_device(wrap_master_key_for_device(mk, public), private) == mk
+
+    def test_wrong_device_key_fails(self):
+        mk = generate_master_key()
+        _, public = generate_device_keypair()
+        other_private, _ = generate_device_keypair()
+        with pytest.raises(CloudCryptoError):
+            unwrap_master_key_for_device(wrap_master_key_for_device(mk, public), other_private)
+
+
+class TestRecoveryPhrase:
+    def test_phrase_is_valid_bip39(self):
+        phrase = generate_recovery_phrase()
+        assert len(phrase.split()) == 12
+        assert validate_recovery_phrase(phrase)
+
+    def test_typo_fails_checksum(self):
+        words = generate_recovery_phrase().split()
+        words[0] = "abandon" if words[0] != "abandon" else "ability"
+        assert not validate_recovery_phrase(" ".join(words))
+
+    def test_round_trip(self):
+        mk = generate_master_key()
+        phrase = generate_recovery_phrase()
+        assert unwrap_master_key_with_phrase(wrap_master_key_with_phrase(mk, phrase), phrase) == mk
+
+    def test_normalization_tolerates_case_and_whitespace(self):
+        mk = generate_master_key()
+        phrase = generate_recovery_phrase()
+        wrap = wrap_master_key_with_phrase(mk, phrase)
+        sloppy = f"  {phrase.upper().replace(' ', '   ')} \n"
+        assert unwrap_master_key_with_phrase(wrap, sloppy) == mk
+
+    def test_wrong_phrase_fails(self):
+        wrap = wrap_master_key_with_phrase(generate_master_key(), generate_recovery_phrase())
+        with pytest.raises(CloudCryptoError):
+            unwrap_master_key_with_phrase(wrap, generate_recovery_phrase())
+
+    def test_malformed_kdf_params_fail(self):
+        wrap = wrap_master_key_with_phrase(generate_master_key(), generate_recovery_phrase())
+        broken = RecoveryWrap(wrapped_key=wrap.wrapped_key, kdf_salt=wrap.kdf_salt, kdf_params="not json")
+        with pytest.raises(CloudCryptoError):
+            unwrap_master_key_with_phrase(broken, generate_recovery_phrase())
+
+
+class TestEnvelope:
+    def test_round_trip(self):
+        mk = generate_master_key()
+        plaintext = b"capture transcript \xf0\x9f\x8e\x99 and some audio bytes" * 100
+        envelope = encrypt_blob(plaintext, mk, **SLOT)
+        assert envelope[:4] == ENVELOPE_MAGIC
+        assert envelope[4] == ALG_XCHACHA20_POLY1305
+        assert decrypt_blob(envelope, mk, **SLOT) == plaintext
+
+    def test_fresh_content_key_per_blob(self):
+        mk = generate_master_key()
+        assert encrypt_blob(b"same", mk, **SLOT) != encrypt_blob(b"same", mk, **SLOT)
+
+    def test_wrong_master_key_fails(self):
+        envelope = encrypt_blob(b"secret", generate_master_key(), **SLOT)
+        with pytest.raises(CloudCryptoError):
+            decrypt_blob(envelope, generate_master_key(), **SLOT)
+
+    @pytest.mark.parametrize(
+        "slot",
+        [
+            {**SLOT, "object_id": "11111111-2222-3333-4444-555555555555"},
+            {**SLOT, "role": "avatar"},
+            {**SLOT, "version": 4},
+        ],
+    )
+    def test_wrong_slot_fails_aad(self, slot):
+        mk = generate_master_key()
+        envelope = encrypt_blob(b"secret", mk, **SLOT)
+        with pytest.raises(CloudCryptoError):
+            decrypt_blob(envelope, mk, **slot)
+
+    def test_tampered_ciphertext_fails(self):
+        mk = generate_master_key()
+        envelope = bytearray(encrypt_blob(b"secret", mk, **SLOT))
+        envelope[-1] ^= 0x01
+        with pytest.raises(CloudCryptoError):
+            decrypt_blob(bytes(envelope), mk, **SLOT)
+
+    def test_bad_magic_and_truncation_fail(self):
+        mk = generate_master_key()
+        envelope = encrypt_blob(b"secret", mk, **SLOT)
+        with pytest.raises(CloudCryptoError):
+            decrypt_blob(b"NOPE" + envelope[4:], mk, **SLOT)
+        with pytest.raises(CloudCryptoError):
+            decrypt_blob(envelope[:20], mk, **SLOT)
+
+    def test_unknown_algorithm_fails(self):
+        mk = generate_master_key()
+        envelope = bytearray(encrypt_blob(b"secret", mk, **SLOT))
+        envelope[4] = 99
+        with pytest.raises(CloudCryptoError):
+            decrypt_blob(bytes(envelope), mk, **SLOT)
+
+    def test_empty_plaintext_round_trips(self):
+        mk = generate_master_key()
+        assert decrypt_blob(encrypt_blob(b"", mk, **SLOT), mk, **SLOT) == b""

backend/tests/test_cloud_roundtrip_integration.py

@@ -0,0 +1,104 @@
+"""End-to-end round-trip against a real voicebox-cloud dev server.
+
+Skipped unless VOICEBOX_CLOUD_TEST_API + VOICEBOX_CLOUD_TEST_KEY are set:
+
+    cd voicebox-cloud && pnpm dev:db && pnpm db:migrate && pnpm dev:api
+    # create an account + API key (web app or seed script), then:
+    VOICEBOX_CLOUD_TEST_API=http://localhost:17593 \\
+    VOICEBOX_CLOUD_TEST_KEY=voicebox_… \\
+    pytest backend/tests/test_cloud_roundtrip_integration.py -v
+
+Exercises the scaffolded server for real: device registration, recovery
+escrow, encrypted push (presigned PUT + commit), sync pull, decrypt — and
+verifies the ciphertext at rest is opaque.
+"""
+
+import base64
+import hashlib
+import json
+import os
+import uuid
+
+import pytest
+
+from backend.services import cloud_crypto
+from backend.services.cloud_api import CloudApiClient
+
+API_URL = os.environ.get("VOICEBOX_CLOUD_TEST_API")
+API_KEY = os.environ.get("VOICEBOX_CLOUD_TEST_KEY")
+
+pytestmark = pytest.mark.skipif(
+    not (API_URL and API_KEY),
+    reason="set VOICEBOX_CLOUD_TEST_API and VOICEBOX_CLOUD_TEST_KEY to run against a dev server",
+)
+
+
+async def test_full_roundtrip():
+    master_key = cloud_crypto.generate_master_key()
+    client_id = str(uuid.uuid4())
+    record_plain = json.dumps({"transcript_raw": "hello from the integration test", "language": "en"}).encode()
+    audio_plain = os.urandom(64_000)  # stands in for capture audio
+
+    async with CloudApiClient(API_URL, API_KEY) as client:
+        # Device + escrow round-trip.
+        private_key, public_key = cloud_crypto.generate_device_keypair()
+        registered = await client.register_device("integration-test", base64.b64encode(public_key).decode())
+        device_id = registered["deviceId"]
+        wrapped = cloud_crypto.wrap_master_key_for_device(master_key, public_key)
+        await client.put_wrapped_key(device_id, base64.b64encode(wrapped).decode())
+        fetched = await client.get_wrapped_key(device_id)
+        assert cloud_crypto.unwrap_master_key_for_device(base64.b64decode(fetched), private_key) == master_key
+
+        # Push: encrypt locally, upsert metadata, PUT ciphertext, commit.
+        object_id_placeholder = client_id  # AAD object binding uses the client id pre-push
+        record_env = cloud_crypto.encrypt_blob(
+            record_plain, master_key, object_id=object_id_placeholder, role="record", version=1
+        )
+        audio_env = cloud_crypto.encrypt_blob(
+            audio_plain, master_key, object_id=object_id_placeholder, role="audio", version=1
+        )
+        pushed = await client.push_object(
+            kind="capture",
+            client_id=client_id,
+            version=1,
+            record={"hash": hashlib.sha256(record_env).hexdigest(), "size": len(record_env)},
+            assets=[
+                {
+                    "role": "audio",
+                    "clientAssetId": f"{client_id}-audio",
+                    "hash": hashlib.sha256(audio_env).hexdigest(),
+                    "size": len(audio_env),
+                }
+            ],
+        )
+        uploads = {u["for"]: u["url"] for u in pushed["uploads"]}
+        await client.upload_blob(uploads["record"], record_env)
+        await client.upload_blob(uploads[f"asset:{client_id}-audio"], audio_env)
+        await client.commit_object(pushed["objectId"])
+
+        # Pull: cursor 0 must include our object; ciphertext decrypts to the original.
+        changes = await client.get_changes(since=pushed["seq"] - 1, limit=10)
+        change = next(c for c in changes["changes"] if c["clientId"] == client_id)
+        assert change["kind"] == "capture"
+        assert not change["deleted"]
+
+        record_cipher = await client.download_blob(change["record"]["url"])
+        assert record_cipher == record_env  # opaque, byte-identical ciphertext at rest
+        assert record_plain not in record_cipher
+        assert (
+            cloud_crypto.decrypt_blob(record_cipher, master_key, object_id=client_id, role="record", version=1)
+            == record_plain
+        )
+
+        (asset,) = change["assets"]
+        audio_cipher = await client.download_blob(asset["url"])
+        assert (
+            cloud_crypto.decrypt_blob(audio_cipher, master_key, object_id=client_id, role="audio", version=1)
+            == audio_plain
+        )
+
+        # Tombstone propagates.
+        await client.delete_object(pushed["objectId"])
+        changes = await client.get_changes(since=changes["cursor"], limit=10)
+        tombstone = next(c for c in changes["changes"] if c["clientId"] == client_id)
+        assert tombstone["deleted"] is True

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"

bun.lock

@@ -57,6 +57,7 @@
         "react-dom": "^18.3.0",
         "react-hook-form": "^7.53.0",
         "react-i18next": "^17.0.4",
+        "react-qr-code": "^2.0.18",
         "react-sound-visualizer": "^1.4.0",
         "tailwind-merge": "^2.5.4",
         "wavesurfer.js": "^7.0.0",
@@ -1005,6 +1006,8 @@
 
     "punycode": ["[email protected]", "", {}, "sha512-vYt7UD1U9Wg6138shLtLOvdAu+8DsC/ilFtEVHcH+wydcSpNE20AfSOduf6MkRFahL5FY7X1oU7nKVZFtfq8Fg=="],
 
+    "qr.js": ["[email protected]", "", {}, "sha512-c4iYnWb+k2E+vYpRimHqSu575b1/wKl4XFeJGpFmrJQz5I88v9aY2czh7s0w36srfCM1sXgC/xpoJz5dJfq+OQ=="],
+
     "queue-microtask": ["[email protected]", "", {}, "sha512-NuaNSa6flKT5JaSYQzJok04JzTL1CA6aGhv5rfLW3PgqA+M2ChpZQnAC8h8i4ZFkBS8X5RqkDBHA7r4hej3K9A=="],
 
     "react": ["[email protected]", "", { "dependencies": { "loose-envify": "^1.1.0" } }, "sha512-wS+hAgJShR0KhEvPJArfuPVN1+Hz1t0Y6n5jLrGQbkb4urgPE/0Rve+1kMB1v/oWgHgm4WIcV+i7F2pTVj+2iQ=="],
@@ -1019,6 +1022,8 @@
 
     "react-loaders": ["[email protected]", "", { "dependencies": { "classnames": "^2.2.3" }, "peerDependencies": { "prop-types": ">=15.6.0", "react": ">=15" } }, "sha512-4igMNqs9Fb3d4Z+0UHIGQNJsw/37gX0nUO8QxupnEKRn1dtyYC1LGwk5GuaoDciMQCQc/MmPwb4Fn6ZfdoX1FQ=="],
 
+    "react-qr-code": ["[email protected]", "", { "dependencies": { "prop-types": "^15.8.1", "qr.js": "0.0.0" }, "peerDependencies": { "react": "*" } }, "sha512-v1Jqz7urLMhkO6jkgJuBYhnqvXagzceg3qJUWayuCK/c6LTIonpWbwxR1f1APGd4xrW/QcQEovNrAojbUz65Tg=="],
+
     "react-refresh": ["[email protected]", "", {}, "sha512-z6F7K9bV85EfseRCp2bzrpyQ0Gkw1uLoCel9XBVWPg/TjRj94SkJzUTGfOa4bs7iJvBWtQG0Wq7wnI0syw3EBQ=="],
 
     "react-remove-scroll": ["[email protected]", "", { "dependencies": { "react-remove-scroll-bar": "^2.3.7", "react-style-singleton": "^2.2.3", "tslib": "^2.1.0", "use-callback-ref": "^1.3.3", "use-sidecar": "^1.1.3" }, "peerDependencies": { "@types/react": "*", "react": "^16.8.0 || ^17.0.0 || ^18.0.0 || ^19.0.0 || ^19.0.0-rc" }, "optionalPeers": ["@types/react"] }, "sha512-Iqb9NjCCTt6Hf+vOdNIZGdTiH1QSqr27H/Ek9sv/a97gfueI/5h1s3yRi1nngzMUaOOToin5dI1dXKdXiF+u0Q=="],