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.

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()