fix(client): preserve index-only snapshot state

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}")

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."""