fix: isolate review resources, shell authority, and goal gate evidence

agent/background_review.py

@@ -22,6 +22,7 @@
 import json
 import logging
 import os
+import uuid
 from typing import Any, Dict, List, Optional
 
 from agent.thread_scoped_output import thread_scoped_silence
@@ -886,6 +887,10 @@ def _unregister_review_agent(agent_ref) -> None:
                 # rebuild path, but these pins guarantee parity even
                 # if a future code path bypasses the cache.
                 review_agent.session_start = agent.session_start
+            # Cache/transcript attribution is not resource ownership. Pin a
+            # private task namespace before borrowing the parent's session id;
+            # both tool dispatch and close() must use this same namespace.
+            review_agent._resource_owner_task_id = str(uuid.uuid4())
             review_agent.session_id = agent.session_id
             # The fork shares the parent's live session_id (pinned above for
             # prefix-cache parity). It is single-lifecycle and calls close()
@@ -973,6 +978,7 @@ def _unregister_review_agent(agent_ref) -> None:
                     else messages_snapshot
                 )
                 review_agent.run_conversation(
+                    task_id=review_agent._resource_owner_task_id,
                     user_message=(
                         prompt
                         + "\n\nYou can only call memory and skill "

hermes_cli/goals.py

@@ -29,7 +29,6 @@
 
 from __future__ import annotations
 
-import hashlib
 import json
 import logging
 import os
@@ -445,8 +444,8 @@ class GoalGate:
     attempts: int = 0
     last_exit_code: Optional[int] = None
     last_output_tail: str = ""
-    # Workspace fingerprint at the time of the last FAILED run — used to skip
-    # re-running an identical gate when nothing changed since it failed.
+    # Legacy serialized field, retained for compatibility only. Never reuse
+    # a gate result based on a workspace fingerprint.
     last_failed_fingerprint: str = ""
 
     def to_dict(self) -> Dict[str, Any]:
@@ -467,36 +466,6 @@ def from_dict(cls, data: Optional[Dict[str, Any]]) -> "GoalGate":
         )
 
 
-def workspace_fingerprint(cwd: Optional[str] = None) -> str:
-    """Cheap workspace change fingerprint for unchanged-gate skip.
-
-    Uses ``git status --porcelain`` + ``git rev-parse HEAD`` when inside a git
-    repo (covers tracked edits, stages, and commits). Outside git, returns
-    an empty string — an empty fingerprint never matches, so gates simply
-    always re-run (safe fallback, no behavior regression for non-repo work).
-    """
-    workdir = cwd or os.getcwd()
-    try:
-        head = subprocess.run(
-            ["git", "rev-parse", "HEAD"],
-            capture_output=True, text=True, encoding="utf-8", errors="replace",
-            timeout=10, cwd=workdir,
-        )
-        if head.returncode != 0:
-            return ""
-        status = subprocess.run(
-            ["git", "status", "--porcelain"],
-            capture_output=True, text=True, encoding="utf-8", errors="replace",
-            timeout=30, cwd=workdir,
-        )
-        if status.returncode != 0:
-            return ""
-        blob = head.stdout.strip() + "\n" + status.stdout
-        return hashlib.sha256(blob.encode("utf-8", "replace")).hexdigest()
-    except Exception:
-        return ""
-
-
 def run_gate(gate: GoalGate, *, cwd: Optional[str] = None) -> Tuple[bool, int, str]:
     """Run one gate command. Returns ``(passed, exit_code, output_tail)``.
 
@@ -1496,26 +1465,16 @@ def _check_gates(self) -> Optional[Dict[str, Any]]:
         either a continuation carrying the gate's output (attempts left)
         or an auto-pause (retries exhausted).
 
-        An unchanged workspace since the last failure of the same gate is
-        NOT re-run — the recorded failure is replayed and the attempt count
-        advances, so a stalled agent can't spin re-running an identical red
-        suite (mirrors Prime-Agent's unchanged-gate rule).
+        Always execute fresh evidence. Git status/HEAD cannot identify edits
+        to an already-dirty file, external state, or another worktree targeted
+        by the command. Retry and turn budgets still bound repeated failures.
         """
         state = self._state
         if state is None or not state.gates:
             return None
 
-        fingerprint = workspace_fingerprint()
         for gate in state.gates:
-            unchanged = (
-                bool(fingerprint)
-                and gate.last_exit_code not in (None, 0)
-                and gate.last_failed_fingerprint == fingerprint
-            )
-            if unchanged:
-                passed, exit_code, tail = False, int(gate.last_exit_code or -1), gate.last_output_tail
-            else:
-                passed, exit_code, tail = run_gate(gate)
+            passed, exit_code, tail = run_gate(gate)
             gate.last_exit_code = exit_code
             gate.last_output_tail = tail
             if passed:
@@ -1524,8 +1483,7 @@ def _check_gates(self) -> Optional[Dict[str, Any]]:
                 continue
 
             gate.attempts += 1
-            gate.last_failed_fingerprint = fingerprint
-            skipped_note = " (workspace unchanged since last failure — not re-run)" if unchanged else ""
+            gate.last_failed_fingerprint = ""
 
             if gate.attempts > gate.max_retries:
                 state.status = "paused"
@@ -1564,7 +1522,7 @@ def _check_gates(self) -> Optional[Dict[str, Any]]:
                 "reason": f"gate failed (exit {exit_code}): $ {gate.command}",
                 "message": (
                     f"✗ Quality gate failed ({state.turns_used}/{state.max_turns} turns, "
-                    f"attempt {gate.attempts}/{gate.max_retries}){skipped_note}: $ {gate.command}"
+                    f"attempt {gate.attempts}/{gate.max_retries}): $ {gate.command}"
                 ),
             }
 
