granthi-sync/client/granthi_sync_client.py

#!/usr/bin/env python3
"""granthi-sync client daemon.

Commands:
  link  [--server URL] [--token ZITADEL_TOKEN] [--device NAME]
        Zitadel OAuth device flow against https://id.shre.ai (or --token for
        the headless path), then POST /v1/link on the granthi-link service.
        Stores {server, gitea_base, login, token, token_name} in
        ~/.granthi-sync/config.json (0600).
  list  [pattern]
        Table of every repo the linked token can see on the forge -- which
        is exactly the set the forge grants this account, never a
        client-side filter over someone else's repos -- with the local
        folder each one is already synced to (if any). `pattern` narrows
        the table by name (substring, or a glob like 'work-*').
  get   <repo|owner/repo> [--into DIR] [--mode M]  |  --all [--into DIR]
        Clone a forge repo and register it for `watch` -- the download half
        of `add`. Remote is named 'granthi' at clone time and the token is
        supplied by the credential helper, never embedded in the URL.
        `--all` pulls every repo this account can see, skipping the ones
        already linked on this machine.
  add   <folder> [--name N] [--private/--public] [--mode M] [--force]
        git init (branch main) if needed, create the cloud repo via
        /v1/repos, add remote 'granthi', initial commit + push. The token
        is supplied via a git credential helper (this script's hidden
        `git-credential` subcommand), never embedded in the remote URL.
        Seeds a starter .gitignore and refuses an unexpectedly large folder
        unless --force.
  watch [--interval 30] [--once]
        Poll linked folders. Per folder, by MODE (below): `mirror`
        autocommits local changes and pushes them; `snapshot` never commits
        for you and instead mirrors the working tree to a per-device backup
        ref. Both ff-pull when the remote is strictly ahead. DIVERGED
        branches are logged + recorded and SKIPPED -- never force, never
        merge (same policy as the estate's gitea_sync.py mesh) -- but a
        diverged folder is still backed up. SIGTERM-clean.
  snapshots <folder>
        List the restore points held for a folder, newest first.
  restore <folder> --at <TIMESTAMP|SHA> [--into DIR]
        Materialise one restore point into a NEW directory. Never writes
        over the working tree.
  status
        Table of linked folders, mode, last sync, divergence flags.

MODES
  mirror    The folder IS the repo. Local changes are autocommitted as
            `sync: <ISO ts>` and pushed; every commit is a restore point.
            Chosen automatically for a plain folder that `add` turned into
            a repo -- the "my documents live in the cloud" case.
  snapshot  The folder has a real git history that belongs to the user.
            NOTHING is committed on their behalf. Each pass builds a commit
            object from the working tree without touching HEAD, the index
            or any file, and pushes it to
            refs/granthi-backup/<device>/<ts>. Uncommitted, unmerged,
            half-finished work is therefore off the machine with a
            timestamp to restore from, while the user's own history stays
            exactly as they left it. Chosen automatically for a folder that
            was already a git repo, and for anything `get` clones.

Stdlib + git CLI only.
"""

import argparse
import base64
import fnmatch
import json
import os
import re
import shlex
import signal
import subprocess
import sys
import tempfile
import time
import urllib.error
import urllib.parse
import urllib.request
import uuid
from datetime import datetime, timedelta, timezone

VERSION = "1.2.0"

CONFIG_DIR = os.path.expanduser(os.environ.get("GRANTHI_SYNC_HOME", "~/.granthi-sync"))
CONFIG_PATH = os.path.join(CONFIG_DIR, "config.json")

ZITADEL_BASE = "https://id.shre.ai"
DEVICE_CLIENT_ID = "386909715541590022"
# ^ Zitadel native app "granthi-sync-device" (appId 386909715541524486) in
#   project granthi-forge (386906525790109702); device-code + refresh grants.
# Public by default: a genuinely new computer cannot be expected to join a
# private network before it can sign in. The tailnet address still works and
# is what internal machines should pass to --server.
DEFAULT_SERVER = os.environ.get("GRANTHI_LINK_SERVER",
                                "https://granthi-link.shre.ai")
TAILNET_SERVER = "http://100.111.127.127:3042"
DEVICE_SCOPE = "openid profile email"

FORGE_PAGE_LIMIT = 50
FORGE_MAX_PAGES = 40  # 2000 repos; a guard against an unbounded paging loop

# Backup refs live OUTSIDE refs/heads on purpose: verified on the beta forge
# (Gitea 1.27.2) that a custom namespace is accepted, is readable through
# ls-remote, and does NOT appear in the branch list or the branch dropdown.
# Under refs/heads a machine taking a snapshot every 30s would bury the
# user's real branches.
BACKUP_NS = "refs/granthi-backup"
SNAPSHOT_TS_FMT = "%Y%m%dT%H%M%SZ"

# Retention. Unbounded snapshots are a disk leak with no way to use them, so
# thin them by age: everything recent, then hourly, then daily.
KEEP_ALL_HOURS = 24
KEEP_HOURLY_DAYS = 7
PRUNE_EVERY_SECONDS = 3600

# `add` guard rails. A plain `git add -A` on the wrong folder commits every
# byte in it, so refuse the surprising cases unless the user insists.
ADD_MAX_BYTES = 512 * 1024 * 1024
ADD_MAX_FILES = 20000
MARKER_FILE = ".granthi-sync.json"

# Seeded only when the folder has no .gitignore of its own. The point is the
# secret-shaped entries: the rest is noise reduction.
DEFAULT_GITIGNORE = """\
# Seeded by granthi-sync. Edit freely -- it is yours now.
.DS_Store
Thumbs.db
*.swp
node_modules/
__pycache__/
.venv/
venv/

# Never sync credentials.
.env
.env.*
*.key
*.pem
*.p12
id_rsa
id_ed25519
credentials.json
secrets.json
"""


def log(msg):
    ts = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
    print(f"[{ts}] {msg}", flush=True)


# --------------------------------------------------------------------------
# HTTP helpers (patchable in tests)
# --------------------------------------------------------------------------

def http_json(method, url, headers=None, body=None, form=None, timeout=30):
    """Returns (status, parsed-json-or-{'raw': text})."""
    data = None
    hdrs = dict(headers or {})
    # Cloudflare in front of id.shre.ai 403s the default Python-urllib UA.
    hdrs.setdefault("User-Agent", f"granthi-sync/{VERSION}")
    if form is not None:
        data = urllib.parse.urlencode(form).encode()
        hdrs.setdefault("Content-Type", "application/x-www-form-urlencoded")
    elif body is not None:
        data = json.dumps(body).encode()
        hdrs.setdefault("Content-Type", "application/json")
    req = urllib.request.Request(url, data=data, headers=hdrs, method=method)
    try:
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            raw, status = resp.read(), resp.status
    except urllib.error.HTTPError as e:
        raw, status = e.read(), e.code
    except (urllib.error.URLError, OSError) as e:
        return 599, {"error": str(e)}
    try:
        return status, json.loads(raw) if raw else {}
    except ValueError:
        return status, {"raw": raw.decode(errors="replace")}


