fix(ai): migrate evidence gate to QPK v2 task service

.github/workflows/evidence-gate.yml

@@ -71,17 +71,10 @@ jobs:
           python -m pip install -e . pandas
           python -m pip install --no-deps -e external/QuantPlatformKit
 
-      - name: Checkout AIAuditBridge
-        uses: actions/checkout@v6
-        with:
-          repository: QuantStrategyLab/AIAuditBridge
-          ref: main
-          path: external/AIAuditBridge
-
       - name: Evaluate Evidence Gate
         env:
           GITHUB_BASE_REF: ${{ github.base_ref }}
-          AIAUDIT_BRIDGE_ROOT: external/AIAuditBridge
-          CODEX_AUDIT_SERVICE_URL: ${{ secrets.CODEX_AUDIT_SERVICE_URL }}
-          AI_GATEWAY_SERVICE_URL: ${{ vars.AI_GATEWAY_SERVICE_URL }}
+          AI_SERVICE_URL: ${{ vars.AI_SERVICE_URL }}
+          AI_SERVICE_AUDIENCE: ${{ vars.AI_SERVICE_AUDIENCE }}
+          AI_SERVICE_REVIEWERS_JSON: ${{ vars.AI_SERVICE_REVIEWERS_JSON }}
         run: python scripts/gate_evidence_package.py

scripts/gate_evidence_package.py

@@ -3,10 +3,12 @@
 
 from __future__ import annotations
 
+import json
 import os
 import re
 import subprocess
 import sys
