diff --git a/README.md b/README.md index 82b6dd6..32a7dea 100644 --- a/README.md +++ b/README.md @@ -61,3 +61,5 @@ it does not call a model or verify production authentication. ## License See [LICENSE](LICENSE). + +Legacy Codex CLI audit transport is retired. A configured Codex endpoint makes the audit unavailable before any SDK call or API fallback. Deterministic signals and all capital/execution restrictions remain unchanged. Consumers must qualify a task-service migration before adopting this runtime change; no replacement provider is enabled here. diff --git a/src/quant_strategy_plugins/ai_audit.py b/src/quant_strategy_plugins/ai_audit.py index cfcd5c7..3a994e8 100644 --- a/src/quant_strategy_plugins/ai_audit.py +++ b/src/quant_strategy_plugins/ai_audit.py @@ -324,21 +324,8 @@ def _complete_with_endpoint( def _codex_via_gateway(prompt: str, model: str, timeout_seconds: float) -> str: - """Execute via AiGateway service — delegates to CodexAdapter on VPS.""" - try: - from ai_gateway_client import AiGatewayClient, GatewayConfig - config = GatewayConfig.from_env() - client = AiGatewayClient(config) - result = client.execute(prompt, mode="review_only", model=model, timeout=timeout_seconds) - if result.success and str(result.output or "").strip(): - return str(result.output) - raise AiAuditError("ai_gateway_rejected") - except ImportError: - raise AiAuditError("ai_gateway_client_unavailable") from None - except AiAuditError: - raise - except Exception: - raise AiAuditError("ai_gateway_request_failed") from None + """Retired legacy Codex endpoint; never start a CLI or send a job.""" + raise AiAuditError("legacy_cli_retired") def _llm_via_gateway(prompt: str, model: str, provider: str, timeout_seconds: float) -> tuple[str, bool]: @@ -681,6 +668,9 @@ def _run_ai_audit( "attempts": [], } + if any(endpoint.normalized().provider == PROVIDER_CODEX for endpoint in endpoints): + return {**base_payload, "status": "failed", "error": "legacy_cli_retired", "attempts": []} + attempts: list[dict[str, Any]] = [] for endpoint in endpoints: try: diff --git a/tests/test_ai_audit.py b/tests/test_ai_audit.py index d542cf9..65b2d6b 100644 --- a/tests/test_ai_audit.py +++ b/tests/test_ai_audit.py @@ -1,7 +1,6 @@ import sys import types import json -import io import tomllib from importlib.metadata import distribution from pathlib import Path @@ -75,54 +74,6 @@ def test_optional_ai_extra_installs_the_declared_gateway_sdk() -> None: assert GatewayConfig.__module__ == "ai_gateway_client.config" -@pytest.mark.parametrize("entry", [ai_audit.run_crisis_ai_audit, ai_audit.run_taco_ai_audit]) -def test_installed_sdk_consumes_codex_job_without_feedback_or_api_fallback(monkeypatch, entry): - from ai_gateway_client import gateway_client - - _clear_ai_audit_env(monkeypatch) - monkeypatch.setenv("CODEX_AUDIT_SERVICE_URL", "https://gateway.invalid") - monkeypatch.setenv("AI_GATEWAY_SOURCE_REPO", "QuantStrategyLab/UsEquitySnapshotPipelines") - monkeypatch.setattr(gateway_client, "_fetch_oidc_token", lambda _audience: "synthetic") - monkeypatch.setattr(gateway_client.time, "sleep", lambda _seconds: None) - calls = [] - job_id = "synthetic-job-000000000000" - - def urlopen(request, **_kwargs): - calls.append((request.get_method(), request.full_url)) - if request.get_method() == "POST": - assert request.full_url == "https://gateway.invalid/v1/ai/execute/jobs" - payload = json.loads(request.data) - assert payload["mode"] == "review_only" - assert payload["model"] == "gpt-6-astra" - assert payload["source_repository"] == "QuantStrategyLab/UsEquitySnapshotPipelines" - assert "watch_only" in payload["prompt"] - response = {"status": "queued", "job_id": job_id} - else: - assert request.full_url == f"https://gateway.invalid/v1/ai/execute/jobs/{job_id}" - response = {"status": "succeeded", "output": json.dumps({ - "verdict": "review", "confidence": 0.8, "summary": "synthetic research opinion", - "mode": "live", "execution_controls": {"broker_order_allowed": True}, - })} - return io.BytesIO(json.dumps(response).encode()) - - monkeypatch.setattr(gateway_client.urllib.request, "urlopen", urlopen) - feedback = [] - monkeypatch.setattr(ai_audit, "_report_shadow_disagreement", lambda **fields: feedback.append(fields)) - deterministic = {"profile": "synthetic", "canonical_route": "no_action", "suggested_action": "watch_only"} - payload = entry(deterministic, enabled=True, codex_enabled=True, codex_model="gpt-6-astra") - - assert payload["status"] == "advisory" - assert payload["summary"] == "synthetic research opinion" - assert payload["deterministic_route"] == "no_action" - assert payload["final_route_unchanged"] is True - assert payload["mode"] == "shadow_only" - assert payload["execution_controls"]["broker_order_allowed"] is False - assert payload["execution_controls"]["live_allocation_mutation_allowed"] is False - assert payload["execution_controls"]["allocation_recommendation_allowed"] is False - assert len(payload["attempts"]) == 1 - assert [method for method, _url in calls] == ["POST", "GET"] - assert feedback == [] - assert deterministic["canonical_route"] == "no_action" def test_ai_audit_uses_generic_anthropic_api_key(monkeypatch) -> None: @@ -235,7 +186,7 @@ def test_gateway_uses_default_model_without_local_provider_key(monkeypatch) -> N assert calls == [("openai", "gpt-5.4-mini")] -def test_gateway_success_calls_analyze_and_execute(monkeypatch) -> None: +def test_gateway_analysis_remains_available_but_codex_execution_is_retired(monkeypatch) -> None: calls: list[tuple[str, tuple, dict]] = [] class SuccessfulGatewayClient: @@ -260,64 +211,13 @@ def execute(self, *args, **kwargs): ) assert _llm_via_gateway("audit", "gpt-test", "openai", 3.0) == ("analysis", False) - assert _codex_via_gateway("review", "codex-test", 4.0) == "review" + with pytest.raises(AiAuditError, match="legacy_cli_retired"): + _codex_via_gateway("review", "codex-test", 4.0) assert calls == [ ("analyze", ("audit",), {"model": "gpt-test", "timeout": 3.0}), - ("execute", ("review",), {"mode": "review_only", "model": "codex-test", "timeout": 4.0}), ] -@pytest.mark.parametrize("entry", [ai_audit.run_crisis_ai_audit, ai_audit.run_taco_ai_audit]) -def test_codex_execution_is_advisory_without_feedback_or_route_changes(monkeypatch, entry): - from copy import deepcopy - - _clear_ai_audit_env(monkeypatch) - monkeypatch.setenv("CODEX_AUDIT_SERVICE_URL", "https://gateway.invalid") - output = json.dumps({ - "verdict": "review", "confidence": 0.8, "summary": "synthetic research opinion", - "status": "verified", "final_route_unchanged": False, "mode": "live", - "execution_controls": {"broker_order_allowed": True}, - }) - calls = [] - - class SuccessfulCodexClient: - def __init__(self, _config): - pass - - def execute(self, prompt, **kwargs): - calls.append((prompt, kwargs)) - return types.SimpleNamespace(success=True, output=output, provider="codex") - - def analyze(self, *_args, **_kwargs): - raise AssertionError("Codex advisory must not fall back to a paid API") - - monkeypatch.setitem(sys.modules, "ai_gateway_client", types.SimpleNamespace( - AiGatewayClient=SuccessfulCodexClient, - GatewayConfig=types.SimpleNamespace(from_env=lambda: object()), - )) - feedback = [] - monkeypatch.setattr(ai_audit, "_report_shadow_disagreement", lambda **fields: feedback.append(fields)) - deterministic = {"profile": "synthetic", "canonical_route": "no_action", "suggested_action": "watch_only"} - original = deepcopy(deterministic) - - payload = entry(deterministic, enabled=True, codex_enabled=True, codex_model="codex-test") - - assert payload["status"] == "advisory" - assert payload["summary"] == "synthetic research opinion" - assert payload["confidence"] == 0.8 - assert len(calls) == 1 - assert calls[0][1]["mode"] == "review_only" - assert calls[0][1]["model"] == "codex-test" - assert len(payload["attempts"]) == 1 - assert payload["attempts"][0]["status"] == "advisory" - assert payload["selected_endpoint"]["provider"] == "codex" - assert feedback == [] - assert deterministic == original - assert payload["final_route_unchanged"] is True - assert payload["mode"] == "shadow_only" - assert payload["execution_controls"]["broker_order_allowed"] is False - assert payload["execution_controls"]["live_allocation_mutation_allowed"] is False - assert payload["execution_controls"]["allocation_recommendation_allowed"] is False @pytest.mark.parametrize("actual_provider", ["anthropic", ""]) @@ -371,7 +271,7 @@ def execute(self, *_args, **_kwargs): direct_calls: list[str] = [] monkeypatch.setattr(ai_audit, direct_fallback, lambda *_args: direct_calls.append("direct") or "{}", raising=False) - with pytest.raises(AiAuditError, match="ai_gateway_request_failed") as exc_info: + with pytest.raises(AiAuditError, match="ai_gateway_request_failed" if gateway_call is _llm_via_gateway else "legacy_cli_retired") as exc_info: if gateway_call is _llm_via_gateway: gateway_call("audit", "test-model", "openai", 1.0) else: @@ -393,7 +293,7 @@ def test_gateway_client_import_failure_never_uses_direct_fallback(monkeypatch, g direct_calls: list[str] = [] monkeypatch.setattr(ai_audit, direct_fallback, lambda *_args: direct_calls.append("direct") or "{}", raising=False) - with pytest.raises(AiAuditError, match="ai_gateway_client_unavailable"): + with pytest.raises(AiAuditError, match="ai_gateway_client_unavailable" if gateway_call is _llm_via_gateway else "legacy_cli_retired"): if gateway_call is _llm_via_gateway: gateway_call("audit", "test-model", "openai", 1.0) else: @@ -530,3 +430,24 @@ def test_original_prompts_distinguish_false_zero_and_missing(builder, boolean_fi assert user["data_freshness"] == "0" assert user["as_of"] == "" assert source == {**dict.fromkeys(boolean_fields, False), "data_freshness": 0} + + +@pytest.mark.parametrize("entry", [ai_audit.run_crisis_ai_audit, ai_audit.run_taco_ai_audit]) +@pytest.mark.parametrize("provider", ["codex", " CoDeX "]) +def test_retired_codex_route_never_calls_sdk_or_paid_fallback(monkeypatch, entry, provider): + from unittest.mock import Mock + monkeypatch.setenv("CODEX_AUDIT_SERVICE_URL", "https://gateway.invalid") + monkeypatch.setattr(ai_audit, "build_ai_audit_endpoints", lambda **_: ( + ai_audit.AiAuditEndpoint("codex", "", provider=provider), + ai_audit.AiAuditEndpoint("primary", "synthetic", provider="openai"), + )) + complete, feedback = Mock(), Mock() + monkeypatch.setattr(ai_audit, "_complete_with_endpoint", complete) + monkeypatch.setattr(ai_audit, "_report_shadow_disagreement", feedback) + source = {"profile": "synthetic", "canonical_route": "no_action"} + result = entry(source, enabled=True) + assert result["status"] == "failed" and result["error"] == "legacy_cli_retired" + assert result["attempts"] == [] and result["final_route_unchanged"] is True + assert result["execution_controls"]["broker_order_allowed"] is False + complete.assert_not_called() + feedback.assert_not_called() diff --git a/tests/test_crisis_response_shadow_plugin.py b/tests/test_crisis_response_shadow_plugin.py index 07a44ef..42bb0fe 100644 --- a/tests/test_crisis_response_shadow_plugin.py +++ b/tests/test_crisis_response_shadow_plugin.py @@ -229,7 +229,7 @@ def test_shadow_signal_ai_audit_skips_without_gateway(monkeypatch) -> None: assert payload["ai_audit"]["final_route_unchanged"] is True -def test_shadow_signal_ai_audit_prefers_gateway_codex_provider(monkeypatch) -> None: +def test_shadow_signal_ai_audit_blocks_retired_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()) @@ -261,10 +261,11 @@ def fake_completion(endpoint, _messages, _timeout_seconds): ) audit = payload["ai_audit"] - assert calls == [("codex", "codex")] - assert audit["status"] == "ok" - assert audit["selected_endpoint"]["provider"] == "codex" - assert audit["verdict"] == "agree" + assert calls == [] + assert audit["status"] == "failed" + assert audit["error"] == "legacy_cli_retired" + assert audit["final_route_unchanged"] is True + assert audit["execution_controls"]["broker_order_allowed"] is False def test_shadow_signal_evidence_uses_configured_benchmark_drawdown() -> None: