fix(client): preserve index-only snapshot state

Hit live today. An operator moved an email onto a different forge account; the
next sign-in tried to create a user with that address, Gitea answered
422 "e-mail already in use", and the generic path turned it into a bare 502.
The person approved a device code and got a number that reads like an outage,
when the real answer was "that address already belongs to somebody here".

create_user now distinguishes that case, and the binding rules translate it to
409 with a message naming the address, the login it tried to create, and the
two ways out: have an operator bind your identity to the existing account, or
use a different address.

211 tests (was 209): the stub now enforces unique emails like real Gitea, one
test asserts the 409 names the address and login and never says 502, and one
asserts an ordinary create is unaffected.

SETUP.md

@@ -0,0 +1,163 @@
+# Setting up Granthi on a new computer
+
+Everything here is one command at a time. You need `git` and `python3`
+(3.9 or newer). There is nothing to install — no package, no build.
+
+## Read this first: what you will see when you sign in
+
+Signing in does **not** hand you an account you already had somewhere else.
+Granthi looks up your shre-id identity, and:
+
+* if it has seen you before, you get the same forge account as last time;
+* if it has not, and the login it would use is free, it makes you a **new,
+  empty account**.
+
+So the first time you sign in on a new computer, `granthi-sync list` may show
+**nothing**. That is not a failure. It means you are a new account and nobody
+has given you access to anything yet.
+
+Granthi refuses to hand you an existing account just because the name lines
+up. It will only join you to one if that account's email is the same as your
+verified sign-in email. That rule is what stops somebody else claiming your
+files, so it is worth the small surprise on day one.
+
+**Two ways to end up with your files, and it is worth choosing on purpose:**
+
+| | what happens | when to pick it |
+|---|---|---|
+| **Fresh account** (recommended) | You sign in, get a new empty account, and somebody shares the repos you need with you | Your everyday laptop. The token stored on the machine can only do what that account can do. |
+| **Join an existing account** | An operator sets that account's email to your verified sign-in email first; then signing in joins you to it and you see everything it owns straight away | You want one identity everywhere and accept that the laptop then carries whatever that account can do |
+
+The fresh account is recommended for a plain reason: the token `link` saves on
+the machine acts as that account. If you join an admin account, every computer
+you sync from is carrying admin.
+
+## 1. Get the program
+
+    git clone https://granthi.shre.ai/nirpa/granthi-sync.git
+    cd granthi-sync
+
+No account is needed for this step.
+
+## 2. Sign in
+
+    ./bin/granthi-sync link
+
+It prints a web address and a short code. Open the address, type the code, and
+approve — you have five minutes. When it finishes, your account exists and a
+key is saved at `~/.granthi-sync/config.json`, readable only by you. You never
+see or type a forge password.
+
+If the code runs out, **nothing was created** — no account, no key, no
+half-finished state. Run `link` again.
+
+You do not need to be on any private network. The client talks to
+`https://granthi-link.shre.ai`.
+
+## 3. See what you have
+
+    ./bin/granthi-sync list
+    ./bin/granthi-sync list work-          # only names containing "work-"
+
+This list comes from the forge, so it is exactly what you are allowed to see —
+not a guess made on this computer.
+
+If it is empty, see the note at the top: you are a new account. Ask whoever
+owns the repos to run `granthi-sync share <your-login> --repo <name>`, or to
+add you to the right organisation.
+
+## 4. Bring folders down
+
+    ./bin/granthi-sync get notes                  # by name
+    ./bin/granthi-sync get Nirlabinc/notes        # when two teams both have a "notes"
+    ./bin/granthi-sync get --all --into ~/work    # everything you are allowed to see
+
+If a plain name matches two repos, Granthi stops and shows you both rather
+than guessing which one you meant.
+
+## 5. Or send a folder up
+
+    ./bin/granthi-sync add ~/Documents/notes
+
+It becomes a private repo. Two things happen for your protection:
+
+* a starter `.gitignore` is written if the folder has none, covering `.env`,
+  `*.key`, `*.pem`, `id_rsa` and similar — those will **not** be synced;
+* a folder holding more than 20,000 files or 512 MB is refused unless you add
+  `--force`, so you cannot upload your whole disk by accident.
+
+## 6. Keep it syncing
+
+Run it in a terminal:
+
+    ./bin/granthi-sync watch
+
+Or leave it running in the background on a Mac:
+
+    cp client/launchd/ai.granthi.sync.plist ~/Library/LaunchAgents/
+
+Open that copied file and replace `/PATH/TO/granthi-sync` with the real path
+to this folder. Then:
+
+    launchctl bootstrap gui/$UID ~/Library/LaunchAgents/ai.granthi.sync.plist
+
+Its log is `/tmp/granthi-sync.log`.
+
+### What syncing does to your folder
+
+It depends on what the folder was:
+
+* **A folder that was already a git project** — Granthi **never commits for
+  you**. Your history stays yours. Work you have not committed is copied to
+  the cloud with a timestamp so it is not only on this laptop.
+* **A plain folder** — Granthi saves changes for you as it goes, and each save
+  is a point you can go back to.
+
+Either way, if your copy and the cloud copy have both moved on, Granthi
+**stops and tells you**. It will never merge or overwrite your work to make
+the conflict go away.
+
+## 7. Going back to an earlier version
+
+    ./bin/granthi-sync snapshots ~/Documents/notes
+    ./bin/granthi-sync restore  ~/Documents/notes --at 20260823T142530Z
+
+Restore writes to a **new folder**. It never overwrites what you have now —
+if you are restoring, you have enough problems already.
+
+You can restore from a computer that never had the folder, including one
+replacing a laptop that is gone.
+
+## 8. Sharing
+
+    ./bin/granthi-sync share bob --repo notes                 # they have an account
+    ./bin/granthi-sync share bob --repo notes --revoke
+    ./bin/granthi-sync shared --repo notes                    # who can see it
+    ./bin/granthi-sync invite carol@example.com --repo notes  # they do not yet
+
+An invite creates nothing until it is accepted. If the person never signs in,
+nothing was ever made for them.
+
+## 9. Keeping track of your computers
+
+    ./bin/granthi-sync devices     # every computer signed in to your account
+    ./bin/granthi-sync logout      # sign this computer out
+    ./bin/granthi-sync logout --device <id>   # sign out a lost one
+    ./bin/granthi-sync activity    # sign-ins and sign-outs, with times
+
+Signing a computer out takes effect **at the server**, immediately. The key on
+that machine stops working on its very next attempt — it is not a setting that
+machine could ignore. That is what makes it useful for a laptop you have lost.
+
+Your folders are left alone when you sign out; only the key is removed.
+
+## If something looks wrong
+
+* **`list` is empty** — you are a new account, nobody has shared anything yet.
+* **"Repository not found"** — you may not have access to that repo; run
+  `list` to see what you do have.
+* **"this device's access has been revoked"** — someone signed this computer
+  out. Run `link` again.
+* **A folder shows DIVERGED** — your copy and the cloud copy both changed.
+  Granthi is waiting for you on purpose. Nothing is lost; sort it out in git
+  or on the forge web page.

