feat(approvals): add write-file approval mode

agent/agent_runtime_helpers.py

@@ -2350,17 +2350,24 @@ def invoke_tool(agent, function_name: str, function_args: dict, effective_task_i
 
     from hermes_cli.middleware import run_tool_execution_middleware
 
-    return run_tool_execution_middleware(
-        function_name,
-        function_args,
-        lambda next_args: _execute(next_args if isinstance(next_args, dict) else function_args),
-        original_args=function_args,
-        task_id=effective_task_id or "",
-        session_id=getattr(agent, "session_id", "") or "",
-        tool_call_id=tool_call_id or "",
-        turn_id=getattr(agent, "_current_turn_id", "") or "",
-        api_request_id=getattr(agent, "_current_api_request_id", "") or "",
-    )
+    try:
+        return run_tool_execution_middleware(
+            function_name,
+            function_args,
+            lambda next_args: _execute(next_args if isinstance(next_args, dict) else function_args),
+            original_args=function_args,
+            task_id=effective_task_id or "",
+            session_id=getattr(agent, "session_id", "") or "",
+            tool_call_id=tool_call_id or "",
+            turn_id=getattr(agent, "_current_turn_id", "") or "",
+            api_request_id=getattr(agent, "_current_api_request_id", "") or "",
+        )
+    finally:
+        try:
+            from tools.file_tools import revoke_file_mutation_once_capability
+            revoke_file_mutation_once_capability(tool_call_id or "")
+        except Exception:
+            pass
 
 
 

agent/tool_executor.py

@@ -423,6 +423,19 @@ def execute_tool_calls_concurrent(agent, assistant_message, messages: list, effe
             tool_call_id=getattr(tool_call, "id", "") or "",
         )
 
+        # Bind approval observability/capability identity while this call's
+        # pre-gates run. The resulting per-call capability map is propagated
+        # into its worker context below.
+        _approval_tokens = None
+        try:
+            from tools.approval import set_current_observability_context
+            _approval_tokens = set_current_observability_context(
+                turn_id=getattr(agent, "_current_turn_id", "") or "",
+                tool_call_id=getattr(tool_call, "id", "") or "",
+            )
+        except Exception:
+            pass
+
         # ── Block evaluation (BEFORE checkpoint preflight) ───────────
         # We must know whether the tool will execute before touching
         # checkpoint state (dedup slot, real snapshots).
@@ -515,6 +528,19 @@ def execute_tool_calls_concurrent(agent, assistant_message, messages: list, effe
                 except Exception:
                     pass
 
+        if block_result is not None:
+            try:
+                from tools.file_tools import revoke_file_mutation_once_capability
+                revoke_file_mutation_once_capability(getattr(tool_call, "id", "") or "")
+            except Exception:
+                pass
+        if _approval_tokens is not None:
+            try:
+                from tools.approval import reset_current_observability_context
+                reset_current_observability_context(_approval_tokens)
+            except Exception:
+                pass
+
         parsed_calls.append((tool_call, function_name, function_args, middleware_trace, block_result, blocked_by_guardrail))
 
     # ── Logging / callbacks ──────────────────────────────────────────
@@ -814,6 +840,16 @@ def execute_tool_calls_concurrent(agent, assistant_message, messages: list, effe
                     cancel_futures=abandon_executor,
                 )
     finally:
+        # Revoke every call slot after normal completion and, critically, on
+        # timeout/interrupt/submit abort. Capabilities are shared objects across
+        # copied worker contexts, so revocation here also invalidates a worker
+        # that has not yet reached registry dispatch.
+        try:
+            from tools.file_tools import revoke_file_mutation_once_capability
+            for tc, *_rest in parsed_calls:
+                revoke_file_mutation_once_capability(getattr(tc, "id", "") or "")
+        except Exception:
+            pass
         if spinner:
             # Build a summary message for the spinner stop
             completed = sum(1 for r in results if r is not None)
@@ -1126,9 +1162,13 @@ def execute_tool_calls_sequential(agent, assistant_message, messages: list, effe
         _execution_blocked = _block_msg is not None or _guardrail_block_decision is not None
 
         if _execution_blocked:
-            # Tool blocked by plugin or guardrail policy — skip counters,
-            # callbacks, checkpointing, activity mutation, and real execution.
-            pass
+            # Tool blocked after plugin approval: revoke this call's one-shot
+            # before any copied context or later retry can redeem it.
+            try:
+                from tools.file_tools import revoke_file_mutation_once_capability
+                revoke_file_mutation_once_capability(getattr(tool_call, "id", "") or "")
+            except Exception:
+                pass
         # Reset nudge counters when the relevant tool is actually used
         elif function_name == "memory":
             agent._turns_since_memory = 0

hermes_cli/config.py

@@ -2534,6 +2534,10 @@ DEFAULT_CONFIG = {
         "mode": "smart",
         "timeout": 60,
         "cron_mode": "deny",
+        # File mutation tools (write_file and patch):
+        #   allow — run normally (default; preserves existing behavior)
+        #   ask   — require the shared human approval gate, even under YOLO
+        "write_file": "allow",
         # User-defined deny rules: fnmatch globs matched against terminal
         # commands. A match blocks the command unconditionally — BEFORE the
         # --yolo / /yolo / mode=off bypass — making this the user-editable
@@ -8132,6 +8136,12 @@ def set_config_value(key: str, value: str):
             file=sys.stderr,
         )
         sys.exit(1)
+    if key == "approvals.write_file" and value not in {"allow", "ask"}:
+        print(
+            "Invalid value for approvals.write_file: expected exactly 'allow' or 'ask'.",
+            file=sys.stderr,
+        )
+        sys.exit(1)
     # Check if it's an API key (goes to .env)
     api_keys = [
         'OPENROUTER_API_KEY', 'OPENAI_API_KEY', 'ANTHROPIC_API_KEY', 'VOICE_TOOLS_OPENAI_KEY',

hermes_cli/plugins.py

@@ -2259,10 +2259,35 @@ def resolve_pre_tool_block(
     if details.action == "approve":
         try:
             from tools.approval import request_tool_approval
+            is_core_file_mutation = False
+            if tool_name in {"write_file", "patch"}:
+                from tools.file_tools import (
+                    is_core_file_mutation_entry,
+                    preflight_file_mutation,
+                )
+                is_core_file_mutation = is_core_file_mutation_entry(tool_name)
+                if is_core_file_mutation:
+                    from hermes_cli.config import cfg_get, load_config
+                    is_core_file_mutation = (
+                        cfg_get(load_config(), "approvals", "write_file", default="allow")
+                        == "ask"
+                    )
+            if is_core_file_mutation:
+                validation_error = preflight_file_mutation(
+                    tool_name, args or {}, task_id or "default"
+                )
+                if validation_error:
+                    return validation_error
             result = request_tool_approval(
                 tool_name,
                 details.message or "",
-                rule_key=details.rule_key or tool_name,
+                # A file-mutation plugin escalation shares the built-in policy
+                # grain so one session/always choice satisfies both gates.
+                rule_key=(
+                    "write_file"
+                    if is_core_file_mutation
+                    else (details.rule_key or tool_name)
+                ),
             )
         except Exception:
             # Fail-closed: if the gate itself errors, block rather than
@@ -2273,6 +2298,12 @@ def resolve_pre_tool_block(
                 result.get("message")
                 or f"BLOCKED: plugin approval required for {tool_name}"
             )
+        if is_core_file_mutation and result.get("approval_scope") == "once":
+            from tools.file_tools import grant_file_mutation_once_capability, registry
+            entry = registry.get_entry(tool_name)
+            grant_file_mutation_once_capability(
+                entry, tool_name, args or {}, tool_call_id=tool_call_id or ""
+            )
     return None
 
 

model_tools.py

@@ -1216,16 +1216,28 @@ def handle_function_call(
         # ACP/Zed edit approval runs before any file mutation.  The requester
         # is bound via ContextVar only for ACP sessions, so CLI/gateway paths
         # are unaffected when it is unset.
+        edit_block_message = None
+        edit_dispatch_allowed = False
         try:
             from acp_adapter.edit_approval import maybe_require_edit_approval
 
             edit_block_message = maybe_require_edit_approval(function_name, function_args)
             if edit_block_message is not None:
                 return edit_block_message
+            edit_dispatch_allowed = True
         except Exception as _edit_approval_err:
             logger.debug("ACP edit approval guard error: %s", _edit_approval_err)
             if function_name in {"write_file", "patch"}:
                 return json.dumps({"error": "Edit approval denied: approval guard failed"}, ensure_ascii=False)
+        finally:
+            # If this downstream guard did not hand off to core dispatch, revoke
+            # the plugin-approved one-shot even when the guard itself raised.
+            if not edit_dispatch_allowed:
+                try:
+                    from tools.file_tools import revoke_file_mutation_once_capability
+                    revoke_file_mutation_once_capability(tool_call_id or "")
+                except Exception:
+                    pass
 
         # Notify the read-loop tracker when a non-read/search tool runs,
         # so the *consecutive* counter resets (reads after other work are fine).
@@ -1295,6 +1307,11 @@ def handle_function_call(
                     reset_current_observability_context(_approval_tokens)
                 except Exception:
                     pass
+            try:
+                from tools.file_tools import revoke_file_mutation_once_capability
+                revoke_file_mutation_once_capability(tool_call_id or "")
+            except Exception:
+                pass
         duration_ms = int((time.monotonic() - _dispatch_start) * 1000)
 
         _emit_post_tool_call_hook(

tests/hermes_cli/test_set_config_value.py

@@ -108,6 +108,12 @@ class TestConfigYamlRouting:
         assert "docker" in config
         assert "terminal" not in _read_env(_isolated_hermes_home)
 
+    @pytest.mark.parametrize("value", ["deny", "always", "off", "true", "ASK "])
+    def test_write_file_approval_rejects_noncanonical_values(self, value, _isolated_hermes_home):
+        with pytest.raises(SystemExit):
+            set_config_value("approvals.write_file", value)
+        assert not (_isolated_hermes_home / "config.yaml").exists()
+
     def test_terminal_image_goes_to_config(self, _isolated_hermes_home):
         """TERMINAL_DOCKER_IMAGE doesn't match _API_KEY or _TOKEN, so config.yaml."""
         set_config_value("terminal.docker_image", "python:3.12")

tests/tools/test_write_file_approval_mode.py

@@ -0,0 +1,820 @@
+"""Narrow approvals.write_file policy for the file mutation pair."""
+
+from __future__ import annotations
+
+import json
+import os
+import subprocess
+import sys
+import threading
+from contextvars import copy_context
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch as mock_patch
+
+import pytest
+
+import tools.approval as approval
+import tools.file_tools as file_tools
+from hermes_cli.config import DEFAULT_CONFIG
+from tools.registry import registry
+
+
+@pytest.fixture(autouse=True)
+def _approval_state(monkeypatch):
+    monkeypatch.setattr(approval, "get_current_session_key", lambda default="default": "write-test")
+    monkeypatch.setattr(approval, "is_approved", lambda session, pattern: False)
+    monkeypatch.setattr(approval, "is_current_session_yolo_enabled", lambda: False)
+    monkeypatch.setattr(approval, "_YOLO_MODE_FROZEN", False)
+    monkeypatch.setattr("tools.terminal_tool._get_approval_callback", lambda: None)
+    monkeypatch.setattr(file_tools, "_check_file_reqs", lambda: True)
+
+
+def _mode(monkeypatch, mode: str) -> None:
+    monkeypatch.setattr(
+        "hermes_cli.config.load_config",
+        lambda: {"approvals": {"write_file": mode}},
+    )
+
+
+def _result(raw: str | dict) -> dict:
+    return json.loads(raw) if isinstance(raw, str) else raw
+
+
+def test_default_config_keeps_file_mutations_allowed():
+    assert DEFAULT_CONFIG["approvals"]["write_file"] == "allow"
+
+
+def test_allow_executes_write_file_and_patch_without_approval(tmp_path, monkeypatch):
+    _mode(monkeypatch, "allow")
+    monkeypatch.setattr(
+        approval,
+        "request_tool_approval",
+        lambda *args, **kwargs: pytest.fail("allow must not request approval"),
+    )
+    target = tmp_path / "example.txt"
+
+    assert "error" not in _result(registry.dispatch("write_file", {"path": str(target), "content": "old"}))
+    assert "error" not in _result(
+        registry.dispatch(
+            "patch",
+            {"mode": "replace", "path": str(target), "old_string": "old", "new_string": "new"},
+        )
+    )
+    assert target.read_text() == "new"
+
+
+@pytest.mark.parametrize("tool_name", ["write_file", "patch"])
+def test_ask_uses_shared_gate_once_and_blocks_mutation(tmp_path, monkeypatch, tool_name):
+    _mode(monkeypatch, "ask")
+    calls = []
+
+    def deny(name, reason, **kwargs):
+        calls.append((name, reason, kwargs))
+        return {"approved": False, "message": "BLOCKED: user denied write"}
+
+    monkeypatch.setattr(approval, "request_tool_approval", deny)
+    target = tmp_path / "blocked.txt"
+    if tool_name == "write_file":
+        args = {"path": str(target), "content": "new"}
+    else:
+        target.write_text("old")
+        args = {"mode": "replace", "path": str(target), "old_string": "old", "new_string": "new"}
+
+    result = _result(registry.dispatch(tool_name, args))
+
+    assert result["error"] == "BLOCKED: user denied write"
+    assert len(calls) == 1
+    assert calls[0][0] == tool_name
+    assert calls[0][2] == {"rule_key": "write_file", "honor_yolo": False}
+    assert (not target.exists()) if tool_name == "write_file" else target.read_text() == "old"
+
+
+def test_read_and_search_are_not_gated(tmp_path, monkeypatch):
+    _mode(monkeypatch, "ask")
+    monkeypatch.setattr(
+        approval,
+        "request_tool_approval",
+        lambda *args, **kwargs: pytest.fail("read-only file tools must not request approval"),
+    )
+    target = tmp_path / "example.txt"
+    target.write_text("needle")
+
+    assert "needle" in registry.dispatch("read_file", {"path": str(target)})
+    assert "needle" in registry.dispatch(
+        "search_files", {"pattern": "needle", "path": str(tmp_path)}
+    )
+
+
+def test_agent_dispatch_prompts_once_despite_execution_middleware(tmp_path, monkeypatch):
+    _mode(monkeypatch, "ask")
+    calls = []
+    monkeypatch.setattr(
+        approval,
+        "request_tool_approval",
+        lambda *args, **kwargs: calls.append((args, kwargs)) or {"approved": True, "message": None},
+    )
+    target = tmp_path / "agent-dispatch.txt"
+    from model_tools import handle_function_call
+
+    result = _result(handle_function_call("write_file", {"path": str(target), "content": "ok"}))
+
+    assert "error" not in result
+    assert target.read_text() == "ok"
+    assert len(calls) == 1
+
+
+def test_ask_cli_path_prompts_and_executes_once(tmp_path, monkeypatch):
+    _mode(monkeypatch, "ask")
+    monkeypatch.setattr(approval, "_is_interactive_cli", lambda: True)
+    monkeypatch.setattr(approval, "_is_gateway_approval_context", lambda: False)
+    prompts = []
+    monkeypatch.setattr(
+        approval,
+        "prompt_dangerous_approval",
+        lambda target, description, **kwargs: prompts.append((target, description)) or "once",
+    )
+    target = tmp_path / "cli.txt"
+
+    result = _result(registry.dispatch("write_file", {"path": str(target), "content": "ok"}))
+
+    assert "error" not in result
+    assert target.read_text() == "ok"
+    assert len(prompts) == 1
+
+
+def test_ask_gateway_path_submits_pending_and_does_not_write(tmp_path, monkeypatch):
+    _mode(monkeypatch, "ask")
+    monkeypatch.setattr(approval, "_is_interactive_cli", lambda: False)
+    monkeypatch.setattr(approval, "_is_gateway_approval_context", lambda: True)
+    submitted = []
+    monkeypatch.setattr(approval, "submit_pending", lambda session, data: submitted.append((session, data)))
+    target = tmp_path / "gateway.txt"
+
+    result = _result(registry.dispatch("write_file", {"path": str(target), "content": "no"}))
+
+    assert result["error"]
+    assert not target.exists()
+    assert len(submitted) == 1
+    assert submitted[0][1]["pattern_key"] == "plugin_rule:write_file"
+
+
+@pytest.mark.parametrize("yolo_scope", ["process", "session"])
+def test_ask_is_enforced_even_in_yolo(tmp_path, monkeypatch, yolo_scope):
+    _mode(monkeypatch, "ask")
+    monkeypatch.setattr(approval, "_YOLO_MODE_FROZEN", yolo_scope == "process")
+    monkeypatch.setattr(
+        approval, "is_current_session_yolo_enabled", lambda: yolo_scope == "session"
+    )
+    monkeypatch.setattr(approval, "_is_interactive_cli", lambda: True)
+    monkeypatch.setattr(approval, "_is_gateway_approval_context", lambda: False)
+    prompts = []
+    monkeypatch.setattr(
+        approval,
+        "prompt_dangerous_approval",
+        lambda *args, **kwargs: prompts.append(args) or "deny",
+    )
+    target = tmp_path / "yolo.txt"
+
+    result = _result(registry.dispatch("write_file", {"path": str(target), "content": "no"}))
+
+    assert result["error"]
+    assert not target.exists()
+    assert len(prompts) == 1
+
+
+def test_ask_without_human_fails_closed(tmp_path, monkeypatch):
+    _mode(monkeypatch, "ask")
+    monkeypatch.setattr(approval, "_is_interactive_cli", lambda: False)
+    monkeypatch.setattr(approval, "_is_gateway_approval_context", lambda: False)
+    monkeypatch.setattr(approval, "env_var_enabled", lambda name: False)
+    target = tmp_path / "headless.txt"
+
+    result = _result(registry.dispatch("write_file", {"path": str(target), "content": "no"}))
+
+    assert "no interactive user or gateway" in result["error"].lower()
+    assert not target.exists()
+
+
+def test_native_sensitive_path_guard_runs_before_approval(monkeypatch):
+    _mode(monkeypatch, "ask")
+    monkeypatch.setattr(
+        approval, "request_tool_approval",
+        lambda *args, **kwargs: pytest.fail("invalid write must not prompt"),
+    )
+    protected = "/etc/hermes-write-approval-test"
+
+    result = _result(registry.dispatch("write_file", {"path": protected, "content": "no"}))
+
+    assert "sensitive system path" in result["error"].lower()
+
+
+def test_native_syntax_gate_runs_before_approval(tmp_path, monkeypatch):
+    _mode(monkeypatch, "ask")
+    monkeypatch.setattr(
+        approval, "request_tool_approval",
+        lambda *args, **kwargs: pytest.fail("invalid write must not prompt"),
+    )
+    target = tmp_path / "invalid.json"
+
+    result = _result(registry.dispatch("write_file", {"path": str(target), "content": '{"broken":'}))
+
+    assert "json" in result["error"].lower()
+    assert not target.exists()
+
+
+def test_invalid_explicit_mode_fails_closed(tmp_path, monkeypatch):
+    _mode(monkeypatch, "deny")
+    target = tmp_path / "invalid.txt"
+
+    result = _result(registry.dispatch("write_file", {"path": str(target), "content": "ok"}))
+
+    assert "invalid approvals.write_file" in result["error"].lower()
+    assert not target.exists()
+
+
+def test_config_read_exception_fails_closed(tmp_path, monkeypatch):
+    monkeypatch.setattr(
+        "hermes_cli.config.load_config",
+        lambda: (_ for _ in ()).throw(OSError("unreadable config")),
+    )
+    target = tmp_path / "config-error.txt"
+
+    result = _result(registry.dispatch("write_file", {"path": str(target), "content": "no"}))
+
+    assert "could not resolve approvals.write_file" in result["error"].lower()
+    assert not target.exists()
+
+
+def test_profile_reload_replaces_permanent_write_grant(monkeypatch):
+    monkeypatch.undo()
+    approval._session_approved.clear()
+    approval._permanent_approved.clear()
+    configs = iter([
+        {"command_allowlist": ["plugin_rule:write_file"]},
+        {"command_allowlist": []},
+    ])
+    monkeypatch.setattr("hermes_cli.config.load_config", lambda: next(configs))
+
+    approval.load_permanent_allowlist()
+    assert approval.is_approved("s", "plugin_rule:write_file")
+    approval.load_permanent_allowlist()
+    assert not approval.is_approved("s", "plugin_rule:write_file")
+
+
+def test_profile_reload_config_failure_clears_prior_grant(monkeypatch):
+    monkeypatch.undo()
+    approval.load_permanent({"plugin_rule:write_file"})
+    monkeypatch.setattr(
+        "hermes_cli.config.load_config",
+        lambda: (_ for _ in ()).throw(OSError("bad profile config")),
+    )
+
+    approval.load_permanent_allowlist()
+
+    assert not approval.is_approved("s", "plugin_rule:write_file")
+
+
+def test_plugin_approve_overlap_uses_one_shared_session_decision(tmp_path, monkeypatch):
+    monkeypatch.undo()
+    _mode(monkeypatch, "ask")
+    monkeypatch.setattr(file_tools, "_check_file_reqs", lambda: True)
+    monkeypatch.setattr(approval, "get_current_session_key", lambda default="default": "overlap")
+    monkeypatch.setattr(approval, "_is_interactive_cli", lambda: True)
+    monkeypatch.setattr(approval, "_is_gateway_approval_context", lambda: False)
+    approval._session_approved.clear()
+    approval._permanent_approved.clear()
+    prompts = []
+    monkeypatch.setattr(
+        approval, "prompt_dangerous_approval",
+        lambda *args, **kwargs: prompts.append(args) or "session",
+    )
+    import hermes_cli.plugins as plugins
+    monkeypatch.setattr(
+        plugins, "_get_pre_tool_call_directive_details",
+        lambda *args, **kwargs: plugins._PreToolCallDirective(
+            action="approve", message="plugin wants confirmation", rule_key="plugin-key"
+        ),
+    )
+    target = tmp_path / "overlap.txt"
+    from model_tools import handle_function_call
+
+    result = _result(handle_function_call("write_file", {"path": str(target), "content": "ok"}))
+
+    assert "error" not in result
+    assert target.read_text() == "ok"
+    assert len(prompts) == 1
+
+
+def test_plugin_approve_once_is_consumed_without_second_prompt(tmp_path, monkeypatch):
+    """Once bridges only this dispatch; it is neither dropped nor persisted."""
+    monkeypatch.undo()
+    _mode(monkeypatch, "ask")
+    monkeypatch.setattr(file_tools, "_check_file_reqs", lambda: True)
+    monkeypatch.setattr(approval, "get_current_session_key", lambda default="default": "once")
+    monkeypatch.setattr(approval, "_is_interactive_cli", lambda: True)
+    monkeypatch.setattr(approval, "_is_gateway_approval_context", lambda: False)
+    approval._session_approved.clear()
+    approval._permanent_approved.clear()
+    prompts = []
+    monkeypatch.setattr(
+        approval, "prompt_dangerous_approval",
+        lambda *args, **kwargs: prompts.append(args) or "once",
+    )
+    import hermes_cli.plugins as plugins
+    monkeypatch.setattr(
+        plugins, "_get_pre_tool_call_directive_details",
+        lambda *args, **kwargs: plugins._PreToolCallDirective(
+            action="approve", message="plugin wants confirmation", rule_key="plugin-key"
+        ),
+    )
+    from model_tools import handle_function_call
+    first = tmp_path / "first.txt"
+    second = tmp_path / "second.txt"
+
+    assert "error" not in _result(handle_function_call("write_file", {"path": str(first), "content": "1"}))
+    assert "error" not in _result(handle_function_call("write_file", {"path": str(second), "content": "2"}))
+
+    assert len(prompts) == 2
+    assert not approval.is_approved("once", "plugin_rule:write_file")
+
+
+def test_plugin_once_capability_is_cleaned_when_downstream_guard_blocks(
+    tmp_path, monkeypatch
+):
+    monkeypatch.undo()
+    _mode(monkeypatch, "ask")
+    monkeypatch.setattr(file_tools, "_check_file_reqs", lambda: True)
+    calls = []
+    monkeypatch.setattr(
+        approval,
+        "request_tool_approval",
+        lambda *args, **kwargs: calls.append(args) or {
+            "approved": True, "message": None, "approval_scope": "once"
+        },
+    )
+    import hermes_cli.plugins as plugins
+    monkeypatch.setattr(
+        plugins,
+        "_get_pre_tool_call_directive_details",
+        lambda *args, **kwargs: plugins._PreToolCallDirective(
+            action="approve", message="plugin wants confirmation", rule_key="plugin-key"
+        ),
+    )
+    monkeypatch.setattr(
+        "acp_adapter.edit_approval.maybe_require_edit_approval",
+        lambda *args, **kwargs: "blocked after plugin",
+    )
+    from model_tools import handle_function_call
+    args = {"path": str(tmp_path / "blocked.txt"), "content": "ok"}
+
+    assert "blocked after plugin" in handle_function_call("write_file", args)
+    assert len(calls) == 1
+    monkeypatch.setattr(
+        approval,
+        "request_tool_approval",
+        lambda *args, **kwargs: calls.append(args) or {
+            "approved": False, "message": "denied"
+        },
+    )
+
+    result = _result(registry.dispatch("write_file", args))
+    assert result["error"]
+    assert len(calls) == 2
+    assert not (tmp_path / "blocked.txt").exists()
+
+
+def test_plugin_once_capability_is_cleaned_when_acp_guard_raises_before_retry(
+    tmp_path, monkeypatch
+):
+    """ACP exceptions cannot leave plugin once consent redeemable by a retry."""
+    monkeypatch.undo()
+    _mode(monkeypatch, "ask")
+    monkeypatch.setattr(file_tools, "_check_file_reqs", lambda: True)
+    calls = []
+    monkeypatch.setattr(
+        approval,
+        "request_tool_approval",
+        lambda *args, **kwargs: calls.append(args) or {
+            "approved": True,
+            "message": None,
+            "approval_scope": "once",
+        },
+    )
+    import hermes_cli.plugins as plugins
+    monkeypatch.setattr(
+        plugins,
+        "_get_pre_tool_call_directive_details",
+        lambda *args, **kwargs: plugins._PreToolCallDirective(
+            action="approve", message="plugin wants confirmation", rule_key="plugin-key"
+        ),
+    )
+    monkeypatch.setattr(
+        "acp_adapter.edit_approval.maybe_require_edit_approval",
+        lambda *args, **kwargs: (_ for _ in ()).throw(RuntimeError("hostile ACP failure")),
+    )
+    from model_tools import handle_function_call
+    target = tmp_path / "acp-exception-retry.txt"
+    args = {"path": str(target), "content": "must-not-write"}
+
+    first = _result(handle_function_call("write_file", args))
+    assert "approval guard failed" in first["error"].lower()
+    assert len(calls) == 1
+    assert not target.exists()
+
+    monkeypatch.setattr(
+        approval,
+        "request_tool_approval",
+        lambda *args, **kwargs: calls.append(args) or {
+            "approved": False,
+            "message": "denied on retry",
+        },
+    )
+
+    retry = _result(registry.dispatch("write_file", args))
+    assert retry["error"] == "denied on retry"
+    assert len(calls) == 2
+    assert not target.exists()
+
+
+@pytest.mark.parametrize("acp_outcome", ["block", "raise"])
+def test_handle_function_call_same_id_retry_cannot_replay_plugin_once_after_acp_failure(
+    tmp_path, monkeypatch, acp_outcome
+):
+    """The public dispatch ID must identify both the grant and ACP cleanup."""
+    monkeypatch.undo()
+    _mode(monkeypatch, "ask")
+    monkeypatch.setattr(file_tools, "_check_file_reqs", lambda: True)
+    import hermes_cli.plugins as plugins
+    plugin_requires_approval = True
+
+    def plugin_directive(*args, **kwargs):
+        if plugin_requires_approval:
+            return plugins._PreToolCallDirective(
+                action="approve", message="plugin confirmation", rule_key="plugin-key"
+            )
+        return plugins._PreToolCallDirective(action="allow", message=None, rule_key=None)
+
+    monkeypatch.setattr(
+        plugins, "_get_pre_tool_call_directive_details", plugin_directive
+    )
+    decisions = iter([
+        {"approved": True, "message": None, "approval_scope": "once"},
+        {"approved": False, "message": "denied on same-id retry"},
+    ])
+    prompts = []
+
+    def decide(*args, **kwargs):
+        prompts.append((args, kwargs))
+        return next(decisions)
+
+    monkeypatch.setattr(approval, "request_tool_approval", decide)
+    acp_calls = 0
+
+    def acp_guard(*args, **kwargs):
+        nonlocal acp_calls
+        acp_calls += 1
+        if acp_calls > 1:
+            return None
+        if acp_outcome == "raise":
+            raise RuntimeError("hostile ACP failure")
+        return "blocked after plugin approval"
+
+    monkeypatch.setattr(
+        "acp_adapter.edit_approval.maybe_require_edit_approval", acp_guard
+    )
+    from model_tools import handle_function_call
+    target = tmp_path / f"same-id-{acp_outcome}.txt"
+    args = {"path": str(target), "content": "must-not-write"}
+    tool_call_id = f"write-{acp_outcome}-1"
+
+    first = handle_function_call("write_file", args, tool_call_id=tool_call_id)
+    assert "blocked" in first.lower() or "approval guard failed" in first.lower()
+    assert len(prompts) == 1
+    assert not target.exists()
+
+    # Remove the plugin escalation for the retry: the built-in gate itself now
+    # denies. A stale once grant would skip that gate and mutate immediately.
+    plugin_requires_approval = False
+    retry = _result(
+        handle_function_call("write_file", args, tool_call_id=tool_call_id)
+    )
+    assert retry["error"] == "denied on same-id retry"
+    assert len(prompts) == 2
+    assert acp_calls == 2
+    assert not target.exists()
+
+
+def test_once_capability_cannot_be_replayed_from_copied_or_parent_context():
+    args = {"path": "/tmp/capability.txt", "content": "ok"}
+    entry = registry.get_entry("write_file")
+    reset_token = file_tools.grant_file_mutation_once_capability(entry, "write_file", args)
+    sibling = copy_context()
+    try:
+        assert file_tools._consume_file_mutation_once_capability(entry, "write_file", args)
+        assert not sibling.run(
+            file_tools._consume_file_mutation_once_capability, entry, "write_file", args
+        )
+        assert not file_tools._consume_file_mutation_once_capability(entry, "write_file", args)
+    finally:
+        file_tools.reset_file_mutation_once_capability(reset_token)
+
+
+def test_once_capability_is_bound_to_exact_entry_tool_and_arguments():
+    args = {"path": "/tmp/capability.txt", "content": "ok"}
+    entry = registry.get_entry("write_file")
+    reset_token = file_tools.grant_file_mutation_once_capability(entry, "write_file", args)
+    try:
+        assert not file_tools._consume_file_mutation_once_capability(
+            registry.get_entry("patch"), "write_file", args
+        )
+        assert not file_tools._consume_file_mutation_once_capability(entry, "patch", args)
+        assert not file_tools._consume_file_mutation_once_capability(
+            entry, "write_file", dict(args)
+        )
+        args["content"] = "mutated"
+        assert not file_tools._consume_file_mutation_once_capability(entry, "write_file", args)
+    finally:
+        file_tools.reset_file_mutation_once_capability(reset_token)
+
+
+def test_same_named_custom_registry_tool_never_uses_builtin_gate(monkeypatch):
+    from tools.registry import ToolRegistry
+    custom = ToolRegistry()
+    custom.register(
+        name="write_file", toolset="custom", schema={},
+        handler=lambda args, **kwargs: "custom-ok",
+    )
+    _mode(monkeypatch, "ask")
+    monkeypatch.setattr(
+        approval, "request_tool_approval",
+        lambda *args, **kwargs: pytest.fail("custom tool is not the core mutation tool"),
+    )
+
+    assert custom.dispatch("write_file", {}) == "custom-ok"
+
+
+def test_symlink_swap_during_write_approval_is_revalidated(tmp_path, monkeypatch):
+    _mode(monkeypatch, "ask")
+    safe = tmp_path / "safe.txt"
+    sensitive = tmp_path / "sensitive.txt"
+    safe.write_text("safe")
+    sensitive.write_text("secret")
+    link = tmp_path / "target.txt"
+    link.symlink_to(safe)
+
+    def approve_and_swap(*args, **kwargs):
+        link.unlink()
+        link.symlink_to(sensitive)
+        return {"approved": True, "message": None}
+
+    monkeypatch.setattr(approval, "request_tool_approval", approve_and_swap)
+    real_sensitive = file_tools._check_sensitive_path
+    monkeypatch.setattr(
+        file_tools, "_check_sensitive_path",
+        lambda path, task_id="default": (
+            "Refusing swapped sensitive target"
+            if os.path.realpath(path) == str(sensitive)
+            else real_sensitive(path, task_id)
+        ),
+    )
+
+    result = _result(registry.dispatch("write_file", {"path": str(link), "content": "changed"}))
+
+    assert "swapped sensitive" in result["error"].lower()
+    assert sensitive.read_text() == "secret"
+
+
+def test_v4a_uses_canonical_target_when_symlink_swaps_at_apply_boundary(
+    tmp_path, monkeypatch
+):
+    _mode(monkeypatch, "allow")
+    safe = tmp_path / "safe.txt"
+    sensitive = tmp_path / "sensitive.txt"
+    safe.write_text("old")
+    sensitive.write_text("old")
+    link = tmp_path / "target.txt"
+    link.symlink_to(safe)
+    patch_body = (
+        "*** Begin Patch\n"
+        f"*** Update File: {link}\n"
+        "@@\n-old\n+new\n"
+        "*** End Patch"
+    )
+    from tools.file_operations import ShellFileOperations
+    real_apply = ShellFileOperations.patch_v4a_operations
+
+    def swap_then_apply(self, operations):
+        link.unlink()
+        link.symlink_to(sensitive)
+        return real_apply(self, operations)
+
+    monkeypatch.setattr(ShellFileOperations, "patch_v4a_operations", swap_then_apply)
+
+    result = _result(registry.dispatch("patch", {"mode": "patch", "patch": patch_body}))
+
+    assert "error" not in result
+    assert safe.read_text() == "new"
+    assert sensitive.read_text() == "old"
+
+
+def test_real_cli_rejects_invalid_write_approval_mode(tmp_path):
+    env = os.environ.copy()
+    env["HERMES_HOME"] = str(tmp_path)
+    repo = os.path.dirname(os.path.dirname(os.path.dirname(__file__)))
+    proc = subprocess.run(
+        [sys.executable, "-m", "hermes_cli.main", "config", "set", "approvals.write_file", "deny"],
+        cwd=repo, env=env, text=True, capture_output=True,
+    )
+
+    assert proc.returncode != 0
+    assert "allow" in (proc.stdout + proc.stderr)
+    assert not (tmp_path / "config.yaml").exists()
+
+
+def test_permanent_grants_are_concurrent_hermes_home_scoped(tmp_path, monkeypatch):
+    monkeypatch.undo()
+    from hermes_constants import reset_hermes_home_override, set_hermes_home_override
+    barrier = threading.Barrier(2)
+    results = {}
+    home_a, home_b = tmp_path / "a", tmp_path / "b"
+
+    def worker(name, home, grant):
+        token = set_hermes_home_override(home)
+        try:
+            approval.load_permanent({"plugin_rule:write_file"} if grant else set())
+            barrier.wait()
+            results[name] = approval.is_approved("shared-session", "plugin_rule:write_file")
+        finally:
+            reset_hermes_home_override(token)
+
+    a = threading.Thread(target=worker, args=("a", home_a, True))
+    b = threading.Thread(target=worker, args=("b", home_b, False))
+    a.start()
+    b.start()
+    a.join()
+    b.join()
+
+    assert results == {"a": True, "b": False}
+
+
+def test_always_is_shared_across_write_file_and_patch(tmp_path, monkeypatch):
+    monkeypatch.undo()
+    _mode(monkeypatch, "ask")
+    monkeypatch.setattr(file_tools, "_check_file_reqs", lambda: True)
+    monkeypatch.setattr(approval, "get_current_session_key", lambda default="default": "shared")
+    monkeypatch.setattr(approval, "_is_interactive_cli", lambda: True)
+    monkeypatch.setattr(approval, "_is_gateway_approval_context", lambda: False)
+    monkeypatch.setattr(approval, "save_permanent_allowlist", lambda patterns: None)
+    approval._session_approved.clear()
+    approval._permanent_approved.clear()
+    prompts = []
+    monkeypatch.setattr(
+        approval, "prompt_dangerous_approval",
+        lambda *args, **kwargs: prompts.append(args) or "always",
+    )
+    target = tmp_path / "shared.txt"
+
+    assert "error" not in _result(registry.dispatch("write_file", {"path": str(target), "content": "old"}))
+    assert "error" not in _result(registry.dispatch("patch", {
+        "mode": "replace", "path": str(target), "old_string": "old", "new_string": "new",
+    }))
+
+    assert target.read_text() == "new"
+    assert len(prompts) == 1
+
+
+def _write_agent(monkeypatch):
+    """Build the real AIAgent executor with only the file tool exposed."""
+    from run_agent import AIAgent
+
+    tool_defs = [{
+        "type": "function",
+        "function": {
+            "name": "write_file",
+            "description": "write",
+            "parameters": {"type": "object", "properties": {}},
+        },
+    }]
+    with (
+        mock_patch("run_agent.get_tool_definitions", return_value=tool_defs),
+        mock_patch("run_agent.check_toolset_requirements", return_value={}),
+        mock_patch(
+            "hermes_cli.config.load_config",
+            return_value={"approvals": {"write_file": "ask"}},
+        ),
+        mock_patch("run_agent.OpenAI"),
+    ):
+        agent = AIAgent(
+            api_key="test-key-1234567890",
+            base_url="https://openrouter.ai/api/v1",
+            quiet_mode=True,
+            skip_context_files=True,
+            skip_memory=True,
+        )
+    agent.client = MagicMock()
+    setattr(agent, "_cached_system_prompt", "test")
+    setattr(agent, "_use_prompt_caching", False)
+    return agent
+
+
+def _write_call(call_id, path, content="ok"):
+    return SimpleNamespace(
+        id=call_id,
+        type="function",
+        function=SimpleNamespace(
+            name="write_file",
+            arguments=json.dumps({"path": str(path), "content": content}),
+        ),
+    )
+
+
+def _approve_plugin(monkeypatch):
+    import hermes_cli.plugins as plugins
+
+    monkeypatch.setattr(
+        plugins,
+        "_get_pre_tool_call_directive_details",
+        lambda *args, **kwargs: plugins._PreToolCallDirective(
+            action="approve", message="plugin confirmation", rule_key="plugin-key"
+        ),
+    )
+
+
+def test_real_concurrent_agent_two_writes_prompt_once_per_call(tmp_path, monkeypatch):
+    """Batch pre-gates must not overwrite the first call's one-shot grant."""
+    monkeypatch.undo()
+    monkeypatch.setattr(file_tools, "_check_file_reqs", lambda: True)
+    _approve_plugin(monkeypatch)
+    prompts = []
+    monkeypatch.setattr(
+        approval,
+        "request_tool_approval",
+        lambda *args, **kwargs: prompts.append((args, kwargs)) or {
+            "approved": True, "message": None, "approval_scope": "once"
+        },
+    )
+    agent = _write_agent(monkeypatch)
+    _mode(monkeypatch, "ask")
+    first, second = tmp_path / "first.txt", tmp_path / "second.txt"
+    message = SimpleNamespace(
+        content="",
+        tool_calls=[
+            _write_call("write-1", first, "one"),
+            _write_call("write-2", second, "two"),
+        ],
+    )
+    results = []
+
+    agent._execute_tool_calls_concurrent(message, results, "task-write")
+
+    assert first.read_text() == "one"
+    assert second.read_text() == "two"
+    assert len(prompts) == 2
+    assert [item["tool_call_id"] for item in results] == ["write-1", "write-2"]
+
+
+def test_real_sequential_guardrail_block_revokes_call_grant_before_retry(
+    tmp_path, monkeypatch
+):
+    """A pre-gate approval cannot survive a downstream guardrail block."""
+    monkeypatch.undo()
+    monkeypatch.setattr(file_tools, "_check_file_reqs", lambda: True)
+    _approve_plugin(monkeypatch)
+    prompts = []
+    monkeypatch.setattr(
+        approval,
+        "request_tool_approval",
+        lambda *args, **kwargs: prompts.append((args, kwargs)) or {
+            "approved": True, "message": None, "approval_scope": "once"
+        },
+    )
+    agent = _write_agent(monkeypatch)
+    _mode(monkeypatch, "ask")
+    target = tmp_path / "retry.txt"
+    call = _write_call("blocked-write", target)
+    message = SimpleNamespace(content="", tool_calls=[call])
+    blocked_results = []
+    guardrails = getattr(agent, "_tool_guardrails")
+    guardrails.before_call = lambda *args, **kwargs: SimpleNamespace(
+        allows_execution=False,
+        message="blocked downstream",
+        action="block",
+        code="test",
+    )
+    agent._guardrail_block_result = lambda decision: json.dumps({"error": decision.message})
+
+    agent._execute_tool_calls_sequential(message, blocked_results, "task-write")
+    assert not target.exists()
+    assert len(prompts) == 1
+
+    guardrails.before_call = lambda *args, **kwargs: SimpleNamespace(
+        allows_execution=True,
+        message=None,
+        action="allow",
+        code=None,
+    )
+    retry_results = []
+    agent._execute_tool_calls_sequential(message, retry_results, "task-write")
+
+    assert target.read_text() == "ok"
+    assert len(prompts) == 2

tools/approval.py

@@ -21,6 +21,7 @@ import tempfile
 import threading
 import time
 import unicodedata
+from collections.abc import MutableSet
 from typing import Optional
 from hermes_cli.config import cfg_get
 
@@ -1478,7 +1479,51 @@ _lock = threading.Lock()
 _pending: dict[str, dict] = {}
 _session_approved: dict[str, set] = {}
 _session_yolo: set[str] = set()
-_permanent_approved: set = set()
+
+
+class _ProfileScopedPermanentApprovals(MutableSet):
+    """Set-compatible permanent grants isolated by context-local Hermes home."""
+
+    def __init__(self):
+        self._by_home: dict[str, set] = {}
+
+    @staticmethod
+    def _key() -> str:
+        from hermes_constants import get_hermes_home
+        try:
+            return str(get_hermes_home().resolve())
+        except OSError:
+            return str(get_hermes_home().absolute())
+
+    def _current(self) -> set:
+        return self._by_home.setdefault(self._key(), set())
+
+    def __contains__(self, value):
+        return value in self._current()
+
+    def __iter__(self):
+        return iter(tuple(self._current()))
+
+    def __len__(self):
+        return len(self._current())
+
+    def add(self, value):
+        self._current().add(value)
+
+    def discard(self, value):
+        self._current().discard(value)
+
+    def clear(self):
+        self._current().clear()
+
+    def update(self, values):
+        self._current().update(values)
+
+    def copy(self) -> set:
+        return self._current().copy()
+
+
+_permanent_approved = _ProfileScopedPermanentApprovals()
 
 # =========================================================================
 # Blocking gateway approval (mirrors CLI's synchronous input() flow)
@@ -1652,8 +1697,9 @@ def approve_permanent(pattern_key: str):
 
 
 def load_permanent(patterns: set):
-    """Bulk-load permanent allowlist entries from config."""
+    """Replace permanent entries with the active profile's allowlist."""
     with _lock:
+        _permanent_approved.clear()
         _permanent_approved.update(patterns)
 
 
@@ -1709,15 +1755,16 @@ def load_permanent_allowlist() -> set:
         from hermes_cli.config import load_config
         config = load_config()
         patterns = set(config.get("command_allowlist", []) or [])
-        if patterns:
-            load_permanent(patterns)
+        load_permanent(patterns)
         return patterns
     except Exception as e:
+        # Profile transition/load failure must not retain the previous profile's grants.
+        load_permanent(set())
         logger.warning("Failed to load permanent allowlist: %s", e)
         return set()
 
 
-def save_permanent_allowlist(patterns: set):
+def save_permanent_allowlist(patterns):
     """Save permanently allowed command patterns to config."""
     try:
         from hermes_cli.config import load_config, save_config
@@ -2107,6 +2154,7 @@ def _run_approval_gate(
     autoapprove_log_prefix: str,
     fail_closed_when_no_human: bool = False,
     no_human_block_message: str = "",
+    honor_yolo: bool = True,
 ) -> dict:
     """Shared human-approval gate for a flagged action (command or tool).
 
@@ -2151,7 +2199,7 @@ def _run_approval_gate(
     # --yolo bypasses all approval prompts (session- or process-scoped).
     # Hardline blocks are handled by the caller BEFORE this gate, so yolo
     # here only skips the recoverable approval layer.
-    if _YOLO_MODE_FROZEN or is_current_session_yolo_enabled():
+    if honor_yolo and (_YOLO_MODE_FROZEN or is_current_session_yolo_enabled()):
         return {"approved": True, "message": None}
 
     session_key = get_current_session_key()
@@ -2269,7 +2317,7 @@ def _run_approval_gate(
                 approve_session(session_key, pattern_key)
                 approve_permanent(pattern_key)
                 save_permanent_allowlist(_permanent_approved)
-            return {"approved": True, "message": None}
+            return {"approved": True, "message": None, "approval_scope": choice}
 
         # No notify callback (e.g. API server without an attached chat):
         # queue for /approve /deny review, agent sees approval_required.
@@ -2312,7 +2360,7 @@ def _run_approval_gate(
         approve_permanent(pattern_key)
         save_permanent_allowlist(_permanent_approved)
 
-    return {"approved": True, "message": None}
+    return {"approved": True, "message": None, "approval_scope": choice}
 
 
 def _should_skip_container_guards(env_type: str, has_host_access: bool = False) -> bool:
@@ -2405,14 +2453,14 @@ def request_tool_approval(
     *,
     rule_key: str = "",
     approval_callback=None,
+    honor_yolo: bool = True,
 ) -> dict:
     """Escalate an arbitrary tool call to the human-approval gate.
 
-    This is the entry point for a plugin ``pre_tool_call`` hook that returns
-    ``{"action": "approve", "message": ...}``: instead of the plugin vetoing
-    the call (``action: block``) or silently allowing it, it asks the SAME
-    human gate that Tier-2 dangerous shell patterns use. The LLM cannot skip
-    or bypass this — the tool call is intercepted before execution.
+    This is the entry point for any tool-specific policy (including a plugin
+    ``pre_tool_call`` hook returning ``{"action": "approve", ...}``) that must
+    use the SAME human gate as Tier-2 dangerous shell patterns. The LLM cannot
+    skip this — the tool call is intercepted before execution.
 
     It reuses the existing approval primitives (session/permanent allowlist,
     ``prompt_dangerous_approval`` for CLI, ``submit_pending`` for the gateway
@@ -2432,6 +2480,8 @@ def request_tool_approval(
             on the same tool).
         approval_callback: Optional CLI callback for interactive prompts
             (same contract as ``check_dangerous_command``).
+        honor_yolo: When false, this explicit policy still asks under process-
+            or session-scoped YOLO. Used by ``approvals.write_file: ask``.
 
     Returns:
         ``{"approved": True, "message": None}`` when allowed, or
@@ -2481,8 +2531,9 @@ def request_tool_approval(
         no_human_block_message=(
             f"BLOCKED: Tool '{tool_name}' requires approval ({description}) "
             "but no interactive user or gateway is present to approve it. "
-            "A plugin flagged this action for human confirmation."
+            "This action requires human confirmation."
         ),
+        honor_yolo=honor_yolo,
     )
 
 

tools/file_operations.py

@@ -1697,9 +1697,12 @@ class ShellFileOperations(FileOperations):
         if parse_error:
             return PatchResult(error=f"Failed to parse patch: {parse_error}")
         
-        # Apply operations
-        result = apply_v4a_operations(operations, self)
-        return result
+        return self.patch_v4a_operations(operations)
+
+    def patch_v4a_operations(self, operations) -> PatchResult:
+        """Apply already parsed operations whose paths may be canonicalized."""
+        from tools.patch_parser import apply_v4a_operations
+        return apply_v4a_operations(operations, self)
     
     def _check_lint(self, path: str, content: Optional[str] = None) -> LintResult:
         """

tools/file_tools.py

@@ -8,6 +8,10 @@ import os
 import posixpath
 import sys
 import threading
+from contextvars import ContextVar
+from dataclasses import dataclass, field
+import hashlib
+import json as _json
 from pathlib import Path, PurePosixPath
 
 from agent.file_safety import get_read_block_error
@@ -1664,9 +1668,207 @@ def _mark_verification_stale(
         logger.debug("verification stale marker failed", exc_info=True)
 
 
+def preflight_file_mutation(tool_name: str, args: dict, task_id: str = "default") -> str | None:
+    """Run native file safety/shape checks without performing a mutation."""
+    args = args if isinstance(args, dict) else {}
+    cross_profile = bool(args.get("cross_profile", False))
+    if tool_name == "write_file":
+        path, content = args.get("path"), args.get("content")
+        if not isinstance(path, str) or not path:
+            return "path required"
+        if not isinstance(content, str):
+            return "content must be a string"
+        paths = [path]
+        if _is_internal_file_tool_content(content):
+            return "Refusing to write internal read_file display text as file content."
+        from tools.file_operations import LINTERS_INPROC, _FAIL_CLOSED_INPROC_EXTS
+        ext = os.path.splitext(path)[1].lower()
+        linter = LINTERS_INPROC.get(ext) if ext in _FAIL_CLOSED_INPROC_EXTS else None
+        if linter is not None:
+            ok, lint_err = linter(content)
+            if not ok and lint_err != "__SKIP__":
+                return f"Refusing to write '{path}': candidate content fails {ext} syntax validation ({lint_err})."
+    elif tool_name == "patch":
+        mode = args.get("mode", "replace")
+        if mode == "replace":
+            path = args.get("path")
+            if not isinstance(path, str) or not path:
+                return "path required"
+            if args.get("old_string") is None or args.get("new_string") is None:
+                return "old_string and new_string required"
+            paths = [path]
+        elif mode == "patch":
+            patch_content = args.get("patch")
+            if not isinstance(patch_content, str) or not patch_content:
+                return "patch content required"
+            from tools.patch_parser import parse_v4a_patch
+            operations, parse_error = parse_v4a_patch(patch_content)
+            if parse_error:
+                return f"Failed to parse patch: {parse_error}"
+            paths = []
+            for operation in operations:
+                for attr in ("path", "new_path"):
+                    value = getattr(operation, attr, None)
+                    if value:
+                        paths.append(str(value))
+            from tools.path_security import has_traversal_component
+            for candidate in paths:
+                if has_traversal_component(candidate):
+                    return f"V4A patch header contains '..' traversal: {candidate!r}."
+        else:
+            return f"Unknown mode: {mode}"
+    else:
+        return None
+    for candidate in paths:
+        sensitive_err = _check_sensitive_path(candidate, task_id)
+        if sensitive_err:
+            return sensitive_err
+        if not cross_profile:
+            cross_warning = _check_cross_profile_path(candidate, task_id)
+            if cross_warning:
+                return cross_warning
+    return None
+
+
+def _mutation_args_fingerprint(args: dict) -> bytes:
+    encoded = _json.dumps(
+        args, sort_keys=True, separators=(",", ":"), ensure_ascii=False, default=repr
+    ).encode("utf-8")
+    return hashlib.sha256(encoded).digest()
+
+
+@dataclass
+class _FileMutationOnceCapability:
+    entry: object
+    handler: object
+    tool_name: str
+    args: dict
+    fingerprint: bytes
+    consumed: bool = False
+    lock: threading.Lock = field(default_factory=threading.Lock)
+
+    def consume(self, entry: object, tool_name: str, args: dict) -> bool:
+        with self.lock:
+            if self.consumed:
+                return False
+            if (
+                entry is not self.entry
+                or getattr(entry, "handler", None) is not self.handler
+                or tool_name != self.tool_name
+                or args is not self.args
+                or _mutation_args_fingerprint(args) != self.fingerprint
+            ):
+                return False
+            self.consumed = True
+            return True
+
+    def revoke(self) -> None:
+        with self.lock:
+            self.consumed = True
+
+
+_file_mutation_once_capabilities: ContextVar[dict[str, _FileMutationOnceCapability]] = ContextVar(
+    "file_mutation_once_capabilities", default={}
+)
+
+
+def _file_mutation_dispatch_id(tool_call_id: str = "") -> str:
+    if tool_call_id:
+        return tool_call_id
+    try:
+        from tools.approval import _approval_tool_call_id
+        return _approval_tool_call_id.get() or "__direct_dispatch__"
+    except Exception:
+        return "__direct_dispatch__"
+
+
+def grant_file_mutation_once_capability(
+    entry, tool_name: str, args: dict, tool_call_id: str = ""
+):
+    """Grant one capability in a call-specific slot shared with copied contexts."""
+    capability = _FileMutationOnceCapability(
+        entry=entry,
+        handler=getattr(entry, "handler", None),
+        tool_name=tool_name,
+        args=args,
+        fingerprint=_mutation_args_fingerprint(args),
+    )
+    capabilities = dict(_file_mutation_once_capabilities.get())
+    capabilities[_file_mutation_dispatch_id(tool_call_id)] = capability
+    return _file_mutation_once_capabilities.set(capabilities)
+
+
+def reset_file_mutation_once_capability(token) -> None:
+    for capability in _file_mutation_once_capabilities.get().values():
+        capability.revoke()
+    _file_mutation_once_capabilities.reset(token)
+
+
+def revoke_file_mutation_once_capability(tool_call_id: str = "") -> None:
+    capability = _file_mutation_once_capabilities.get().get(
+        _file_mutation_dispatch_id(tool_call_id)
+    )
+    if capability is not None:
+        capability.revoke()
+
+
+def _consume_file_mutation_once_capability(entry, tool_name: str, args: dict) -> bool:
+    dispatch_id = _file_mutation_dispatch_id()
+    capabilities = _file_mutation_once_capabilities.get()
+    capability = capabilities.get(dispatch_id)
+    consumed = bool(capability and capability.consume(entry, tool_name, args))
+    if consumed:
+        remaining = dict(capabilities)
+        remaining.pop(dispatch_id, None)
+        _file_mutation_once_capabilities.set(remaining)
+    return consumed
+
+
+def is_core_file_mutation_entry(tool_name: str) -> bool:
+    """Identify the actual core registry entry, not a same-named replacement."""
+    expected = {"write_file": _handle_write_file, "patch": _handle_patch}.get(tool_name)
+    entry = registry.get_entry(tool_name)
+    return expected is not None and entry is not None and entry.handler is expected
+
+
+def _require_file_mutation_approval(tool_name: str, args: dict) -> str | None:
+    """Apply the narrow write policy after validation and before mutation."""
+    try:
+        from hermes_cli.config import cfg_get, load_config
+        mode = str(
+            cfg_get(load_config(), "approvals", "write_file", default="allow")
+        ).strip().lower()
+    except Exception as exc:
+        return f"BLOCKED: could not resolve approvals.write_file setting: {exc}"
+    if mode not in {"allow", "ask"}:
+        return (
+            f"BLOCKED: invalid approvals.write_file mode {mode!r}; "
+            "expected 'allow' or 'ask'"
+        )
+    if mode == "allow":
+        return None
+    entry = registry.get_entry(tool_name)
+    if entry is not None and _consume_file_mutation_once_capability(entry, tool_name, args):
+        return None
+    try:
+        from tools.approval import request_tool_approval
+        result = request_tool_approval(
+            tool_name,
+            f"File mutation requested via {tool_name}",
+            rule_key="write_file",
+            honor_yolo=False,
+        )
+    except Exception as exc:
+        return f"BLOCKED: write tool approval gate failed for {tool_name}: {exc}"
+    if result.get("approved"):
+        return None
+    return str(result.get("message") or f"BLOCKED: approval required for {tool_name}")
+
+
 def write_file_tool(path: str, content: str, task_id: str = "default",
                     cross_profile: bool = False,
-                    session_id: str | None = None) -> str:
+                    session_id: str | None = None,
+                    _dispatch_args: dict | None = None) -> str:
     """Write content to a file.
 
     ``cross_profile`` opts out of the soft cross-Hermes-profile guard. The
@@ -1688,6 +1890,21 @@ def write_file_tool(path: str, content: str, task_id: str = "default",
             "Strip read_file line-number prefixes or reconstruct the intended "
             "file contents before writing."
         )
+    # Native syntax validation precedes the approval prompt. The lower-level
+    # writer repeats this at the side-effect boundary (defense in depth).
+    from tools.file_operations import LINTERS_INPROC, _FAIL_CLOSED_INPROC_EXTS
+    ext = os.path.splitext(path)[1].lower()
+    inproc_linter = LINTERS_INPROC.get(ext) if ext in _FAIL_CLOSED_INPROC_EXTS else None
+    if inproc_linter is not None:
+        ok, lint_err = inproc_linter(content)
+        if not ok and lint_err != "__SKIP__":
+            return tool_error(
+                f"Refusing to write '{path}': candidate content fails {ext} "
+                f"syntax validation ({lint_err}). The file was NOT created or modified."
+            )
+    approval_error = _require_file_mutation_approval("write_file", _dispatch_args or {})
+    if approval_error:
+        return tool_error(approval_error)
     try:
         # Resolve once for the registry lock + stale check.  Failures here
         # fall back to the legacy path — write proceeds, per-task staleness
@@ -1698,6 +1915,13 @@ def write_file_tool(path: str, content: str, task_id: str = "default",
             _resolved = None
 
         if _resolved is None:
+            final_error = preflight_file_mutation(
+                "write_file",
+                {"path": path, "content": content, "cross_profile": cross_profile},
+                task_id,
+            )
+            if final_error:
+                return tool_error(final_error)
             stale_warning = _check_file_staleness(path, task_id)
             file_ops = _get_file_ops(task_id)
             result = file_ops.write_file(path, content)
@@ -1713,6 +1937,13 @@ def write_file_tool(path: str, content: str, task_id: str = "default",
         # subagents can't interleave on the same file.  Different paths
         # remain fully parallel.
         with file_state.lock_path(_resolved):
+            final_error = preflight_file_mutation(
+                "write_file",
+                {"path": path, "content": content, "cross_profile": cross_profile},
+                task_id,
+            )
+            if final_error:
+                return tool_error(final_error)
             # Cross-agent staleness wins over per-task warning when both
             # fire — its message names the sibling subagent.
             cross_warning = file_state.check_stale(task_id, _resolved)
@@ -1750,7 +1981,8 @@ def write_file_tool(path: str, content: str, task_id: str = "default",
 def patch_tool(mode: str = "replace", path: str = None, old_string: str = None,
                new_string: str = None, replace_all: bool = False, patch: str = None,
                task_id: str = "default", cross_profile: bool = False,
-               session_id: str | None = None) -> str:
+               session_id: str | None = None,
+               _dispatch_args: dict | None = None) -> str:
     """Patch a file using replace mode or V4A patch format.
 
     ``cross_profile`` opts out of the soft cross-Hermes-profile guard for
@@ -1810,6 +2042,25 @@ def patch_tool(mode: str = "replace", path: str = None, old_string: str = None,
             cross_warning = _check_cross_profile_path(_p, task_id)
             if cross_warning:
                 return tool_error(cross_warning)
+    # Parse/shape checks above and protected/cross-profile checks must all fail
+    # before asking. No file side effect has occurred yet.
+    if mode == "replace":
+        if not path:
+            return tool_error("path required")
+        if old_string is None or new_string is None:
+            return tool_error("old_string and new_string required")
+    elif mode == "patch":
+        if not patch:
+            return tool_error("patch content required")
+        from tools.patch_parser import parse_v4a_patch
+        _operations, parse_error = parse_v4a_patch(patch)
+        if parse_error:
+            return tool_error(f"Failed to parse patch: {parse_error}")
+    else:
+        return tool_error(f"Unknown mode: {mode}")
+    approval_error = _require_file_mutation_approval("patch", _dispatch_args or {})
+    if approval_error:
+        return tool_error(approval_error)
     try:
         # Resolve paths for locking.  Ordered + deduplicated so concurrent
         # callers lock in the same order — prevents deadlock on overlapping
@@ -1834,6 +2085,15 @@ def patch_tool(mode: str = "replace", path: str = None, old_string: str = None,
             for _r in _resolved_paths:
                 _locks.enter_context(file_state.lock_path(_r))
 
+            final_args = {
+                "mode": mode, "path": path, "old_string": old_string,
+                "new_string": new_string, "replace_all": replace_all,
+                "patch": patch, "cross_profile": cross_profile,
+            }
+            final_error = preflight_file_mutation("patch", final_args, task_id)
+            if final_error:
+                return tool_error(final_error)
+
             # Collect warnings — cross-agent registry first (names sibling),
             # then per-task tracker as a fallback.
             stale_warnings: list[str] = []
@@ -1870,7 +2130,25 @@ def patch_tool(mode: str = "replace", path: str = None, old_string: str = None,
             elif mode == "patch":
                 if not patch:
                     return tool_error("patch content required")
-                result = file_ops.patch_v4a(patch)
+                from tools.patch_parser import parse_v4a_patch
+                operations, parse_error = parse_v4a_patch(patch)
+                if parse_error:
+                    return tool_error(f"Failed to parse patch: {parse_error}")
+                for operation in operations:
+                    operation.file_path = (
+                        _path_to_resolved.get(operation.file_path) or operation.file_path
+                    )
+                    if operation.new_path:
+                        operation.new_path = (
+                            _path_to_resolved.get(operation.new_path) or operation.new_path
+                        )
+                apply_operations = getattr(type(file_ops), "patch_v4a_operations", None)
+                if apply_operations is None:
+                    # Compatibility for third-party/test FileOperations doubles
+                    # that still implement only the historical text API.
+                    result = file_ops.patch_v4a(patch)
+                else:
+                    result = apply_operations(file_ops, operations)
             else:
                 return tool_error(f"Unknown mode: {mode}")
 
@@ -2171,6 +2449,7 @@ def _handle_write_file(args, **kw):
         path=args["path"], content=args["content"], task_id=tid,
         cross_profile=bool(args.get("cross_profile", False)),
         session_id=kw.get("session_id"),
+        _dispatch_args=args,
     )
 
 
@@ -2182,6 +2461,7 @@ def _handle_patch(args, **kw):
         replace_all=args.get("replace_all", False), patch=args.get("patch"), task_id=tid,
         cross_profile=bool(args.get("cross_profile", False)),
         session_id=kw.get("session_id"),
+        _dispatch_args=args,
     )
 
 

tools/registry.py

@@ -26,7 +26,6 @@ from typing import Callable, Dict, List, Optional, Set
 
 logger = logging.getLogger(__name__)
 
-
 def _is_registry_register_call(node: ast.AST) -> bool:
     """Return True when *node* is a ``registry.register(...)`` call expression."""
     if not isinstance(node, ast.Expr) or not isinstance(node.value, ast.Call):

website/docs/user-guide/configuration.md

@@ -1878,7 +1878,8 @@ Control how Hermes handles potentially dangerous commands:
 
 ```yaml
 approvals:
-  mode: smart   # smart | manual | off
+  mode: smart        # smart | manual | off
+  write_file: allow  # allow | ask; applies to write_file and patch
 ```
 
 | Mode | Behavior |
@@ -1889,6 +1890,8 @@ approvals:
 
 Smart mode is particularly useful for reducing approval fatigue — it lets the agent work more autonomously on safe operations while still catching genuinely destructive commands.
 
+`approvals.write_file` is a separate, narrow file-mutation policy. Its default is `allow` for backward compatibility. Set it to `ask` to route both `write_file` and `patch` through the shared CLI/gateway/Desktop/TUI human gate, even in YOLO mode. It does not affect `read_file`, `search_files`, memory or skill write approval. Protected-path, cross-profile, traversal, malformed-request, and fail-closed syntax checks run before any prompt and are repeated where applicable at the mutation boundary.
+
 :::warning
 Setting `approvals.mode: off` disables all safety checks for terminal commands. Only use this in trusted, sandboxed environments.
 :::

website/docs/user-guide/security.md

@@ -33,6 +33,7 @@ approvals:
   mode: smart                     # smart | manual | off
   timeout: 60                     # seconds to wait for user response (default: 60)
   cron_mode: deny                 # deny | approve — what cron jobs do when they hit a dangerous command
+  write_file: allow               # allow | ask — approval for write_file and patch
   mcp_reload_confirm: true        # /reload-mcp asks before invalidating the MCP tool cache
   destructive_slash_confirm: true # /clear, /new, /reset, /undo prompt before discarding state
 ```
@@ -44,6 +45,7 @@ The full set of keys:
 | `mode` | `smart` | Approval policy for dangerous shell commands — see the table below. |
 | `timeout` | `60` | Seconds Hermes waits for an approval reply before timing out. |
 | `cron_mode` | `deny` | How [cron jobs](./features/cron.md) behave headlessly when they trigger a dangerous-command prompt. `deny` blocks the command (the agent must find another path); `approve` auto-approves everything in cron context. |
+| `write_file` | `allow` | Approval policy for the file mutation pair, `write_file` and `patch`. `allow` preserves normal behavior; `ask` requires the shared human approval UI across CLI, gateway, Desktop, and TUI, even under YOLO. Read-only `read_file` and `search_files` are unaffected. Native protected-path, cross-profile, traversal, malformed-request, and fail-closed syntax checks run before any prompt and are repeated where applicable at the mutation boundary. |
 | `mcp_reload_confirm` | `true` | When true, `/reload-mcp` asks before rebuilding the MCP tool set. Rebuilding invalidates the provider prompt cache (tool schemas live in the system prompt), so the next message re-sends full input tokens. Users who click **Always Approve** flip this key to `false`. |
 | `destructive_slash_confirm` | `true` | When true, destructive session slash commands (`/clear`, `/new`, `/reset`, `/undo`) prompt before discarding conversation state. Three-option dialog (Approve Once / Always Approve / Cancel) routed through native yes/no buttons on Telegram, Discord, and Slack; text fallback elsewhere. Users who click **Always Approve** flip this key to `false`. TUI uses its own modal overlay (set `HERMES_TUI_NO_CONFIRM=1` to opt out there). |
 
@@ -59,7 +61,7 @@ Setting `approvals.mode: off` disables all safety prompts. Use only in trusted e
 
 ### YOLO Mode
 
-YOLO mode bypasses **all** dangerous command approval prompts for the current session. It can be activated three ways:
+YOLO mode bypasses dangerous command approval prompts for the current session. The explicit `approvals.write_file: ask` policy is an exception: file mutations still require human approval. YOLO can be activated three ways:
 
 1. **CLI flag**: Start a session with `hermes --yolo` or `hermes chat --yolo`
 2. **Slash command**: Type `/yolo` during a session to toggle it on/off