fix(agent): re-gate later terminal approvals after an earlier batch failure (#113158)

Desktop terminal batching pre-collects the approval for every command in
a run before any of them executes, so the user consents to a batch in
which all commands are expected to run. When an earlier command then
fails (or is denied/blocked), that informed consent no longer describes
the world the later commands will run in — yet the executor still
consumed the pre-made decision as though nothing had happened.

- _TerminalBatch.failure_seen: set by the sequential publisher after a
  slot's failed (or blocked) result is committed — i.e. after the failure
  the model actually sees, never from a wedged worker's late result.
- consume_prepared_guard drops the prepared decision for any later slot
  once a failure is published and returns None, so the guard runs its
  live flow again: tirith scan, allowlist, and a fresh human approval
  request when the command still warrants one. Nothing is auto-denied
  and an explicit denial of the failing command remains authoritative.
- The flag is sticky for the batch: a later success must not un-stale an
  approval collected before an even earlier failure.

Success-path batching is unchanged: with no failure, prepared decisions
are consumed exactly as before (byte-for-byte the same flow).

agent/terminal_approval_batch.py

@@ -98,6 +98,12 @@ class _TerminalBatch:
         self.authorization_gate = _ConcurrentToolAuthorizationGate()
         self.executor = DaemonThreadPoolExecutor(max_workers=len(parsed))
         self.slots = [_TerminalSlot(self, pc, i) for i, pc in enumerate(parsed)]
+        # Set once ANY slot in the batch has published a failed result (or a
+        # denied/blocked one). Informed consent is per the batch state the user
+        # SAW: after a failure, a later slot's pre-collected approval no longer
+        # describes the world its command will run in, so it must not be
+        # consumed — the guard re-runs live instead (#113158).
+        self.failure_seen = False
 
     def start(self):
         from agent.tool_executor import _resolve_sequential_tool_timeout
@@ -211,6 +217,16 @@ def consume_prepared_guard(command, env_type, has_host_access):
     if slot is None or slot.preparing:
         return None
     slot.check_cancelled()
+    # Re-gate after an earlier slot in the same batch failed (#113158): the
+    # user approved a batch where every command was expected to run; once one
+    # failed, that informed consent is stale for the commands after it, so
+    # drop the pre-made decision and let the guard run its live flow (tirith
+    # scan, allowlist, human approval). Nothing is auto-denied: an explicit
+    # human answer still wins; the prepared (often auto/policy) decision is
+    # simply not consumed.
+    if slot.batch.failure_seen and slot.decision is not None:
+        slot.decision = None
+        return None
     from tools.approval_context import _approval_tool_call_id
     if (_approval_tool_call_id.get() != slot.parsed.ref(slot.batch.task_id).call_id
             or slot.guard_key != (command, env_type, has_host_access)):
@@ -224,6 +240,20 @@ def preparing_terminal_approval():
     return slot is not None and slot.preparing
 
 
+def mark_batch_outcome(failed: bool) -> None:
+    """Record that the batch's current slot published a failed result.
+
+    Called by the sequential publisher AFTER a result is committed, so the
+    flag lands only for failures the model actually sees (a wedged worker's
+    late result never publishes). Sticky for the batch: one failure re-gates
+    every later prepared slot (#113158); successes leave it alone — a later
+    success must not un-stale an approval after an even earlier failure.
+    """
+    batch = _batch.get()
+    if batch is not None and failed:
+        batch.failure_seen = True
+
+
 def validate_prepared_terminal(args):
     slot = _slot.get()
     if slot is not None:

agent/tool_executor.py

@@ -1801,6 +1801,13 @@ def _publish_sequential_result(agent, messages: list, ref: _ToolCallRef, managed
         return False
     function_result, display_function_result, risk_metadata = committed
 
+    # Terminal approval batching (#113158): once this slot's failure is
+    # published, the informed consent collected for later slots describes a
+    # batch state that no longer holds — flag it so consume_prepared_guard
+    # drops their pre-made decisions and the live guard flow re-runs.
+    from agent.terminal_approval_batch import mark_batch_outcome
+    mark_batch_outcome(_is_error_result or bool(managed.blocked))
+
     _emit_tool_complete_and_risk(agent, ref, display_function_result, risk_metadata, managed.blocked)
     if _tool_progress_enabled(agent):
         _print_tool_completed(agent, index, tool_duration, function_result)

tests/agent/test_terminal_approval_batch.py

@@ -3,6 +3,7 @@
 import json
 import queue
 import threading
+import time
 from contextlib import ExitStack
 from types import SimpleNamespace
 from unittest.mock import patch
@@ -230,3 +231,95 @@ def test_cancelled_preparation_drains_requests_without_reusing_once(tmp_path, mo
                 retry_worker.join(5)
             cleanup_vm(key)
             clear_session_vars(tokens)
+
+
+def test_failed_command_re_gates_later_prepared_approvals(tmp_path, monkeypatch):
+    """#113158: the batch collects every approval before any command runs. When an
+    earlier command in the batch FAILS, the informed consent gathered for the later
+    ones describes a world that no longer holds: the prepared decision must be
+    discarded and the guard re-run live (a fresh approval request), instead of being
+    consumed as though nothing had failed."""
+    from tools.terminal_scope import reset_terminal_scope, set_terminal_scope
+    from tools.terminal_tool_lifecycle import cleanup_vm
+
+    monkeypatch.delenv("HERMES_DESKTOP", raising=False)
+    monkeypatch.setenv("HERMES_EXEC_ASK", "1")
+    monkeypatch.setenv("TERMINAL_ENV", "local")
+    monkeypatch.setenv("TERMINAL_CWD", str(tmp_path))
+    monkeypatch.setattr("tools.approval_context._get_approval_mode", lambda: "manual")
+    monkeypatch.setattr("tools.approval._tirith_scan", lambda command: {"action": "allow"})
+    monkeypatch.setattr("agent.title_generator.maybe_auto_title", lambda *a, **kw: None)
+    key = "regate-terminal-batch"
+    # First command fails (exit 1); second would succeed if allowed to run.
+    effect = tmp_path / "effect-later.txt"
+    commands = {"failing": "rm -rf absent-first; false",
+                "later": f"rm -rf absent-second; printf ran > '{effect}'"}
+    agent = _agent()
+    published = queue.Queue()
+    approval.register_gateway_notify(key, published.put)
+    from tui_gateway import server
+    monkeypatch.setattr(server, "_sessions", {key: {
+        "session_key": key, "source": "desktop", "agent": agent, "cwd": str(tmp_path),
+    }})
+    tokens = server._set_session_context(key)
+    calls = [_call(call_id, commands[call_id]) for call_id in ("failing", "later")]
+    executed = []
+    messages = []
+    errors = []
+    agent._flush_messages_to_session_db = lambda *a, **kw: True
+
+    def started(call_id, name, args):
+        executed.append((call_id, args["command"]))
+
+    agent.tool_start_callback = started
+
+    def run():
+        try:
+            agent._execute_tool_calls(SimpleNamespace(tool_calls=calls), messages, key)
+        except BaseException as exc:
+            errors.append(exc)
+
+    with ExitStack() as scope:
+        scope.callback(reset_terminal_scope, set_terminal_scope({"TERMINAL_ENV": "local", "TERMINAL_CWD": str(tmp_path)}))
+        worker = threading.Thread(target=propagate_context_to_thread(run), daemon=True)
+        worker.start()
+        try:
+            # Preparation collects BOTH approvals before any command runs.
+            first_request = published.get(timeout=10)
+            later_request = published.get(timeout=5)
+            assert {first_request["command"], later_request["command"]} == set(commands.values())
+            assert executed == []
+            assert approval.resolve_gateway_approval(key, "once", request_id=later_request["request_id"]) == 1
+            assert approval.resolve_gateway_approval(key, "once", request_id=first_request["request_id"]) == 1
+
+            # The failing command runs and its failure is published.
+            worker_join_marker = time.monotonic()
+            while not executed and time.monotonic() - worker_join_marker < 10:
+                time.sleep(0.05)
+            assert executed == [("failing", commands["failing"])]
+
+            # The later command must NOT run on the pre-collected approval: the
+            # guard re-runs and publishes a FRESH approval request, and the
+            # shell only starts once that one is answered.
+            regated = published.get(timeout=10)
+            assert regated["command"] == commands["later"]
+            assert regated["request_id"] != later_request["request_id"]
+            assert not effect.exists()
+            assert approval.resolve_gateway_approval(key, "once", request_id=regated["request_id"]) == 1
+
+            worker.join(timeout=15)
+            assert not worker.is_alive()
+            assert errors == []
+            assert executed == [("failing", commands["failing"]), ("later", commands["later"])]
+            assert [m["tool_call_id"] for m in messages] == [c.id for c in calls]
+            results = {m["tool_call_id"]: json.loads(m["content"]) for m in messages}
+            assert results["failing"]["exit_code"] == 1
+            assert results["later"]["exit_code"] == 0
+            assert effect.exists() and effect.read_text() == "ran"
+            assert approval.list_gateway_approvals(key) == []
+        finally:
+            agent.interrupt("test cleanup")
+            approval.unregister_gateway_notify(key)
+            worker.join(timeout=5)
+            cleanup_vm(key)
+            clear_session_vars(tokens)