client/granthi_sync_client.py

@@ -104,6 +104,10 @@ FORGE_MAX_PAGES = 40  # 2000 repos; a guard against an unbounded paging loop
 # Under refs/heads a machine taking a snapshot every 30s would bury the
 # user's real branches.
 BACKUP_NS = "refs/granthi-backup"
+_SNAPSHOT_IDENT = {"GIT_AUTHOR_NAME": "granthi-sync",
+                   "GIT_AUTHOR_EMAIL": "[email protected]",
+                   "GIT_COMMITTER_NAME": "granthi-sync",
+                   "GIT_COMMITTER_EMAIL": "[email protected]"}
 SNAPSHOT_TS_FMT = "%Y%m%dT%H%M%SZ"
 
 # Retention. Unbounded snapshots are a disk leak with no way to use them, so
@@ -292,24 +296,44 @@ def build_snapshot(folder):
         tree = tree.strip()
         if not tree:
             return None
+        head_tree = ""
         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
+            head_tree = head_tree.strip()
         ts = datetime.now(timezone.utc).isoformat(timespec="seconds")
         args = ["commit-tree", tree, "-m", f"granthi snapshot: {ts}"]
         if head:
             args += ["-p", head]
+        # The worktree tree alone loses STAGED-ONLY work. Stage a hunk, edit the
+        # file further, lose the laptop, and the snapshot holds only the later
+        # worktree version -- the carefully staged one is gone. git itself keeps
+        # index and worktree as separate states, so the backup must too.
+        # The user's real index is read WITHOUT touching it, and when it differs
+        # from both HEAD and the worktree it rides along as a second parent, so
+        # it is reachable from the snapshot. (codex review, P2.)
+        rc_idx, index_tree = git(folder, "write-tree", check=False)
+        index_tree = index_tree.strip()
+        staged_state_added = False
+        if rc_idx == 0 and index_tree and index_tree != tree:
+            if index_tree != head_tree:
+                icommit_args = ["commit-tree", index_tree, "-m",
+                                f"granthi snapshot (staged): {ts}"]
+                if head:
+                    icommit_args += ["-p", head]
+                rc_ic, icommit = git(folder, *icommit_args, check=False,
+                                     env=_SNAPSHOT_IDENT)
+                if rc_ic == 0 and icommit.strip():
+                    args += ["-p", icommit.strip()]
+                    staged_state_added = True
+        if head_tree == tree and not staged_state_added:
+            # The worktree may match HEAD while the real index still holds a
+            # staged-only state. Only skip after both states were inspected.
+            return None
         # 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]"})