# --------------------------------------------------------------------------
# Config
# --------------------------------------------------------------------------

def load_config():
    try:
        with open(CONFIG_PATH) as f:
            return json.load(f)
    except FileNotFoundError:
        return {}


def device_id(cfg):
    """Stable per-installation id, minted once and persisted.

    The hostname is neither stable (people rename laptops) nor unique (every
    fresh mac is 'Mac mini'), so it cannot key a backup ref or a device
    registry -- two machines would write each other's snapshots. Callers get
    the id from here; `save_config` is theirs to call.
    """
    existing = cfg.get("device_id")
    if existing:
        return existing
    cfg["device_id"] = uuid.uuid4().hex
    return cfg["device_id"]


def save_config(cfg):
    os.makedirs(CONFIG_DIR, mode=0o700, exist_ok=True)
    tmp = CONFIG_PATH + ".tmp"
    # O_CREAT with mode 0600 -- the file is never observable with wider
    # permissions (a write-then-chmod sequence leaves a umask-sized window
    # in which the token is world-readable).
    fd = os.open(tmp, os.O_CREAT | os.O_WRONLY | os.O_TRUNC, 0o600)
    with os.fdopen(fd, "w") as f:
        json.dump(cfg, f, indent=2)
    os.replace(tmp, CONFIG_PATH)


# --------------------------------------------------------------------------
# git helpers
# --------------------------------------------------------------------------

def git(folder, *args, check=True, env=None):
    """Run git in folder. Returns (rc, stdout). Never uses --force."""
    run_env = None
    if env:
        run_env = dict(os.environ)
        run_env.update(env)
    proc = subprocess.run(["git", "-C", folder] + list(args),
                          capture_output=True, text=True, env=run_env)
    if check and proc.returncode != 0:
        raise RuntimeError(
            f"git {' '.join(args)} failed in {folder}: {proc.stderr.strip()}")
    return proc.returncode, proc.stdout.strip()


def ensure_repo(folder):
    if os.path.isdir(os.path.join(folder, ".git")):
        return
    rc, _ = git(folder, "init", "-b", "main", check=False)
    if rc != 0:  # git < 2.28 fallback
        git(folder, "init")
        git(folder, "symbolic-ref", "HEAD", "refs/heads/main")


def autocommit(folder):
    """Commit all local changes as 'sync: <ISO ts>'. Returns True if a
    commit was made."""
    _, status = git(folder, "status", "--porcelain")
    if not status:
        return False
    git(folder, "add", "-A")
    ts = datetime.now(timezone.utc).isoformat(timespec="seconds")
    git(folder, "commit", "-m", f"sync: {ts}")
    return True


# --------------------------------------------------------------------------
# snapshots -- back up work that was never committed
# --------------------------------------------------------------------------

def _head_sha(folder):
    rc, sha = git(folder, "rev-parse", "--verify", "--quiet", "HEAD",
                  check=False)
    return sha.strip() if rc == 0 and sha.strip() else None


def build_snapshot(folder):
    """Commit the working tree WITHOUT touching HEAD, the index or a file.

    A scratch index is loaded from HEAD, `add -A` stages the working tree
    into *that* index, and commit-tree writes a commit object parented on
    HEAD. The user's own index and history are untouched -- they can be
    mid-rebase with a dirty tree and this still records what is on disk.

    Returns (commit_sha, tree_sha) or None when the tree is empty.
    """
    head = _head_sha(folder)
    git_dir = git(folder, "rev-parse", "--absolute-git-dir")[1].strip()
    fd, tmp_index = tempfile.mkstemp(prefix="granthi-index-", dir=git_dir)
    os.close(fd)
    os.unlink(tmp_index)  # git wants to create it itself
    env = {"GIT_INDEX_FILE": tmp_index}
    try:
        if head:
            git(folder, "read-tree", head, env=env)
        # Honours .gitignore, so a seeded ignore file keeps secrets out of
        # snapshots exactly as it keeps them out of commits.
        git(folder, "add", "-A", env=env)
        _, tree = git(folder, "write-tree", env=env)
        tree = tree.strip()
        if not tree:
            return None
        if head:
            _, head_tree = git(folder, "rev-parse", f"{head}^{{tree}}")
            if head_tree.strip() == tree:
                # Nothing uncommitted: HEAD already holds this exact content.
                return None
        ts = datetime.now(timezone.utc).isoformat(timespec="seconds")
        args = ["commit-tree", tree, "-m", f"granthi snapshot: {ts}"]
        if head:
            args += ["-p", head]
        # Snapshots are parented on HEAD and nothing else -- deliberately NOT
        # chained to the previous snapshot. Chaining would keep every old
        # snapshot reachable from the newest one, so pruning a ref would free
        # nothing and retention would be decorative.
        _, commit = git(folder, *args,
                        env={"GIT_AUTHOR_NAME": "granthi-sync",
                             "GIT_AUTHOR_EMAIL": "[email protected]",
                             "GIT_COMMITTER_NAME": "granthi-sync",
                             "GIT_COMMITTER_EMAIL": "[email protected]"})
        return commit.strip(), tree
    finally:
        if os.path.exists(tmp_index):
            os.unlink(tmp_index)


def snapshot_ref(dev, ts=None):
    ts = ts or datetime.now(timezone.utc).strftime(SNAPSHOT_TS_FMT)
    return f"{BACKUP_NS}/{dev}/{ts}", ts


def push_snapshot(folder, dev, remote="granthi"):
    """Take a snapshot and push it to this device's backup namespace.

    Returns the ref pushed, or None when there was nothing new to back up.
    """
    built = build_snapshot(folder)
    if not built:
        return None
    commit, tree = built
    if tree == _last_snapshot_tree(folder, dev):
        return None  # working tree unchanged since the last backup
    ref, _ = snapshot_ref(dev)
    git(folder, "push", remote, f"{commit}:{ref}")
    _remember_snapshot_tree(folder, dev, tree)
    return ref


def _tree_marker_path(folder, dev):
    git_dir = git(folder, "rev-parse", "--absolute-git-dir")[1].strip()
    return os.path.join(git_dir, f"granthi-last-snapshot-{dev}")


def _last_snapshot_tree(folder, dev):
    try:
        with open(_tree_marker_path(folder, dev)) as f:
            return f.read().strip()
    except OSError:
        return None


def _remember_snapshot_tree(folder, dev, tree):
    try:
        with open(_tree_marker_path(folder, dev), "w") as f:
            f.write(tree)
    except OSError:
        pass  # a lost marker costs one redundant push, never correctness


