fix(agent): block partial compression markers before dispatch

(cherry picked from commit a558bd0aabf8e779fd4e12a2273ba6dda768454f)

agent/compression_marker.py

@@ -32,9 +32,17 @@ _COMPRESSION_MARKER_RE = re.compile(
     .replace(re.escape("{total:,}"), r"\d[\d,]*")
 )
 
+# A marker cut by a later boundary may never reach the fixed sentence above. The
+# dispatch guard therefore recognizes a minted marker as soon as its first numeric
+# count is present. Requiring that rendered count keeps the bare prefix/template
+# editable in source and documentation.
+_COMPRESSION_MARKER_ARTIFACT_RE = re.compile(
+    re.escape(_COMPRESSION_MARKER_PREFIX) + r"\s+\d[\d,]*"
+)
+
 # #121548 — every OTHER model-visible elision (turn text, summaries, skill bodies, diagnostics)
-# mints the args marker's first sentence only (so the dispatch-boundary guard's
-# _COMPRESSION_MARKER_RE catches a copy from any renderer) and fits small caps like clarify's 199.
+# mints the args marker's first sentence only (so the dispatch-boundary artifact
+# matcher catches a copy from any renderer) and fits small caps like clarify's 199.
 _ELISION_MARKER_TEMPLATE = _COMPRESSION_MARKER_TEMPLATE.split(". ", 1)[0] + ".⟫"
 
 

agent/tool_dispatch_helpers.py

@@ -16,7 +16,7 @@ import re
 from pathlib import Path
 from typing import Any, Dict, List, Optional
 
-from agent.compression_marker import _COMPRESSION_MARKER_RE
+from agent.compression_marker import _COMPRESSION_MARKER_ARTIFACT_RE
 from agent.message_metadata import stamp_message_timestamp
 from agent.tool_result_classification import (
     FILE_MUTATING_TOOL_NAMES as _FILE_MUTATING_TOOLS,
@@ -72,9 +72,9 @@ _REDIRECT_OVERWRITE = re.compile(r'[^>]>[^>]|^>[^>]')
 def _context_pruned_argument_paths(tool_name: str, args: Any) -> list[str]:
     """Paths whose values contain model-visible context-compression artifacts.
 
-    The compressor's current marker carries numeric omitted/total counts. Match
-    that rendered shape rather than the prefix alone so Hermes can still edit
-    source/docs that mention the compression marker constant or its template.
+    A minted marker is identified by its prefix plus the first rendered numeric
+    count. This catches a marker cut short before its fixed sentence while still
+    letting Hermes edit source/docs that mention the bare prefix or template.
     Unknown/plugin/MCP tools stay effect-capable by default; known read-only
     tools may inspect or quote compressed history.
     """
@@ -85,7 +85,7 @@ def _context_pruned_argument_paths(tool_name: str, args: Any) -> list[str]:
 
     def _walk(value: Any, path: str) -> None:
         if isinstance(value, str):
-            if _COMPRESSION_MARKER_RE.search(value):
+            if _COMPRESSION_MARKER_ARTIFACT_RE.search(value):
                 found.append(path)
             return
         if isinstance(value, dict):

tests/agent/test_tool_call_guardrail_runtime.py

@@ -408,6 +408,34 @@ def test_context_pruned_effectful_call_blocks_before_dispatch():
 
     assert _context_pruned_argument_paths("write_file", {"content": "x" * 201 + "...[truncated]"}) == []
     assert _context_pruned_argument_paths("write_file", {"content": f"see {_COMPRESSION_MARKER_PREFIX} docs"}) == []
+    assert _context_pruned_argument_paths(
+        "write_file",
+        {"content": f"template {_COMPRESSION_MARKER_PREFIX} {{omitted:,}} of {{total:,}}"},
+    ) == []
+
+
+@pytest.mark.parametrize("suffix", [" 1,800", " 1,800 of", " 1,800 of 2,000 chars omitted"])
+def test_partial_counted_context_marker_blocks_before_dispatch(suffix):
+    from agent.compression_marker import _COMPRESSION_MARKER_PREFIX
+
+    agent = _make_agent("test_effectful_write")
+    pruned = {"body": f"prefix {_COMPRESSION_MARKER_PREFIX}{suffix}"}
+    tc = _mock_tool_call(
+        "test_effectful_write", json.dumps(pruned, ensure_ascii=False), "c-pruned-partial"
+    )
+    msg = SimpleNamespace(content="", tool_calls=[tc])
+    messages = []
+
+    with (
+        patch("hermes_cli.plugins._dispatch_pre_tool_call_hooks", return_value=(None, pruned)),
+        patch("model_tools.handle_function_call", return_value="SHOULD_NOT_RUN") as dispatch,
+    ):
+        agent._execute_tool_calls_sequential(msg, messages, "task-1")
+
+    dispatch.assert_not_called()
+    payload = json.loads(messages[0]["content"])
+    assert payload["error"] == "suspected_pruned_tool_arguments"
+    assert payload["argument_paths"] == ["$.body"]
 
 
 def test_read_only_tool_may_quote_current_context_prune_marker():