+        _, commit = git(folder, *args, env=_SNAPSHOT_IDENT)
         return commit.strip(), tree
     finally:
         if os.path.exists(tmp_index):
@@ -330,14 +354,29 @@ def push_snapshot(folder, dev, remote="granthi"):
     if not built:
         return None
     commit, tree = built
-    if tree == _last_snapshot_tree(folder, dev):
-        return None  # working tree unchanged since the last backup
+    state = _snapshot_state(folder, tree)
+    if state == _last_snapshot_tree(folder, dev):
+        return None  # worktree and staged state unchanged since the last backup
     ref, _ = snapshot_ref(dev)
     git(folder, "push", remote, f"{commit}:{ref}")
-    _remember_snapshot_tree(folder, dev, tree)
+    _remember_snapshot_tree(folder, dev, state)
     return ref
 
 
+def _snapshot_state(folder, worktree_tree):
+    """A stable deduplication key for both worktree and staged-only content."""
+    head = _head_sha(folder)
+    head_tree = ""
+    if head:
+        head_tree = git(folder, "rev-parse", f"{head}^{{tree}}")[1].strip()
+    rc_idx, index_tree = git(folder, "write-tree", check=False)
+    index_tree = index_tree.strip()
+    if (rc_idx == 0 and index_tree
+            and index_tree != worktree_tree and index_tree != head_tree):
+        return f"{worktree_tree}:{index_tree}"
+    return worktree_tree
+
+
 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}")
@@ -599,8 +638,13 @@ def device_flow():
         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")
+    # flush=True is not cosmetic. Python buffers stdout when it is not a
+    # terminal, so `granthi-sync link | tee setup.log`, a wrapper script, or
+    # anything capturing output shows NOTHING while the code silently expires
+    # five minutes later. Hit for real on 2026-08-23 driving a first sign-in.
+    print(f"\nTo link this device, open:\n\n  {resp.get('verification_uri_complete') or resp.get('verification_uri')}\n",
+          flush=True)
+    print(f"and enter code: {resp['user_code']}\n", flush=True)
     interval = int(resp.get("interval", 5))
     deadline = time.time() + int(resp.get("expires_in", 300))
     while time.time() < deadline:
@@ -833,7 +877,7 @@ def clone_one(cfg, full_name, dest, mode=None):
     return meta
 
 
