feat(sync): wire state.json + divergence detection + dialogs

state.py: per-instance sync state. <pack>/.cloud-sync/state.json
(mode 600) records last_pulled_snapshot_id + last_pulled_at +
host_tag. Versioned schema. clear() on remote-empty.

sync.pull decision tree (replaces the unconditional restore):

  no token file
    → prompt_login_qt; on Skip return 0 (don't block launch)
  no state + remote empty
    → no-op
  no state + remote non-empty
    → restore (first-run on this machine)
  state.id == remote.id
    → skip restore (up to date)
  state.id != remote.id, no in-scope local edits since state.at
    → restore (fast-forward)
  state.id != remote.id, in-scope local edits since state.at
    → prompt_conflict_qt
      keep_local  → don't restore; push will overwrite cloud
      use_remote  → restore + update state
      cancel      → exit 1

sync.push: --json output parsed for snapshot_id; state.json updated
to that id after a successful backup. Skips silently if no token.

_find_modified_in_scope: walks include roots, filters via
_matches_any (restic-style globs: dir/, **/dir/, **/*.glob).
Stops at 50 hits; we only need 'any' + a sample for the dialog.

_format_dt: hand-rolled (no GNU-vs-Windows strftime quirks) →
'Thursday, October 21, 2021 at 7:12 PM'.

Restic JSON parsing helpers: _parse_snapshots, _parse_restic_time
(handles nanosecond precision), _parse_backup_summary.

tests/test_state.py: 19 new tests covering state read/write, scope-
aware mtime walk, exclude glob matching, restic output parsers.
Total: 52 green.

cloud_sync/state.py

@@ -0,0 +1,84 @@
+"""Per-instance sync state.
+
+Tracks the snapshot id this pack was last synced to and when. Lives at
+``<pack-folder>/.cloud-sync/state.json`` (mode 600).
+
+Purpose: divergence detection. On ``pull``, if the remote latest id
+differs from ``last_pulled_snapshot_id`` AND any in-scope local file
+has mtime > ``last_pulled_at``, the local and remote diverged from a
+common ancestor — surface the conflict dialog.
+"""
+
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from pathlib import Path
+
+
+SCHEMA_VERSION = 1
+
+
+@dataclass(frozen=True)
+class State:
+    last_pulled_snapshot_id: str
+    last_pulled_at: datetime
+    host_tag: str = "cloud-sync"
+
+
+def state_path(pack_folder: Path) -> Path:
+    return pack_folder / ".cloud-sync" / "state.json"
+
+
+def read(pack_folder: Path) -> State | None:
+    """Return parsed state or None if file missing / unreadable / wrong schema."""
+    p = state_path(pack_folder)
+    if not p.exists():
+        return None
+    try:
+        data = json.loads(p.read_text(encoding="utf-8"))
+    except (OSError, json.JSONDecodeError):
+        return None
+    if data.get("schema") != SCHEMA_VERSION:
+        return None
+    try:
+        return State(
+            last_pulled_snapshot_id=data["last_pulled_snapshot_id"],
+            last_pulled_at=_parse_iso(data["last_pulled_at"]),
+            host_tag=data.get("host_tag", "cloud-sync"),
+        )
+    except (KeyError, ValueError):
+        return None
+
+
+def write(pack_folder: Path, state: State) -> None:
+    """Persist state. Creates parent dir + sets mode 600."""
+    p = state_path(pack_folder)
+    p.parent.mkdir(parents=True, exist_ok=True)
+    payload = {
+        "schema": SCHEMA_VERSION,
+        "last_pulled_snapshot_id": state.last_pulled_snapshot_id,
+        "last_pulled_at": state.last_pulled_at.astimezone(timezone.utc)
+        .isoformat()
+        .replace("+00:00", "Z"),
+        "host_tag": state.host_tag,
+    }
+    p.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8")
+    p.chmod(0o600)
+
+
+def clear(pack_folder: Path) -> None:
+    """Remove state.json if present. Used when remote has zero snapshots."""
+    p = state_path(pack_folder)
+    p.unlink(missing_ok=True)
+
+
+def _parse_iso(s: str) -> datetime:
+    """Parse ISO-8601 with trailing Z or +HH:MM, return tz-aware UTC."""
+    if s.endswith("Z"):
+        s = s[:-1] + "+00:00"
+    dt = datetime.fromisoformat(s)
+    if dt.tzinfo is None:
+        dt = dt.replace(tzinfo=timezone.utc)
+    return dt.astimezone(timezone.utc)

cloud_sync/sync.py

@@ -1,38 +1,52 @@
-"""pull + push entry points.
+"""pull + push entry points + divergence detection.
 
-Both subprocess restic against ``rest:<scheme>://<id>:<password>@<host>/<id>/``
-where the password is HTTP basic auth ONLY. Restic repos are initialised with
-``--insecure-no-password`` so no encryption-at-rest password exists; protection
-relies on:
+Backend: restic-rest-server. URL form:
+``rest:<scheme>://<id>:<password>@<host>/<id>/``
 
-  1. TLS in transit at the reverse proxy
-  2. ``--private-repos`` + htpasswd per user at restic-rest-server
-  3. ``--append-only`` to prevent client-side deletion
-  4. Disk-level encryption (LUKS) on the host
+Password = HTTP basic auth ONLY. Restic repos are initialised with
+``--insecure-no-password``; encryption-at-rest delegated to LUKS on the
+host disk + TLS at the reverse proxy.
 
-Defense-in-depth via repo encryption was dropped because the threat model
-(homelab, operator-trusted) doesn't justify the password-coordination cost.
+Divergence detection (pull only):
+
+  1. Fetch remote latest snapshot id + time via ``snapshots --json --latest 1``.
+  2. Read ``<pack>/.cloud-sync/state.json`` for the last-pulled snapshot id.
+  3. Branches:
+       - no state.json AND remote empty           → no-op
+       - no state.json AND remote non-empty       → first-run restore
+       - state.last_pulled == remote.id           → up to date, skip
+       - state.last_pulled != remote.id, no local edits since
+         state.last_pulled_at                     → fast-forward restore
+       - state.last_pulled != remote.id, local edits since
+         state.last_pulled_at                     → CONFLICT dialog
 """
 
 from __future__ import annotations
 
+import fnmatch
+import json
 import sys
 import urllib.parse
+from datetime import datetime, timezone
 from pathlib import Path
 
-from . import restic, scope as scopemod
+from . import restic, scope as scopemod, state as statemod
 from .cli import Args
 from .creds import read_credentials
 from .ui import HeadlessProgress, Progress
 
 
 def pull(args: Args, progress: Progress | None = None) -> int:
-    """Restore latest snapshot's files into pack_folder.
-
-    If the repo has no snapshots yet, this is a no-op (first run on this
-    machine; nothing to restore).
-    """
     ui = progress or HeadlessProgress()
+
+    # First-run login. If the user declines, skip cloud sync without
+    # blocking the launch (return 0 — non-fatal for Prism PreLaunch).
+    if not args.token_file.exists():
+        if not _prompt_login_and_save(args, ui):
+            ui.set_status("Cloud sync skipped")
+            print("cloud-sync: no token; skipping pull")
+            return 0
+
     ui.set_status("Reading credentials…")
     discord_id, password = read_credentials(args.token_file)
 
@@ -56,18 +70,56 @@ def pull(args: Args, progress: Progress | None = None) -> int:
             file=sys.stderr,
         )
         return 2
-    stripped = out.strip()
-    if stripped in ("", "null", "[]"):
+
+    snapshots = _parse_snapshots(out)
+    if not snapshots:
+        statemod.clear(args.pack_folder)
         ui.set_status("No snapshots yet — nothing to pull")
-        print(
-            "cloud-sync: no snapshots yet for this user "
-            "(first run on this machine?); nothing to pull"
-        )
+        print("cloud-sync: no snapshots yet for this user; nothing to pull")
         return 0
 
-    scope = scopemod.load(args.pack_folder)
-    _, exclude_from = scopemod.materialize_for_restic(args.pack_folder, scope)
+    remote = snapshots[0]
+    remote_id = remote["id"]
+    remote_time = _parse_restic_time(remote["time"])
 
+    local_state = statemod.read(args.pack_folder)
+    scope = scopemod.load(args.pack_folder)
+
+    decision: str
+    if local_state is None:
+        decision = "use_remote"
+    elif local_state.last_pulled_snapshot_id == remote_id:
+        ui.set_status("Cloud is up to date")
+        print("cloud-sync: already at latest snapshot")
+        return 0
+    else:
+        modified = _find_modified_in_scope(
+            args.pack_folder, scope, local_state.last_pulled_at
+        )
+        if not modified:
+            decision = "use_remote"
+        else:
+            decision = _ask_conflict(modified, remote_time)
+            if decision is None:
+                # UI unavailable in headless mode → conservative: cancel
+                ui.set_status("Conflict detected; no UI available")
+                print(
+                    "cloud-sync: conflict detected (remote moved + local edits) "
+                    "but headless mode can't prompt; aborting",
+                    file=sys.stderr,
+                )
+                return 1
+
+    if decision == "cancel":
+        ui.set_status("Launch cancelled")
+        return 1
+    if decision == "keep_local":
+        ui.set_status("Keeping local; push will overwrite cloud on exit")
+        print("cloud-sync: keeping local copy")
+        return 0
+
+    # decision == "use_remote"
+    _, exclude_from = scopemod.materialize_for_restic(args.pack_folder, scope)
     ui.set_status("Restoring files…")
     code, _ = restic.run(
         binary,
@@ -85,14 +137,27 @@ def pull(args: Args, progress: Progress | None = None) -> int:
     if code != 0:
         print(f"cloud-sync: restic restore failed (exit {code})", file=sys.stderr)
         return 2
+
+    statemod.write(
+        args.pack_folder,
+        statemod.State(
+            last_pulled_snapshot_id=remote_id,
+            last_pulled_at=datetime.now(timezone.utc),
+        ),
+    )
     ui.set_status("Pull complete")
     print("cloud-sync: pull ok")
     return 0
 
 
 def push(args: Args, progress: Progress | None = None) -> int:
-    """Snapshot the in-scope files into the user's repo."""
     ui = progress or HeadlessProgress()
+
+    if not args.token_file.exists():
+        ui.set_status("No cloud token; skipping push")
+        print("cloud-sync: no token; skipping push")
+        return 0
+
     ui.set_status("Reading credentials…")
     discord_id, password = read_credentials(args.token_file)
 
@@ -105,7 +170,7 @@ def push(args: Args, progress: Progress | None = None) -> int:
     files_from, exclude_from = scopemod.materialize_for_restic(args.pack_folder, scope)
 
     ui.set_status("Uploading snapshot…")
-    code, _ = restic.run(
+    code, out = restic.run(
         binary,
         [
             "-r", repo, "--insecure-no-password",
@@ -114,6 +179,7 @@ def push(args: Args, progress: Progress | None = None) -> int:
             "--exclude-file", str(exclude_from),
             "--host", "cloud-sync",
             "--tag", "auto",
+            "--json",
         ],
         env=env,
         cwd=args.pack_folder,
@@ -124,22 +190,211 @@ def push(args: Args, progress: Progress | None = None) -> int:
     if code != 0:
         print(f"cloud-sync: restic backup failed (exit {code})", file=sys.stderr)
         return 2
+
+    new_id = _parse_backup_summary(out)
+    if new_id is not None:
+        statemod.write(
+            args.pack_folder,
+            statemod.State(
+                last_pulled_snapshot_id=new_id,
+                last_pulled_at=datetime.now(timezone.utc),
+            ),
+        )
     ui.set_status("Push complete")
     print("cloud-sync: push ok")
     return 0
 
 
 # ---------------------------------------------------------------------------
-# helpers
+# divergence detection
 # ---------------------------------------------------------------------------
 
 
-def _restic_repo(base_url: str, discord_id: str, password: str) -> str:
-    """Build rest:<scheme>://<id>:<pw>@<host>/<id>/
+def _find_modified_in_scope(
+    pack_folder: Path, scope: scopemod.Scope, since: datetime
+) -> list[tuple[Path, datetime]]:
+    """Return (rel_path, mtime) for in-scope files newer than ``since``.
 
-    URL-embedded basic auth is universally supported by restic; alternative
-    env vars (RESTIC_REST_USERNAME, RESTIC_REST_PASSWORD) require 0.16+.
+    Walks each include root, skips paths matching any exclude glob. Stops
+    early at 50 hits (we only need to know "any" + sample some for the
+    conflict dialog). False positives are safer than false negatives —
+    a spurious conflict shows a dismissable dialog; a missed conflict
+    silently overwrites the user's edits.
     """
+    since_ts = since.timestamp()
+    hits: list[tuple[Path, datetime]] = []
+    for include in scope.include:
+        root = pack_folder / include.rstrip("/")
+        if not root.exists():
+            continue
+        candidates: list[Path] = [root] if root.is_file() else _walk_files(root)
+        for f in candidates:
+            try:
+                mtime = f.stat().st_mtime
+            except OSError:
+                continue
+            if mtime <= since_ts:
+                continue
+            rel = f.relative_to(pack_folder)
+            if _matches_any(rel, scope.exclude):
+                continue
+            hits.append((rel, datetime.fromtimestamp(mtime, tz=timezone.utc)))
+            if len(hits) >= 50:
+                return hits
+    return hits
+
+
+def _walk_files(root: Path):
+    try:
+        for p in root.rglob("*"):
+            if p.is_file():
+                yield p
+    except OSError:
+        return
+
+
+def _matches_any(rel: Path, patterns: list[str]) -> bool:
+    """Restic-style glob match against the relative path.
+
+    Subset of restic's exclude semantics that covers our default scope:
+      ``foo/``         — foo itself OR anything under it
+      ``**/foo/``      — any ancestor directory named foo, recursively
+      ``foo/bar*``     — fnmatch against the full relative path
+      ``**/*.log``     — fnmatch the basename (and any tail subpath)
+    """
+    rel_str = str(rel)
+    parts = rel.parts
+    for pat in patterns:
+        if pat.endswith("/"):
+            core = pat.rstrip("/")
+            if core.startswith("**/"):
+                tail = core[len("**/") :]
+                if tail in parts:
+                    return True
+                continue
+            if rel_str == core or rel_str.startswith(core + "/"):
+                return True
+            continue
+        if fnmatch.fnmatch(rel_str, pat):
+            return True
+        if pat.startswith("**/"):
+            suffix = pat[len("**/") :]
+            if fnmatch.fnmatch(rel.name, suffix):
+                return True
+            for i in range(len(parts)):
+                if fnmatch.fnmatch("/".join(parts[i:]), suffix):
+                    return True
+    return False
+
+
+def _ask_conflict(
+    modified: list[tuple[Path, datetime]],
+    remote_time: datetime,
+) -> str | None:
+    """Show the conflict dialog. Returns choice or None if no UI available."""
+    try:
+        from .ui_qt import prompt_conflict_qt
+    except ImportError:
+        return None
+    newest = max(modified, key=lambda h: h[1])
+    return prompt_conflict_qt(
+        local_modified=_format_dt(newest[1]),
+        remote_modified=_format_dt(remote_time),
+        save_label="Minecraft save",
+    )
+
+
+def _prompt_login_and_save(args: Args, ui: Progress) -> bool:
+    """First-run login. Returns True if a token was saved, False if skipped."""
+    try:
+        from .ui_qt import prompt_login_qt
+    except ImportError:
+        ui.set_status("No token and no UI; can't prompt")
+        print(
+            "cloud-sync: no token at "
+            f"{args.token_file} and no Qt UI available",
+            file=sys.stderr,
+        )
+        return False
+    token = prompt_login_qt()
+    if token is None:
+        return False
+    args.token_file.parent.mkdir(parents=True, exist_ok=True)
+    args.token_file.write_text(token + "\n", encoding="utf-8")
+    args.token_file.chmod(0o600)
+    return True
+
+
+# ---------------------------------------------------------------------------
+# restic output parsing
+# ---------------------------------------------------------------------------
+
+
+def _parse_snapshots(out: str) -> list[dict]:
+    s = out.strip()
+    if not s or s == "null":
+        return []
+    try:
+        data = json.loads(s)
+    except json.JSONDecodeError:
+        return []
+    return data if isinstance(data, list) else []
+
+
+def _parse_restic_time(s: str) -> datetime:
+    """Restic emits e.g. ``2026-06-04T18:33:21.123456789Z``."""
+    if s.endswith("Z"):
+        s = s[:-1] + "+00:00"
+    # Python's fromisoformat doesn't accept nanosecond precision — trim to micro.
+    if "." in s:
+        head, _, tail = s.partition(".")
+        frac, _, tz = tail.partition("+")
+        if tz:
+            tz = "+" + tz
+        else:
+            frac, _, tz = tail.partition("-")
+            if tz:
+                tz = "-" + tz
+        s = f"{head}.{frac[:6]}{tz}"
+    return datetime.fromisoformat(s).astimezone(timezone.utc)
+
+
+def _parse_backup_summary(out: str) -> str | None:
+    """restic backup --json emits one JSON line per event; the final
+    ``summary`` event carries ``snapshot_id``."""
+    for line in reversed(out.splitlines()):
+        line = line.strip()
+        if not line:
+            continue
+        try:
+            ev = json.loads(line)
+        except json.JSONDecodeError:
+            continue
+        if ev.get("message_type") == "summary" and "snapshot_id" in ev:
+            return str(ev["snapshot_id"])
+    return None
+
+
+# ---------------------------------------------------------------------------
+# misc helpers
+# ---------------------------------------------------------------------------
+
+
+def _format_dt(dt: datetime) -> str:
+    """Format a tz-aware datetime as 'Thursday, October 21, 2021 at 7:12 PM'.
+
+    Hand-rolled instead of ``strftime("%-d")`` because GNU strftime's
+    leading-zero stripping syntax is platform-specific (``%-d`` vs ``%#d``).
+    """
+    local = dt.astimezone()
+    weekday = local.strftime("%A")
+    month = local.strftime("%B")
+    hour = local.hour % 12 or 12
+    ampm = local.strftime("%p")
+    return f"{weekday}, {month} {local.day}, {local.year} at {hour}:{local.minute:02d} {ampm}"
+
+
+def _restic_repo(base_url: str, discord_id: str, password: str) -> str:
     raw = base_url.strip()
     if raw.startswith("rest:"):
         raw = raw[len("rest:"):]