@@ -2136,7 +2094,6 @@ def _log(msg: str) -> None:
     "parse_contract",
     "draft_contract",
     "run_gate",
-    "workspace_fingerprint",
     "CONTINUATION_PROMPT_TEMPLATE",
     "CONTINUATION_PROMPT_WITH_SUBGOALS_TEMPLATE",
     "CONTINUATION_PROMPT_WITH_CONTRACT_TEMPLATE",

run_agent.py

@@ -4312,7 +4312,13 @@ def close(self) -> None:
         Safe to call multiple times (idempotent).  Each cleanup step is
         independently guarded so a failure in one does not prevent the rest.
         """
-        task_id = getattr(self, "session_id", None) or ""
+        # Review forks borrow session_id for prompt-cache attribution, never
+        # ownership of the parent's processes/environments/browser/CUA state.
+        task_id = (
+            getattr(self, "_resource_owner_task_id", None)
+            or getattr(self, "session_id", None)
+            or ""
+        )
 
         # 1. Kill background processes for this task
         try:

tests/hermes_cli/test_goal_gate_fresh_evidence.py

@@ -0,0 +1,60 @@
+"""Native goal gates must execute fresh evidence, not replay git-status receipts."""
+
+import subprocess
+import sys
+
+import pytest
+
+from hermes_cli.goals import GoalManager
+
+
+def git(path, *args):
+    return subprocess.run(
+        ["git", "-C", str(path), *args], check=True, capture_output=True, text=True,
+    ).stdout
+
+
+@pytest.mark.parametrize("other_worktree", [False, True])
+def test_gate_reruns_after_repeated_dirty_edits(tmp_path, monkeypatch, other_worktree):
+    repo = tmp_path / "repo"
+    repo.mkdir()
+    git(repo, "init")
+    gate_script = repo / "gate.py"
+    gate_script.write_text(
+        "from pathlib import Path\n"
+        "value = Path(__file__).with_name('value').read_text().strip()\n"
+        "print('actual gate value=' + value)\n"
+        "raise SystemExit(int(value))\n"
+    )
+    (repo / "value").write_text("0")
+    git(repo, "add", "gate.py", "value")
+    git(repo, "-c", "user.name=Fixture", "-c", "user.email=fixture@example.invalid",
+        "commit", "-m", "fixture")
+    target = repo
+    if other_worktree:
+        target = tmp_path / "target"
+        git(repo, "worktree", "add", "--detach", str(target), "HEAD")
+    monkeypatch.chdir(repo)
+    mgr = GoalManager(session_id="gate-fresh-evidence")
+    mgr.set("fixture goal")
+    mgr.add_gate(f'"{sys.executable}" "{target / "gate.py"}"')
+
+    (target / "value").write_text("1")
+    first_status = git(repo, "status", "--porcelain")
+    assert mgr._check_gates()["verdict"] == "gate_failed"
+    gate = mgr.state.gates[0]
+    assert gate.last_exit_code == 1
+    assert "actual gate value=1" in gate.last_output_tail
+
+    # Same dirty-path listing and HEAD, different bytes (or different worktree).
+    (target / "value").write_text("2")
+    assert git(repo, "status", "--porcelain") == first_status
+    assert mgr._check_gates()["verdict"] == "gate_failed"
+    assert gate.last_exit_code == 2, "replayed stale failure instead of executing gate"
+    assert "actual gate value=2" in gate.last_output_tail
+
+    (target / "value").write_text("0")
+    assert mgr._check_gates() is None
+    assert gate.last_exit_code == 0
+    assert "actual gate value=0" in gate.last_output_tail
+    assert gate.attempts == 0

tests/hermes_cli/test_goal_gates.py

@@ -161,8 +161,7 @@ def test_status_line_mentions_gates():
 def test_failing_gate_short_circuits_judge():
     mgr = _mgr_with_goal("gate-fail-sid")
     mgr.add_gate("exit 5")
-    with patch("hermes_cli.goals.judge_goal") as mock_judge, \
-         patch("hermes_cli.goals.workspace_fingerprint", return_value=""):
+    with patch("hermes_cli.goals.judge_goal") as mock_judge:
         decision = mgr.evaluate_after_turn("I think it's done!")
     mock_judge.assert_not_called()
     assert decision["verdict"] == "gate_failed"
@@ -190,8 +189,7 @@ def test_gate_retry_exhaustion_pauses_goal():
     mgr = _mgr_with_goal("gate-exhaust-sid")
     mgr.add_gate("exit 1")
     mgr.state.gates[0].max_retries = 2
-    with patch("hermes_cli.goals.judge_goal") as mock_judge, \
-         patch("hermes_cli.goals.workspace_fingerprint", return_value=""):
+    with patch("hermes_cli.goals.judge_goal") as mock_judge:
         d1 = mgr.evaluate_after_turn("attempt one")
         d2 = mgr.evaluate_after_turn("attempt two")
         d3 = mgr.evaluate_after_turn("attempt three")
@@ -204,38 +202,38 @@ def test_gate_retry_exhaustion_pauses_goal():
     assert "gate" in (mgr.state.paused_reason or "")
 
 
-def test_unchanged_workspace_skips_rerun():
+def test_unchanged_workspace_reruns_gate_with_fresh_diagnostics():
     mgr = _mgr_with_goal("gate-unchanged-sid")
     mgr.add_gate("exit 1")
-    with patch("hermes_cli.goals.workspace_fingerprint", return_value="fp-1"), \
-         patch("hermes_cli.goals.judge_goal"):
+    with patch("hermes_cli.goals.judge_goal"):
         mgr.evaluate_after_turn("turn 1")
-        # Second turn, same fingerprint — run_gate must NOT run again.
-        with patch("hermes_cli.goals.run_gate") as mock_run:
+        with patch("hermes_cli.goals.run_gate", return_value=(False, 2, "fresh failure")) as mock_run:
             d2 = mgr.evaluate_after_turn("turn 2")
-        mock_run.assert_not_called()
+        mock_run.assert_called_once()
     assert d2["verdict"] == "gate_failed"
-    assert "unchanged" in d2["message"]
+    assert "fresh failure" in d2["continuation_prompt"]
+    assert mgr.state.gates[0].last_exit_code == 2
 
 
-def test_changed_workspace_reruns_gate():
-    mgr = _mgr_with_goal("gate-changed-sid")
-    mgr.add_gate("exit 1")
-    with patch("hermes_cli.goals.judge_goal"):
-        with patch("hermes_cli.goals.workspace_fingerprint", return_value="fp-1"):
-            mgr.evaluate_after_turn("turn 1")
-        with patch("hermes_cli.goals.workspace_fingerprint", return_value="fp-2"), \
-             patch("hermes_cli.goals.run_gate", return_value=(False, 1, "still red")) as mock_run:
-            mgr.evaluate_after_turn("turn 2")
-        mock_run.assert_called_once()
+def test_legacy_failure_fingerprint_cannot_skip_passing_rerun():
+    mgr = _mgr_with_goal("gate-legacy-sid")
+    mgr.add_gate("true")
+    gate = mgr.state.gates[0]
+    gate.last_exit_code = 1
+    gate.last_output_tail = "old failure"
+    gate.last_failed_fingerprint = "legacy-fingerprint"
+    save_goal(mgr.session_id, mgr.state)
+    reloaded = GoalManager(session_id=mgr.session_id)
+    assert reloaded._check_gates() is None
+    assert reloaded.state.gates[0].last_exit_code == 0
+    assert reloaded.state.gates[0].last_failed_fingerprint == ""
 
 
 def test_gate_continuation_respects_turn_budget():
     mgr = GoalManager(session_id="gate-budget-sid", default_max_turns=1)
     mgr.set("budget goal")
     mgr.add_gate("exit 1")
-    with patch("hermes_cli.goals.judge_goal"), \
-         patch("hermes_cli.goals.workspace_fingerprint", return_value=""):
+    with patch("hermes_cli.goals.judge_goal"):
         decision = mgr.evaluate_after_turn("only turn")
     assert decision["status"] == "paused"
     assert decision["should_continue"] is False

tests/run_agent/test_review_resource_ownership.py

@@ -0,0 +1,78 @@
+"""Review teardown must not own the live parent's tool resources."""
+
+from types import SimpleNamespace
+
+import pytest
+
+import run_agent
+from agent import background_review
+from run_agent import AIAgent
+from tests.run_agent.test_background_review import _bare_agent
+
+
+@pytest.mark.parametrize("crash", [False, True])
+def test_real_review_lifecycle_closes_only_review_resources(monkeypatch, crash):
+    parent = _bare_agent()
+    resources = {
+        kind: {parent.session_id: SimpleNamespace(alive=True)}
+        for kind in ("job", "environment", "browser", "cua")
+    }
+    parent_resources = {kind: rows[parent.session_id] for kind, rows in resources.items()}
+    captured = {}
+
+    def release(kind, task_id):
+        resource = resources[kind].pop(task_id, None)
+        if resource:
+            resource.alive = False
+
+    # Stateful resource fixtures: no actual process killing or backend calls.
+    from tools.process_registry import process_registry
+    import tools.computer_use as cua
+    monkeypatch.setattr(process_registry, "kill_all", lambda task_id: release("job", task_id))
+    monkeypatch.setattr(run_agent, "cleanup_vm", lambda task_id: release("environment", task_id))
+    monkeypatch.setattr(run_agent, "cleanup_browser", lambda task_id: release("browser", task_id))
+    monkeypatch.setattr(cua, "release_computer_use_session", lambda task_id: release("cua", task_id))
+    monkeypatch.setattr("hermes_cli.mem_trim.trim_memory", lambda **kw: None)
+    monkeypatch.setattr(background_review, "_resolve_review_runtime", lambda agent: {"routed": False})
+    monkeypatch.setattr("model_tools.get_tool_definitions", lambda **kw: [])
+
+    class Review(AIAgent):
+        def __init__(self, **kwargs):
+            # Isolate provider initialization; retain the REAL lifecycle/close.
+            self.session_id = "review-constructor-session"
+            self._session_messages = []
+            self.client = None
+            self._session_db = None
+            captured["review"] = self
+
+        def run_conversation(self, **kwargs):
+            captured["ran"] = True
+            captured["task_id"] = kwargs.get("task_id") or "review-generated-turn"
+            captured["prompt"] = self._cached_system_prompt
+            captured["session_id"] = self.session_id
+            captured["resources"] = {}
+            for kind, rows in resources.items():
+                resource = SimpleNamespace(alive=True)
+                rows[captured["task_id"]] = resource
+                captured["resources"][kind] = resource
+            if crash:
+                raise RuntimeError("fixture review failure")
+
+        def shutdown_memory_provider(self):
+            pass
+
+    monkeypatch.setattr(run_agent, "AIAgent", Review)
+    # Exercise the real review worker, including its success/exception finally.
+    background_review._run_review_in_thread(parent, [], "fixture review")
+
+    assert captured.get("ran"), "review did not reach its execution seam"
+    assert captured["prompt"] == parent._cached_system_prompt
+    assert captured["session_id"] == parent.session_id  # cache attribution unchanged
+    assert all(resource.alive for resource in parent_resources.values()), "review closed parent resources"
+    assert captured["task_id"] != parent.session_id
+    assert all(not resource.alive for resource in captured["resources"].values()), "review leaked its resources"
+    assert parent._active_children == []
+    assert parent._background_review_agent is None
+    # Repeated close remains confined to this review's namespace.
+    captured["review"].close()
+    assert all(resource.alive for resource in parent_resources.values())