-def resolve_granted(cfg, want, repos=None):
+def resolve_granted(cfg, want, repos=None, truncated=False):
     """Turn what the user typed into the repo the forge actually grants them.
 
     `get notes` used to mean `<your-login>/notes` and nothing else. On a real
@@ -852,7 +896,7 @@ def resolve_granted(cfg, want, repos=None):
         return parse_repo_arg(want, cfg["login"])
     if repos is None:
         try:
-            repos, _truncated = list_repos(cfg)
+            repos, truncated = list_repos(cfg)
         except (SystemExit, ValueError, OSError) as e:
             # Best effort. If the listing is unreachable, a name the user
             # typed in full must still clone, and a bare name should degrade
@@ -871,8 +915,18 @@ def resolve_granted(cfg, want, repos=None):
             f"{want!r} is ambiguous — {len(matches)} repos you can see have "
             f"that name:\n{listed}\nRe-run with the owner, e.g. "
             f"granthi-sync get {sorted(matches)[0]}")
-    # Nothing granted by that name. Fall back to the caller's own namespace so
-    # the message names a concrete repo instead of a guess.
+    # Nothing granted by that name. If the listing was TRUNCATED the answer is
+    # unknown rather than absent -- falling back to <login>/<name> could clone a
+    # different repo that happens to exist under your own account. Refuse and
+    # ask for the owner. (codex review, P3.)
+    if truncated:
+        raise SystemExit(
+            f"could not confirm {want!r}: your repo list was truncated at "
+            f"{FORGE_MAX_PAGES * FORGE_PAGE_LIMIT} repos, so a match may exist "
+            f"beyond it. Re-run with the owner, e.g. "
+            f"granthi-sync get <owner>/{want}")
+    # Otherwise the listing was complete and simply has no such repo; name a
+    # concrete one so the clone error is precise.
     return parse_repo_arg(want, cfg["login"])
 
 
@@ -1336,13 +1390,14 @@ def cmd_bootstrap(args):
     into = os.path.abspath(args.into or ".")
     print(f"workspace: {name}\n")
 
-    granted_repos, _trunc = list_repos(cfg)
+    granted_repos, granted_truncated = list_repos(cfg)
     granted = {r.get("full_name") for r in granted_repos}
     pulled = skipped = denied = 0
     for repo in ws["repos"]:
         want = repo["name"]
         try:
-            full = resolve_granted(cfg, want, repos=granted_repos)
+            full = resolve_granted(cfg, want, repos=granted_repos,
+                                   truncated=granted_truncated)
         except SystemExit as e:
             log(f"AMBIGUOUS {want}: {e}")
             denied += 1

server/granthi_link.py

@@ -80,6 +80,9 @@ TEST_MODE_ENV = "GRANTHI_LINK_ALLOW_TEST_MODE"
 
 # Sentinel: create_user hit a 409 (someone else created the login first).
 USER_CREATE_CONFLICT = object()
+# Sentinel: the address already belongs to another forge account, which is a
+# 409 the caller can act on -- not a 502 that reads like the service is down.
+EMAIL_IN_USE = object()
 
 LOGIN_SAFE = re.compile(r"[^a-zA-Z0-9._-]+")
 
@@ -292,6 +295,14 @@ def check_config_perms(path, euid=None):
 # Identity map: zitadel sub -> gitea login (JSON, 0600, atomic writes)
 # --------------------------------------------------------------------------
 
+def _email_in_use_message(login, userinfo):
+    email = userinfo.get("email") or "your address"
+    return (f"cannot create the account '{login}': {email} already belongs to "
+            f"a different account on this forge. If that other account is "
+            f"yours, an operator must bind your identity to it; if it is not, "
+            f"use a different address.")
+
+
 class IdentityStore:
     """Persistent map of Zitadel `sub` -> Gitea login binding records.
 
@@ -404,6 +415,23 @@ class IdentityStore:
             self._write(data)
         return pending
 
+    def consume_invite(self, email, repo):
+        """Drop ONE applied grant, leaving any that failed still pending.
+
+        The whole-list `take_invites` is what made a transient forge error
+        permanent; this removes only what actually landed.
+        """
+        data = self._load()
+        book = data.get("invites") or {}
+        key = (email or "").strip().lower()
+        entry = [e for e in book.get(key, []) if e.get("repo") != repo]
+        if entry:
+            book[key] = entry
+        else:
+            book.pop(key, None)
+        data["invites"] = book
+        self._write(data)
+
     def peek_invites(self, email):
         book = self._load().get("invites") or {}
         return list(book.get((email or "").strip().lower(), []))
@@ -660,6 +688,14 @@ class LinkService:
             headers=self._admin_hdr(), body=body)
         if status == 409:
             return USER_CREATE_CONFLICT
+        if status == 422 and "e-mail already in use" in str(resp).lower():
+            # The address belongs to a DIFFERENT forge account. Reported as its
+            # own case because the generic path turns it into a bare 502, and
+            # "502" sends the person looking for an outage when the real answer
+            # is "that address is already somebody's account here". Hit live on
+            # 2026-08-23: an operator moved an email onto another account and
+            # the next sign-in failed with nothing but the number.
+            return EMAIL_IN_USE
         if status != 201:
             return f"gitea admin user create failed (HTTP {status}): {resp}"
         return None