def list_snapshots(folder, dev=None, remote="granthi"):
    """Backup refs on the remote, newest first.

    Read from the REMOTE, not a local cache: the point of the feature is
    surviving the loss of this machine, so what the server holds is the
    answer that matters.

    `dev=None` means EVERY device, and that is what the read paths pass.
    The three scopes differ on purpose:

      * writing is device-scoped, so two machines never overwrite each
        other's snapshots;
      * pruning is device-scoped, so machine A never applies its own clock
        and retention to machine B's refs;
      * reading is NOT scoped, because the case this feature exists for is
        "the laptop died". A new machine has a new id, and scoping the read
        to it would show an empty list while the backups sit on the forge.
    """
    pattern = f"{BACKUP_NS}/{dev}/*" if dev else f"{BACKUP_NS}/*"
    rc, out = git(folder, "ls-remote", remote, pattern, check=False)
    if rc != 0:
        return []
    prefix = f"{BACKUP_NS}/{dev}/" if dev else f"{BACKUP_NS}/"
    snaps = []
    for line in out.splitlines():
        sha, _, ref = line.partition("\t")
        ref = ref.strip()
        if not ref.startswith(prefix):
            continue
        rest = ref[len(f"{BACKUP_NS}/"):]
        device, _, ts = rest.partition("/")
        if not ts:
            continue
        snaps.append({"sha": sha.strip(), "ref": ref, "ts": ts,
                      "device": device})
    return sorted(snaps, key=lambda s: s["ts"], reverse=True)


def _parse_ts(ts):
    try:
        return datetime.strptime(ts, SNAPSHOT_TS_FMT).replace(tzinfo=timezone.utc)
    except ValueError:
        return None


def snapshots_to_prune(snaps, now=None):
    """Retention: keep everything for KEEP_ALL_HOURS, then one per hour for
    KEEP_HOURLY_DAYS, then one per day. Returns the refs to delete.

    Unparseable timestamps are KEPT. A ref this version does not understand
    is not evidence it is worthless, and deleting the unrecognised is how a
    backup system loses the one thing someone needed.
    """
    now = now or datetime.now(timezone.utc)
    keep, drop, seen = [], [], set()
    for snap in sorted(snaps, key=lambda s: s["ts"], reverse=True):
        when = _parse_ts(snap["ts"])
        if when is None:
            keep.append(snap)
            continue
        age = now - when
        if age <= timedelta(hours=KEEP_ALL_HOURS):
            keep.append(snap)
            continue
        bucket = (when.strftime("%Y%m%dT%H")
                  if age <= timedelta(days=KEEP_HOURLY_DAYS)
                  else when.strftime("%Y%m%d"))
        if bucket in seen:
            drop.append(snap)
        else:
            seen.add(bucket)
            keep.append(snap)
    return [s["ref"] for s in drop]


def prune_snapshots(folder, dev, remote="granthi", now=None):
    """Delete thinned-out backup refs on the remote. Returns how many went."""
    stale = snapshots_to_prune(list_snapshots(folder, dev, remote), now=now)
    if not stale:
        return 0
    # Deleting a ref is not force-pushing over anyone's work: these refs are
    # written by this device alone and hold no history but its own.
    for batch_start in range(0, len(stale), 50):
        batch = stale[batch_start:batch_start + 50]
        git(folder, "push", remote, *[f":{ref}" for ref in batch], check=False)
    return len(stale)


def sync_state(folder, remote="granthi", branch="main"):
    """Classify local vs remote after a fetch.

    Returns one of: 'no-remote-branch', 'in-sync', 'local-ahead',
    'remote-ahead-ff', 'diverged'.
    """
    rc, _ = git(folder, "rev-parse", "--verify", "--quiet",
                f"refs/remotes/{remote}/{branch}", check=False)
    if rc != 0:
        return "no-remote-branch"
    _, local = git(folder, "rev-parse", branch)
    _, remote_sha = git(folder, "rev-parse", f"refs/remotes/{remote}/{branch}")
    if local == remote_sha:
        return "in-sync"
    rc, base = git(folder, "merge-base", branch,
                   f"refs/remotes/{remote}/{branch}", check=False)
    if rc != 0:
        return "diverged"  # unrelated histories: treat as divergence
    base = base.strip()
    if base == remote_sha:
        return "local-ahead"
    if base == local:
        return "remote-ahead-ff"
    return "diverged"


def worktree_dirty(folder):
    _, status = git(folder, "status", "--porcelain")
    return bool(status)


def sync_folder(folder, remote="granthi", branch="main", mode="mirror",
                dev=None):
    """One sync pass for one folder. Returns (outcome, detail).

    outcome in {'synced', 'pushed', 'clean', 'diverged', 'error'}.
    Policy: ff-only pulls, plain pushes. NEVER --force, NEVER merge.

    In `snapshot` mode nothing is committed on the user's behalf; their
    uncommitted work is pushed to a backup ref instead, and that happens on
    EVERY outcome -- including diverged and including a tree too dirty to
    ff-pull. Those are exactly the moments work is most at risk, so they are
    the last moments to skip a backup.
    """
    try:
        backed_up = None
        committed = False
        if mode == "mirror":
            committed = autocommit(folder)
        git(folder, "fetch", remote)
        if mode == "snapshot" and dev:
            backed_up = push_snapshot(folder, dev, remote)

        def note(base):
            if backed_up:
                return f"{base}; backed up {backed_up.rsplit('/', 1)[-1]}"
            return base

        state = sync_state(folder, remote, branch)
        ff_pulled = False
        if state == "remote-ahead-ff":
            if mode == "snapshot" and worktree_dirty(folder):
                # A ff-pull would fail (or clobber) with local edits present.
                # The edits are already safe on the backup ref, so report and
                # leave the working tree exactly as the user left it.
                return "clean", note("remote ahead; local edits present, "
                                     "not pulling")
            git(folder, "merge", "--ff-only", f"refs/remotes/{remote}/{branch}")
            ff_pulled = True
            state = sync_state(folder, remote, branch)
        if state == "diverged":
            return "diverged", note("local and remote both advanced; "
                                    "skipping (no force, no merge)")
        if state in ("local-ahead", "no-remote-branch"):
            git(folder, "push", "-u", remote, branch)
            return "pushed", note("committed+pushed" if committed else "pushed")
        # state == "in-sync"
        if ff_pulled:
            return "synced", note("ff-pulled")
        return "clean", note("in-sync")
    except RuntimeError as e:
        return "error", str(e)


# --------------------------------------------------------------------------
# credential helper (git calls back into this script; token never in URL)
# --------------------------------------------------------------------------

def cmd_git_credential(argv):
    op = argv[0] if argv else "get"
    if op != "get":
        return 0  # ignore store/erase
    attrs = {}
    for line in sys.stdin:
        line = line.strip()
        if not line:
            break
        k, _, v = line.partition("=")
        attrs[k] = v
    cfg = load_config()
    gitea = cfg.get("gitea_base", "")
    want = urllib.parse.urlparse(gitea)
    if attrs.get("host") == want.netloc:
        print(f"username={cfg.get('login', '')}")
        print(f"password={cfg.get('token', '')}")
    return 0


def credential_helper_value():
    """Shell command git runs for credentials. Both paths are shlex-quoted:
    a Python or script path containing spaces (or shell metacharacters)
    must neither break the helper nor inject into the shell."""
    return "!{} {} git-credential".format(
        shlex.quote(sys.executable),
        shlex.quote(os.path.abspath(__file__)))


