fix(audit): preserve advisory content without feedback authority

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,

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)

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,15 +98,15 @@ 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,
             "summary": "Evidence supports the deterministic crisis route; keep deterministic controls in charge.",
             "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,15 +156,15 @@ 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,
             "summary": "Anthropic fallback found the evidence plausible but wants operator review.",
             "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,15 +237,15 @@ 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,
             "summary": "Codex audit agrees with the deterministic route.",
             "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(

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,15 +90,15 @@ 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,
             "summary": "Event source and rebound confirmation support the deterministic manual-review route.",
             "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,15 +122,18 @@ 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"
     assert audit["verdict"] == "agree"
     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: