Skip to content

Commit d483ea1

Browse files
Pigbibiclaudecursoragent
authored
feat(dual-review): wire promotion and drift gates to pipeline (#35)
* feat(dual-review): wire promotion and drift gates to pipeline Add Codex primary + gateway secondary reviewers, end-to-end pipeline scripts, and drift critical-strategy batch runner for evidence-gate CI. Co-Authored-By: Claude <noreply@anthropic.com> Co-authored-by: Cursor <cursoragent@cursor.com> * fix(dual-review): address Codex review blockers for gate wiring Resolve drift pipeline path, make --from-evidence CLI usable, inject evidence_summary into promotion context, and harden primary review errors. Co-Authored-By: Claude <noreply@anthropic.com> Co-authored-by: Cursor <cursoragent@cursor.com> * fix(dual-review): harden gate defaults and gateway routing Fail closed when Codex primary is missing, validate --from-evidence packages, stop using CODEX_AUDIT_SERVICE_URL as gateway URL, and isolate per-provider gateway analyze failures. Co-Authored-By: Claude <noreply@anthropic.com> Co-authored-by: Cursor <cursoragent@cursor.com> * fix(ci): fallback to direct API when Codex service exec fails Co-Authored-By: Claude <noreply@anthropic.com> Co-authored-by: Cursor <cursoragent@cursor.com> * Revert "fix(ci): fallback to direct API when Codex service exec fails" This reverts commit aa7390c. --------- Co-authored-by: Claude <noreply@anthropic.com> Co-authored-by: Cursor <cursoragent@cursor.com>
1 parent 2b75c68 commit d483ea1

7 files changed

Lines changed: 639 additions & 0 deletions

‎scripts/run_drift_dual_review.py‎

Lines changed: 110 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,110 @@
1+
#!/usr/bin/env python3
2+
"""Run dual-review pipeline for critical drift detections (task 11b)."""
3+
4+
from __future__ import annotations
5+
6+
import argparse
7+
import json
8+
import os
9+
import subprocess
10+
import sys
11+
from pathlib import Path
12+
from typing import Any
13+
14+
15+
def _critical_drifts(domain: str) -> list[dict[str, Any]]:
16+
from quant_platform_kit.strategy_lifecycle.drift_detector import run_drift_detection
17+
18+
results = run_drift_detection(domain)
19+
payloads: list[dict[str, Any]] = []
20+
for item in results:
21+
status = getattr(getattr(item, "status", None), "value", "")
22+
if status != "critical":
23+
continue
24+
payloads.append(
25+
{
26+
"trigger": "drift",
27+
"strategy_profile": item.strategy_profile,
28+
"domain": item.domain,
29+
"drift_score": item.drift_score,
30+
"context": {
31+
"domain": item.domain,
32+
"drift_score": item.drift_score,
33+
"repository": os.environ.get("GITHUB_REPOSITORY", ""),
34+
},
35+
}
36+
)
37+
return payloads
38+
39+
40+
def main(argv: list[str] | None = None) -> int:
41+
parser = argparse.ArgumentParser(description="Dual-review for critical drift strategies.")
42+
parser.add_argument("--domain", default=os.environ.get("STRATEGY_DOMAIN", "").strip())
43+
parser.add_argument("--dispatch", action="store_true")
44+
parser.add_argument("--dry-run", action="store_true")
45+
parser.add_argument(
46+
"--aab-root",
47+
default=os.environ.get("AIAUDIT_BRIDGE_ROOT", "external/AIAuditBridge"),
48+
)
49+
args = parser.parse_args(argv)
50+
51+
domain = str(args.domain or "").strip()
52+
if not domain:
53+
print(json.dumps({"ok": False, "error": "domain_required"}))
54+
return 1
55+
56+
if str(os.environ.get("DUAL_REVIEW_GATE_SKIP", "")).strip().lower() in {"1", "true", "yes"}:
57+
print(json.dumps({"ok": True, "skipped": ["dual_review_gate_disabled"], "count": 0}))
58+
return 0
59+
60+
critical = _critical_drifts(domain)
61+
if not critical:
62+
print(json.dumps({"ok": True, "count": 0, "results": []}))
63+
return 0
64+
65+
aab_root = Path(args.aab_root).resolve()
66+
pipeline = aab_root / "scripts" / "run_dual_review_pipeline.py"
67+
if not pipeline.is_file():
68+
print(json.dumps({"ok": False, "error": f"pipeline_not_found: {pipeline}"}))
69+
return 1
70+
71+
results: list[dict[str, Any]] = []
72+
worst = 0
73+
for item in critical:
74+
cmd = [
75+
sys.executable,
76+
str(pipeline),
77+
"--trigger",
78+
"drift",
79+
"--strategy-profile",
80+
item["strategy_profile"],
81+
"--context-json",
82+
json.dumps(item["context"]),
83+
]
84+
if args.dispatch:
85+
cmd.append("--dispatch")
86+
if args.dry_run:
87+
cmd.append("--dry-run")
88+
proc = subprocess.run(
89+
cmd,
90+
cwd=str(aab_root),
91+
env={**os.environ, "PYTHONPATH": str(aab_root)},
92+
capture_output=True,
93+
text=True,
94+
check=False,
95+
)
96+
try:
97+
body = json.loads(proc.stdout) if proc.stdout.strip() else {"ok": False, "error": proc.stderr}
98+
except json.JSONDecodeError:
99+
body = {"ok": False, "error": proc.stdout or proc.stderr}
100+
body["exit_code"] = proc.returncode
101+
results.append(body)
102+
worst = max(worst, proc.returncode)
103+
104+
summary = {"ok": True, "domain": domain, "count": len(results), "results": results}
105+
print(json.dumps(summary, ensure_ascii=False, indent=2))
106+
return worst
107+
108+
109+
if __name__ == "__main__":
110+
raise SystemExit(main())

‎scripts/run_dual_review_pipeline.py‎

Lines changed: 206 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,206 @@
1+
#!/usr/bin/env python3
2+
"""End-to-end dual-review pipeline: Codex primary → GPT+Claude secondary → dispatch."""
3+
4+
from __future__ import annotations
5+
6+
import argparse
7+
import json
8+
import os
9+
from pathlib import Path
10+
from typing import Any
11+
12+
from service.dual_review import VERDICT_DISAGREEMENT, VERDICT_FAIL
13+
from service.dual_review_dispatch import dispatch_dual_review_result
14+
from service.dual_review_orchestrator import orchestrate_from_payload
15+
from service.dual_review_primary import (
16+
build_primary_prompt,
17+
primary_review_available,
18+
run_codex_primary_review,
19+
)
20+
from service.dual_review_triggers import resolve_trigger
21+
22+
23+
def _load_json(value: str) -> dict[str, Any]:
24+
path = Path(value)
25+
if path.is_file():
26+
loaded = json.loads(path.read_text(encoding="utf-8"))
27+
else:
28+
loaded = json.loads(value)
29+
if not isinstance(loaded, dict):
30+
raise ValueError("expected JSON object")
31+
return loaded
32+
33+
34+
def _load_evidence_package(path: Path) -> dict[str, Any]:
35+
if not path.is_file():
36+
raise ValueError(f"evidence file not found: {path}")
37+
try:
38+
loaded = json.loads(path.read_text(encoding="utf-8"))
39+
except json.JSONDecodeError as exc:
40+
raise ValueError(f"evidence file is not valid JSON: {path}") from exc
41+
if not isinstance(loaded, dict):
42+
raise ValueError(f"evidence file must be a JSON object: {path}")
43+
return loaded
44+
45+
46+
def _profile_from_evidence(path: Path) -> str:
47+
evidence = _load_evidence_package(path)
48+
profile = str(evidence.get("strategy_profile") or evidence.get("profile") or "").strip()
49+
return profile or path.stem
50+
51+
52+
def _evidence_summary_from_path(path: Path) -> dict[str, Any]:
53+
evidence = _load_evidence_package(path)
54+
return {
55+
k: evidence.get(k)
56+
for k in (
57+
"strategy_profile",
58+
"status",
59+
"oos_sharpe",
60+
"max_drawdown",
61+
"hit_rate",
62+
"evidence_version",
63+
)
64+
if evidence.get(k) not in (None, "")
65+
}
66+
67+
68+
def _build_payload(
69+
*,
70+
trigger: str,
71+
strategy_profile: str,
72+
context: dict[str, Any],
73+
primary_review: dict[str, Any],
74+
) -> dict[str, Any]:
75+
payload: dict[str, Any] = {
76+
"trigger": trigger,
77+
"strategy_profile": strategy_profile,
78+
"primary_review": primary_review,
79+
}
80+
payload.update(context)
81+
return payload
82+
83+
84+
def _pipeline_enabled() -> bool:
85+
if str(os.environ.get("DUAL_REVIEW_GATE_SKIP", "")).strip().lower() in {"1", "true", "yes"}:
86+
return False
87+
return True
88+
89+
90+
def _primary_skip_allowed() -> bool:
91+
return str(os.environ.get("DUAL_REVIEW_GATE_ALLOW_SKIP", "")).strip().lower() in {
92+
"1",
93+
"true",
94+
"yes",
95+
}
96+
97+
98+
def run_pipeline(
99+
*,
100+
trigger: str,
101+
strategy_profile: str,
102+
context: dict[str, Any],
103+
primary_review: dict[str, Any] | None = None,
104+
evidence_path: Path | None = None,
105+
dispatch: bool = False,
106+
dry_run: bool = False,
107+
) -> dict[str, Any]:
108+
if not _pipeline_enabled():
109+
return {"ok": True, "skipped": ["dual_review_gate_disabled"]}
110+
111+
if primary_review is None:
112+
if not primary_review_available():
113+
if _primary_skip_allowed():
114+
return {"ok": True, "skipped": ["codex_primary_unconfigured"]}
115+
return {"ok": False, "error": "codex_primary_unconfigured"}
116+
prompt = build_primary_prompt(
117+
trigger=trigger,
118+
strategy_profile=strategy_profile,
119+
context=context,
120+
evidence_path=evidence_path,
121+
)
122+
primary_review = run_codex_primary_review(prompt=prompt)
123+
124+
payload = _build_payload(
125+
trigger=trigger,
126+
strategy_profile=strategy_profile,
127+
context=context,
128+
primary_review=primary_review,
129+
)
130+
if resolve_trigger(payload) is None:
131+
return {"ok": False, "error": "invalid_trigger", "payload": payload}
132+
133+
outcome = orchestrate_from_payload(payload)
134+
if outcome is None:
135+
return {"ok": False, "error": "orchestration_failed", "payload": payload}
136+
137+
result = outcome.to_dict()
138+
if dispatch:
139+
result["dispatch"] = dispatch_dual_review_result(outcome, dry_run=dry_run)
140+
result["ok"] = True
141+
return result
142+
143+
144+
def _exit_code(result: dict[str, Any]) -> int:
145+
if not result.get("ok"):
146+
return 1
147+
if result.get("skipped"):
148+
return 0
149+
outcome = str(result.get("outcome") or "")
150+
if outcome in {VERDICT_DISAGREEMENT, VERDICT_FAIL}:
151+
return 2
152+
return 0
153+
154+
155+
def main(argv: list[str] | None = None) -> int:
156+
parser = argparse.ArgumentParser(description="Run Codex primary + dual API secondary review pipeline.")
157+
parser.add_argument("--trigger", choices=("promotion", "hit_rate", "drift"))
158+
parser.add_argument("--strategy-profile")
159+
parser.add_argument("--context-json", default="{}", help="Inline JSON or file path for trigger context")
160+
parser.add_argument("--evidence-file", help="Evidence package path (promotion)")
161+
parser.add_argument("--primary-review", help="Precomputed primary review JSON (skip Codex)")
162+
parser.add_argument("--dispatch", action="store_true")
163+
parser.add_argument("--dry-run", action="store_true")
164+
parser.add_argument(
165+
"--from-evidence",
166+
help="Shorthand: promotion review for evidence package (sets trigger=promotion)",
167+
)
168+
args = parser.parse_args(argv)
169+
170+
context = _load_json(args.context_json)
171+
evidence_path = Path(args.evidence_file) if args.evidence_file else None
172+
trigger = args.trigger
173+
profile = args.strategy_profile
174+
175+
if args.from_evidence:
176+
evidence_path = Path(args.from_evidence)
177+
try:
178+
trigger = "promotion"
179+
profile = _profile_from_evidence(evidence_path)
180+
context.setdefault("repository", os.environ.get("GITHUB_REPOSITORY", ""))
181+
context.setdefault("old_status", "shadow_candidate")
182+
context.setdefault("new_status", "live_candidate")
183+
summary = _evidence_summary_from_path(evidence_path)
184+
if summary:
185+
context.setdefault("evidence_summary", summary)
186+
except ValueError as exc:
187+
parser.error(str(exc))
188+
elif not trigger or not profile:
189+
parser.error("--trigger and --strategy-profile are required unless --from-evidence is set")
190+
191+
primary = _load_json(args.primary_review) if args.primary_review else None
192+
result = run_pipeline(
193+
trigger=trigger,
194+
strategy_profile=profile,
195+
context=context,
196+
primary_review=primary,
197+
evidence_path=evidence_path,
198+
dispatch=args.dispatch,
199+
dry_run=args.dry_run,
200+
)
201+
print(json.dumps(result, ensure_ascii=False, indent=2))
202+
return _exit_code(result)
203+
204+
205+
if __name__ == "__main__":
206+
raise SystemExit(main())

0 commit comments

Comments
 (0)