@@ -716,6 +752,8 @@ class LinkService:
                                  "was not created by this service; refusing "
                                  "to re-create"}, None
                 err = self.create_user(login, userinfo)
+                if err is EMAIL_IN_USE:
+                    return 409, {"error": _email_in_use_message(login, userinfo)}, None
                 if err and err is not USER_CREATE_CONFLICT:
                     return 502, {"error": err}, None
                 LOG.info("re-created service-managed gitea user %s", login)
@@ -741,6 +779,8 @@ class LinkService:
                     return 409, {"error": "login exists and is not linked "
                                           "to this identity"}, None
                 LOG.info("user %s created concurrently; continuing", login)
+            elif err is EMAIL_IN_USE:
+                return 409, {"error": _email_in_use_message(login, userinfo)}, None
             elif err:
                 return 502, {"error": err}, None
             else:
@@ -1038,8 +1078,13 @@ class LinkService:
             # the only thing tying the promise to this person.
             return []
         with self.state.lock:
-            pending = self.state.take_invites(email)
+            pending = self.state.peek_invites(email)
         applied = []
+        # PEEK, not take. Consuming the invite first means a transient 502 from
+        # the forge destroys it: the person links successfully, gets no access,
+        # and re-linking never retries because the promise is gone. Only the
+        # grants that actually landed are removed, so a failure is retried on
+        # the next link instead of being silently lost. (codex review, P2.)
         for grant in pending:
             status, resp = http_json(
                 "PUT",
@@ -1050,6 +1095,8 @@ class LinkService:
                 body={"permission": grant.get("permission", "write")})
             if status in (200, 204):
                 applied.append(grant["repo"])
+                with self.state.lock:
+                    self.state.consume_invite(email, grant["repo"])
                 self.audit.write("invite.applied", login=login,
                                  repo=grant["repo"],
                                  permission=grant.get("permission"),

tests/test_client.py

@@ -1284,5 +1284,127 @@ class TestResolveGranted(unittest.TestCase):
                 client.resolve_granted(self.cfg, "Nirlabinc/Ai-Assistant"),
                 "Nirlabinc/Ai-Assistant")
 
