98 lines
2.8 KiB
Python
98 lines
2.8 KiB
Python
#!/usr/bin/env python3
|
|
"""Execute the bounded five-lane SRE comparison and paid-provider canary."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import asyncio
|
|
import json
|
|
import re
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
def _bootstrap_api_import(repo_root: Path) -> Path:
|
|
candidates = (
|
|
repo_root / "apps" / "api", # source checkout
|
|
repo_root, # production API image: /app/src
|
|
)
|
|
for api_root in candidates:
|
|
if (api_root / "src").is_dir():
|
|
if str(api_root) not in sys.path:
|
|
sys.path.insert(0, str(api_root))
|
|
return api_root
|
|
raise RuntimeError("paid_provider_canary_api_source_root_not_found")
|
|
|
|
|
|
_REPO_ROOT = Path(__file__).resolve().parents[2]
|
|
_API_ROOT = _bootstrap_api_import(_REPO_ROOT)
|
|
|
|
|
|
async def _init_runtime() -> None:
|
|
from src.core.redis_client import init_redis_pool
|
|
|
|
await init_redis_pool()
|
|
|
|
|
|
async def _close_runtime() -> None:
|
|
from src.core.redis_client import close_redis_pool
|
|
|
|
await close_redis_pool()
|
|
|
|
|
|
async def _execute_validation(**kwargs):
|
|
from src.services.paid_provider_canary_validation import (
|
|
run_paid_provider_canary_validation,
|
|
)
|
|
|
|
return await run_paid_provider_canary_validation(**kwargs)
|
|
|
|
|
|
async def _run_validation(**kwargs):
|
|
"""Run with the same Redis lifecycle the API process normally provides."""
|
|
|
|
await _init_runtime()
|
|
try:
|
|
return await _execute_validation(**kwargs)
|
|
finally:
|
|
await _close_runtime()
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--run-ref", required=True)
|
|
parser.add_argument("--operator-id", default="codex-controlled-canary")
|
|
parser.add_argument("--authorization-ref", required=True)
|
|
args = parser.parse_args()
|
|
try:
|
|
receipt = asyncio.run(
|
|
_run_validation(
|
|
run_ref=args.run_ref,
|
|
operator_id=args.operator_id,
|
|
authorization_ref=args.authorization_ref,
|
|
)
|
|
)
|
|
except BaseException as exc:
|
|
error_code = str(exc)
|
|
if not re.fullmatch(r"[a-z][a-z0-9_]{0,119}", error_code):
|
|
error_code = f"canary_failed_{type(exc).__name__}"
|
|
print(
|
|
json.dumps(
|
|
{
|
|
"schema_version": "paid_provider_sre_canary_error_v1",
|
|
"status": "blocked_with_safe_next_action",
|
|
"error_code": error_code[:120],
|
|
"raw_prompt_persisted": False,
|
|
"raw_response_persisted": False,
|
|
"secret_value_returned": False,
|
|
},
|
|
sort_keys=True,
|
|
)
|
|
)
|
|
return 2
|
|
print(json.dumps(receipt, ensure_ascii=False, sort_keys=True))
|
|
return 0 if receipt["independent_verifier"]["passed"] else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|