fix(guardrails): mark every file-tool loop refusal and exempt it on the live failure seam

The salvaged commit adds the `guardrail_refusal` marker to the read-dedup block
and exempts it in `classify_tool_failure`. That classifier is only the fallback:
the executor (`agent/tool_executor.py::_commit_tool_result`) passes
`failed=_detect_tool_failure(...)` from `agent/display.py` into
`after_call`, so on the real path the refusal was still counted and the
escalation to `repeated_exact_failure_block` still fired (live probe: block at
call 8 of an identical read after 5 refusals, on the salvaged head as on main).

- `agent/tool_result_classification.py`: `GUARDRAIL_REFUSAL_KEY` +
  `is_guardrail_refusal()` so the marker has one definition and both
  classifiers share the predicate.
- `agent/display.py::_detect_tool_failure`: exempt refusals (the live seam).
- `tools/file_tools.py`: the two sibling loop refusals -- the consecutive-read
  BLOCK in `read_file_tool` and the repeated-search BLOCK in `search_tool` --
  carry the same marker; they are emitted via `tool_error` too and escalated
  the same way.
- Tests trimmed to two invariants covering all three refusals on both
  classifiers, plus the negative (a genuine error still counts; a non-boolean
  marker is ignored).

Approval-denied write blocks (`file_tools_write_guards`) deliberately keep
counting: a user's denial retried unchanged is exactly the loop the streak
exists to stop.

agent/display.py

@@ -17,7 +17,7 @@ from urllib.parse import urlsplit
 
 from utils import safe_json_loads
 from agent.redact import redact_sensitive_text
-from agent.tool_result_classification import file_mutation_result_landed
+from agent.tool_result_classification import file_mutation_result_landed, is_guardrail_refusal
 
 logger = logging.getLogger(__name__)
 
@@ -915,6 +915,11 @@ def _detect_tool_failure(tool_name: str, result: Any) -> tuple[bool, str]:
     if result is None or file_mutation_result_landed(tool_name, result):
         return False, ""
     data = result if isinstance(result, dict) else safe_json_loads(result)
+    # A harness REFUSAL of a redundant call (repeated identical read/search) is not a
+    # failed call. This is the ``failed`` the executor hands the loop guardrail, so
+    # counting it would escalate refusals into ``repeated_exact_failure_block``.
+    if is_guardrail_refusal(data):
+        return False, ""
 
     # A denied/timed-out approval carries one human sentence; show it instead of the model-facing
     # "BLOCKED: ... Do NOT retry" text (which stays in the JSON for the model).

agent/tool_guardrails.py

@@ -14,7 +14,7 @@ from dataclasses import asdict, dataclass, field, fields
 from typing import Any, Mapping
 
 from utils import safe_json_loads