+import tomllib
 from pathlib import Path
 
 GATE_STAGES = frozenset(
@@ -94,38 +96,39 @@ def _validate_with_promotion_standard(path: Path) -> tuple[bool, list[str]]:
 
 
 def _run_promotion_dual_review(evidence_files: list[Path]) -> int:
-    root = Path(os.environ.get("AIAUDIT_BRIDGE_ROOT", "external/AIAuditBridge"))
-    script = root / "scripts" / "run_dual_review_pipeline.py"
-    if not script.is_file():
-        print("[evidence-gate] dual-review skipped: AIAuditBridge not found")
-        return 0
+    """Advisory V2 task-service quorum; any unavailable or pending result fails closed."""
     if str(os.environ.get("DUAL_REVIEW_GATE_SKIP", "")).strip().lower() in {"1", "true", "yes"}:
         print("[evidence-gate] dual-review skipped by DUAL_REVIEW_GATE_SKIP")
         return 0
-
-    worst = 0
+    try:
+        from quant_platform_kit.strategy_lifecycle.task_review import review_material
+    except ImportError:
+        print("[evidence-gate] review task runtime unavailable", file=sys.stderr)
+        return 3
     for path in evidence_files:
-        proc = subprocess.run(
-            [
-                sys.executable,
-                str(script),
-                "--from-evidence",
-                str(path),
-                "--dispatch",
-            ],
-            cwd=str(root),
-            env={**os.environ, "PYTHONPATH": str(root)},
-            check=False,
-        )
-        print(f"[evidence-gate] dual-review {path.name} exit={proc.returncode}")
-        if proc.stdout:
-            print(proc.stdout.strip())
-        if proc.stderr:
-            print(proc.stderr.strip(), file=sys.stderr)
-        worst = max(worst, proc.returncode)
-    if worst >= 2:
-        print("::error::Dual-review blocked promotion (disagreement or reject)", file=sys.stderr)
-        return 1
+        try:
+            if path.is_symlink() or not path.is_file() or path.stat().st_size > 200000:
+                raise ValueError("invalid evidence file")
+            text = path.read_text(encoding="utf-8")
+            material = tomllib.loads(text) if path.suffix == ".toml" else json.loads(text)
+            if not isinstance(material, dict):
+                raise ValueError("invalid evidence object")
+            caller_ref = os.environ.get("GITHUB_SHA", "")
+            if re.fullmatch(r"[0-9a-f]{40}", caller_ref) is None:
+                raise ValueError("caller revision required")
+            result = review_material(
+                material, operation_id="crypto-evidence:" + caller_ref + ":" + path.as_posix(),
+                required_roles=("reviewer-primary", "reviewer-secondary", "reviewer-independent"),
+            )
+        except Exception:
+            print("[evidence-gate] review task unavailable", file=sys.stderr)
+            return 3
+        if result.get("status") != "completed":
+            print("[evidence-gate] review quorum pending or unavailable", file=sys.stderr)
+            return 3
+        if result.get("outcome") != "agree_approve" or result.get("advisory_only") is not True:
+            print("::error::Evidence review needs human resolution", file=sys.stderr)
+            return 1
     return 0
 
 

tests/test_evidence_gate_review.py

@@ -0,0 +1,73 @@
+import importlib.util
+import json
+import sys
+import types
+from pathlib import Path
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+def _load_gate():
+    spec = importlib.util.spec_from_file_location("gate_evidence_package", ROOT / "scripts" / "gate_evidence_package.py")
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+def _install_reviewer(monkeypatch, result):
+    calls = []
+    module = types.ModuleType("quant_platform_kit.strategy_lifecycle.task_review")
+
+    def review_material(material, *, operation_id, required_roles):
+        calls.append((operation_id, tuple(required_roles)))
+        if isinstance(result, Exception):
+            raise result
+        return result
+
+    module.review_material = review_material
+    monkeypatch.setitem(sys.modules, "quant_platform_kit.strategy_lifecycle.task_review", module)
+    return calls
+
+
+@pytest.fixture
+def evidence(tmp_path, monkeypatch):
+    monkeypatch.setenv("GITHUB_SHA", "a" * 40)
+    monkeypatch.delenv("DUAL_REVIEW_GATE_SKIP", raising=False)
+    path = tmp_path / "synthetic.json"
+    path.write_text(json.dumps({"strategy_profile": "synthetic"}), encoding="utf-8")
+    return path
+
+
+def test_unanimous_advisory_approval_passes(monkeypatch, evidence):
+    calls = _install_reviewer(monkeypatch, {"status": "completed", "outcome": "agree_approve", "advisory_only": True})
+    assert _load_gate()._run_promotion_dual_review([evidence]) == 0
+    assert len(calls) == 1 and len(set(calls[0][1])) == 3
+
+
+@pytest.mark.parametrize("result,code", [
+    ({"status": "unavailable", "reason": "review_routes_incomplete", "advisory_only": True}, 3),
+    ({"status": "pending", "tasks": [], "advisory_only": True}, 3),
+    (RuntimeError("service down"), 3),
+    ({"status": "completed", "outcome": "requires_human", "advisory_only": True}, 1),
+    ({"status": "completed", "outcome": "agree_reject", "advisory_only": True}, 1),
+    ({"status": "completed", "outcome": "agree_approve"}, 1),
+])
+def test_review_fails_closed(monkeypatch, evidence, result, code):
+    _install_reviewer(monkeypatch, result)
+    assert _load_gate()._run_promotion_dual_review([evidence]) == code
+
+
+def test_missing_caller_revision_fails_closed(monkeypatch, evidence):
+    _install_reviewer(monkeypatch, {"status": "completed", "outcome": "agree_approve", "advisory_only": True})
+    monkeypatch.setenv("GITHUB_SHA", "not-a-sha")
+    assert _load_gate()._run_promotion_dual_review([evidence]) == 3
+
+
+def test_workflow_uses_v2_task_service_only():
+    workflow = (ROOT / ".github" / "workflows" / "evidence-gate.yml").read_text(encoding="utf-8")
+    for retired in ("AIAuditBridge", "CODEX_AUDIT_SERVICE_URL", "AI_GATEWAY_SERVICE_URL"):
+        assert retired not in workflow
+    assert "AI_SERVICE_URL: ${{ vars.AI_SERVICE_URL }}" in workflow
+    assert "id-token: write" in workflow