def install_credential_helper(folder):
    """Make OUR helper the only one this repo consults.

    credential.helper is a LIST that accumulates across system, global and
    repo config, and git asks every helper in order. On a stock mac there are
    already two (`osxkeychain` from Xcode's gitconfig, `store` from many
    people's ~/.gitconfig), and they lose both ways:

      * reading -- a stale entry for the forge host answers first, so pushes
        fail with "Failed to authenticate user" long after the token was
        rotated, and nothing in this tool's config explains why;
      * writing -- git calls `approve` on every helper after a successful
        auth, so `store` copies the forge token into ~/.git-credentials in
        PLAINTEXT. Keeping the token in a 0600 file and out of remote URLs
        is pointless if git hands it to a plaintext store on first use.

    An empty value resets the inherited list, so replace-all-then-add leaves
    exactly one helper: this script.
    """
    git(folder, "config", "--replace-all", "credential.helper", "")
    git(folder, "config", "--add", "credential.helper",
        credential_helper_value())


# --------------------------------------------------------------------------
# commands
# --------------------------------------------------------------------------

def device_flow():
    """Zitadel OAuth 2.0 device authorization grant. Returns access token."""
    status, resp = http_json(
        "POST", f"{ZITADEL_BASE}/oauth/v2/device_authorization",
        form={"client_id": DEVICE_CLIENT_ID, "scope": DEVICE_SCOPE})
    if status != 200:
        raise SystemExit(f"device authorization failed (HTTP {status}): {resp}")
    print(f"\nTo link this device, open:\n\n  {resp.get('verification_uri_complete') or resp.get('verification_uri')}\n")
    print(f"and enter code: {resp['user_code']}\n")
    interval = int(resp.get("interval", 5))
    deadline = time.time() + int(resp.get("expires_in", 300))
    while time.time() < deadline:
        time.sleep(interval)
        status, tok = http_json(
            "POST", f"{ZITADEL_BASE}/oauth/v2/token",
            form={"grant_type": "urn:ietf:params:oauth:grant-type:device_code",
                  "device_code": resp["device_code"],
                  "client_id": DEVICE_CLIENT_ID})
        if status == 200:
            return tok["access_token"]
        err = tok.get("error", "")
        if err == "authorization_pending":
            continue
        if err == "slow_down":
            interval += 5
            continue
        raise SystemExit(f"device flow failed: {tok}")
    raise SystemExit("device flow timed out (code expired)")


def cmd_link(args):
    zitadel_token = args.token or device_flow()
    cfg = load_config()
    dev = device_id(cfg)
    status, resp = http_json(
        "POST", f"{args.server.rstrip('/')}/v1/link",
        body={"zitadel_access_token": zitadel_token,
              "device_name": args.device,
              # Sent so the service can keep a device registry (and so a
              # device can later be listed or revoked by something other
              # than a hostname). A service that does not know the field
              # ignores it, so the client can lead.
              "device_id": dev})
    if status != 200:
        raise SystemExit(f"link failed (HTTP {status}): {resp}")
    if resp.get("device_id"):
        # The service is authoritative for the id it filed the device under
        # (an older client that sent none gets one back).
        cfg["device_id"] = resp["device_id"]
    cfg.update({"server": args.server.rstrip("/"),
                "gitea_base": resp["gitea_base"],
                "login": resp["login"],
                "token": resp["token"],
                "token_name": resp["token_name"]})
    cfg.setdefault("folders", {})
    save_config(cfg)
    log(f"linked as {resp['login']} on {resp['gitea_base']} "
        f"(token {resp['token_name']}, device {dev[:8]}); "
        f"config: {CONFIG_PATH}")
    return 0


def require_linked(cfg):
    """Every forge-touching command fails the same way on an unlinked box."""
    if "token" not in cfg:
        raise SystemExit("not linked yet -- run: granthi-sync link")
    return cfg


def forge_get(cfg, path, params=None):
    """Authenticated GET against the forge the link handed us. The client
    already holds a scoped user token, so read paths need no granthi-link
    round-trip."""
    url = f"{cfg['gitea_base'].rstrip('/')}{path}"
    if params:
        url = f"{url}?{urllib.parse.urlencode(params)}"
    return http_json("GET", url,
                     headers={"Authorization": f"token {cfg['token']}"})


def list_repos(cfg):
    """Every repo the linked token can see, following pagination.

    Returns (repos, truncated). `truncated` is True when FORGE_MAX_PAGES was
    hit -- a bounded page must never be presented as 'that is all of them'.
    """
    def fetch(page):
        status, resp = forge_get(cfg, "/api/v1/user/repos",
                                 {"page": page, "limit": FORGE_PAGE_LIMIT})
        if status != 200:
            raise SystemExit(f"listing repos failed (HTTP {status}): {resp}")
        return resp if isinstance(resp, list) else resp.get("data", [])

    repos, page = [], 1
    while page <= FORGE_MAX_PAGES:
        batch = fetch(page)
        repos.extend(batch)
        if len(batch) < FORGE_PAGE_LIMIT:
            return repos, False
        page += 1
    # Every page up to the cap was full, which does not by itself mean more
    # exist: a total that is an exact multiple of the page size ends on a
    # full page. One sentinel fetch tells "complete" from "truncated".
    return repos, bool(fetch(FORGE_MAX_PAGES + 1))


def match_repo(repo, pattern):
    """Case-insensitive: glob if the pattern has glob syntax, else substring.

    Matched against full_name AND the bare name, so 'notes' finds
    'alice/notes' without anyone having to think about owners.
    """
    if not pattern:
        return True
    full = (repo.get("full_name") or "").lower()
    name = (repo.get("name") or full.rsplit("/", 1)[-1]).lower()
    pat = pattern.lower()
    if any(c in pat for c in "*?["):
        return fnmatch.fnmatch(full, pat) or fnmatch.fnmatch(name, pat)
    return pat in full or pat in name


def cmd_list(args):
    cfg = require_linked(load_config())
    repos, truncated = list_repos(cfg)
    pattern = getattr(args, "pattern", None)
    if pattern:
        # Filter for display only. The set came from the forge under this
        # account's own token, so it is already the permitted set -- this
        # narrows what is shown, it does not widen what is reachable.
        repos = [r for r in repos if match_repo(r, pattern)]
        if not repos:
            print(f"no repos matching {pattern!r}"
                  + (" (and the listing was truncated)" if truncated else ""))
            return 0
    if not repos:
        print("no repos on the forge for this account")
        return 0
    # Which of them are already on this machine, so the table answers
    # "what can I pull down?" and not just "what exists?".
    # Keyed on full_name, not name: an account that can see both alice/cloud
    # and bob/cloud would otherwise show both as local when only one is.
    # Folders written before full_name was recorded fall back to <login>/<name>.
    local_by_full = {}
    for folder, m in cfg.get("folders", {}).items():
        full = m.get("full_name") or f"{cfg.get('login')}/{m.get('name')}"
        local_by_full[full] = folder
    rows = [("REPO", "VIS", "UPDATED", "LOCAL FOLDER")]
    for r in sorted(repos, key=lambda r: r.get("full_name") or ""):
        rows.append((r.get("full_name") or "?",
                     "private" if r.get("private") else "public",
                     (r.get("updated_at") or "")[:10],
                     local_by_full.get(r.get("full_name"), "-")))
    widths = [max(len(row[i]) for row in rows) for i in range(4)]
    for row in rows:
        print("  ".join(c.ljust(w) for c, w in zip(row, widths)))
    if truncated:
        print(f"\n... more repos exist: stopped after {FORGE_MAX_PAGES} pages "
              f"of {FORGE_PAGE_LIMIT}. This list is NOT complete.")
    return 0