+
+class TestDeviceFlowOutputIsVisible(unittest.TestCase):
+    """The code has a five-minute life. If it is sitting in a buffer, the user
+    never sees it and it expires -- which is what happened on 2026-08-23 while
+    driving a first real sign-in through a wrapper."""
+
+    def test_the_url_and_code_are_flushed_immediately(self):
+        seen = []
+        real_print = print
+
+        def spy(*a, **kw):
+            seen.append(kw.get("flush", False))
+            return real_print(*a, **{k: v for k, v in kw.items() if k != "flush"})
+
+        resp = {"verification_uri_complete": "https://id.example/device?user_code=AB-CD",
+                "user_code": "AB-CD", "device_code": "dc", "interval": 0,
+                "expires_in": 0}
+        with mock.patch.object(client, "http_json", lambda *a, **k: (200, resp)), \
+             mock.patch("builtins.print", spy), \
+             mock.patch.object(client.time, "sleep", lambda *_: None):
+            with self.assertRaises(SystemExit):   # expires_in 0 -> times out
+                client.device_flow()
+        self.assertTrue(seen, "device_flow printed nothing")
+        self.assertTrue(all(seen[:2]),
+                        "the URL and code must be printed with flush=True")
+
+
+class TestCodexReviewFindings(GitScenarioBase):
+    """Regressions for the three issues codex found in today's merged work."""
+
+    def test_staged_only_work_survives_a_snapshot(self):
+        """[P2] Stage a hunk, edit further, lose the laptop: the staged version
+        must still be recoverable, not just the later worktree one."""
+        self.write(self.local, "a.txt", "committed")
+        run_git(self.local, "add", "-A")
+        run_git(self.local, "commit", "-m", "base")
+        self.write(self.local, "a.txt", "THE CAREFULLY STAGED VERSION")
+        run_git(self.local, "add", "a.txt")          # staged
+        self.write(self.local, "a.txt", "later scratch edit")   # worktree moved on
+
+        commit, tree = client.build_snapshot(self.local)
+
+        # the worktree state is the snapshot's own tree
+        self.assertEqual(run_git(self.local, "show", f"{commit}:a.txt"),
+                         "later scratch edit")
+        # ...and the staged state is reachable through the extra parent
+        parents = run_git(self.local, "log", "-1", "--format=%P", commit).split()
+        staged = [p for p in parents
+                  if run_git(self.local, "show", f"{p}:a.txt")
+                  == "THE CAREFULLY STAGED VERSION"]
+        self.assertTrue(staged, f"staged version unreachable from {parents}")
+
+    def test_a_snapshot_does_not_disturb_the_index(self):
+        self.write(self.local, "a.txt", "one")
+        run_git(self.local, "add", "-A")
+        run_git(self.local, "commit", "-m", "base")
+        self.write(self.local, "a.txt", "staged")
+        run_git(self.local, "add", "a.txt")
+        before = run_git(self.local, "status", "--porcelain")
+        client.build_snapshot(self.local)
+        self.assertEqual(run_git(self.local, "status", "--porcelain"), before)
+
+    def test_index_only_work_survives_a_pushed_snapshot(self):
+        """A staged version remains backed up when worktree bytes equal HEAD."""
+        self.write(self.local, "a.txt", "committed")
+        run_git(self.local, "add", "-A")
+        run_git(self.local, "commit", "-m", "base")
+        run_git(self.local, "push", "-u", "granthi", "main")
+        head_before = run_git(self.local, "rev-parse", "HEAD")
+
+        self.write(self.local, "a.txt", "INDEX-ONLY STAGED VERSION")
+        run_git(self.local, "add", "a.txt")
+        self.write(self.local, "a.txt", "committed")
+        status_before = run_git(self.local, "status", "--porcelain")
+        index_before = run_git(self.local, "write-tree")
+        with open(os.path.join(self.local, "a.txt")) as f:
+            file_before = f.read()
+        self.assertEqual(status_before, "MM a.txt")
+
+        ref = client.push_snapshot(self.local, "dev-index-only")
+
+        self.assertIsNotNone(ref)
+        snapshot = run_git(self.bare, "rev-parse", ref)
+        parents = run_git(self.bare, "log", "-1", "--format=%P", snapshot).split()
+        self.assertTrue(
+            any(run_git(self.bare, "show", f"{parent}:a.txt")
+                == "INDEX-ONLY STAGED VERSION" for parent in parents),
+            f"staged version unreachable from {parents}")
+        self.assertEqual(run_git(self.local, "rev-parse", "HEAD"), head_before)
+        self.assertEqual(run_git(self.local, "write-tree"), index_before)
+        self.assertEqual(run_git(self.local, "status", "--porcelain"), status_before)
+        with open(os.path.join(self.local, "a.txt")) as f:
+            self.assertEqual(f.read(), file_before)
+
+        self.write(self.local, "a.txt", "A NEW STAGED VERSION")
+        run_git(self.local, "add", "a.txt")
+        self.write(self.local, "a.txt", "committed")
+        second_ref = f"refs/granthi-backup/dev-index-only/second"
+        with mock.patch.object(client, "snapshot_ref",
+                               return_value=(second_ref, "second")):
+            self.assertEqual(
+                client.push_snapshot(self.local, "dev-index-only"), second_ref)
+        second = run_git(self.bare, "rev-parse", second_ref)
+        second_parents = run_git(
+            self.bare, "log", "-1", "--format=%P", second).split()
+        self.assertTrue(
+            any(run_git(self.bare, "show", f"{parent}:a.txt")
+                == "A NEW STAGED VERSION" for parent in second_parents),
+            f"updated staged version unreachable from {second_parents}")
+
+    def test_a_truncated_listing_refuses_instead_of_guessing(self):
+        """[P3] A name that is merely beyond the page cap must not resolve to a
+        different repo that happens to exist under your own account."""
+        cfg = {"login": "alice", "gitea_base": "http://forge.example", "token": "t"}
+        with self.assertRaises(SystemExit) as e:
+            client.resolve_granted(cfg, "notes", repos=[], truncated=True)
+        self.assertIn("truncated", str(e.exception))
+        # a complete listing still falls back, because absence is then real
+        self.assertEqual(
+            client.resolve_granted(cfg, "notes", repos=[], truncated=False),
+            "alice/notes")
+
 if __name__ == "__main__":
     unittest.main()

tests/test_server.py