-from agent.tool_result_classification import file_mutation_result_landed
+from agent.tool_result_classification import file_mutation_result_landed, is_guardrail_refusal
 
 
 IDEMPOTENT_TOOL_NAMES = frozenset({
@@ -219,12 +219,11 @@ def classify_tool_failure(tool_name: str, result: str | None) -> tuple[bool, str
     if result is None or file_mutation_result_landed(tool_name, result):
         return False, ""
 
-    # A body this harness emitted to REFUSE a call is not a call that failed.
-    # It carries `"error"` for the model's benefit, which is exactly what the
-    # substring test below keys on, so counting it would let a refusal raise
-    # the failure streak that produces the next, harder refusal.
-    data = safe_json_loads(result)
-    if isinstance(data, dict) and data.get("guardrail_refusal") is True:
+    # A harness REFUSAL of a redundant call (repeated identical read/search) carries
+    # ``"error"`` for the model's benefit -- exactly what the substring test below keys
+    # on -- but nothing failed; counting it lets the cheap refusal feed the streak that
+    # fires the next, harder one. Mirrored in ``agent.display._detect_tool_failure``.
+    if is_guardrail_refusal(result):
         return False, ""
 
     if tool_name == "terminal":

agent/tool_result_classification.py

@@ -23,6 +23,24 @@ def tool_may_have_side_effect(tool_name: str) -> bool:
     return tool_name not in NO_EFFECT_TOOL_NAMES
 
 
+# Set by a tool that REFUSED a call the harness judged redundant (repeated identical
+# read/search). The body still carries ``"error"`` so the model reads it as a stop
+# signal, but nothing failed: failure classifiers must not count it, or the cheap
+# refusal feeds the streak that fires ``repeated_exact_failure_block``.
+GUARDRAIL_REFUSAL_KEY = "guardrail_refusal"
+
+
+def is_guardrail_refusal(result: Any) -> bool:
+    """Return True when ``result`` (JSON string or parsed dict) is a harness refusal."""
+    data = result
+    if isinstance(result, str):
+        try:
+            data = json.loads(result.strip())
+        except Exception:
+            return False
+    return isinstance(data, dict) and data.get(GUARDRAIL_REFUSAL_KEY) is True
+
+
 def file_mutation_result_landed(tool_name: str, result: Any) -> bool:
     """Return True when a file mutation result proves the write landed."""
     if tool_name not in FILE_MUTATING_TOOL_NAMES or not isinstance(result, str):

tests/agent/test_tool_guardrails.py

@@ -378,25 +378,37 @@ def test_supervised_task_platforms_keep_warning_only_default():
         assert cfg.hard_stop_enabled is True, platform
 
 
-def test_a_harness_refusal_is_not_counted_as_a_tool_failure():
-    """The read-dedup block carries `"error"` for the model's benefit, which is
-    exactly what `classify_tool_failure`'s substring test keys on. Counting it
-    let a refusal raise the failure streak that produces the next, harder
-    refusal -- an escalation to `repeated_exact_failure_block` reporting N
-    failures that never happened. Observed on a real session: 13 of 14 results
-    classified as failed were this block, against one genuine tool failure."""
-    from tools.file_tools import _dedup_stub_or_block
+def test_harness_refusals_are_not_tool_failures_on_either_classifier(tmp_path):
+    """Every loop refusal the file tools emit (read dedup block, consecutive-read block,
+    repeated-search block) carries `"error"` for the model's benefit -- exactly what the
+    substring tests key on. Neither `classify_tool_failure` nor the executor's live seam
+    `_detect_tool_failure` may count them, or the cheap refusal feeds the streak that
+    fires `repeated_exact_failure_block` over calls that never failed."""
+    from agent.display import _detect_tool_failure
+    from tools.file_tools import _dedup_stub_or_block, read_file_tool, search_tool
 
-    task_data = {"dedup_hits": {}}
-    key = ("/repo/responses.ts", 1, 999)
+    target = tmp_path / "responses.ts"
+    target.write_text("export const x = 1;\n" * 30, encoding="utf-8")
+
+    task = {"dedup_hits": {}}
     for _ in range(3):
-        blocked = _dedup_stub_or_block(task_data, key, "/repo/responses.ts")
+        dedup_block = _dedup_stub_or_block(task, (str(target), 1, 999), str(target))
+    consecutive_block = [read_file_tool(str(target), offset=1, limit=5, task_id="t-read") for _ in range(4)][-1]
+    search_block = [search_tool("const x", path=str(tmp_path), task_id="t-search") for _ in range(4)][-1]
 
-    assert json.loads(blocked)["guardrail_refusal"] is True
-    assert classify_tool_failure("read_file", blocked) == (False, "")
+    for refusal in (dedup_block, consecutive_block, search_block):
+        assert json.loads(refusal)["error"].startswith("BLOCKED"), refusal
+        assert classify_tool_failure("read_file", refusal) == (False, ""), refusal
+        assert _detect_tool_failure("read_file", refusal) == (False, ""), refusal
 
 
 def test_a_real_tool_error_is_still_a_failure():
     """The exemption is keyed on the marker, not on the word: a body that
     genuinely failed still counts, or the streak that stops a real loop is gone."""
-    assert classify_tool_failure("read_file", '{"error": "ENOENT: no such file"}')[0] is True
+    from agent.display import _detect_tool_failure
+
+    real = '{"error": "ENOENT: no such file"}'
+    assert classify_tool_failure("read_file", real)[0] is True
+    assert _detect_tool_failure("read_file", real)[0] is True
+    # The marker is only honoured as the literal boolean, never as truthy prose.
+    assert classify_tool_failure("read_file", '{"error": "x", "guardrail_refusal": "yes"}')[0] is True

tools/file_tools.py

@@ -19,6 +19,7 @@ from contextlib import ExitStack
 from pathlib import Path
 
 from agent.file_safety import get_nt_namespace_error, get_read_block_error
+from agent.tool_result_classification import GUARDRAIL_REFUSAL_KEY
 from tools.binary_extensions import has_binary_extension
 from tools.file_operations import (
     ShellFileOperations, normalize_read_pagination, normalize_search_pagination)
@@ -503,13 +504,10 @@ def _dedup_stub_or_block(task_data: dict, dedup_key: tuple, path: str) -> str:
             "the information you already have.",
             path=path,
             already_read=hits + 1,
-            # This body is a REFUSAL the harness chose, not a failure the tool
-            # hit. Without the marker the guardrail's own block feeds its
-            # failure counter (`classify_tool_failure` keys on the literal
-            # `"error"`), so refusing a repeated read escalates to
-            # `repeated_exact_failure_block` and reports N failures that never
-            # happened.
-            guardrail_refusal=True)
+            # A REFUSAL the harness chose, not a failure the tool hit: without the
+            # marker the failure classifiers count the block and a repeated read
+            # escalates to `repeated_exact_failure_block` over calls that never failed.
+            **{GUARDRAIL_REFUSAL_KEY: True})
 
     return json.dumps({
         "status": "unchanged",
@@ -702,7 +700,8 @@ def read_file_tool(path: str, offset: int = 1, limit: int = DEFAULT_READ_LIMIT,
                 "The content has NOT changed. You already have this information. "
                 "STOP re-reading and proceed with your task.",
                 path=path,
-                already_read=count)
+                already_read=count,
+                **{GUARDRAIL_REFUSAL_KEY: True})
         if count >= 3:
             result_dict["_warning"] = (
                 f"You have read this exact file region {count} times consecutively. "
@@ -1017,7 +1016,8 @@ def search_tool(pattern: str, target: str = "content", path: str = ".",
                 "The results have NOT changed. You already have this information. "
                 "STOP re-searching and proceed with your task.",
                 pattern=pattern,
-                already_searched=count)
+                already_searched=count,
+                **{GUARDRAIL_REFUSAL_KEY: True})
 
         # Raw string before _resolve_path_for_task: resolving is the NTLM-leak
         # trigger and the task-base join would hide the prefix (see read_file_tool).