_SEGMENT_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*$")


def parse_repo_arg(repo, login):
    """'<name>' or '<owner>/<name>' -> full_name. Rejects anything else.

    The result is concatenated into a URL and a filesystem path, so a segment
    carrying '?', '#', '..', an encoded slash, or an extra path component
    could redirect the clone or the remote that gets persisted. Validate
    rather than quote: the forge's own naming rules are this narrow anyway.
    """
    parts = repo.split("/") if "/" in repo else [login, repo]
    if len(parts) != 2 or not all(_SEGMENT_RE.match(p) and p not in (".", "..")
                                  for p in parts):
        raise SystemExit(
            f"invalid repo {repo!r}: expected <name> or <owner>/<name> using "
            f"letters, digits, '.', '_' or '-'")
    return "/".join(parts)


def read_marker(folder):
    """Intent `add` left in the repo, if any: {'mode': ...}.

    Carrying the mode in the repo is what makes a plain synced folder behave
    the same on the second machine. Without it, `get` would have to guess,
    and guessing 'mirror' on someone's real project is the destructive
    direction.
    """
    try:
        with open(os.path.join(folder, MARKER_FILE)) as f:
            data = json.load(f)
        return data if isinstance(data, dict) else {}
    except (OSError, ValueError):
        return {}


def clone_one(cfg, full_name, dest, mode=None):
    """Clone one repo and register it. Returns the folder meta written."""
    name = full_name.rsplit("/", 1)[-1]
    if os.path.exists(dest) and os.listdir(dest):
        raise SystemExit(f"refusing to clone into a non-empty path: {dest}")
    clone_url = f"{cfg['gitea_base'].rstrip('/')}/{full_name}.git"
    # -c supplies the helper *during* the clone -- install_credential_helper
    # cannot run first because the repo does not exist yet -- and git also
    # persists it into the new repo's config. --origin names the remote
    # 'granthi' up front so `watch` picks the folder up without a rename.
    proc = subprocess.run(
        # The empty -c resets the inherited helper list (see
        # install_credential_helper) so a stale keychain/store entry cannot
        # answer for the forge host during the clone, and the token cannot
        # leak into a plaintext store afterwards.
        ["git", "clone",
         "-c", "credential.helper=",
         "-c", f"credential.helper={credential_helper_value()}",
         "--origin", "granthi", clone_url, dest],
        capture_output=True, text=True)
    if proc.returncode != 0:
        raise SystemExit(f"clone failed: {proc.stderr.strip()}")
    install_credential_helper(dest)  # idempotent; guarantees persistence
    # symbolic-ref, not rev-parse: an empty repo has an unborn HEAD.
    rc, branch = git(dest, "symbolic-ref", "--short", "HEAD", check=False)
    if rc != 0 or not branch:
        branch = "main"
    # Default to `snapshot`: a cloned repo carries history that is the
    # user's, and autocommitting onto it is the one mistake that cannot be
    # undone quietly. `add` marks the folders that genuinely want mirroring.
    chosen = mode or read_marker(dest).get("mode") or "snapshot"
    meta = {
        "name": name, "full_name": full_name, "branch": branch,
        "mode": chosen,
        "last_sync": datetime.now(timezone.utc).isoformat(timespec="seconds"),
        "diverged": False}
    cfg.setdefault("folders", {})[dest] = meta
    save_config(cfg)
    log(f"cloned {full_name} -> {dest} (branch {branch}, mode {chosen}); "
        f"`granthi-sync watch` will keep it synced")
    return meta


def cmd_get(args):
    cfg = require_linked(load_config())
    device_id(cfg)
    if getattr(args, "all", False):
        return _get_all(cfg, args)
    if not args.repo:
        raise SystemExit("give a repo name, or --all")
    full_name = parse_repo_arg(args.repo, cfg["login"])
    name = full_name.rsplit("/", 1)[-1]
    clone_one(cfg, full_name, os.path.abspath(args.into or name), args.mode)
    return 0


def _get_all(cfg, args):
    """Pull every repo this account can see.

    'Only the repos they are granted' needs no client-side permission logic:
    /api/v1/user/repos is evaluated by the forge against this account's own
    scoped token, so the list IS the grant. Anything else would be a second
    opinion about someone else's authorisation.
    """
    repos, truncated = list_repos(cfg)
    base = os.path.abspath(args.into or ".")
    folders = cfg.get("folders", {})
    # full_name already living at a path, so "already here" can be told apart
    # from "a different repo wants that same path".
    owner_of = {os.path.abspath(p): m.get("full_name")
                for p, m in folders.items()}
    cloned = skipped = failed = 0
    for repo in sorted(repos, key=lambda r: r.get("full_name") or ""):
        full_name = repo.get("full_name")
        if not full_name:
            continue
        owner, _, name = full_name.rpartition("/")
        dest = os.path.join(base, name)
        if owner_of.get(dest) == full_name:
            log(f"skip {full_name}: already present at {dest}")
            skipped += 1
            continue
        if dest in owner_of or (os.path.exists(dest) and os.listdir(dest)):
            # alice/notes and bob/notes both want <base>/notes. Reporting the
            # second as "already here" would be a silent collision -- the user
            # would believe they had pulled both.
            taken_by = owner_of.get(dest)
            qualified = os.path.join(base, f"{owner}-{name}" if owner else name)
            if owner_of.get(qualified) == full_name or (
                    os.path.exists(qualified) and os.listdir(qualified)):
                log(f"skip {full_name}: already present at {qualified}")
                skipped += 1
                continue
            log(f"name clash on {dest}"
                + (f" (held by {taken_by})" if taken_by else "")
                + f": cloning {full_name} to {qualified} instead")
            dest = qualified
        try:
            clone_one(cfg, full_name, dest, args.mode)
            owner_of[os.path.abspath(dest)] = full_name
            cloned += 1
        except (SystemExit, RuntimeError) as e:
            # One bad repo must not abandon the rest. clone_one raises
            # SystemExit for a refused destination and RuntimeError from any
            # failing git call -- both are one repo's problem, not the run's.
            log(f"FAILED {full_name}: {e}")
            failed += 1
    log(f"get --all: {cloned} cloned, {skipped} already here, {failed} failed")
    if truncated:
        log(f"WARNING: the forge listing stopped after {FORGE_MAX_PAGES} "
            f"pages of {FORGE_PAGE_LIMIT} -- this was NOT every repo you "
            f"have access to.")
    return 1 if failed else 0


