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

@@ -296,11 +296,10 @@ 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:
@@ -314,11 +313,9 @@ def build_snapshot(folder):
         # 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:
-            head_tree_now = ""
-            if head:
-                head_tree_now = git(folder, "rev-parse", f"{head}^{{tree}}")[1].strip()
-            if index_tree != head_tree_now:
+            if index_tree != head_tree:
                 icommit_args = ["commit-tree", index_tree, "-m",
                                 f"granthi snapshot (staged): {ts}"]
                 if head:
@@ -327,6 +324,11 @@ def build_snapshot(folder):
                                      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
@@ -352,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}")

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.
 
@@ -677,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
@@ -733,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)
@@ -758,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:

tests/test_client.py

@@ -1346,6 +1346,54 @@ class TestCodexReviewFindings(GitScenarioBase):
         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."""

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"]})
@@ -1001,6 +1005,34 @@ class TestInviteSurvivesAFailedGrant(ServiceTestBase):
         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()