@@ -77,6 +77,10 @@ class StubUpstream(BaseHTTPRequestHandler):
         if self.path == "/api/v1/admin/users":
             if body["username"] in st["users"]:
                 return self._json(409, {"message": "user already exists"})
+            if body.get("email") in st["users"].values():
+                # real Gitea: emails are unique across accounts
+                return self._json(422, {"message":
+                    f"e-mail already in use [email: {body['email']}]"})
             st["users"][body["username"]] = body["email"]
             st["created"].append(body)
             return self._json(201, {"login": body["username"]})
@@ -960,6 +964,75 @@ class TestInvites(ServiceTestBase):
         _, mine = self.svc.audit_read({"token": self.alice["token"]})
         self.assertIn("invite", [e["event"] for e in mine["events"]])
 
+
+class TestInviteSurvivesAFailedGrant(ServiceTestBase):
+    """[P2, codex] A transient forge error must not destroy the promise."""
+
+    def setUp(self):
+        super().setUp()
+        self.enable_test_mode()
+        self.alice = self.stub_link("s1", "alice", email="[email protected]",
+                                    verified=True, device_id="dev-a")[1]
+        StubUpstream.state["repo_owner"]["alice/notes"] = "alice"
+        StubUpstream.state["repo_owner"]["alice/reports"] = "alice"
+
+    def test_a_failed_grant_leaves_the_invite_pending_for_next_time(self):
+        self.svc.invite({"token": self.alice["token"], "email": "[email protected]",
+                         "repos": ["notes"]})
+        # the forge loses the repo mid-flight -> the PUT 404s
+        StubUpstream.state["repo_owner"].pop("alice/notes")
+        status, resp = self.stub_link("s9", "carol", email="[email protected]",
+                                      verified=True, device_id="dev-c")
+        self.assertEqual(status, 200)                 # sign-in still succeeds
+        self.assertIsNone(resp.get("granted_repos"))
+        # the promise is STILL THERE rather than silently consumed
+        self.assertEqual([g["repo"] for g in self.svc.state.peek_invites("[email protected]")],
+                         ["alice/notes"])
+        # and it lands on the next link, once the repo is back
+        StubUpstream.state["repo_owner"]["alice/notes"] = "alice"
+        status, resp = self.stub_link("s9", "carol", email="[email protected]",
+                                      verified=True, device_id="dev-c2")
+        self.assertEqual(resp.get("granted_repos"), ["alice/notes"])
+        self.assertEqual(self.svc.state.peek_invites("[email protected]"), [])
+
+    def test_a_partial_failure_only_consumes_what_landed(self):
+        self.svc.invite({"token": self.alice["token"], "email": "[email protected]",
+                         "repos": ["notes", "reports"]})
+        StubUpstream.state["repo_owner"].pop("alice/reports")   # one of two fails
+        _, resp = self.stub_link("s10", "dan", email="[email protected]",
+                                 verified=True, device_id="dev-d")
+        self.assertEqual(resp.get("granted_repos"), ["alice/notes"])
+        self.assertEqual([g["repo"] for g in self.svc.state.peek_invites("[email protected]")],
+                         ["alice/reports"])
+
+
+class TestDuplicateEmailIsExplained(ServiceTestBase):
+    """A 502 sends someone looking for an outage. The real answer is that the
+    address already belongs to another account here -- say so. (Hit live on
+    2026-08-23 when an operator moved an email onto a different account.)"""
+
+    def setUp(self):
+        super().setUp()
+        self.enable_test_mode()
+        # an existing account already holds the address
+        StubUpstream.state["users"]["existing"] = "[email protected]"
+
+    def test_it_is_a_409_that_names_the_problem(self):
+        status, resp = self.stub_link("s-new", "brandnew", email="[email protected]",
+                                      verified=True, device_id="dev-x")
+        self.assertEqual(status, 409, resp)
+        msg = resp["error"]
+        self.assertIn("[email protected]", msg)
+        self.assertIn("already belongs to a different account", msg)
+        self.assertIn("brandnew", msg)          # names the login it tried
+        self.assertNotIn("502", msg)
+
+    def test_a_normal_create_is_unaffected(self):
+        status, resp = self.stub_link("s-ok", "fresh", email="[email protected]",
+                                      verified=True, device_id="dev-y")
+        self.assertEqual(status, 200, resp)
+        self.assertIn("token", resp)
+
 if __name__ == "__main__":
     unittest.main()