def measure_folder(folder, max_files=ADD_MAX_FILES):
    """(files, bytes) that would actually sync, stopping past max_files.

    Ignore rules are honoured, because the guard's own advice is "add a
    .gitignore for what should not sync" -- and advice that changes nothing
    is worse than no advice. git is asked with a THROWAWAY git dir outside
    the folder, so a refused `add` leaves no .git behind in a directory the
    user never agreed to turn into a repo.

    Bounded on purpose: the caller only needs to know whether the folder is
    surprisingly big, and walking a 2 TB drive to answer that would be the
    same mistake as syncing it.
    """
    listing = _git_would_sync(folder)
    if listing is None:
        return _walk_folder(folder, max_files)
    files = total = 0
    for rel in listing:
        path = os.path.join(folder, rel)
        try:
            if os.path.islink(path):
                continue
            total += os.path.getsize(path)
        except OSError:
            continue
        files += 1
        if files > max_files:
            break
    return files, total


def _git_would_sync(folder):
    """Paths git would take from `folder`, honouring .gitignore. None if git
    could not answer, so the caller falls back to a plain walk."""
    scratch = tempfile.mkdtemp(prefix="granthi-measure-")
    try:
        git_dir = os.path.join(scratch, "git")
        init = subprocess.run(["git", "init", "-q", "--bare", git_dir],
                              capture_output=True, text=True)
        if init.returncode != 0:
            return None
        proc = subprocess.run(
            ["git", f"--git-dir={git_dir}", f"--work-tree={folder}",
             "status", "--porcelain", "--untracked-files=all",
             "--ignored=no", "-z"],
            capture_output=True, text=True)
        if proc.returncode != 0:
            return None
        out = []
        for entry in proc.stdout.split("\0"):
            if len(entry) > 3:
                out.append(entry[3:])
        return out
    except OSError:
        return None
    finally:
        subprocess.run(["rm", "-rf", scratch], capture_output=True)


def _walk_folder(folder, max_files):
    files = total = 0
    for root, dirs, names in os.walk(folder):
        dirs[:] = [d for d in dirs if d != ".git"]
        for n in names:
            path = os.path.join(root, n)
            try:
                if os.path.islink(path):
                    continue
                total += os.path.getsize(path)
            except OSError:
                continue
            files += 1
            if files > max_files:
                return files, total
    return files, total


def seed_gitignore(folder):
    """Write a starter .gitignore when the folder has none. Returns bool.

    `add -A` takes everything it is not told to leave, and the folders
    people sync are exactly the ones holding a stray .env or id_rsa. An
    existing .gitignore is never touched -- it is the user's.
    """
    path = os.path.join(folder, ".gitignore")
    if os.path.exists(path):
        return False
    with open(path, "w") as f:
        f.write(DEFAULT_GITIGNORE)
    return True


def detect_mode(folder, had_git):
    """`mirror` for a plain folder we turned into a repo; `snapshot` for one
    that already had a history worth protecting."""
    if not had_git:
        return "mirror"
    return "snapshot" if _head_sha(folder) else "mirror"


def write_marker(folder, mode):
    with open(os.path.join(folder, MARKER_FILE), "w") as f:
        json.dump({"mode": mode, "created_by": f"granthi-sync/{VERSION}"},
                  f, indent=2)
        f.write("\n")


def cmd_add(args):
    cfg = require_linked(load_config())
    folder = os.path.abspath(args.folder)
    if not os.path.isdir(folder):
        raise SystemExit(f"no such folder: {folder}")
    name = args.name or re.sub(r"[^a-zA-Z0-9._-]+", "-", os.path.basename(folder))
    had_git = os.path.isdir(os.path.join(folder, ".git"))
    if not args.force:
        files, size = measure_folder(folder)
        if files > ADD_MAX_FILES or size > ADD_MAX_BYTES:
            raise SystemExit(
                f"{folder} holds {files} files / {size // (1024 * 1024)} MB, "
                f"over the {ADD_MAX_FILES} file / "
                f"{ADD_MAX_BYTES // (1024 * 1024)} MB guard.\n"
                f"Add a .gitignore for what should not sync, point at a "
                f"narrower folder, or re-run with --force if this is really "
                f"what you want.")
    ensure_repo(folder)
    mode = args.mode or detect_mode(folder, had_git)
    if seed_gitignore(folder):
        log(f"seeded a starter .gitignore in {folder} -- review it; "
            f"anything listed there will NOT be synced")
    if mode == "mirror":
        # Only mirror folders carry the marker: it tells the next machine's
        # `get` that autocommitting here is wanted. A snapshot folder is
        # someone's real project and gets no file dropped into it.
        write_marker(folder, mode)
    install_credential_helper(folder)
    status, resp = http_json(
        "POST", f"{cfg['server']}/v1/repos",
        body={"token": cfg["token"], "name": name, "private": args.private})
    if status == 200:
        clone_url = resp["clone_url"]
    elif status == 409:
        clone_url = f"{cfg['gitea_base']}/{cfg['login']}/{name}.git"
        log(f"cloud repo {name} already exists; reusing")
    else:
        raise SystemExit(f"repo create failed (HTTP {status}): {resp}")
    rc, _ = git(folder, "remote", "get-url", "granthi", check=False)
    if rc == 0:
        git(folder, "remote", "set-url", "granthi", clone_url)
    else:
        git(folder, "remote", "add", "granthi", clone_url)
    # An existing repo may sit on master, or on a feature branch. Pushing
    # its work to a hardcoded 'main' would publish the wrong thing under the
    # right name.
    rc, branch = git(folder, "symbolic-ref", "--short", "HEAD", check=False)
    branch = branch.strip() if rc == 0 and branch.strip() else "main"
    if mode == "mirror":
        autocommit(folder)
    git(folder, "push", "-u", "granthi", branch)
    dev = device_id(cfg)
    if mode == "snapshot":
        ref = push_snapshot(folder, dev)
        if ref:
            log(f"backed up uncommitted work to {ref}")
    cfg.setdefault("folders", {})[folder] = {
        "name": name, "full_name": f"{cfg['login']}/{name}", "branch": branch,
        "mode": mode,
        "last_sync": datetime.now(timezone.utc).isoformat(timespec="seconds"),
        "diverged": False}
    save_config(cfg)
    log(f"linked folder {folder} -> {clone_url} (mode {mode}, branch {branch})")
    if mode == "snapshot":
        log("mode snapshot: your commits stay yours -- granthi-sync will "
            "never commit for you here, and backs up uncommitted work to "
            f"{BACKUP_NS}/{dev[:8]}…/<timestamp>")
    return 0


_STOP = False


def _sigterm(signum, frame):
    global _STOP
    _STOP = True
    log(f"signal {signum} received; finishing current pass then exiting")


