From b7fcf5b8c71d668f696e03592662299ff72d8bd2 Mon Sep 17 00:00:00 2001 From: Enzo Adami Date: Wed, 10 Jun 2026 21:59:13 -0400 Subject: [PATCH] feat(guardrails): add repeated-mutation halt and destructive-overwrite guard MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two data-protection guards for the per-turn tool guardrail controller, both born from real incidents on a long-running local agent: 1. Repeated-mutation halt: the existing no-progress guard only watches idempotent (read-only) tools, and the failure counters only advance on failed calls. A mutating tool that keeps *succeeding* on the exact same call is never caught — we observed a model issue the identical write_file 38 times in one turn. The guard counts identical successful mutating calls per signature (tool + args, content included, so legitimate iterative edits never match), warns at the existing no_progress warn threshold and halts at the block threshold. 2. Destructive-overwrite guard: blocks a write_file that would replace a non-empty file (>= safe_write_min_bytes, default 200) with content smaller than safe_write_shrink_ratio (default 0.5) of its current size. This is data-loss prevention, not loop detection: it is independent of hard_stop_enabled, does not halt the turn (the model can recover in place by writing the full intended content), and an identical re-issue confirms intent so genuine shrinks still go through. Motivated by a model accidentally blanking a status file and a full report with empty scaffold content. New config keys under tool_loop_guardrails (all with safe defaults): safe_write_enabled, safe_write_min_bytes, safe_write_shrink_ratio. Existing test suite passes unchanged (tests/run_agent/ test_tool_call_guardrail_runtime.py, 9/9). Co-Authored-By: Claude Fable 5 --- agent/tool_guardrails.py | 122 +++++++++++++++ .../test_mutation_and_safe_write_guards.py | 139 ++++++++++++++++++ 2 files changed, 261 insertions(+) create mode 100644 tests/run_agent/test_mutation_and_safe_write_guards.py diff --git a/agent/tool_guardrails.py b/agent/tool_guardrails.py index 0332796922884..dbc57b93d9f84 100644 --- a/agent/tool_guardrails.py +++ b/agent/tool_guardrails.py @@ -10,6 +10,7 @@ import hashlib import json +import os from dataclasses import dataclass, field from typing import Any, Mapping @@ -79,6 +80,14 @@ class ToolCallGuardrailConfig: no_progress_block_after: int = 5 idempotent_tools: frozenset[str] = field(default_factory=lambda: IDEMPOTENT_TOOL_NAMES) mutating_tools: frozenset[str] = field(default_factory=lambda: MUTATING_TOOL_NAMES) + # Destructive-overwrite guard (cf. 2026-06-09: the companion blanked STATUS.md + # and a full research report by overwriting them with empty/scaffold content). + # Independent of hard_stop_enabled — this is data-loss prevention, not loop + # detection. Blocks write_file when a non-empty file (>= min_bytes) would be + # replaced by content < shrink_ratio of its current size. + safe_write_enabled: bool = True + safe_write_min_bytes: int = 200 + safe_write_shrink_ratio: float = 0.5 @classmethod def from_mapping(cls, data: Mapping[str, Any] | None) -> "ToolCallGuardrailConfig": @@ -121,6 +130,13 @@ def from_mapping(cls, data: Mapping[str, Any] | None) -> "ToolCallGuardrailConfi hard_stop_after.get("idempotent_no_progress", data.get("no_progress_block_after")), defaults.no_progress_block_after, ), + safe_write_enabled=_as_bool(data.get("safe_write_enabled"), defaults.safe_write_enabled), + safe_write_min_bytes=_positive_int( + data.get("safe_write_min_bytes"), defaults.safe_write_min_bytes + ), + safe_write_shrink_ratio=_as_ratio( + data.get("safe_write_shrink_ratio"), defaults.safe_write_shrink_ratio + ), ) @@ -232,6 +248,8 @@ def reset_for_turn(self) -> None: self._exact_failure_counts: dict[ToolCallSignature, int] = {} self._same_tool_failure_counts: dict[str, int] = {} self._no_progress: dict[ToolCallSignature, tuple[str, int]] = {} + self._mutation_repeats: dict[ToolCallSignature, int] = {} + self._shrink_confirmed: set[ToolCallSignature] = set() self._halt_decision: ToolGuardrailDecision | None = None @property @@ -240,6 +258,10 @@ def halt_decision(self) -> ToolGuardrailDecision | None: def before_call(self, tool_name: str, args: Mapping[str, Any] | None) -> ToolGuardrailDecision: signature = ToolCallSignature.from_call(tool_name, _coerce_args(args)) + if self.config.safe_write_enabled and tool_name == "write_file": + shrink = self._destructive_overwrite_decision(_coerce_args(args), signature) + if shrink is not None: + return shrink if not self.config.hard_stop_enabled: return ToolGuardrailDecision(tool_name=tool_name, signature=signature) @@ -349,6 +371,51 @@ def after_call( if not self._is_idempotent(tool_name): self._no_progress.pop(signature, None) + # Repeated IDENTICAL successful mutation = cognitive loop (cf. bug 38x + # write_file, 2026-06-08, Claude). The read-only no_progress guard above + # excludes mutating tools, and the failure counters only move on failed=True, + # so a tool that keeps "succeeding" on the exact same call is never caught. + # Keyed on signature (tool + args, content included) → never flags a + # legitimate iterative edit, since different content = different signature. + # Reuses the idempotent_no_progress thresholds. Resets each turn. + if tool_name in self.config.mutating_tools: + repeat = self._mutation_repeats.get(signature, 0) + 1 + self._mutation_repeats[signature] = repeat + if ( + self.config.hard_stop_enabled + and repeat >= self.config.no_progress_block_after + ): + decision = ToolGuardrailDecision( + action="halt", + code="repeated_mutation_halt", + message=( + f"Stopped {tool_name}: the identical successful call was " + f"repeated {repeat} times this turn with no change. This is a " + "loop (cf. write_file 38x). Read the file, run a validation " + "(bash -n / py_compile), and change approach before retrying." + ), + tool_name=tool_name, + count=repeat, + signature=signature, + ) + self._halt_decision = decision + return decision + if ( + self.config.warnings_enabled + and repeat >= self.config.no_progress_warn_after + ): + return ToolGuardrailDecision( + action="warn", + code="repeated_mutation_warning", + message=( + f"{tool_name} repeated the identical call {repeat} times this " + "turn with no change. Verify the result (read the file, run a " + "syntax check) instead of writing the same thing again." + ), + tool_name=tool_name, + count=repeat, + signature=signature, + ) return ToolGuardrailDecision(tool_name=tool_name, signature=signature) result_hash = _result_hash(result) @@ -379,6 +446,51 @@ def _is_idempotent(self, tool_name: str) -> bool: return False return tool_name in self.config.idempotent_tools + def _destructive_overwrite_decision( + self, args: Mapping[str, Any], signature: ToolCallSignature + ) -> ToolGuardrailDecision | None: + """Block a write_file that overwrites a non-empty file with empty / much + smaller content (the accidental-blanking failure: STATUS.md and a full + research report were wiped this way on 2026-06-09). + + Independent of hard_stop_enabled — data-loss prevention, not loop + detection. Does NOT set _halt_decision: only this call is blocked, so the + model can recover in-turn by writing the intended full content. Re-issuing + the identical call once confirms intent (a genuine shrink then proceeds). + """ + path = args.get("path") or args.get("file_path") + content = args.get("content") + if not isinstance(path, str) or not path or not isinstance(content, str): + return None + try: + existing = os.path.getsize(os.path.expanduser(path)) + except OSError: + return None # new or unreadable file → nothing to protect + if existing < self.config.safe_write_min_bytes: + return None # trivially small existing file → allow + new_bytes = len(content.encode("utf-8")) + if new_bytes >= existing * self.config.safe_write_shrink_ratio: + return None # not a drastic shrink → allow + if signature in self._shrink_confirmed: + return None # identical re-issue = explicit confirmation → allow + self._shrink_confirmed.add(signature) + pct = int(round(new_bytes * 100 / existing)) if existing else 0 + return ToolGuardrailDecision( + action="block", + code="destructive_overwrite_block", + message=( + f"Blocked write_file: this overwrites {path} ({existing} bytes) with " + f"only {new_bytes} bytes ({pct}% of current). This matches the " + "accidental-blanking failure (STATUS.md and a full report were wiped " + "this way on 2026-06-09). Read the file first and write the full " + "intended content. If you truly mean to shrink/replace it, re-issue " + "this exact call once to confirm." + ), + tool_name="write_file", + count=new_bytes, + signature=signature, + ) + def toolguard_synthetic_result(decision: ToolGuardrailDecision) -> str: """Build a synthetic role=tool content string for a blocked tool call.""" @@ -471,5 +583,15 @@ def _positive_int(value: Any, default: int) -> int: return parsed if parsed >= 1 else default +def _as_ratio(value: Any, default: float) -> float: + if value is None: + return default + try: + parsed = float(value) + except (TypeError, ValueError): + return default + return parsed if 0.0 < parsed <= 1.0 else default + + def _sha256(value: str) -> str: return hashlib.sha256(value.encode("utf-8")).hexdigest() diff --git a/tests/run_agent/test_mutation_and_safe_write_guards.py b/tests/run_agent/test_mutation_and_safe_write_guards.py new file mode 100644 index 0000000000000..79f183dd13bfc --- /dev/null +++ b/tests/run_agent/test_mutation_and_safe_write_guards.py @@ -0,0 +1,139 @@ +"""Tests for the repeated-mutation halt and the destructive-overwrite guard.""" + +from agent.tool_guardrails import ( + ToolCallGuardrailConfig, + ToolCallGuardrailController, +) + +OK_RESULT = '{"status": "ok"}' + + +def _controller(**overrides): + data = { + "warnings_enabled": True, + "hard_stop_enabled": True, + "no_progress_warn_after": 3, + "no_progress_block_after": 5, + } + data.update(overrides) + return ToolCallGuardrailController(ToolCallGuardrailConfig.from_mapping(data)) + + +# --------------------------------------------------------- repeated-mutation halt + + +def test_identical_successful_mutation_warns_then_halts(): + controller = _controller() + args = {"path": "/tmp/does-not-exist-loop.md", "content": "same content"} + decisions = [ + controller.after_call("write_file", args, OK_RESULT, failed=False) + for _ in range(5) + ] + assert [d.action for d in decisions[:2]] == ["allow", "allow"] + assert decisions[2].action == "warn" + assert decisions[2].code == "repeated_mutation_warning" + assert decisions[4].action == "halt" + assert decisions[4].code == "repeated_mutation_halt" + assert controller.halt_decision is not None + + +def test_iterative_edits_with_different_content_never_flag(): + controller = _controller() + for i in range(8): + decision = controller.after_call( + "write_file", + {"path": "/tmp/iterative-edit.md", "content": f"version {i}"}, + OK_RESULT, + failed=False, + ) + assert decision.action == "allow", (i, decision.action, decision.code) + + +def test_mutation_counter_resets_each_turn(): + controller = _controller() + args = {"path": "/tmp/x.md", "content": "y"} + for _ in range(4): + controller.after_call("write_file", args, OK_RESULT, failed=False) + controller.reset_for_turn() + decision = controller.after_call("write_file", args, OK_RESULT, failed=False) + assert decision.action == "allow" + + +def test_mutation_halt_requires_hard_stop_enabled(): + controller = _controller(hard_stop_enabled=False) + args = {"path": "/tmp/x.md", "content": "y"} + last = None + for _ in range(7): + last = controller.after_call("write_file", args, OK_RESULT, failed=False) + assert last.action in ("allow", "warn") + assert controller.halt_decision is None + + +# --------------------------------------------------------- destructive-overwrite + + +def _victim(tmp_path, size=1000): + victim = tmp_path / "victim.md" + victim.write_text("x" * size) + return victim + + +def test_blocks_drastic_shrink_without_halting_turn(tmp_path): + controller = _controller() + victim = _victim(tmp_path) + decision = controller.before_call( + "write_file", {"path": str(victim), "content": "stub"} + ) + assert decision.action == "block" + assert decision.code == "destructive_overwrite_block" + # block, not halt: the model can recover in-turn with the full content + assert controller.halt_decision is None + + +def test_identical_reissue_confirms_intent(tmp_path): + controller = _controller() + victim = _victim(tmp_path) + args = {"path": str(victim), "content": "intentional shrink"} + assert controller.before_call("write_file", args).action == "block" + assert controller.before_call("write_file", args).action == "allow" + + +def test_allows_small_files_new_files_and_growth(tmp_path): + controller = _controller() + small = tmp_path / "small.md" + small.write_text("x" * 50) # below safe_write_min_bytes + assert ( + controller.before_call("write_file", {"path": str(small), "content": ""}).action + == "allow" + ) + assert ( + controller.before_call( + "write_file", {"path": str(tmp_path / "new.md"), "content": "hello"} + ).action + == "allow" + ) + victim = _victim(tmp_path) + assert ( + controller.before_call( + "write_file", {"path": str(victim), "content": "x" * 900} + ).action + == "allow" + ) + + +def test_safe_write_independent_of_hard_stop(tmp_path): + controller = _controller(hard_stop_enabled=False) + victim = _victim(tmp_path) + decision = controller.before_call( + "write_file", {"path": str(victim), "content": "stub"} + ) + assert decision.action == "block" + + +def test_safe_write_flag_off_allows(tmp_path): + controller = _controller(safe_write_enabled=False) + victim = _victim(tmp_path) + decision = controller.before_call( + "write_file", {"path": str(victim), "content": ""} + ) + assert decision.action == "allow"