diff --git a/src/quant_strategy_plugins/ai_audit.py b/src/quant_strategy_plugins/ai_audit.py index e4968b6..d650ae5 100644 --- a/src/quant_strategy_plugins/ai_audit.py +++ b/src/quant_strategy_plugins/ai_audit.py @@ -4,14 +4,8 @@ import logging import os import re -import subprocess -import tempfile -import time -import urllib.error -import urllib.request from collections.abc import Callable, Mapping, Sequence from dataclasses import dataclass -from pathlib import Path from typing import Any _logger = logging.getLogger(__name__) @@ -28,8 +22,6 @@ DEFAULT_ANTHROPIC_VERSION = "2023-06-01" DEFAULT_AI_AUDIT_TIMEOUT_SECONDS = 15.0 AI_AUDIT_SCHEMA_VERSION = "strategy_plugin_ai_audit.v1" -DEFAULT_MAX_RETRIES = 2 -DEFAULT_BACKOFF_BASE_SECONDS = 1.0 SANITIZE_MAX_FIELD_LENGTH = 2000 # Patterns that may appear in upstream error responses and must be scrubbed. @@ -141,38 +133,6 @@ def _scrub_api_key_from_text(text: str) -> str: return text -def _should_retry(status_code: int | None) -> bool: - return status_code is not None and (status_code == 429 or status_code >= 500) - - -def _retry_with_backoff( - fn: Callable[[], str], - *, - max_retries: int = DEFAULT_MAX_RETRIES, - base_seconds: float = DEFAULT_BACKOFF_BASE_SECONDS, -) -> str: - """Call *fn* with exponential backoff on retriable HTTP errors.""" - last_exc: Exception | None = None - for attempt in range(max_retries + 1): - try: - return fn() - except AiAuditError as exc: - last_exc = exc - cause = exc.__cause__ - status = None - if isinstance(cause, urllib.error.HTTPError): - status = cause.code - if not _should_retry(status) or attempt >= max_retries: - raise - wait = base_seconds * (2 ** attempt) - _logger.warning( - "ai_audit attempt %d/%d failed with status %s; retrying in %.1fs", - attempt + 1, max_retries + 1, status, wait, - ) - time.sleep(wait) - raise last_exc # type: ignore[misc] - - def build_ai_audit_endpoints( *, api_key: str | None = None, @@ -327,132 +287,18 @@ def build_ai_audit_endpoints( ).normalized() ) - return tuple(endpoints) - - -def _chat_completions_url(base_url: str) -> str: - url = str(base_url or DEFAULT_AI_AUDIT_BASE_URL).strip().rstrip("/") - if url.endswith("/chat/completions"): - return url - return f"{url}/chat/completions" - + if os.environ.get("CODEX_AUDIT_SERVICE_URL", "").strip() and not endpoints: + endpoints.append( + AiAuditEndpoint( + name="primary", + api_key="", + provider=PROVIDER_OPENAI, + base_url="gateway", + model=primary_model or DEFAULT_AI_AUDIT_MODEL, + ).normalized() + ) -def _openai_compatible_chat_completion( - endpoint: AiAuditEndpoint, - messages: Sequence[Mapping[str, str]], - timeout_seconds: float, -) -> str: - endpoint = endpoint.normalized() - body = json.dumps( - { - "model": endpoint.model, - "messages": list(messages), - "temperature": 0, - "max_tokens": 700, - } - ).encode("utf-8") - request = urllib.request.Request( - _chat_completions_url(endpoint.base_url), - data=body, - headers={ - "Authorization": f"Bearer {endpoint.api_key}", - "Content-Type": "application/json", - }, - method="POST", - ) - def _call() -> str: - try: - with urllib.request.urlopen(request, timeout=float(timeout_seconds)) as response: - response_body = response.read().decode("utf-8") - except urllib.error.HTTPError as exc: - detail = exc.read().decode("utf-8", errors="replace")[:500] - detail = _scrub_api_key_from_text(detail) - raise AiAuditError(f"HTTP {exc.code}: {detail}") from exc - except (urllib.error.URLError, OSError, ValueError) as exc: - raise AiAuditError(f"network or encoding error: {_scrub_api_key_from_text(str(exc))}") from exc - - payload = json.loads(response_body) - choices = payload.get("choices") if isinstance(payload, Mapping) else None - if not choices: - raise AiAuditError("empty completion choices") - first = choices[0] - if not isinstance(first, Mapping): - raise AiAuditError("invalid completion choice") - message = first.get("message") - if isinstance(message, Mapping): - content = message.get("content") - else: - content = first.get("text") - text = str(content or "").strip() - if not text: - raise AiAuditError("empty completion content") - return text - - return _retry_with_backoff(_call) - - -def _anthropic_messages_url(base_url: str) -> str: - url = str(base_url or DEFAULT_ANTHROPIC_BASE_URL).strip().rstrip("/") - if url.endswith("/messages"): - return url - return f"{url}/messages" - - -def _anthropic_messages_completion( - endpoint: AiAuditEndpoint, - messages: Sequence[Mapping[str, str]], - timeout_seconds: float, -) -> str: - endpoint = endpoint.normalized() - system_parts = [str(message.get("content") or "") for message in messages if message.get("role") == "system"] - user_messages = [ - {"role": str(message.get("role") or "user"), "content": str(message.get("content") or "")} - for message in messages - if message.get("role") != "system" - ] - body = json.dumps( - { - "model": endpoint.model, - "max_tokens": 700, - "system": "\n\n".join(part for part in system_parts if part.strip()), - "messages": user_messages, - } - ).encode("utf-8") - request = urllib.request.Request( - _anthropic_messages_url(endpoint.base_url), - data=body, - headers={ - "x-api-key": endpoint.api_key, - "anthropic-version": endpoint.api_version or DEFAULT_ANTHROPIC_VERSION, - "Content-Type": "application/json", - }, - method="POST", - ) - def _call() -> str: - try: - with urllib.request.urlopen(request, timeout=float(timeout_seconds)) as response: - response_body = response.read().decode("utf-8") - except urllib.error.HTTPError as exc: - detail = exc.read().decode("utf-8", errors="replace")[:500] - detail = _scrub_api_key_from_text(detail) - raise AiAuditError(f"HTTP {exc.code}: {detail}") from exc - except (urllib.error.URLError, OSError, ValueError) as exc: - raise AiAuditError(f"network or encoding error: {_scrub_api_key_from_text(str(exc))}") from exc - - payload = json.loads(response_body) - content = payload.get("content") if isinstance(payload, Mapping) else None - if not isinstance(content, Sequence) or isinstance(content, (str, bytes, bytearray)): - raise AiAuditError("Anthropic response did not include content") - text_parts = [ - str(block.get("text") or "").strip() - for block in content - if isinstance(block, Mapping) and block.get("type") == "text" and str(block.get("text") or "").strip() - ] - if not text_parts: - raise AiAuditError("Anthropic response did not include text content") - return "\n\n".join(text_parts) - - return _retry_with_backoff(_call) + return tuple(endpoints) def _complete_with_endpoint( @@ -462,24 +308,17 @@ def _complete_with_endpoint( ) -> str: endpoint = endpoint.normalized() - # Route through AiGateway when CODEX_AUDIT_SERVICE_URL is configured. # API keys live on the VPS — no keys in plugin config needed. gateway_url = os.environ.get("CODEX_AUDIT_SERVICE_URL", "").strip() - if gateway_url: - prompt = "\n\n".join( - f"{str(m.get('role') or 'user').upper()}:\n{str(m.get('content') or '').strip()}" - for m in messages if str(m.get("content") or "").strip() - ) - if endpoint.provider == PROVIDER_CODEX: - return _codex_via_gateway(prompt, endpoint.model, timeout_seconds) - return _llm_via_gateway(prompt, endpoint.model, endpoint.provider, timeout_seconds) - - # Fallback: direct API / subprocess calls + if not gateway_url: + raise AiAuditError("ai_gateway_unavailable") + prompt = "\n\n".join( + f"{str(m.get('role') or 'user').upper()}:\n{str(m.get('content') or '').strip()}" + for m in messages if str(m.get("content") or "").strip() + ) if endpoint.provider == PROVIDER_CODEX: - return _codex_exec_completion(endpoint, messages, timeout_seconds) - if endpoint.provider == PROVIDER_ANTHROPIC: - return _anthropic_messages_completion(endpoint, messages, timeout_seconds) - return _openai_compatible_chat_completion(endpoint, messages, timeout_seconds) + return _codex_via_gateway(prompt, endpoint.model, timeout_seconds) + return _llm_via_gateway(prompt, endpoint.model, endpoint.provider, timeout_seconds) def _codex_via_gateway(prompt: str, model: str, timeout_seconds: float) -> str: @@ -489,17 +328,15 @@ def _codex_via_gateway(prompt: str, model: str, timeout_seconds: float) -> str: config = GatewayConfig.from_env() client = AiGatewayClient(config) result = client.execute(prompt, mode="review_only", model=model, timeout=timeout_seconds) - if result.success: - return result.output - raise AiAuditError(result.error) + if result.success and str(result.output or "").strip(): + return str(result.output) + raise AiAuditError("ai_gateway_rejected") except ImportError: - return _codex_exec_direct(prompt, timeout_seconds) - except Exception as exc: - _logger.warning( - "ai_audit gateway codex call failed: %s; falling back to direct", - _scrub_api_key_from_text(str(exc)), - ) - return _codex_exec_direct(prompt, timeout_seconds) + raise AiAuditError("ai_gateway_client_unavailable") from None + except AiAuditError: + raise + except Exception: + raise AiAuditError("ai_gateway_request_failed") from None def _llm_via_gateway(prompt: str, model: str, provider: str, timeout_seconds: float) -> str: @@ -509,63 +346,28 @@ def _llm_via_gateway(prompt: str, model: str, provider: str, timeout_seconds: fl config = GatewayConfig.from_env() client = AiGatewayClient(config) result = client.analyze(prompt, model=model, timeout=timeout_seconds) - if result.success: - return result.output - raise AiAuditError(result.error) + actual_provider = str(getattr(result, "provider", "") or "").strip().lower() + if actual_provider != provider: + raise AiAuditError("ai_gateway_provider_mismatch") + if result.success and str(result.output or "").strip(): + return str(result.output) + raise AiAuditError("ai_gateway_rejected") except ImportError: - return _llm_direct(prompt, model, provider, timeout_seconds) - except Exception as exc: - _logger.warning( - "ai_audit gateway analyze call failed: %s; falling back to direct", - _scrub_api_key_from_text(str(exc)), - ) - return _llm_direct(prompt, model, provider, timeout_seconds) + raise AiAuditError("ai_gateway_client_unavailable") from None + except AiAuditError: + raise + except Exception: + raise AiAuditError("ai_gateway_request_failed") from None def _llm_direct(prompt: str, model: str, provider: str, timeout_seconds: float) -> str: - """Direct API call fallback when gateway is unavailable.""" - endpoint = AiAuditEndpoint( - name="fallback", api_key="", provider=provider, - base_url="", model=model, - ).normalized() - messages: tuple[Mapping[str, str], ...] = ({"role": "user", "content": prompt},) - if provider == PROVIDER_ANTHROPIC: - return _anthropic_messages_completion(endpoint, messages, timeout_seconds) - return _openai_compatible_chat_completion(endpoint, messages, timeout_seconds) + del prompt, model, provider, timeout_seconds + raise AiAuditError("direct_ai_completion_forbidden") def _codex_exec_direct(prompt: str, timeout_seconds: float) -> str: - """Direct codex exec fallback when gateway is unavailable.""" - with tempfile.TemporaryDirectory(prefix="qsp-ai-audit-") as temp_dir: - output_path = Path(temp_dir) / "codex-final-message.md" - command = ["codex", "exec", "--cd", temp_dir, "--output-last-message", str(output_path), "-"] - try: - result = subprocess.run( - command, input=prompt, text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, - timeout=float(timeout_seconds), check=False, env=_scrubbed_codex_env(), - ) - except FileNotFoundError as exc: - raise AiAuditError("codex command was not found") from exc - except subprocess.TimeoutExpired as exc: - raise AiAuditError(f"codex command timed out after {timeout_seconds:g}s") from exc - if result.returncode != 0: - detail = _bounded_text(result.stdout or "", limit=300) - raise AiAuditError(f"codex command failed with exit code {result.returncode}: {detail}") - text = output_path.read_text(encoding="utf-8").strip() if output_path.exists() else "" - if not text: - text = str(result.stdout or "").strip() - if not text: - raise AiAuditError("codex command returned empty output") - return text - - -def _scrubbed_codex_env() -> dict[str, str]: - secret_markers = ("TOKEN", "SECRET", "PASSWORD", "PRIVATE_KEY", "CREDENTIAL", "API_KEY") - return { - key: value - for key, value in os.environ.items() - if not any(marker in key.upper() for marker in secret_markers) - } + del prompt, timeout_seconds + raise AiAuditError("direct_ai_completion_forbidden") def _codex_exec_completion( @@ -573,47 +375,8 @@ def _codex_exec_completion( messages: Sequence[Mapping[str, str]], timeout_seconds: float, ) -> str: - del endpoint - prompt = "\n\n".join( - f"{str(message.get('role') or 'user').upper()}:\n{str(message.get('content') or '').strip()}" - for message in messages - if str(message.get("content") or "").strip() - ) - with tempfile.TemporaryDirectory(prefix="qsp-ai-audit-") as temp_dir: - output_path = Path(temp_dir) / "codex-final-message.md" - command = [ - "codex", - "exec", - "--cd", - temp_dir, - "--output-last-message", - str(output_path), - "-", - ] - try: - result = subprocess.run( - command, - input=prompt, - text=True, - stdout=subprocess.PIPE, - stderr=subprocess.STDOUT, - timeout=float(timeout_seconds), - check=False, - env=_scrubbed_codex_env(), - ) - except FileNotFoundError as exc: - raise AiAuditError("codex command was not found") from exc - except subprocess.TimeoutExpired as exc: - raise AiAuditError(f"codex command timed out after {timeout_seconds:g}s") from exc - if result.returncode != 0: - detail = _bounded_text(result.stdout or "", limit=300) - raise AiAuditError(f"codex command failed with exit code {result.returncode}: {detail}") - text = output_path.read_text(encoding="utf-8").strip() if output_path.exists() else "" - if not text: - text = str(result.stdout or "").strip() - if not text: - raise AiAuditError("codex command returned empty output") - return text + del endpoint, messages, timeout_seconds + raise AiAuditError("direct_ai_completion_forbidden") def _extract_json_object(value: str | Mapping[str, Any]) -> dict[str, Any]: @@ -893,6 +656,20 @@ def _run_ai_audit( "notification_profile": "shadow_only", }, } + if not os.environ.get("CODEX_AUDIT_SERVICE_URL", "").strip(): + return { + **base_payload, + "status": "skipped", + "skip_reason": "gateway_unavailable", + "attempts": [], + } + if completion_client is not None: + return { + **base_payload, + "status": "skipped", + "skip_reason": "custom_completion_client_forbidden", + "attempts": [], + } if not endpoints: return { **base_payload, @@ -901,11 +678,10 @@ def _run_ai_audit( "attempts": [], } - client = completion_client or _complete_with_endpoint attempts: list[dict[str, Any]] = [] for endpoint in endpoints: try: - raw_response = client(endpoint, messages, float(timeout_seconds)) + raw_response = _complete_with_endpoint(endpoint, messages, float(timeout_seconds)) audit_response = _normalize_ai_audit_response(_extract_json_object(raw_response)) attempts.append({**endpoint.report(), "status": "ok"}) diff --git a/tests/test_ai_audit.py b/tests/test_ai_audit.py index 0300dc6..30a725c 100644 --- a/tests/test_ai_audit.py +++ b/tests/test_ai_audit.py @@ -1,4 +1,18 @@ -from quant_strategy_plugins.ai_audit import _failure_text, _scrub_api_key_from_text, build_ai_audit_endpoints +import sys +import types + +import pytest + +from quant_strategy_plugins import ai_audit +from quant_strategy_plugins.ai_audit import ( + AiAuditError, + _codex_via_gateway, + _failure_text, + _llm_via_gateway, + _run_ai_audit, + _scrub_api_key_from_text, + build_ai_audit_endpoints, +) def _clear_ai_audit_env(monkeypatch) -> None: @@ -12,6 +26,7 @@ def _clear_ai_audit_env(monkeypatch) -> None: "QSP_STRATEGY_PLUGIN_AI_AUDIT_ANTHROPIC_API_KEY", "QSP_CRISIS_AI_AUDIT_ANTHROPIC_API_KEY", "ANTHROPIC_API_KEY", + "CODEX_AUDIT_SERVICE_URL", ): monkeypatch.delenv(key, raising=False) @@ -64,3 +79,177 @@ def test_ai_audit_failure_text_redacts_secret_values() -> None: assert "password=[REDACTED]" in text assert password_value not in text + + +def test_ai_audit_skips_when_gateway_is_unavailable(monkeypatch) -> None: + _clear_ai_audit_env(monkeypatch) + calls: list[str] = [] + + payload = _run_ai_audit( + {"canonical_route": "true_crisis", "suggested_action": "defend"}, + audit_kind="crisis_response_shadow", + messages=({"role": "user", "content": "audit"},), + enabled=True, + api_key="sk-primary", + codex_enabled=False, + completion_client=lambda *_args: calls.append("direct") or "{}", + ) + + assert payload["status"] == "skipped" + assert payload["skip_reason"] == "gateway_unavailable" + assert payload["final_route_unchanged"] is True + assert calls == [] + + +def test_ai_audit_rejects_custom_completion_client(monkeypatch) -> None: + monkeypatch.setenv("CODEX_AUDIT_SERVICE_URL", "https://gateway.example") + calls: list[str] = [] + + payload = _run_ai_audit( + {"canonical_route": "true_crisis", "suggested_action": "defend"}, + audit_kind="crisis_response_shadow", + messages=({"role": "user", "content": "audit"},), + enabled=True, + completion_client=lambda *_args: calls.append("custom") or "{}", + ) + + assert payload["status"] == "skipped" + assert payload["skip_reason"] == "custom_completion_client_forbidden" + assert calls == [] + + +def test_gateway_uses_default_model_without_local_provider_key(monkeypatch) -> None: + _clear_ai_audit_env(monkeypatch) + monkeypatch.setenv("CODEX_AUDIT_SERVICE_URL", "https://gateway.example") + calls: list[tuple[str, str]] = [] + + monkeypatch.setattr( + ai_audit, + "_complete_with_endpoint", + lambda endpoint, *_args: calls.append((endpoint.provider, endpoint.model)) + or '{"verdict":"agree","summary":"ok","risk_flags":[],"evidence_gaps":[],"confidence":0.5}', + ) + payload = _run_ai_audit( + {"canonical_route": "true_crisis", "suggested_action": "defend"}, + audit_kind="crisis_response_shadow", + messages=({"role": "user", "content": "audit"},), + enabled=True, + codex_enabled=False, + ) + + assert payload["status"] == "ok" + assert calls == [("openai", "gpt-5.4-mini")] + + +def test_gateway_success_calls_analyze_and_execute(monkeypatch) -> None: + calls: list[tuple[str, tuple, dict]] = [] + + class SuccessfulGatewayClient: + def __init__(self, _config) -> None: + pass + + def analyze(self, *args, **kwargs): + calls.append(("analyze", args, kwargs)) + return types.SimpleNamespace(success=True, output="analysis", provider="openai") + + def execute(self, *args, **kwargs): + calls.append(("execute", args, kwargs)) + return types.SimpleNamespace(success=True, output="review", provider="codex") + + monkeypatch.setitem( + sys.modules, + "ai_gateway_client", + types.SimpleNamespace( + AiGatewayClient=SuccessfulGatewayClient, + GatewayConfig=types.SimpleNamespace(from_env=lambda: object()), + ), + ) + + assert _llm_via_gateway("audit", "gpt-test", "openai", 3.0) == "analysis" + assert _codex_via_gateway("review", "codex-test", 4.0) == "review" + assert calls == [ + ("analyze", ("audit",), {"model": "gpt-test", "timeout": 3.0}), + ("execute", ("review",), {"mode": "review_only", "model": "codex-test", "timeout": 4.0}), + ] + + +@pytest.mark.parametrize("actual_provider", ["anthropic", ""]) +def test_gateway_provider_mismatch_fails_closed(monkeypatch, actual_provider: str) -> None: + class MismatchedGatewayClient: + def __init__(self, _config) -> None: + pass + + def analyze(self, *_args, **_kwargs): + return types.SimpleNamespace(success=True, output="analysis", provider=actual_provider) + + monkeypatch.setitem( + sys.modules, + "ai_gateway_client", + types.SimpleNamespace( + AiGatewayClient=MismatchedGatewayClient, + GatewayConfig=types.SimpleNamespace(from_env=lambda: object()), + ), + ) + + with pytest.raises(AiAuditError, match="ai_gateway_provider_mismatch"): + _llm_via_gateway("audit", "gpt-test", "openai", 3.0) + + +@pytest.mark.parametrize( + ("gateway_call", "direct_fallback"), + [ + (_llm_via_gateway, "_llm_direct"), + (_codex_via_gateway, "_codex_exec_direct"), + ], +) +def test_gateway_failure_never_uses_direct_fallback(monkeypatch, gateway_call, direct_fallback) -> None: + class FailingGatewayClient: + def __init__(self, _config) -> None: + pass + + def analyze(self, *_args, **_kwargs): + raise RuntimeError("provider token=supersecret123 failed") + + def execute(self, *_args, **_kwargs): + raise RuntimeError("provider token=supersecret123 failed") + + monkeypatch.setitem( + sys.modules, + "ai_gateway_client", + types.SimpleNamespace( + AiGatewayClient=FailingGatewayClient, + GatewayConfig=types.SimpleNamespace(from_env=lambda: object()), + ), + ) + direct_calls: list[str] = [] + monkeypatch.setattr(ai_audit, direct_fallback, lambda *_args: direct_calls.append("direct") or "{}") + + with pytest.raises(AiAuditError, match="ai_gateway_request_failed") as exc_info: + if gateway_call is _llm_via_gateway: + gateway_call("audit", "test-model", "openai", 1.0) + else: + gateway_call("audit", "test-model", 1.0) + + assert "supersecret123" not in str(exc_info.value) + assert direct_calls == [] + + +@pytest.mark.parametrize( + ("gateway_call", "direct_fallback"), + [ + (_llm_via_gateway, "_llm_direct"), + (_codex_via_gateway, "_codex_exec_direct"), + ], +) +def test_gateway_client_import_failure_never_uses_direct_fallback(monkeypatch, gateway_call, direct_fallback) -> None: + monkeypatch.setitem(sys.modules, "ai_gateway_client", None) + direct_calls: list[str] = [] + monkeypatch.setattr(ai_audit, direct_fallback, lambda *_args: direct_calls.append("direct") or "{}") + + with pytest.raises(AiAuditError, match="ai_gateway_client_unavailable"): + if gateway_call is _llm_via_gateway: + gateway_call("audit", "test-model", "openai", 1.0) + else: + gateway_call("audit", "test-model", 1.0) + + assert direct_calls == [] diff --git a/tests/test_crisis_response_shadow_plugin.py b/tests/test_crisis_response_shadow_plugin.py index b6579d9..c9d4244 100644 --- a/tests/test_crisis_response_shadow_plugin.py +++ b/tests/test_crisis_response_shadow_plugin.py @@ -84,7 +84,8 @@ def test_shadow_signal_routes_financial_credit_crisis_without_live_execution() - assert "ai_audit" not in payload -def test_shadow_signal_ai_audit_uses_fallback_without_changing_route() -> None: +def test_shadow_signal_ai_audit_uses_gateway_fallback_without_changing_route(monkeypatch) -> None: + monkeypatch.setenv("CODEX_AUDIT_SERVICE_URL", "https://gateway.example") prices = _financial_crisis_prices() as_of = str(pd.to_datetime(prices["as_of"]).max().date()) calls: list[str] = [] @@ -105,6 +106,7 @@ def fake_completion(endpoint, messages, timeout_seconds): "human_review_recommended": False, } + monkeypatch.setattr("quant_strategy_plugins.ai_audit._complete_with_endpoint", fake_completion) payload = build_crisis_response_shadow_signal( prices, events=(), @@ -122,7 +124,6 @@ def fake_completion(endpoint, messages, timeout_seconds): ai_audit_fallback_model="fallback-model", ai_audit_codex_enabled=False, ai_audit_timeout_seconds=7.0, - ai_audit_completion_client=fake_completion, ) assert payload["canonical_route"] == ROUTE_TRUE_CRISIS @@ -140,7 +141,8 @@ def fake_completion(endpoint, messages, timeout_seconds): assert audit["attempts"][1]["status"] == "ok" -def test_shadow_signal_ai_audit_uses_anthropic_provider_fallback() -> None: +def test_shadow_signal_ai_audit_uses_gateway_anthropic_fallback(monkeypatch) -> None: + monkeypatch.setenv("CODEX_AUDIT_SERVICE_URL", "https://gateway.example") prices = _financial_crisis_prices() as_of = str(pd.to_datetime(prices["as_of"]).max().date()) calls: list[tuple[str, str]] = [] @@ -159,6 +161,7 @@ def fake_completion(endpoint, _messages, _timeout_seconds): "human_review_recommended": True, } + monkeypatch.setattr("quant_strategy_plugins.ai_audit._complete_with_endpoint", fake_completion) payload = build_crisis_response_shadow_signal( prices, events=(), @@ -174,7 +177,6 @@ def fake_completion(endpoint, _messages, _timeout_seconds): ai_audit_anthropic_api_key="sk-ant", ai_audit_anthropic_model="anthropic-model", ai_audit_anthropic_version="2023-06-01", - ai_audit_completion_client=fake_completion, ) audit = payload["ai_audit"] @@ -187,7 +189,7 @@ def fake_completion(endpoint, _messages, _timeout_seconds): assert audit["final_route_unchanged"] is True -def test_shadow_signal_ai_audit_skips_without_api_key(monkeypatch) -> None: +def test_shadow_signal_ai_audit_skips_without_gateway(monkeypatch) -> None: for key in ( "QSP_STRATEGY_PLUGIN_AI_AUDIT_API_KEY", "QSP_CRISIS_AI_AUDIT_API_KEY", @@ -198,6 +200,7 @@ def test_shadow_signal_ai_audit_skips_without_api_key(monkeypatch) -> None: "QSP_STRATEGY_PLUGIN_AI_AUDIT_ANTHROPIC_API_KEY", "QSP_CRISIS_AI_AUDIT_ANTHROPIC_API_KEY", "ANTHROPIC_API_KEY", + "CODEX_AUDIT_SERVICE_URL", ): monkeypatch.delenv(key, raising=False) @@ -217,11 +220,12 @@ def test_shadow_signal_ai_audit_skips_without_api_key(monkeypatch) -> None: assert payload["canonical_route"] == ROUTE_TRUE_CRISIS assert payload["ai_audit"]["status"] == "skipped" - assert payload["ai_audit"]["skip_reason"] == "missing_api_endpoint" + assert payload["ai_audit"]["skip_reason"] == "gateway_unavailable" assert payload["ai_audit"]["final_route_unchanged"] is True -def test_shadow_signal_ai_audit_prefers_codex_provider() -> None: +def test_shadow_signal_ai_audit_prefers_gateway_codex_provider(monkeypatch) -> None: + monkeypatch.setenv("CODEX_AUDIT_SERVICE_URL", "https://gateway.example") prices = _financial_crisis_prices() as_of = str(pd.to_datetime(prices["as_of"]).max().date()) calls: list[tuple[str, str]] = [] @@ -238,6 +242,7 @@ def fake_completion(endpoint, _messages, _timeout_seconds): "human_review_recommended": False, } + monkeypatch.setattr("quant_strategy_plugins.ai_audit._complete_with_endpoint", fake_completion) payload = build_crisis_response_shadow_signal( prices, events=(), @@ -248,7 +253,6 @@ def fake_completion(endpoint, _messages, _timeout_seconds): rate_symbols=(), ai_audit_enabled=True, ai_audit_codex_enabled=True, - ai_audit_completion_client=fake_completion, ) audit = payload["ai_audit"] diff --git a/tests/test_strategy_plugin_runner.py b/tests/test_strategy_plugin_runner.py index ec05e59..0f2f7b9 100644 --- a/tests/test_strategy_plugin_runner.py +++ b/tests/test_strategy_plugin_runner.py @@ -948,6 +948,7 @@ def test_strategy_plugin_runner_can_enable_ai_audit_without_api_key(tmp_path, mo "QSP_STRATEGY_PLUGIN_AI_AUDIT_ANTHROPIC_API_KEY", "QSP_CRISIS_AI_AUDIT_ANTHROPIC_API_KEY", "ANTHROPIC_API_KEY", + "CODEX_AUDIT_SERVICE_URL", ): monkeypatch.delenv(key, raising=False) config = _shadow_plugin_config(tmp_path) @@ -961,7 +962,7 @@ def test_strategy_plugin_runner_can_enable_ai_audit_without_api_key(tmp_path, mo payload = json.loads((output_dir / "latest_signal.json").read_text(encoding="utf-8")) assert payload["canonical_route"] == "no_action" assert payload["ai_audit"]["status"] == "skipped" - assert payload["ai_audit"]["skip_reason"] == "missing_api_endpoint" + assert payload["ai_audit"]["skip_reason"] == "gateway_unavailable" assert payload["execution_controls"]["ai_audit_shadow_only"] is True @@ -1110,6 +1111,7 @@ def test_strategy_plugin_runner_can_enable_taco_ai_audit_without_api_key(tmp_pat "QSP_STRATEGY_PLUGIN_AI_AUDIT_ANTHROPIC_API_KEY", "QSP_CRISIS_AI_AUDIT_ANTHROPIC_API_KEY", "ANTHROPIC_API_KEY", + "CODEX_AUDIT_SERVICE_URL", ): monkeypatch.delenv(key, raising=False) @@ -1144,7 +1146,7 @@ def test_strategy_plugin_runner_can_enable_taco_ai_audit_without_api_key(tmp_pat latest = json.loads((output_dir / "latest_signal.json").read_text(encoding="utf-8")) assert latest["canonical_route"] == "taco_rebound" assert latest["ai_audit"]["status"] == "skipped" - assert latest["ai_audit"]["skip_reason"] == "missing_api_endpoint" + assert latest["ai_audit"]["skip_reason"] == "gateway_unavailable" assert latest["execution_controls"]["ai_audit_shadow_only"] is True diff --git a/tests/test_taco_rebound_shadow_plugin.py b/tests/test_taco_rebound_shadow_plugin.py index 94780bc..fd5b4b9 100644 --- a/tests/test_taco_rebound_shadow_plugin.py +++ b/tests/test_taco_rebound_shadow_plugin.py @@ -66,7 +66,8 @@ def test_taco_rebound_shadow_routes_geopolitical_deescalation_to_manual_review_n assert payload["event_quality"]["checks"]["rebound_confirmation_satisfied"] is True -def test_taco_rebound_shadow_ai_audit_uses_fallback_without_changing_route() -> None: +def test_taco_rebound_shadow_ai_audit_uses_gateway_fallback_without_changing_route(monkeypatch) -> None: + monkeypatch.setenv("CODEX_AUDIT_SERVICE_URL", "https://gateway.example") prices = _panic_rebound_prices() dates = pd.bdate_range("2026-03-20", periods=12) event = TradeWarEvent( @@ -97,6 +98,7 @@ def fake_completion(endpoint, messages, timeout_seconds): "human_review_recommended": True, } + monkeypatch.setattr("quant_strategy_plugins.ai_audit._complete_with_endpoint", fake_completion) payload = build_taco_rebound_shadow_signal( prices, events=(event,), @@ -111,7 +113,6 @@ def fake_completion(endpoint, messages, timeout_seconds): ai_audit_fallback_model="fallback-model", ai_audit_codex_enabled=False, ai_audit_timeout_seconds=6.0, - ai_audit_completion_client=fake_completion, ) assert payload["canonical_route"] == ROUTE_TACO_REBOUND @@ -130,7 +131,7 @@ def fake_completion(endpoint, messages, timeout_seconds): assert audit["attempts"][1]["status"] == "ok" -def test_taco_rebound_shadow_ai_audit_skips_without_api_key(monkeypatch) -> None: +def test_taco_rebound_shadow_ai_audit_skips_without_gateway(monkeypatch) -> None: for key in ( "QSP_STRATEGY_PLUGIN_AI_AUDIT_API_KEY", "QSP_CRISIS_AI_AUDIT_API_KEY", @@ -141,6 +142,7 @@ def test_taco_rebound_shadow_ai_audit_skips_without_api_key(monkeypatch) -> None "QSP_STRATEGY_PLUGIN_AI_AUDIT_ANTHROPIC_API_KEY", "QSP_CRISIS_AI_AUDIT_ANTHROPIC_API_KEY", "ANTHROPIC_API_KEY", + "CODEX_AUDIT_SERVICE_URL", ): monkeypatch.delenv(key, raising=False) @@ -167,7 +169,7 @@ def test_taco_rebound_shadow_ai_audit_skips_without_api_key(monkeypatch) -> None assert payload["canonical_route"] == ROUTE_TACO_REBOUND assert payload["ai_audit"]["status"] == "skipped" - assert payload["ai_audit"]["skip_reason"] == "missing_api_endpoint" + assert payload["ai_audit"]["skip_reason"] == "gateway_unavailable" assert payload["ai_audit"]["final_route_unchanged"] is True