def watch_pass(now=None):
    cfg = load_config()
    dev = device_id(cfg)
    # Pruning talks to the remote for every folder, so it runs on a slow
    # clock of its own rather than on every 30s pass. The clock is PERSISTED:
    # `watch --once` under launchd or cron is a fresh process every time, and
    # an in-memory timer would make "at most hourly" mean "every run".
    now = now or time.time()
    due = now - float(cfg.get("last_prune") or 0) >= PRUNE_EVERY_SECONDS
    for folder, meta in sorted(cfg.get("folders", {}).items()):
        if not os.path.isdir(folder):
            log(f"SKIP {folder}: missing")
            continue
        mode = meta.get("mode", "mirror")
        outcome, detail = sync_folder(folder, branch=meta.get("branch", "main"),
                                      mode=mode, dev=dev)
        meta["diverged"] = outcome == "diverged"
        if outcome in ("pushed", "synced"):
            meta["last_sync"] = datetime.now(timezone.utc).isoformat(timespec="seconds")
        if outcome == "diverged":
            log(f"DIVERGED {folder}: {detail}")
        elif outcome == "error":
            log(f"ERROR {folder}: {detail}")
        else:
            log(f"{outcome} {folder}: {detail}")
        if due and mode == "snapshot" and outcome != "error":
            try:
                gone = prune_snapshots(folder, dev)
                if gone:
                    log(f"pruned {gone} old restore points in {folder}")
            except RuntimeError as e:
                # Retention failing must never stop the backup itself.
                log(f"prune failed in {folder} (backups unaffected): {e}")
    if due:
        cfg["last_prune"] = now
    save_config(cfg)


def cmd_watch(args):
    signal.signal(signal.SIGTERM, _sigterm)
    signal.signal(signal.SIGINT, _sigterm)
    log(f"granthi-sync watch starting (interval {args.interval}s)")
    while True:
        watch_pass()
        if args.once or _STOP:
            break
        deadline = time.time() + args.interval
        while time.time() < deadline and not _STOP:
            time.sleep(1)
        if _STOP:
            break
    log("watch stopped cleanly")
    return 0


def cmd_devices(args):
    """Every computer signed in to this account."""
    cfg = require_linked(load_config())
    status, resp = http_json("POST", f"{cfg['server']}/v1/devices",
                             body={"token": cfg["token"]})
    if status == 401:
        raise SystemExit("this device's access has been revoked -- "
                         "run: granthi-sync link")
    if status != 200:
        raise SystemExit(f"could not list devices (HTTP {status}): {resp}")
    devices = resp.get("devices") or []
    if not devices:
        print("no devices recorded for this account")
        return 0
    this = cfg.get("device_id")
    rows = [("DEVICE", "ID", "LINKED", "STATUS")]
    for d in devices:
        rows.append((d.get("name") or "?", (d.get("device_id") or "")[:12],
                     (d.get("linked_at") or "")[:19],
                     "REVOKED" if d.get("revoked_at") else
                     ("active (this computer)" if d.get("device_id") == this
                      else "active")))
    widths = [max(len(r[i]) for r in rows) for i in range(len(rows[0]))]
    for r in rows:
        print("  ".join(c.ljust(w) for c, w in zip(r, widths)))
    return 0


def cmd_logout(args):
    """Sign a computer out. Revocation happens at the FORGE -- the token is
    deleted, so that machine's next fetch or push fails at the server. It is
    not a flag a client could ignore, which is the only kind of logout worth
    having for a lost laptop."""
    cfg = require_linked(load_config())
    target = args.device or cfg.get("device_id")
    if not target:
        raise SystemExit("no device id recorded; pass --device (see: "
                         "granthi-sync devices)")
    is_self = target == cfg.get("device_id")
    status, resp = http_json("POST", f"{cfg['server']}/v1/devices/revoke",
                             body={"token": cfg["token"], "device_id": target})
    if status == 404:
        raise SystemExit(f"no device {target} on this account "
                         f"(see: granthi-sync devices)")
    if status != 200:
        raise SystemExit(f"revoke failed (HTTP {status}): {resp}")
    log(f"revoked {target} at the forge ({resp.get('revoked_at')})")
    if is_self:
        # Drop the local token too. The forge already refuses it, but leaving
        # a dead secret on disk is pointless risk -- and `status` should say
        # "not linked" rather than pretend.
        for key in ("token", "token_name"):
            cfg.pop(key, None)
        cfg["logged_out_at"] = datetime.now(timezone.utc).isoformat(
            timespec="seconds")
        save_config(cfg)
        log("local token deleted; syncing stops at the next pass. "
            "Linked folders are left on disk untouched.")
        log("run `granthi-sync link` to sign back in")
    return 0


def cmd_activity(args):
    """Security events for this account, newest first."""
    cfg = require_linked(load_config())
    status, resp = http_json("POST", f"{cfg['server']}/v1/audit",
                             body={"token": cfg["token"],
                                   "limit": args.limit})
    if status != 200:
        raise SystemExit(f"could not read activity (HTTP {status}): {resp}")
    events = resp.get("events") or []
    if not events:
        print("no recorded events for this account")
    for e in events:
        extra = " ".join(f"{k}={v}" for k, v in sorted(e.items())
                         if k not in ("ts", "event", "login"))
        print(f"{e.get('ts')}  {e.get('event'):<22} {extra}")
    if resp.get("note"):
        print(f"\nnote: {resp['note']}")
    return 0


def _folder_meta(cfg, folder):
    path = os.path.abspath(folder)
    meta = cfg.get("folders", {}).get(path)
    if not meta:
        raise SystemExit(f"{path} is not a linked folder "
                         f"(see: granthi-sync status)")
    return path, meta


def cmd_snapshots(args):
    """Restore points, newest first. Both modes answer the same question."""
    cfg = require_linked(load_config())
    folder, meta = _folder_meta(cfg, args.folder)
    this_dev = cfg.get("device_id")
    if meta.get("mode", "mirror") == "mirror":
        rc, out = git(folder, "log", "--format=%H\t%cI\t%s", "-n",
                      str(args.limit), check=False)
        if rc != 0 or not out:
            print("no restore points yet")
            return 0
        print(f"restore points for {folder} (mode mirror: every commit is one)")
        for line in out.splitlines():
            sha, ts, subject = (line.split("\t") + ["", ""])[:3]
            print(f"  {ts}  {sha[:12]}  {subject}")
        return 0
    # Every device's backups, not just this one's -- see list_snapshots.
    snaps = list_snapshots(folder)
    if not snaps:
        print("no restore points yet (nothing uncommitted has been backed up)")
        return 0
    print(f"restore points for {folder} (mode snapshot):")
    for snap in snaps[:args.limit]:
        here = " (this device)" if snap["device"] == this_dev else ""
        print(f"  {snap['ts']}  {snap['sha'][:12]}  "
              f"device {snap['device'][:8]}{here}")
    if len(snaps) > args.limit:
        print(f"  ... {len(snaps) - args.limit} older (use --limit)")
    return 0


