fix(delegate): don't delete a subagent's uncommitted work when git inspection fails

tests/tools/test_subagent_worktree.py

@@ -18,9 +18,9 @@
 from tools import subagent_worktree as sw  # noqa: E402
 
 
-def _git(args, cwd):
+def _git(args, cwd, check=True):
     return subprocess.run(
-        ["git", *args], cwd=cwd, capture_output=True, text=True, check=True
+        ["git", *args], cwd=cwd, capture_output=True, text=True, check=check
     )
 
 
@@ -36,6 +36,14 @@ def _make_repo(root: Path) -> Path:
     return repo
 
 
+def _break_git_index(wt: Path) -> None:
+    """Corrupt a worktree's index so the real git probes exit non-zero."""
+    git_dir = Path(_git(["rev-parse", "--git-dir"], wt).stdout.strip())
+    if not git_dir.is_absolute():
+        git_dir = (wt / git_dir).resolve()
+    (git_dir / "index").write_bytes(b"not-a-valid-git-index\n")
+
+
 class SubagentWorktreeTests(unittest.TestCase):
     def setUp(self):
         self.tmp = Path(tempfile.mkdtemp(prefix="hermes-sw-test-"))
@@ -132,6 +140,156 @@ def test_finalize_keeps_dirty_worktree(self):
         self.assertTrue(payload["dirty"])
         self.assertTrue(os.path.isdir(info["path"]))
 
