fix(agent): re-derive no-human batch approvals from the policy in force at execution

A Desktop terminal batch prepares every command's guard decision before the
first one runs. A decision nobody answered (/yolo, approvals.mode off, the
allowlist, a clean command) was consumed as-is, so switching YOLO or
"Approvals: off" off after the first command still ran the rest unprompted,
while the sequential path re-gates. Only human answers (and a Smart APPROVE
while Smart mode is still selected) are reused now; everything else runs the
live guard.
This commit is contained in:
John Paul Soliva
2026-09-28 19:46:07 +09:00
committed by brooklyn!
parent 0d9579ff71
commit 447fb4955d
2 changed files with 57 additions and 0 deletions

View File

@@ -232,6 +232,13 @@ def consume_prepared_guard(command, env_type, has_host_access):
or slot.guard_key != (command, env_type, has_host_access)):
return None
decision, slot.decision = slot.decision, None # single-use, even for identical calls
# Batching exists to publish the human asks together. An approval nobody answered (/yolo,
# approvals.mode off, the allowlist, a clean command) is policy, and the policy in force NOW
# governs: switching YOLO or "Approvals: off" off mid-batch must stop the later commands.
if decision is not None and decision.get("approved") and not decision.get("user_approved"):
from tools.approval_context import _get_approval_mode
if not (decision.get("smart_approved") and _get_approval_mode() == "smart"):
return None
return decision

View File

@@ -323,3 +323,53 @@ def test_failed_command_re_gates_later_prepared_approvals(tmp_path, monkeypatch)
worker.join(timeout=5)
cleanup_vm(key)
clear_session_vars(tokens)
def test_switching_yolo_off_mid_batch_re_gates_later_commands(tmp_path, monkeypatch):
"""A prepared approval nobody answered is policy: the policy in force at execution governs."""
from tools.terminal_scope import reset_terminal_scope, set_terminal_scope
from tools.terminal_tool_lifecycle import cleanup_vm
monkeypatch.setenv("HERMES_EXEC_ASK", "1")
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 = "yolo-off-terminal-batch"
agent = _agent()
agent._flush_messages_to_session_db = lambda *a, **kw: True
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(f"c{i}", f"rm -rf absent-{i}; touch ran-{i}") for i in range(3)]
# The user switches YOLO off once the first command has run.
agent.tool_start_callback = lambda call_id, name, args: call_id == "c1" and approval.disable_session_yolo(key)
messages, errors = [], []
approval.enable_session_yolo(key)
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:
for _ in range(2):
request = published.get(timeout=15)
assert approval.resolve_gateway_approval(key, "deny", request_id=request["request_id"]) == 1
worker.join(timeout=15)
assert not worker.is_alive() and errors == []
assert sorted(p.name for p in tmp_path.glob("ran-*")) == ["ran-0"]
finally:
agent.interrupt("test cleanup")
approval.unregister_gateway_notify(key)
approval.clear_session(key)
worker.join(timeout=5)
cleanup_vm(key)
clear_session_vars(tokens)