fix(tools): block compacted content in side-effect calls

model_tools.py

@@ -791,6 +791,59 @@ def _resolve_active_context_length() -> int:
 _AGENT_LOOP_TOOLS = {"todo", "memory", "session_search", "delegate_task"}
 _READ_SEARCH_TOOLS = {"read_file", "search_files"}
 
+# Context compressors use these exact marker shapes when shortening historical
+# strings. A later model turn must not mistake that synthetic copy for complete
+# outbound content. Requiring either the compression sentinel or an ellipsis-
+# bracket marker keeps ordinary prose such as "the preview was truncated" valid.
+_SYNTHETIC_TRUNCATION_MARKER_RE = re.compile(
+    r"(?:"
+    r"(?:\.\.\.|…)[ \t]*\[truncated\](?:\.\.\.)?"
+    r"|⟪HERMES-CONTEXT-COMPRESSION:"
+    r")",
+    re.IGNORECASE,
+)
+
+
+def _contains_synthetic_truncation_marker(value: Any) -> bool:
+    """Find compactor markers in JSON-like arguments, including nested leaves."""
+    pending = [value]
+    seen: set[int] = set()
+    while pending:
+        item = pending.pop()
+        if isinstance(item, str):
+            if _SYNTHETIC_TRUNCATION_MARKER_RE.search(item):
+                return True
+            continue
+        if isinstance(item, dict):
+            identity = id(item)
+            if identity in seen:
+                continue
+            seen.add(identity)
+            pending.extend(item.keys())
+            pending.extend(item.values())
+        elif isinstance(item, (list, tuple, set, frozenset)):
+            identity = id(item)
+            if identity in seen:
+                continue
+            seen.add(identity)
+            pending.extend(item)
+    return False
+
+
+def _compacted_side_effect_error(tool_name: str, args: Any) -> Optional[str]:
+    """Return a fail-closed error for marker-bearing write-capable calls."""
+    if not _contains_synthetic_truncation_marker(args):
+        return None
+    entry = registry.get_entry(tool_name)
+    if entry is not None and entry.read_only is True:
+        return None
+    return (
+        f"Blocked potentially side-effecting tool '{tool_name}': its arguments "
+        "contain a synthetic truncation marker from compacted history, so exact "
+        "content cannot be verified. Recover the exact content from the original "
+        "source, then obtain fresh confirmation before retrying. The tool was NOT run."
+    )
+
 
 # =========================================================================
 # Tool error sanitization
@@ -1466,6 +1519,25 @@ def _return_bridge_result(result: Any) -> Any:
                 )
                 return result
 
+        compaction_block = _compacted_side_effect_error(function_name, function_args)
+        if compaction_block is not None:
+            result = tool_error(compaction_block)
+            _emit_post_tool_call_hook(
+                function_name=function_name,
+                function_args=function_args,
+                result=result,
+                task_id=task_id,
+                session_id=session_id,
+                tool_call_id=tool_call_id,
+                turn_id=turn_id,
+                api_request_id=api_request_id,
+                status="blocked",
+                error_type="compacted_tool_arguments",
+                error_message=compaction_block,
+                middleware_trace=list(_tool_middleware_trace),
+            )
+            return result
+
         # 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.
@@ -1543,6 +1615,9 @@ def _return_bridge_result(result: Any) -> Any:
                 # the parent's tool set via the process-global.
                 sandbox_enabled = enabled_tools if enabled_tools is not None else _last_resolved_tool_names
                 def _dispatch(next_args: Dict[str, Any]) -> Any:
+                    block = _compacted_side_effect_error(function_name, next_args)
+                    if block is not None:
+                        return tool_error(block)
                     return registry.dispatch(
                         function_name, next_args,
                         task_id=task_id,
@@ -1551,6 +1626,9 @@ def _dispatch(next_args: Dict[str, Any]) -> Any:
                     )
             else:
                 def _dispatch(next_args: Dict[str, Any]) -> Any:
+                    block = _compacted_side_effect_error(function_name, next_args)
+                    if block is not None:
+                        return tool_error(block)
                     return registry.dispatch(
                         function_name, next_args,
                         task_id=task_id,

tests/test_model_tools.py

@@ -30,7 +30,160 @@ def test_unknown_tool_returns_error(self):
         assert "error" in result
         assert "totally_fake_tool_xyz" in result["error"]
 
+    def test_compacted_tool_arguments_cannot_reach_write_handler(self, monkeypatch):
+        from agent.context_compressor import _truncate_tool_call_args_json
 
+        original = json.dumps({"path": "fixture.txt", "content": "x" * 500})
+        compacted_args = json.loads(_truncate_tool_call_args_json(original))
+        monkeypatch.setattr(
+            "model_tools.registry.dispatch",
+            lambda *_args, **_kwargs: (_ for _ in ()).throw(
+                AssertionError("compacted write must not dispatch")
+            ),
+        )
+
+        result = json.loads(handle_function_call("write_file", compacted_args))
+
+        assert "exact content" in result["error"].lower()
+        assert "fresh confirmation" in result["error"].lower()
+
+    def test_nested_compaction_marker_blocks_write_handler(self, monkeypatch):
+        monkeypatch.setattr(
+            "model_tools.registry.dispatch",
+            lambda *_args, **_kwargs: (_ for _ in ()).throw(
+                AssertionError("nested compacted write must not dispatch")
+            ),
+        )
+
+        result = json.loads(
+            handle_function_call(
+                "write_file",
+                {"path": "fixture.json", "content": {"sections": ["partial...[truncated]"]}},
+            )
+        )
+
+        assert "exact content" in result["error"].lower()
+
+    def test_named_compaction_sentinel_blocks_write_handler(self, monkeypatch):
+        monkeypatch.setattr(
+            "model_tools.registry.dispatch",
+            lambda *_args, **_kwargs: (_ for _ in ()).throw(
+                AssertionError("compaction sentinel must not dispatch")
+            ),
+        )
+        marker = "⟪HERMES-CONTEXT-COMPRESSION: 300 chars omitted here⟫"
+
+        result = json.loads(
+            handle_function_call(
+                "write_file", {"path": "fixture.txt", "content": marker}
+            )
+        )
+
+        assert "exact content" in result["error"].lower()
+
+    def test_complete_write_payload_passes_byte_for_byte(self, monkeypatch):
+        payload = "alpha\nβeta\n"
+        captured = {}
+
+        def dispatch(_name, args, **_kwargs):
+            captured["content"] = args["content"]
+            return json.dumps({"ok": True})
+
+        monkeypatch.setattr("model_tools.registry.dispatch", dispatch)
+
+        assert json.loads(
+            handle_function_call("write_file", {"path": "fixture.txt", "content": payload})
+        ) == {"ok": True}
+        assert captured["content"].encode("utf-8") == payload.encode("utf-8")
+
+    def test_read_only_tool_is_not_blocked_by_compaction_marker(self, monkeypatch):
+        captured = {}
+
+        def dispatch(_name, args, **_kwargs):
+            captured.update(args)
+            return json.dumps({"ok": True})
+
+        monkeypatch.setattr("model_tools.registry.dispatch", dispatch)
+
+        args = {"path": "reports/... [truncated]/index.txt"}
+        assert json.loads(handle_function_call("read_file", args)) == {"ok": True}
+        assert captured == args
+
+    def test_legitimate_truncation_prose_is_not_blocked(self, monkeypatch):
+        payload = "Explain that the preview was truncated by the UI."
+        captured = {}
+
+        def dispatch(_name, args, **_kwargs):
+            captured.update(args)
+            return json.dumps({"ok": True})
+
+        monkeypatch.setattr("model_tools.registry.dispatch", dispatch)
+
+        args = {"path": "fixture.txt", "content": payload}
+        assert json.loads(handle_function_call("write_file", args)) == {"ok": True}
+        assert captured == args
+
+    def test_unknown_outbound_tool_fails_closed_on_marker(self, monkeypatch):
+        monkeypatch.setattr(
+            "model_tools.registry.dispatch",
+            lambda *_args, **_kwargs: (_ for _ in ()).throw(
+                AssertionError("unknown outbound handler must not dispatch")
+            ),
+        )
+
+        result = json.loads(
+            handle_function_call(
+                "send_email",
+                {"to": "recipient.invalid", "body": "partial...[truncated]"},
+            )
+        )
+
+        assert "not run" in result["error"].lower()
+
+    def test_execution_middleware_cannot_inject_marker_into_write(self, monkeypatch):
+        def execution_middleware(**kwargs):
+            return kwargs["next_call"](
+                {**kwargs["args"], "content": "partial...[truncated]"}
+            )
+
+        manager = type(
+            "Manager", (), {"_middleware": {"tool_execution": [execution_middleware]}}
+        )()
+        monkeypatch.setattr("hermes_cli.plugins.get_plugin_manager", lambda: manager)
+        monkeypatch.setattr(
+            "model_tools.registry.dispatch",
+            lambda *_args, **_kwargs: (_ for _ in ()).throw(
+                AssertionError("middleware marker must not reach handler")
+            ),
+        )
+
+        result = json.loads(
+            handle_function_call(
+                "write_file", {"path": "fixture.txt", "content": "complete"}
+            )
+        )
+
+        assert "exact content" in result["error"].lower()
+
+    def test_guard_does_not_mutate_existing_role_alternation(self, monkeypatch):
+        history = [
+            {"role": "user", "content": "write it"},
+            {"role": "assistant", "content": None, "tool_calls": ["call-1"]},
+        ]
+        before = json.dumps(history, ensure_ascii=False, sort_keys=True)
+        monkeypatch.setattr(
+            "model_tools.registry.dispatch",
+            lambda *_args, **_kwargs: (_ for _ in ()).throw(
+                AssertionError("compacted write must not dispatch")
+            ),
+        )
+
+        handle_function_call(
+            "write_file", {"path": "fixture.txt", "content": "partial...[truncated]"}
+        )
+
+        assert json.dumps(history, ensure_ascii=False, sort_keys=True) == before
+        assert [message["role"] for message in history] == ["user", "assistant"]
 
     def test_post_tool_call_receives_non_negative_integer_duration_ms(self):
         """Regression: post_tool_call and transform_tool_result hooks must

tests/tools/test_mcp_trust_gating.py

@@ -220,7 +220,8 @@ def test_registration_records_hints_and_trust(self):
             "trust": "untrusted",
             "tools": {"resources": False, "prompts": False},
         }
-        with patch("tools.registry.registry", ToolRegistry()), \
+        reg = ToolRegistry()
+        with patch("tools.registry.registry", reg), \
              patch("tools.mcp_tool._track_mcp_tool_server"):
             mcp_tool._register_server_tools("srv", server, config)
 
@@ -230,6 +231,12 @@ def test_registration_records_hints_and_trust(self):
         # Anything not exactly True is write-capable.
         assert not hints.get("delete_repo")
         assert not hints.get("no_annotations")
+        list_entry = reg.get_entry("mcp__srv__list_repos")
+        delete_entry = reg.get_entry("mcp__srv__delete_repo")
+        unknown_entry = reg.get_entry("mcp__srv__no_annotations")
+        assert list_entry is not None and list_entry.read_only is True
+        assert delete_entry is not None and delete_entry.read_only is False
+        assert unknown_entry is not None and unknown_entry.read_only is False
 
     def test_dict_annotations_supported(self):
         """Cached/JSON annotations arrive as plain dicts."""

tests/tools/test_registry.py

@@ -108,6 +108,26 @@ def test_returns_openai_format(self):
         names = {d["function"]["name"] for d in defs}
         assert names == {"t1", "t2"}
 
+    def test_read_only_metadata_does_not_change_prompt_schema(self):
+        write_reg = ToolRegistry()
+        read_reg = ToolRegistry()
+        schema = _make_schema("stable")
+        write_reg.register(
+            name="stable", toolset="s1", schema=schema, handler=_dummy_handler
+        )
+        read_reg.register(
+            name="stable",
+            toolset="s1",
+            schema=schema,
+            handler=_dummy_handler,
+            read_only=True,
+        )
+
+        assert write_reg.get_definitions({"stable"}) == read_reg.get_definitions({"stable"})
+        write_entry = write_reg.get_entry("stable")
+        read_entry = read_reg.get_entry("stable")
+        assert write_entry is not None and write_entry.read_only is False
+        assert read_entry is not None and read_entry.read_only is True
 
     def test_reuses_shared_check_fn_once_per_call(self):
         reg = ToolRegistry()

tools/file_tools.py

@@ -2818,7 +2818,7 @@ def _handle_search_files(args, **kw):
         output_mode=args.get("output_mode", "content"), context=args.get("context", 0), task_id=tid)
 
 
-registry.register(name="read_file", toolset="file", schema=READ_FILE_SCHEMA, handler=_handle_read_file, check_fn=_check_file_reqs, emoji="📖", max_result_size_chars=100_000)
+registry.register(name="read_file", toolset="file", schema=READ_FILE_SCHEMA, handler=_handle_read_file, check_fn=_check_file_reqs, emoji="📖", max_result_size_chars=100_000, read_only=True)
 registry.register(name="write_file", toolset="file", schema=WRITE_FILE_SCHEMA, handler=_handle_write_file, check_fn=_check_file_reqs, emoji="✍️", max_result_size_chars=100_000)
 registry.register(name="patch", toolset="file", schema=PATCH_SCHEMA, handler=_handle_patch, check_fn=_check_file_reqs, emoji="🔧", max_result_size_chars=100_000)
-registry.register(name="search_files", toolset="file", schema=SEARCH_FILES_SCHEMA, handler=_handle_search_files, check_fn=_check_file_reqs, emoji="🔎", max_result_size_chars=100_000)
+registry.register(name="search_files", toolset="file", schema=SEARCH_FILES_SCHEMA, handler=_handle_search_files, check_fn=_check_file_reqs, emoji="🔎", max_result_size_chars=100_000, read_only=True)

tools/mcp_tool.py

@@ -7232,6 +7232,7 @@ def _should_register(tool_name: str) -> bool:
                     name, mcp_tool.name, server.tool_timeout
                 ),
                 "check_fn": check_fn,
+                "read_only": _annotation_read_only_hint(mcp_tool),
             }
         )
 
@@ -7255,6 +7256,7 @@ def _should_register(tool_name: str) -> bool:
                     name, server.tool_timeout
                 ),
                 "check_fn": check_fn,
+                "read_only": True,
             }
         )
 
@@ -7362,6 +7364,7 @@ def _should_register(tool_name: str) -> bool:
             check_fn=candidate["check_fn"],
             is_async=False,
             description=candidate["schema"]["description"],
+            read_only=candidate["read_only"],
         )
 
         # The pre-check above is advisory only. Multiple servers connect in
@@ -7519,6 +7522,10 @@ def _should_register(tool_name: str) -> bool:
             check_fn=check_fn,
             is_async=False,
             description=schema["description"],
+            read_only=(
+                isinstance(raw.get("annotations"), dict)
+                and raw["annotations"].get("readOnlyHint") is True
+            ),
         )
         if registry.get_toolset_for_tool(registry_name) != toolset_name:
             continue
@@ -7552,6 +7559,7 @@ def _should_register(tool_name: str) -> bool:
             check_fn=check_fn,
             is_async=False,
             description=schema.get("description") or "",
+            read_only=True,
         )
         if registry.get_toolset_for_tool(util_name) != toolset_name:
             continue