From a20a95cedf216062a41bb51a020311ae813d5113 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sat, 5 Sep 2026 08:54:01 +0800 Subject: [PATCH] fix(audit): consume advisory research without shadow feedback Co-Authored-By: Codex --- src/quant_strategy_plugins/ai_audit.py | 50 ++++++---- tests/test_ai_audit.py | 103 +++++++++++++++++++- tests/test_crisis_response_shadow_plugin.py | 23 +++-- tests/test_taco_rebound_shadow_plugin.py | 15 ++- 4 files changed, 157 insertions(+), 34 deletions(-) diff --git a/src/quant_strategy_plugins/ai_audit.py b/src/quant_strategy_plugins/ai_audit.py index d650ae5..588d73c 100644 --- a/src/quant_strategy_plugins/ai_audit.py +++ b/src/quant_strategy_plugins/ai_audit.py @@ -305,7 +305,7 @@ def _complete_with_endpoint( endpoint: AiAuditEndpoint, messages: Sequence[Mapping[str, str]], timeout_seconds: float, -) -> str: +) -> tuple[str, bool]: endpoint = endpoint.normalized() # API keys live on the VPS — no keys in plugin config needed. @@ -317,7 +317,7 @@ def _complete_with_endpoint( 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 _codex_via_gateway(prompt, endpoint.model, timeout_seconds), False return _llm_via_gateway(prompt, endpoint.model, endpoint.provider, timeout_seconds) @@ -339,8 +339,8 @@ def _codex_via_gateway(prompt: str, model: str, timeout_seconds: float) -> str: raise AiAuditError("ai_gateway_request_failed") from None -def _llm_via_gateway(prompt: str, model: str, provider: str, timeout_seconds: float) -> str: - """Analyze via AiGateway service — delegates to LlmAdapter on VPS.""" +def _llm_via_gateway(prompt: str, model: str, provider: str, timeout_seconds: float) -> tuple[str, bool]: + """Return analysis content and its advisory status, never decision authority.""" try: from ai_gateway_client import AiGatewayClient, GatewayConfig config = GatewayConfig.from_env() @@ -349,8 +349,20 @@ def _llm_via_gateway(prompt: str, model: str, provider: str, timeout_seconds: fl 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) + output = result.output + raw = getattr(result, "raw", None) + note = getattr(result, "note", "") + status = raw.get("status", "ok") if isinstance(raw, dict) else "ok" + policy = raw.get("policy_verdict", status) if isinstance(raw, dict) else status + advisory = (result.success is False and note == "advisory" + and status == "advisory" and policy == "advisory" and isinstance(raw, dict)) + ok = (result.success is True and note == "" + and status == "ok" and policy in ("ok", "eligible")) + if (isinstance(output, str) and output.strip() and not getattr(result, "error", "") + and (raw is None or isinstance(raw, dict)) + and (not isinstance(raw, dict) or raw.get("output", output) == output) + and (ok or advisory)): + return output, advisory raise AiAuditError("ai_gateway_rejected") except ImportError: raise AiAuditError("ai_gateway_client_unavailable") from None @@ -681,22 +693,24 @@ def _run_ai_audit( attempts: list[dict[str, Any]] = [] for endpoint in endpoints: try: - raw_response = _complete_with_endpoint(endpoint, messages, float(timeout_seconds)) + raw_response, advisory = _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"}) - - # Phase 3: report AI vs deterministic disagreement to AiGateway - _report_shadow_disagreement( - audit_kind=audit_kind, - ai_verdict=audit_response.get("verdict", ""), - ai_confidence=audit_response.get("confidence") or 0.0, - deterministic_route=str(deterministic_payload.get("canonical_route") or - deterministic_payload.get("suggested_action") or ""), - ) + status = "advisory" if advisory else "ok" + attempts.append({**endpoint.report(), "status": status}) + + # Advisory content is display-only; it must not mutate feedback state. + if not advisory: + _report_shadow_disagreement( + audit_kind=audit_kind, + ai_verdict=audit_response.get("verdict", ""), + ai_confidence=audit_response.get("confidence") or 0.0, + deterministic_route=str(deterministic_payload.get("canonical_route") or + deterministic_payload.get("suggested_action") or ""), + ) return { **base_payload, - "status": "ok", + "status": status, "selected_endpoint": endpoint.report(), "attempts": attempts, **audit_response, diff --git a/tests/test_ai_audit.py b/tests/test_ai_audit.py index 30a725c..72e9576 100644 --- a/tests/test_ai_audit.py +++ b/tests/test_ai_audit.py @@ -127,7 +127,7 @@ def test_gateway_uses_default_model_without_local_provider_key(monkeypatch) -> N 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}', + or ('{"verdict":"agree","summary":"ok","risk_flags":[],"evidence_gaps":[],"confidence":0.5}', False), ) payload = _run_ai_audit( {"canonical_route": "true_crisis", "suggested_action": "defend"}, @@ -165,7 +165,7 @@ def execute(self, *args, **kwargs): ), ) - assert _llm_via_gateway("audit", "gpt-test", "openai", 3.0) == "analysis" + assert _llm_via_gateway("audit", "gpt-test", "openai", 3.0) == ("analysis", False) assert _codex_via_gateway("review", "codex-test", 4.0) == "review" assert calls == [ ("analyze", ("audit",), {"model": "gpt-test", "timeout": 3.0}), @@ -253,3 +253,102 @@ def test_gateway_client_import_failure_never_uses_direct_fallback(monkeypatch, g gateway_call("audit", "test-model", 1.0) assert direct_calls == [] + + +def _install_analysis_result(monkeypatch, result): + calls = [] + + class FakeGatewayClient: + def __init__(self, _config): + pass + + def analyze(self, *_args, **_kwargs): + calls.append("analyze") + return result + + def execute(self, *_args, **_kwargs): + raise AssertionError("advisory analysis must not execute") + + monkeypatch.setitem(sys.modules, "ai_gateway_client", types.SimpleNamespace( + AiGatewayClient=FakeGatewayClient, + GatewayConfig=types.SimpleNamespace(from_env=lambda: object()), + )) + return calls + + +@pytest.mark.parametrize("status", ["ok", "advisory"]) +@pytest.mark.parametrize("entry", [ai_audit.run_crisis_ai_audit, ai_audit.run_taco_ai_audit]) +def test_research_content_preserves_status_controls_and_feedback_boundary(monkeypatch, status, entry): + import json + from copy import deepcopy + from quant_strategy_plugins.plugin_signal_utils import flatten_for_csv, json_scalar + + output = json.dumps({ + "verdict": "review", "summary": "synthetic research opinion", "confidence": 0.8, + "status": "ok", "final_route_unchanged": False, "mode": "live", + "execution_controls": {"broker_order_allowed": True}, + }) + result = types.SimpleNamespace( + success=status == "ok", output=output, provider="openai", model="test-model", + note="advisory" if status == "advisory" else "", error="", + raw={"status": status, "output": output, + "policy_verdict": "advisory" if status == "advisory" else "eligible"}, + ) + calls = _install_analysis_result(monkeypatch, result) + monkeypatch.setenv("CODEX_AUDIT_SERVICE_URL", "https://gateway.invalid") + monkeypatch.setattr(ai_audit, "build_ai_audit_endpoints", lambda **_kwargs: ( + ai_audit.AiAuditEndpoint("primary", "", model="test-model"), + ai_audit.AiAuditEndpoint("fallback", "", model="fallback-model"), + )) + 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=False) + + assert payload["status"] == status + assert payload["attempts"][0]["status"] == status + assert calls == ["analyze"] # Advisory must not trigger a fallback model. + assert len(feedback) == (1 if status == "ok" else 0) + 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 payload["selected_endpoint"]["provider"] == "openai" + assert deterministic == original + embedded = json_scalar({**deterministic, "ai_audit": payload}) + assert flatten_for_csv(embedded)["ai_audit.status"] == status + assert result.success is (status == "ok") + + +@pytest.mark.parametrize("changes", [ + {"success": True, "note": "advisory", "raw": {"status": "failed"}}, + {"success": True, "raw": {"status": "invalid"}}, + {"success": True, "raw": {"status": "unknown"}}, + {"success": False, "note": "advisory", "raw": {"status": "ok"}}, + {"success": True, "note": "advisory", "raw": {"status": "advisory"}}, + {"success": False, "note": "advisory", "raw": {"status": "advisory", "policy_verdict": "invalid"}}, + {"success": False, "note": "advisory", "raw": None}, + {"success": True, "note": "failed", "raw": None}, + {"success": True, "raw": {"status": "ok", "output": None}}, + {"success": True, "output": 42}, + {"success": True, "output": None}, + {"success": True, "output": " "}, + {"success": True, "output": {"summary": "not text"}}, +]) +def test_invalid_analysis_metadata_and_content_are_rejected(monkeypatch, changes): + fields = dict(success=True, note="", error="", raw=None, provider="openai", output="synthetic text") + fields.update(changes) + _install_analysis_result(monkeypatch, types.SimpleNamespace(**fields)) + with pytest.raises(AiAuditError, match="ai_gateway_rejected"): + _llm_via_gateway("synthetic prompt", "test-model", "openai", 1.0) + + +def test_advisory_provider_mismatch_still_rejected(monkeypatch): + _install_analysis_result(monkeypatch, types.SimpleNamespace( + success=False, note="advisory", error="", provider="anthropic", output="synthetic text", + raw={"status": "advisory", "output": "synthetic text"}, + )) + with pytest.raises(AiAuditError, match="ai_gateway_provider_mismatch"): + _llm_via_gateway("synthetic prompt", "test-model", "openai", 1.0) diff --git a/tests/test_crisis_response_shadow_plugin.py b/tests/test_crisis_response_shadow_plugin.py index c9d4244..07a44ef 100644 --- a/tests/test_crisis_response_shadow_plugin.py +++ b/tests/test_crisis_response_shadow_plugin.py @@ -3,6 +3,7 @@ import json import pandas as pd +import pytest from quant_strategy_plugins.crisis_response_research import ROUTE_NO_ACTION, ROUTE_TRUE_CRISIS from quant_strategy_plugins.crisis_response_shadow_plugin import ( @@ -84,7 +85,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_gateway_fallback_without_changing_route(monkeypatch) -> None: +@pytest.mark.parametrize("advisory", [False, True]) +def test_shadow_signal_ai_audit_uses_gateway_fallback_without_changing_route(monkeypatch, advisory) -> 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()) @@ -96,7 +98,7 @@ def fake_completion(endpoint, messages, timeout_seconds): assert messages[0]["role"] == "system" if endpoint.name == "primary": raise RuntimeError("primary unavailable") - return { + return ({ "verdict": "agree", "route_assessment": "confirm_true_crisis", "confidence": 0.82, @@ -104,7 +106,7 @@ def fake_completion(endpoint, messages, timeout_seconds): "key_risks": ["financial and credit stress are both active"], "data_gaps": [], "human_review_recommended": False, - } + }, advisory) monkeypatch.setattr("quant_strategy_plugins.ai_audit._complete_with_endpoint", fake_completion) payload = build_crisis_response_shadow_signal( @@ -131,14 +133,17 @@ def fake_completion(endpoint, messages, timeout_seconds): assert payload["execution_controls"]["ai_audit_shadow_only"] is True assert calls == ["primary", "fallback"] audit = payload["ai_audit"] - assert audit["status"] == "ok" + expected_status = "advisory" if advisory else "ok" + assert audit["status"] == expected_status assert audit["selected_endpoint"]["name"] == "fallback" assert audit["selected_endpoint"]["model"] == "fallback-model" assert audit["verdict"] == "agree" assert audit["final_route_unchanged"] is True assert audit["deterministic_route"] == ROUTE_TRUE_CRISIS assert audit["attempts"][0]["status"] == "failed" - assert audit["attempts"][1]["status"] == "ok" + assert audit["attempts"][1]["status"] == expected_status + from quant_strategy_plugins.plugin_signal_utils import flatten_for_csv + assert flatten_for_csv(payload)["ai_audit.status"] == expected_status def test_shadow_signal_ai_audit_uses_gateway_anthropic_fallback(monkeypatch) -> None: @@ -151,7 +156,7 @@ def fake_completion(endpoint, _messages, _timeout_seconds): calls.append((endpoint.name, endpoint.provider)) if endpoint.provider == "openai": raise RuntimeError("openai unavailable") - return { + return ({ "verdict": "review", "route_assessment": "needs_human_review", "confidence": 0.64, @@ -159,7 +164,7 @@ def fake_completion(endpoint, _messages, _timeout_seconds): "key_risks": ["rapid drawdown"], "data_gaps": ["macro context not in payload"], "human_review_recommended": True, - } + }, False) monkeypatch.setattr("quant_strategy_plugins.ai_audit._complete_with_endpoint", fake_completion) payload = build_crisis_response_shadow_signal( @@ -232,7 +237,7 @@ def test_shadow_signal_ai_audit_prefers_gateway_codex_provider(monkeypatch) -> N def fake_completion(endpoint, _messages, _timeout_seconds): calls.append((endpoint.name, endpoint.provider)) - return { + return ({ "verdict": "agree", "route_assessment": "codex_confirmed", "confidence": 0.70, @@ -240,7 +245,7 @@ def fake_completion(endpoint, _messages, _timeout_seconds): "key_risks": [], "data_gaps": [], "human_review_recommended": False, - } + }, False) monkeypatch.setattr("quant_strategy_plugins.ai_audit._complete_with_endpoint", fake_completion) payload = build_crisis_response_shadow_signal( diff --git a/tests/test_taco_rebound_shadow_plugin.py b/tests/test_taco_rebound_shadow_plugin.py index fd5b4b9..cf1bdcf 100644 --- a/tests/test_taco_rebound_shadow_plugin.py +++ b/tests/test_taco_rebound_shadow_plugin.py @@ -3,6 +3,7 @@ import json import pandas as pd +import pytest from quant_strategy_plugins.taco_panic_rebound_research import EVENT_KIND_SOFTENING, TradeWarEvent from quant_strategy_plugins.taco_rebound_shadow_plugin import ( @@ -66,7 +67,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_gateway_fallback_without_changing_route(monkeypatch) -> None: +@pytest.mark.parametrize("advisory", [False, True]) +def test_taco_rebound_shadow_ai_audit_uses_gateway_fallback_without_changing_route(monkeypatch, advisory) -> None: monkeypatch.setenv("CODEX_AUDIT_SERVICE_URL", "https://gateway.example") prices = _panic_rebound_prices() dates = pd.bdate_range("2026-03-20", periods=12) @@ -88,7 +90,7 @@ def fake_completion(endpoint, messages, timeout_seconds): assert "TACO rebound plugin" in messages[0]["content"] if endpoint.name == "primary": raise RuntimeError("primary unavailable") - return { + return ({ "verdict": "agree", "route_assessment": "event_rebound_context_supported", "confidence": 0.78, @@ -96,7 +98,7 @@ def fake_completion(endpoint, messages, timeout_seconds): "key_risks": ["headline-driven event context can reverse"], "data_gaps": [], "human_review_recommended": True, - } + }, advisory) monkeypatch.setattr("quant_strategy_plugins.ai_audit._complete_with_endpoint", fake_completion) payload = build_taco_rebound_shadow_signal( @@ -120,7 +122,8 @@ def fake_completion(endpoint, messages, timeout_seconds): assert payload["execution_controls"]["ai_audit_shadow_only"] is True assert calls == ["primary", "fallback"] audit = payload["ai_audit"] - assert audit["status"] == "ok" + expected_status = "advisory" if advisory else "ok" + assert audit["status"] == expected_status assert audit["audit_kind"] == "taco_rebound_shadow" assert audit["selected_endpoint"]["name"] == "fallback" assert audit["selected_endpoint"]["model"] == "fallback-model" @@ -128,7 +131,9 @@ def fake_completion(endpoint, messages, timeout_seconds): assert audit["final_route_unchanged"] is True assert audit["deterministic_route"] == ROUTE_TACO_REBOUND assert audit["attempts"][0]["status"] == "failed" - assert audit["attempts"][1]["status"] == "ok" + assert audit["attempts"][1]["status"] == expected_status + from quant_strategy_plugins.plugin_signal_utils import flatten_for_csv + assert flatten_for_csv(payload)["ai_audit.status"] == expected_status def test_taco_rebound_shadow_ai_audit_skips_without_gateway(monkeypatch) -> None: