ed/manual_slop
Private
Public Access
0
0
Fork 0
Code Activity

refactor(ai_client): migrate _classify_anthropic + _classify_gemini_error to Result[T] (Phase 11 sites 1+2)

Browse Source 
Both classify functions had:
  try:
      sdk = _require_warmed('xxx')
      if isinstance(exc, sdk.SomeException): return ErrorInfo(...)
      ...
  except (ImportError, AttributeError):
      pass
  # body-string matching fallback
  ...

Body: bare 'except: pass' = SS violation (silent recovery).

Migration per TIER1_REVIEW directive (per-site decision):
- Initial attempt: _try_warm_sdk(name) -> Any sentinel (None on failure)
- Audit flagged the sentinel helper as UNCLEAR (Heuristic B requires class
  method with self.attr assignment; module-level sentinel doesn't match)
- Per Phase 9 redo precedent: migrate to Result instead of adding heuristic

Final approach: _try_warm_sdk_result(name) -> Result[Any]
  Returns Result(data=module) on success,
          Result(data=None, errors=[ErrorInfo]) on ImportError/AttributeError.

Classify callers check result.ok and use result.data on success.

Audit: ai_client SS 2 -> 0; UNCLEAR 1 -> 0 (after Result migration).
COMPLIANT 32 -> 33.
This commit is contained in:
ed
2026-06-20 14:10:42 -04:00
parent 48cca536a3
commit 26ebbf7818
2 changed files with 78 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ai_client.py
+24 -9
View File
@@ -284,9 +284,27 @@ def _load_credentials() -> dict[str, Any]:
f"Or set SLOP_CREDENTIALS env var to a custom path."
)
def _classify_anthropic_error(exc: Exception, source: str = "ai_client.anthropic") -> ErrorInfo:
def _try_warm_sdk_result(name: str) -> Result[Any]:
"""Try to get a warmed SDK module. Returns Result[Any].
Lazy-loading sentinel: the caller checks result.ok and uses result.data
on success. On failure, returns Result(errors=[ErrorInfo]). The caller
falls back to body-string matching, preserving the original behavior.
Per Phase 11 anti-sliming protocol: NOT a sentinel-None return; the
caller observes the Result explicitly.
"""
try:
anthropic = _require_warmed("anthropic")
return Result(data=_require_warmed(name))
except (ImportError, AttributeError) as e:
return Result(
data=None,
errors=[ErrorInfo(kind=ErrorKind.INTERNAL, message=f"SDK module '{name}' unavailable: {e}", source=f"ai_client._try_warm_sdk_result", original=e)],
)
def _classify_anthropic_error(exc: Exception, source: str = "ai_client.anthropic") -> ErrorInfo:
sdk_result = _try_warm_sdk_result("anthropic")
if sdk_result.ok:
anthropic = sdk_result.data
if isinstance(exc, anthropic.RateLimitError): return ErrorInfo(kind=ErrorKind.RATE_LIMIT, message=str(exc), source=source, original=exc)
if isinstance(exc, anthropic.AuthenticationError): return ErrorInfo(kind=ErrorKind.AUTH, message=str(exc), source=source, original=exc)
if isinstance(exc, anthropic.PermissionDeniedError): return ErrorInfo(kind=ErrorKind.AUTH, message=str(exc), source=source, original=exc)
@@ -299,24 +317,21 @@ def _classify_anthropic_error(exc: Exception, source: str = "ai_client.anthropic
if status == 402: return ErrorInfo(kind=ErrorKind.BALANCE, message=str(exc), source=source, original=exc)
if "credit" in body or "balance" in body or "billing" in body: return ErrorInfo(kind=ErrorKind.BALANCE, message=str(exc), source=source, original=exc)
if "quota" in body or "limit" in body or "exceeded" in body: return ErrorInfo(kind=ErrorKind.QUOTA, message=str(exc), source=source, original=exc)
except ImportError:
pass
return ErrorInfo(kind=ErrorKind.UNKNOWN, message=str(exc), source=source, original=exc)
def _classify_gemini_error(exc: Exception, source: str = "ai_client.gemini") -> ErrorInfo:
body = str(exc).lower()
try:
gac = _require_warmed("google.api_core.exceptions")
sdk_result = _try_warm_sdk_result("google.api_core.exceptions")
if sdk_result.ok:
gac = sdk_result.data
if isinstance(exc, gac.ResourceExhausted): return ErrorInfo(kind=ErrorKind.QUOTA, message=str(exc), source=source, original=exc)
if isinstance(exc, gac.TooManyRequests): return ErrorInfo(kind=ErrorKind.RATE_LIMIT, message=str(exc), source=source, original=exc)
if isinstance(exc, (gac.Unauthenticated, gac.PermissionDenied)): return ErrorInfo(kind=ErrorKind.AUTH, message=str(exc), source=source, original=exc)
if isinstance(exc, gac.ServiceUnavailable): return ErrorInfo(kind=ErrorKind.NETWORK, message=str(exc), source=source, original=exc)
except (ImportError, AttributeError):
pass
if "429" in body or "quota" in body or "resource exhausted" in body: return ErrorInfo(kind=ErrorKind.QUOTA, message=str(exc), source=source, original=exc)
if "rate" in body and "limit" in body: return ErrorInfo(kind=ErrorKind.RATE_LIMIT, message=str(exc), source=source, original=exc)
if "401" in body or "403" in body or "api key" in body or "unauthenticated" in body: return ErrorInfo(kind=ErrorKind.AUTH, message=str(exc), source=source, original=exc)
if "402" in body or "billing" in body or "balance" in body or "payment" in body: return ErrorInfo(kind=ErrorKind.BALANCE, message=str(exc), source=source, original=exc)
if "402" in body or "billing" in body or "balance" in body or "payment" in body: return ErrorInfo(kind=ErrorKind.BALANCE, message=str(exc), source=source, original=exc)
if "connection" in body or "timeout" in body or "unreachable" in body: return ErrorInfo(kind=ErrorKind.NETWORK, message=str(exc), source=source, original=exc)
return ErrorInfo(kind=ErrorKind.UNKNOWN, message=str(exc), source=source, original=exc)
tests/tier2/phase11_sites12_test.py
+54
View File
@@ -0,0 +1,54 @@
"""Phase 11 sites 1+2: _classify_anthropic_error + _classify_gemini_error.
Both have:
try:
sdk = _require_warmed("xxx")
if isinstance(exc, sdk.SomeException): return ErrorInfo(...)
...
except (ImportError, AttributeError):
pass
# body-string matching fallback
...
Body: pass = SS violation (silent recovery).
Migration: extract a _try_warm_sdk sentinel helper. Caller checks for
None and proceeds. The sentinel helper itself uses 'try: return ...;
except: return None' which may be flagged by the audit as SS initially;
if so, it should be classified as a lazy-loading sentinel (Phase 11 may
need a heuristic addition).
"""
import sys
sys.path.insert(0, ".")
def test_phase11_sites12_try_warm_sdk_result_helper_exists():
import src.ai_client
assert hasattr(src.ai_client, "_try_warm_sdk_result"), \
"_try_warm_sdk_result helper missing"
def test_phase11_sites12_classify_anthropic_uses_helper():
import inspect
import src.ai_client
src_text = inspect.getsource(src.ai_client._classify_anthropic_error)
assert "_try_warm_sdk_result" in src_text, \
"_classify_anthropic_error should use _try_warm_sdk_result helper"
assert "except ImportError" not in src_text, \
"_classify_anthropic_error must NOT have 'except ImportError'"
def test_phase11_sites12_classify_gemini_uses_helper():
import inspect
import src.ai_client
src_text = inspect.getsource(src.ai_client._classify_gemini_error)
assert "_try_warm_sdk_result" in src_text, \
"_classify_gemini_error should use _try_warm_sdk_result helper"
assert "except ImportError" not in src_text and "except (ImportError, AttributeError)" not in src_text, \
"_classify_gemini_error must NOT have raw except ImportError/AttributeError"
def test_phase11_sites12_legacy_preserved():
import src.ai_client
assert callable(getattr(src.ai_client, "_classify_anthropic_error", None))
assert callable(getattr(src.ai_client, "_classify_gemini_error", None))