refactor(compression): fold simplify findings — dedup floor constant, drift-guard test

- 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.
This commit is contained in:
kshitij 2026-08-08 14:24:11 +05:30
parent 39056e8de4
commit d81f2f49ea
2 changed files with 38 additions and 11 deletions

View File

@ -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:

View File

@ -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):