+    def test_finalize_keeps_worktree_when_git_inspection_fails(self):
+        """#88113: a non-zero git status exit must not be read as "clean".
+
+        Corrupting the index makes the real `git status --porcelain` probe
+        exit 128. The old code kept the payload defaults (commits=0,
+        dirty=False) and pruned on them — permanently deleting the child's
+        uncommitted work. A destructive cleanup requires affirmative proof
+        of a clean tree."""
+        repo = _make_repo(self.tmp)
+        info = sw.create_subagent_worktree(str(repo), "inspect-fail1")
+        assert info is not None
+        wt = Path(info["path"])
+        (wt / "UNCOMMITTED-WORK.txt").write_text(
+            "irreplaceable\n", encoding="utf-8"
+        )
+
+        _break_git_index(wt)
+        # Sanity: the probe really fails now.
+        broken = _git(["status", "--porcelain"], wt, check=False)
+        self.assertNotEqual(broken.returncode, 0)
+
+        payload = sw.finalize_subagent_worktree(info)
+
+        self.assertFalse(payload["pruned"])
+        self.assertTrue(os.path.isdir(info["path"]))
+        self.assertTrue((wt / "UNCOMMITTED-WORK.txt").exists())
+        branches = _git(["branch", "--list", info["branch"]], repo).stdout
+        self.assertNotEqual(branches.strip(), "")
+        # The parent agent only ever sees this payload (it cannot read logs),
+        # so the uncertainty must travel in the dict — otherwise "0 commits,
+        # clean" reads as "the child produced nothing" and the work we just
+        # preserved never gets looked at. Assert the actionable invariants
+        # (flag set; note names the worktree + branch), not the prose.
+        self.assertTrue(payload["inspection_failed"])
+        self.assertIn(info["path"], payload["note"])
+        self.assertIn(info["branch"], payload["note"])
+
+    def test_finalize_flags_unproven_state_distinguishably(self):
+        """#88113 follow-up: a failed inspection must not look like "no work".
+
+        Without an explicit flag, "inspection failed, uncommitted work
+        preserved" and "inspected fine, child left nothing" are
+        indistinguishable to the parent agent — so its rational reading of the
+        failure case is the exact wrong conclusion. The contract asserted here
+        is *distinguishability*, not the specific field values (a future
+        change emitting ``commits: None`` for "unknown" would be strictly
+        better and must not break this test).
+        """
+        repo = _make_repo(self.tmp)
+
+        # Case 1: inspection SUCCEEDED, tree genuinely clean, prune disabled.
+        ok_info = sw.create_subagent_worktree(str(repo), "proven-clean")
+        assert ok_info is not None
+        ok_payload = sw.finalize_subagent_worktree(ok_info, prune=False)
+
+        # Case 2: inspection FAILED with real uncommitted work on disk.
+        bad_info = sw.create_subagent_worktree(str(repo), "unproven")
+        assert bad_info is not None
+        bad_wt = Path(bad_info["path"])
+        (bad_wt / "WIP.txt").write_text("real work\n", encoding="utf-8")
+        _break_git_index(bad_wt)
+        bad_payload = sw.finalize_subagent_worktree(bad_info)
+
+        # Both keep the worktree, so "pruned" alone cannot separate them...
+        self.assertFalse(ok_payload["pruned"])
+        self.assertFalse(bad_payload["pruned"])
+        self.assertTrue(os.path.isdir(bad_info["path"]))
+        self.assertTrue((bad_wt / "WIP.txt").exists())
+        # ...the flag must. Tolerant of an always-present-but-False refactor.
+        self.assertFalse(ok_payload.get("inspection_failed", False))
+        self.assertTrue(bad_payload["inspection_failed"])
+        self.assertNotEqual(
+            bool(ok_payload.get("inspection_failed")),
+            bool(bad_payload.get("inspection_failed")),
+        )
+        self.assertFalse((ok_payload.get("note") or "").strip())
+
+    def test_finalize_flags_unproven_state_when_inspection_raises(self):
+        """A raising probe is the same unknown state as a non-zero exit."""
+        repo = _make_repo(self.tmp)
+        info = sw.create_subagent_worktree(str(repo), "raises")
+        assert info is not None
+
+        def _boom(*_a, **_k):
+            raise subprocess.TimeoutExpired(cmd="git", timeout=30)
+
+        with mock.patch.object(sw, "_run_git", side_effect=_boom) as m:
+            payload = sw.finalize_subagent_worktree(info)
+
+        # Prove the patched seam was actually exercised.
+        self.assertGreaterEqual(m.call_count, 1)
+        self.assertFalse(payload["pruned"])
+        self.assertTrue(payload["inspection_failed"])
+        self.assertIn(info["path"], payload["note"])
+        self.assertIn(info["branch"], payload["note"])
+        self.assertTrue(os.path.isdir(info["path"]))
+        branches = _git(["branch", "--list", info["branch"]], repo).stdout
+        self.assertNotEqual(branches.strip(), "")
+
+    def test_finalize_note_disclaims_only_the_unmeasured_field(self):
+        """A partial failure must not claim a MEASURED value is unknown.
+
+        A bad base_commit fails `rev-list` while `status` still succeeds, so
+        ``dirty`` is a real measurement — the note should disclaim ``commits``
+        only, or it misreports in the other direction.
+        """
+        repo = _make_repo(self.tmp)
+        info = sw.create_subagent_worktree(str(repo), "partial")
+        assert info is not None
+        wt = Path(info["path"])
+        (wt / "UNTRACKED.txt").write_text("dirty!\n", encoding="utf-8")
+
+        payload = sw.finalize_subagent_worktree(
+            {**info, "base_commit": "deadbeef" * 5}
+        )
+
+        # status succeeded, so dirty is trustworthy and reported as such.
+        self.assertTrue(payload["dirty"])
+        self.assertTrue(payload["inspection_failed"])
+        self.assertFalse(payload["pruned"])
+        # The note names ONLY the unmeasured field.
+        self.assertIn("commits UNKNOWN", payload["note"])
+        self.assertNotIn("dirty UNKNOWN", payload["note"])
+        self.assertNotIn("commits/dirty", payload["note"])
+        self.assertTrue((wt / "UNTRACKED.txt").exists())
+
+    def test_finalize_keeps_worktree_when_base_commit_missing(self):
+        """An unmeasurable commit count must not authorize deletion.
+
+        With no base_commit the rev-list probe never runs, so
+        ``payload["commits"]`` would keep its unproven 0 default — the same
+        class of bug as #88113. Fail closed instead.
+        """
+        repo = _make_repo(self.tmp)
+        info = sw.create_subagent_worktree(str(repo), "nobase")
+        assert info is not None
+        wt = Path(info["path"])
+        (wt / "CHILD.txt").write_text("work\n", encoding="utf-8")
+        _git(["add", "-A"], wt)
+        _git(["commit", "-q", "-m", "child work"], wt)
+
+        payload = sw.finalize_subagent_worktree({**info, "base_commit": ""})
+
+        self.assertFalse(payload["pruned"])
+        self.assertTrue(payload["inspection_failed"])
+        self.assertTrue(os.path.isdir(info["path"]))
+        self.assertTrue((wt / "CHILD.txt").exists())
+        branches = _git(["branch", "--list", info["branch"]], repo).stdout
+        self.assertNotEqual(branches.strip(), "")
+
     def test_finalize_missing_path_reports_pruned(self):
         payload = sw.finalize_subagent_worktree(
             {"path": str(self.tmp / "gone"), "branch": "b", "repo_root": "",
@@ -166,6 +324,69 @@ def test_context_note_names_path_and_branch(self):
         self.assertIn("WORKTREE ISOLATION", note)
 
 
+class WorktreePayloadSchemaTests(unittest.TestCase):
+    """The parent agent reads ONE schema under ``entry["worktree"]``.
+
+    Two producers write it: ``finalize_subagent_worktree`` and
+    ``delegate_tool``'s fallback for when finalize itself raises. Both now go
+    through the shared factory, so this compares the factory's real output
+    against real finalize output — behavior, not source text.
+    """
+
+    def setUp(self):
+        self.tmp = Path(tempfile.mkdtemp(prefix="hermes-sw-schema-"))
+        self.addCleanup(shutil.rmtree, self.tmp, True)
+
+    def test_unproven_payload_matches_finalize_schema(self):
+        repo = _make_repo(self.tmp)
+        info = sw.create_subagent_worktree(str(repo), "schema")
+        assert info is not None
+
+        happy = sw.finalize_subagent_worktree(info, prune=False)
+        unproven = sw.unproven_worktree_payload(info, "finalize raised: boom")
+
+        # Superset: every happy-path key, plus exactly the two unproven keys.
+        self.assertTrue(set(happy).issubset(set(unproven)))
+        self.assertEqual(
+            set(unproven) - set(happy), {"inspection_failed", "note"}
+        )
+        # Must NOT leak the creation-side internals the parent has no use for
+        # (the pre-fix fallback emitted these instead of the real schema).
+        for leaked in ("repo_root", "base_commit"):
+            self.assertNotIn(leaked, unproven)
+        # And it must carry the actionable content.
+        self.assertTrue(unproven["inspection_failed"])
+        self.assertIn(info["path"], unproven["note"])
+        self.assertIn(info["branch"], unproven["note"])
+
+    def test_delegate_tool_fallback_uses_the_shared_factory(self):
+        """delegate_tool's fallback must emit the flagged schema, not the
+        creation-side metadata dict it used to leak."""
+        from tools import delegate_tool  # noqa: F401  (import-safety check)
+
+        repo = _make_repo(self.tmp)
+        info = sw.create_subagent_worktree(str(repo), "fallback")
+        assert info is not None
+
+        # Drive the real helper the fallback calls.
+        payload = sw.unproven_worktree_payload(info, "finalize raised: boom")
+        self.assertEqual(
+            set(payload),
+            {
+                "path",
+                "branch",
+                "commits",
+                "dirty",
+                "pruned",
+                "inspection_failed",
+                "note",
+            },
+        )
+        self.assertFalse(payload["pruned"])
+        self.assertEqual(payload["commits"], 0)
+        self.assertFalse(payload["dirty"])
+
+
 class DelegationConfigGateTests(unittest.TestCase):
     def test_worktree_isolation_default_off(self):
         from tools import delegate_tool

tools/delegate_tool.py

@@ -2508,8 +2508,35 @@ def _attach_worktree(entry_dict: Dict[str, Any]) -> None:
                 subagent_worktree.finalize_subagent_worktree(_worktree_info)
             )
         except Exception as e:
-            logger.debug("worktree finalize failed: %s", e)
-            entry_dict["worktree"] = dict(_worktree_info)
+            # finalize is written hard not to raise, but if it ever does the
+            # state is unknown — emit the SAME schema the parent expects,
+            # flagged, via the shared factory so the two producers of this
+            # payload can never drift.
+            logger.warning("worktree finalize failed: %s", e)
+            try:
+                from tools import subagent_worktree as _sw
+
+                entry_dict["worktree"] = _sw.unproven_worktree_payload(
+                    _worktree_info, f"finalize raised: {e}"
+                )
+            except Exception:
+                # Import itself failed — inline the same shape rather than
+                # dropping the flag (the parent must still see the warning).
+                entry_dict["worktree"] = {
+                    "path": _worktree_info.get("path", ""),
+                    "branch": _worktree_info.get("branch", ""),
+                    "commits": 0,
+                    "dirty": False,
+                    "pruned": False,
+                    "inspection_failed": True,
+                    "note": (
+                        f"worktree finalize raised ({e}) and the reporting "
+                        "helper was unavailable: 'commits' and 'dirty' are "
+                        "UNKNOWN, not zero/clean. Inspect "
+                        f"{_worktree_info.get('path', '')} before assuming "
+                        "no work."
+                    ),
+                }
 
     try:
         _heartbeat_thread.start()