def cmd_restore(args):
    """Materialise a restore point into a NEW directory.

    Never into the working tree. Someone restoring a backup is already
    having a bad day; overwriting the files they still have would be how a
    recovery tool becomes the second disaster.
    """
    cfg = require_linked(load_config())
    folder, meta = _folder_meta(cfg, args.folder)
    target = args.at
    sha = None
    if meta.get("mode", "mirror") == "snapshot":
        # Across ALL devices: restoring is exactly the moment the device that
        # took the backup may no longer exist.
        for snap in list_snapshots(folder):
            if snap["ts"] == target or snap["sha"].startswith(target):
                sha = snap["sha"]
                git(folder, "fetch", "granthi", f"{snap['ref']}")
                break
    else:
        # Mirror mode lists commits by their %cI timestamp, and that first
        # column is what a person copies. It is not a commit-ish, so match it
        # against the log before falling through to rev-parse -- otherwise
        # `restore --at <what snapshots just printed>` fails and sends the
        # user back to `snapshots`, which prints the same thing again.
        rc, out = git(folder, "log", "--format=%H\t%cI", check=False)
        if rc == 0:
            hits = [line.partition("\t")[0] for line in out.splitlines()
                    if line.partition("\t")[2] == target]
            if len(hits) > 1:
                # Two commits in the same second share a %cI. Silently taking
                # the newest would restore something the user did not pick.
                listed = "\n".join(f"  {h[:12]}" for h in hits)
                raise SystemExit(
                    f"{target} matches {len(hits)} restore points in "
                    f"{folder}. Re-run with one of these ids instead:\n"
                    f"{listed}")
            if hits:
                sha = hits[0]
    if sha is None:
        rc, resolved = git(folder, "rev-parse", "--verify", "--quiet",
                           f"{target}^{{commit}}", check=False)
        if rc != 0 or not resolved.strip():
            raise SystemExit(
                f"no restore point {target!r} for {folder} "
                f"(see: granthi-sync snapshots {folder})")
        sha = resolved.strip()
    # A %cI timestamp carries ':' and '+', which make an awkward folder name.
    suffix = re.sub(r"[^A-Za-z0-9]+", "-", target).strip("-") or sha[:12]
    dest = os.path.abspath(
        args.into or f"{folder.rstrip(os.sep)}-restore-{suffix}")
    if os.path.exists(dest) and os.listdir(dest):
        raise SystemExit(f"refusing to restore over a non-empty path: {dest}")
    os.makedirs(dest, exist_ok=True)
    proc = subprocess.run(["git", "-C", folder, "archive", "--format=tar", sha],
                          capture_output=True)
    if proc.returncode != 0:
        raise SystemExit(
            f"could not read {sha[:12]}: {proc.stderr.decode(errors='replace').strip()}")
    untar = subprocess.run(["tar", "-x", "-C", dest], input=proc.stdout,
                           capture_output=True)
    if untar.returncode != 0:
        raise SystemExit(
            f"could not write {dest}: {untar.stderr.decode(errors='replace').strip()}")
    log(f"restored {sha[:12]} -> {dest} (your working tree was not touched)")
    return 0


def cmd_status(args):
    cfg = load_config()
    if "login" in cfg:
        print(f"linked: {cfg['login']} @ {cfg.get('gitea_base')} "
              f"(token {cfg.get('token_name')})")
    else:
        print("not linked")
    folders = cfg.get("folders", {})
    if not folders:
        print("no folders linked")
        return 0
    rows = [("FOLDER", "REPO", "MODE", "LAST SYNC", "DIVERGED")]
    for folder, meta in sorted(folders.items()):
        rows.append((folder, meta.get("name", "?"),
                     meta.get("mode", "mirror"),
                     meta.get("last_sync", "never"),
                     "YES" if meta.get("diverged") else "no"))
    widths = [max(len(r[i]) for r in rows) for i in range(len(rows[0]))]
    for r in rows:
        print("  ".join(c.ljust(w) for c, w in zip(r, widths)))
    return 0


def main(argv=None):
    p = argparse.ArgumentParser(prog="granthi-sync",
                                description="Granthi folder-to-cloud sync client")
    p.add_argument("--version", action="version", version=VERSION)
    sub = p.add_subparsers(dest="cmd", required=True)

    sp = sub.add_parser("link", help="link this device to your granthi account")
    sp.add_argument("--server", default=DEFAULT_SERVER)
    sp.add_argument("--token", help="ready Zitadel access token (headless path)")
    sp.add_argument("--device", default=os.uname().nodename.split(".")[0])
    sp.set_defaults(fn=cmd_link)

    sp = sub.add_parser("list", help="list forge repos this account can see")
    sp.add_argument("pattern", nargs="?",
                    help="filter by name: substring, or a glob like 'work-*'")
    sp.set_defaults(fn=cmd_list)

    sp = sub.add_parser("get", help="clone a forge repo and keep it synced")
    sp.add_argument("repo", nargs="?", help="repo name, or owner/repo")
    sp.add_argument("--all", action="store_true",
                    help="clone every repo this account can see")
    sp.add_argument("--into",
                    help="target folder (default: ./<repo>; with --all, the "
                         "directory to clone into)")
    sp.add_argument("--mode", choices=("mirror", "snapshot"),
                    help="override the sync mode (default: snapshot, or "
                         "whatever the repo was marked with)")
    sp.set_defaults(fn=cmd_get)

    sp = sub.add_parser("add", help="link a folder and push it to the cloud")
    sp.add_argument("folder")
    sp.add_argument("--name")
    sp.add_argument("--mode", choices=("mirror", "snapshot"),
                    help="override the sync mode (default: mirror for a plain "
                         "folder, snapshot for an existing git repo)")
    sp.add_argument("--force", action="store_true",
                    help="skip the folder-size guard")
    grp = sp.add_mutually_exclusive_group()
    grp.add_argument("--private", dest="private", action="store_true", default=True)
    grp.add_argument("--public", dest="private", action="store_false")
    sp.set_defaults(fn=cmd_add)

    sp = sub.add_parser("snapshots", help="list restore points for a folder")
    sp.add_argument("folder")
    sp.add_argument("--limit", type=int, default=20)
    sp.set_defaults(fn=cmd_snapshots)

    sp = sub.add_parser("restore",
                        help="materialise a restore point into a NEW folder")
    sp.add_argument("folder")
    sp.add_argument("--at", required=True,
                    help="timestamp or sha from `granthi-sync snapshots`")
    sp.add_argument("--into", help="destination (default: <folder>-restore-<at>)")
    sp.set_defaults(fn=cmd_restore)

    sp = sub.add_parser("watch", help="sync loop over linked folders")
    sp.add_argument("--interval", type=int, default=30)
    sp.add_argument("--once", action="store_true", help="single pass then exit")
    sp.set_defaults(fn=cmd_watch)

    sp = sub.add_parser("devices",
                        help="list the computers signed in to this account")
    sp.set_defaults(fn=cmd_devices)

    sp = sub.add_parser("logout",
                        help="sign a computer out (revokes it at the forge)")
    sp.add_argument("--device",
                    help="device id to revoke (default: this computer)")
    sp.set_defaults(fn=cmd_logout)

    sp = sub.add_parser("activity",
                        help="security events for this account")
    sp.add_argument("--limit", type=int, default=50)
    sp.set_defaults(fn=cmd_activity)

    sp = sub.add_parser("status", help="show linked folders")
    sp.set_defaults(fn=cmd_status)

    argv = list(sys.argv[1:] if argv is None else argv)
    if argv and argv[0] == "git-credential":  # hidden helper protocol
        return cmd_git_credential(argv[1:])
    args = p.parse_args(argv)
    return args.fn(args)


if __name__ == "__main__":
    sys.exit(main())