diff --git a/amplifier_module_loop_live/failures.py b/amplifier_module_loop_live/failures.py
index 9510d06..f94127d 100644
--- a/amplifier_module_loop_live/failures.py
+++ b/amplifier_module_loop_live/failures.py
@@ -1,5 +1,7 @@
 """Bounded public turn failures; exception payloads remain private causes."""
 
+import re
+
 from amplifier_core.llm_errors import (
     AuthenticationError,
     ContentFilterError,
@@ -12,6 +14,30 @@
 )
 
 
+def count_diagnostic(error):
+    """Recognize an optional provider contract without importing its SDK."""
+    if (type(error).__module__, type(error).__name__) != (
+            'amplifier_module_provider_openai._token_count', 'TokenCountError'):
+        return None
+    value = getattr(error, 'count_failure', None)
+    if not isinstance(value, dict):
+        return {}
+    result = {}
+    if isinstance(value.get('category'), str) and value['category'] in {'timeout', 'connection', 'rate_limit', 'service',
+                                'authentication', 'permission', 'quota', 'invalid_request', 'invalid_response'}:
+        result['category'] = value['category']
+    result['retryable'] = value.get('retryable') is True and result.get('category') in {
+        'timeout', 'connection', 'rate_limit', 'service'}
+    for key, low, high in [('httpStatus', 400, 599), ('attempts', 1, 3)]:
+        number = value.get(key)
+        if type(number) is int and low <= number <= high:
+            result[key] = number
+    request_id = value.get('requestId')
+    if isinstance(request_id, str) and re.fullmatch(r'[A-Za-z0-9_-]{1,100}', request_id):
+        result['requestId'] = request_id
+    return result
+
+
 def turn_failure(error, stage="manager_turn"):
     """Describe a failure without copying SDK bodies, prompts, paths or keys.
 
@@ -32,7 +58,11 @@ def turn_failure(error, stage="manager_turn"):
                   "native_measurement_unavailable", "native_compaction_failed",
                   "disabled", "request_context_unavailable", "invalid_native_contract",
                   "authoritative_measurement_unavailable"}
-    if compaction:
+    count_failure = count_diagnostic(error)
+    if count_failure is not None:
+        category, stage = 'context_measurement', 'context_preparation'
+        message = 'Could not check conversation size. Saved history and checkpoint are preserved.'
+    elif compaction:
         category, stage = "context_compaction", "context_preparation"
         message = "Context compaction failed. Original history is preserved; repair context preparation before continuing."
     elif isinstance(error, ContextLengthError):
@@ -60,9 +90,11 @@ def turn_failure(error, stage="manager_turn"):
     kind = next((cls.__name__ for cls in type(error).__mro__
                  if cls.__module__ in {"builtins", "amplifier_core.llm_errors"}), "Exception")
     return {**({"error_code": code} if code in safe_codes else {}),
+            **({'count_failure': count_failure} if count_failure is not None else {}),
             "error_type": "CompactionError" if compaction else kind, "error_category": category, "error_stage": stage,
             "error_message": message, "effects": "not_rolled_back", "replayed": False,
-            "retryable": isinstance(error, LLMError) and error.retryable is True}
+            "retryable": (count_failure.get('retryable', False) if count_failure is not None
+                          else isinstance(error, LLMError) and error.retryable is True)}
 
 
 class ManagerTurnError(RuntimeError):
diff --git a/tests/test_failures.py b/tests/test_failures.py
index 2286cbd..1674951 100644
--- a/tests/test_failures.py
+++ b/tests/test_failures.py
@@ -73,3 +73,28 @@ def test_compaction_failure_preserves_bounded_code_without_optional_dependency(c
     assert 'secret' not in json.dumps(failure)
     assert 'private payload' not in json.dumps(failure)
     assert failure['retryable'] is False
+
+
+def test_token_count_failure_has_safe_diagnostic_without_generation_retry():
+    from amplifier_core.llm_errors import LLMError
+    cls = type('TokenCountError', (LLMError,), {'__module__': 'amplifier_module_provider_openai._token_count'})
+    error = cls('private SDK body', provider='openai', retryable=False)
+    error.count_failure = {'category': 'service', 'retryable': True, 'httpStatus': 503,
+                           'attempts': 3, 'requestId': 'req_123', 'body': 'SECRET'}
+    result = turn_failure(error)
+    assert result['error_category'] == 'context_measurement'
+    assert result['error_stage'] == 'context_preparation'
+    assert result['retryable'] is True and error.retryable is False
+    assert result['count_failure'] == {k: v for k, v in error.count_failure.items() if k != 'body'}
+    assert 'SECRET' not in json.dumps(result) and 'private' not in json.dumps(result)
+
+
+def test_token_count_diagnostic_rejects_unknown_values():
+    cls = type('TokenCountError', (RuntimeError,), {'__module__': 'amplifier_module_provider_openai._token_count'})
+    error = cls('secret')
+    error.count_failure = {'category': 'secret', 'retryable': True, 'httpStatus': True,
+                           'attempts': 999, 'requestId': 'https://secret@example.com'}
+    result = turn_failure(error)
+    assert result['count_failure'] == {'retryable': False}
+    assert not result['retryable']
+    assert 'secret' not in json.dumps(result)