diff --git a/gateway/response_filters.py b/gateway/response_filters.py index a5e09d309e0..ccd2db370ff 100644 --- a/gateway/response_filters.py +++ b/gateway/response_filters.py @@ -7,6 +7,7 @@ conversation history. from __future__ import annotations +import unicodedata from typing import Any # Canonical model-emitted control token for intentional silence. @@ -27,6 +28,31 @@ def _canonical_silence_candidate(text: str) -> str: return " ".join(text.strip().upper().split()) +def _strip_edge_silence_punctuation(text: str) -> str: + """Strip stray edge punctuation without erasing marker structure. + + Models sometimes emit ``.NO_REPLY`` or ``*NO_REPLY*`` instead of the exact + marker. Keep square brackets structural so malformed ``[SILENT`` does not + become ``SILENT``. + """ + start = 0 + end = len(text) + while start < end and text[start] not in "[]" and unicodedata.category(text[start]).startswith("P"): + start += 1 + while end > start and text[end - 1] not in "[]" and unicodedata.category(text[end - 1]).startswith("P"): + end -= 1 + return text[start:end].strip() + + +def _canonical_silence_candidates(text: str) -> tuple[str, ...]: + exact = _canonical_silence_candidate(text) + stripped = _strip_edge_silence_punctuation(text.strip()) + if stripped == text.strip(): + return (exact,) + fallback = _canonical_silence_candidate(stripped) + return (exact, fallback) + + def is_intentional_silence_response(response: Any) -> bool: """Return True only when ``response`` is exactly a silence marker. @@ -41,7 +67,7 @@ def is_intentional_silence_response(response: Any) -> bool: return False if len(stripped) > 64: return False - return _canonical_silence_candidate(stripped) in LIVE_GATEWAY_SILENT_MARKERS + return any(candidate in LIVE_GATEWAY_SILENT_MARKERS for candidate in _canonical_silence_candidates(stripped)) def is_intentional_silence_agent_result(agent_result: dict | None, response: Any) -> bool: @@ -74,7 +100,7 @@ def is_partial_silence_marker(text: Any) -> bool: stripped = text.strip() if not stripped or len(stripped) > 64: return False - candidate = _canonical_silence_candidate(stripped) - if not candidate: - return False - return any(marker.startswith(candidate) for marker in LIVE_GATEWAY_SILENT_MARKERS) + for candidate in _canonical_silence_candidates(stripped): + if candidate and any(marker.startswith(candidate) for marker in LIVE_GATEWAY_SILENT_MARKERS): + return True + return False diff --git a/tests/gateway/test_response_filters.py b/tests/gateway/test_response_filters.py index 6453d55ce66..c24b9725cb7 100644 --- a/tests/gateway/test_response_filters.py +++ b/tests/gateway/test_response_filters.py @@ -9,10 +9,17 @@ def test_exact_silence_tokens_are_intentional_silence(): assert is_intentional_silence_response(token) +def test_edge_punctuation_silence_tokens_are_intentional_silence(): + for token in (".NO_REPLY", "*NO_REPLY*", " .NO_REPLY ", "*[SILENT]*", "NO_REPLY."): + assert is_intentional_silence_response(token) + + def test_blank_and_prose_mentions_are_not_silence(): assert not is_intentional_silence_response("") assert not is_intentional_silence_response("Use NO_REPLY when no answer is needed.") assert not is_intentional_silence_response("The reply was [SILENT], intentionally.") + assert not is_intentional_silence_response("😄 NO_REPLY") + assert not is_intentional_silence_response("[SILENT") def test_failed_agent_result_never_counts_as_intentional_silence():