From d81f2f49ea999a4f9af69c15e1fdd5d710a91580 Mon Sep 17 00:00:00 2001
From: kshitij <82637225+kshitijk4poor@users.noreply.github.com>
Date: Sat, 8 Aug 2026 14:24:11 +0530
Subject: [PATCH] =?UTF-8?q?refactor(compression):=20fold=20simplify=20find?=
 =?UTF-8?q?ings=20=E2=80=94=20dedup=20floor=20constant,=20drift-guard=20te?=
 =?UTF-8?q?st?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- Wire the Pass-1 dedup floor (len < 200) to the shared _PRUNE_MIN_CHARS
  constant it was already documented as matching, and use the constant in
  the remaining test literal.
- Restructure the clarify 'resolved' computation (is_answer_shaped +
  sentinel check) instead of compute-then-flip.
- Add a live producer->recognizer drift guard: the REAL oneshot no-user
  callback's output must be recognized as a sentinel, so producer wording
  drift fails a test instead of silently reintroducing false attribution.
- Document the any()-poisoning semantic for multi-select sentinel lists.
---
 agent/context_compressor.py            | 27 ++++++++++++++++----------
 tests/agent/test_context_compressor.py | 22 ++++++++++++++++++++-
 2 files changed, 38 insertions(+), 11 deletions(-)

diff --git a/agent/context_compressor.py b/agent/context_compressor.py
index ea93ba1737c0e..df423d943d945 100644
--- a/agent/context_compressor.py
+++ b/agent/context_compressor.py
@@ -466,7 +466,13 @@ _CLARIFY_NON_RESPONSE_PREFIXES = (
 
 def _is_clarify_non_response_sentinel(response: Any) -> bool:
     """Return True when a clarify ``user_response`` is runtime sentinel prose
-    (timeout / no-user), not an actual user answer."""
+    (timeout / no-user), not an actual user answer.
+
+    For lists, ANY sentinel item poisons the whole response: every real
+    producer returns a scalar sentinel, so a mixed list means forged or
+    corrupt tool content — fall back to the generic path (may lose info,
+    never misattributes).
+    """
     if isinstance(response, str):
         return response.lstrip().startswith(_CLARIFY_NON_RESPONSE_PREFIXES)
     if isinstance(response, list):
@@ -1360,20 +1366,21 @@ def _summarize_tool_result_unguarded(tool_name: str, tool_args: str, tool_conten
         except (json.JSONDecodeError, TypeError):
             result = {}
         response = result.get("user_response") if isinstance(result, dict) else None
-        resolved = (
+        is_answer_shaped = (
             isinstance(response, str) and bool(response)
         ) or (
             isinstance(response, list)
             and bool(response)
             and all(isinstance(item, str) and item for item in response)
         )
-        if resolved and _is_clarify_non_response_sentinel(response):
-            # Timeout / no-user paths embed sentinel prose as user_response
-            # (gateway "[user did not respond within Nm]", oneshot
-            # "[oneshot mode: ...]", CLI "The user did not provide a
-            # response..."). Quoting those as a user answer would be false
-            # attribution — keep them on the generic path.
-            resolved = False
+        # Timeout / no-user paths embed sentinel prose as user_response
+        # (gateway "[user did not respond within Nm]", oneshot
+        # "[oneshot mode: ...]", CLI "The user did not provide a
+        # response..."). Quoting those as a user answer would be false
+        # attribution — keep them on the generic path.
+        resolved = is_answer_shaped and not _is_clarify_non_response_sentinel(
+            response
+        )
         if resolved:
             # Keep ordinary Unicode intact while escaping lone UTF-16
             # surrogates so the compacted message remains UTF-8/SQLite safe.
@@ -3062,7 +3069,7 @@ class ContextCompressor(ContextEngine):
                 # Multimodal dict envelopes ({_multimodal: True, content: [...]}) and
                 # other non-string tool-result shapes can't be hashed/deduped by text.
                 continue
-            if len(content) < 200:
+            if len(content) < _PRUNE_MIN_CHARS:
                 continue
             h = hashlib.md5(content.encode("utf-8", errors="replace")).hexdigest()[:12]
             if h in content_hashes:
diff --git a/tests/agent/test_context_compressor.py b/tests/agent/test_context_compressor.py
index 6df4c84b4bfe2..164f54a652108 100644
--- a/tests/agent/test_context_compressor.py
+++ b/tests/agent/test_context_compressor.py
@@ -149,7 +149,7 @@ class TestSummarizeToolResultClarify:
         summary = pruned_messages[1]["content"]
 
         assert pruned_count == 1
-        assert len(summary) <= 200
+        assert len(summary) <= _PRUNE_MIN_CHARS
         assert summary.encode("utf-8")
         assert "Привет 😀" in summary
         assert "\\ud83d" in summary
@@ -212,6 +212,26 @@ class TestSummarizeToolResultClarify:
 
         assert summary == "[clarify] asked user a question"
 
+    def test_live_oneshot_producer_is_recognized_as_sentinel(self):
+        """Producer→recognizer drift guard: run the REAL oneshot no-user
+        callback and assert its output is filtered. If the producer's wording
+        drifts away from _CLARIFY_NON_RESPONSE_PREFIXES, this fails."""
+        from hermes_cli.oneshot import _oneshot_clarify_callback
+
+        sentinels = (
+            _oneshot_clarify_callback("Deploy when?", choices=["a", "b"]),
+            _oneshot_clarify_callback(
+                "Deploy when?", choices=["a", "b"], multi_select=True
+            ),
+            _oneshot_clarify_callback("Deploy when?"),
+        )
+        for sentinel in sentinels:
+            content = json.dumps({"user_response": sentinel})
+
+            summary = _summarize_tool_result("clarify", "{}", content)
+
+            assert summary == "[clarify] asked user a question", sentinel
+
 
 class TestShouldCompress:
     def test_below_threshold(self, compressor):