feat: emit PR review completion events

.github/workflows/codex_pr_review.yml

@@ -115,6 +115,7 @@ jobs:
         working-directory: source
         run: |
           set -euo pipefail
+          rm -rf "${GITHUB_WORKSPACE}/source/data/output/codex_pr_review"
           bridge_script="${GITHUB_WORKSPACE}/bridge/scripts/run_codex_pr_review.py"
           if [ -f "${bridge_script}" ]; then
             script_path="${bridge_script}"
@@ -154,3 +155,22 @@ jobs:
           name: codex-pr-review-${{ github.event.pull_request.number || github.run_id }}-${{ github.run_id }}
           path: source/data/output/codex_pr_review/
           if-no-files-found: warn
+
+      - name: Emit review completion event
+        if: always()
+        continue-on-error: true
+        env:
+          CODEX_AUDIT_SERVICE_URL: ${{ secrets.CODEX_AUDIT_SERVICE_URL }}
+          CODEX_AUDIT_SERVICE_AUDIENCE: ${{ vars.CODEX_AUDIT_SERVICE_AUDIENCE || 'quant-codex-audit' }}
+          CODEX_REVIEW_STEP_OUTCOME: ${{ steps.review.outcome }}
+          CODEX_REVIEW_DECISION_PATH: ${{ github.workspace }}/source/data/output/codex_pr_review/decision.json
+          CODEX_REVIEW_PR_NUMBER: ${{ github.event.pull_request.number }}
+          CODEX_REVIEW_HEAD_SHA: ${{ github.event.pull_request.head.sha }}
+        run: |
+          set -euo pipefail
+          emitter="${GITHUB_WORKSPACE}/bridge/scripts/emit_pr_review_event.py"
+          if [ ! -f "${emitter}" ]; then
+            echo "::warning::Trusted review event emitter not found."
+            exit 1
+          fi
+          timeout --signal=TERM --kill-after=5s 60s python -I "${emitter}"

cloudflare/codex-audit-proxy/README.md

@@ -6,7 +6,7 @@ This Worker should stay separate from the Pigbibi CodexGateway Worker so origin
 
 The Worker is intentionally thin:
 
-- serves `GET /healthz` locally, proxies legacy `POST /v1/codex-audit`, and proxies async `POST /v1/codex-audit/jobs` plus `GET /v1/codex-audit/jobs/{job_id}`;
+- serves `GET /healthz` locally, proxies legacy/async Codex routes, and permits authenticated `POST /v1/ai/automation/runs` for sanitized review completion events;
 - requires a bearer token before proxying and forwards the GitHub Actions OIDC `Authorization` header to the existing Codex audit service;
 - keeps the VPS origin URL in a Cloudflare Worker secret, not in git;
 - does not store provider keys, GitHub tokens, or Codex credentials.

cloudflare/codex-audit-proxy/src/index.mjs

@@ -13,6 +13,7 @@
     POST /v1/ai/execute
     POST /v1/ai/execute/jobs
     POST /v1/ai/review
+    POST /v1/ai/automation/runs
     POST /v1/ai/feedback/register
     POST /v1/ai/feedback/evaluate
     POST /v1/ai/feedback/shadow
@@ -32,6 +33,7 @@ const ALLOWED_ROUTES = [
   "/v1/ai/execute",
   "/v1/ai/execute/jobs",
   "/v1/ai/review",
+  "/v1/ai/automation/runs",
   "/v1/ai/health",
   "/v1/ai/quota",
   "/v1/ai/changes",
@@ -116,7 +118,7 @@ function methodAllowed(method, pathname) {
   if (pathname === "/v1/ai/changes") return method === "GET";
   // All other routes: POST only
   if (["/v1/ai/analyze", "/v1/ai/execute", "/v1/ai/execute/jobs",
-       "/v1/ai/review", "/v1/ai/feedback/register", "/v1/ai/feedback/evaluate",
+       "/v1/ai/review", "/v1/ai/automation/runs", "/v1/ai/feedback/register", "/v1/ai/feedback/evaluate",
        "/v1/codex-audit", "/v1/codex-audit/jobs"].includes(pathname)) {
     return method === "POST";
   }

cloudflare/codex-audit-proxy/tests/index.test.mjs

@@ -108,3 +108,39 @@ test("worker allows authenticated job status polling subroutes", async () => {
     globalThis.fetch = originalFetch;
   }
 });
+
+test("worker proxies authenticated review event records only via POST", async () => {
+  const originalFetch = globalThis.fetch;
+  globalThis.fetch = async (url, init) => {
+    assert.equal(url, "https://origin.example/v1/ai/automation/runs");
+    assert.equal(init.method, "POST");
+    assert.equal(init.headers.get("Authorization"), "Bearer test-token");
+    return new Response(JSON.stringify({ status: "ok", notification: { status: "sent" } }), {
+      status: 200,
+      headers: { "Content-Type": "application/json" },
+    });
+  };
+
+  try {
+    const response = await worker.fetch(
+      new Request("https://proxy.example/v1/ai/automation/runs", {
+        method: "POST",
+        headers: { Authorization: "Bearer test-token", "Content-Type": "application/json" },
+        body: "{}",
+      }),
+      { CODEX_AUDIT_ORIGIN_URL: "https://origin.example" },
+    );
+    assert.equal(response.status, 200);
+    assert.deepEqual(await response.json(), { status: "ok", notification: { status: "sent" } });
+
+    const rejected = await worker.fetch(
+      new Request("https://proxy.example/v1/ai/automation/runs", {
+        headers: { Authorization: "Bearer test-token" },
+      }),
+      { CODEX_AUDIT_ORIGIN_URL: "https://origin.example" },
+    );
+    assert.equal(rejected.status, 405);
+  } finally {
+    globalThis.fetch = originalFetch;
+  }
+});

scripts/deploy_codex_audit_service.sh

@@ -21,6 +21,7 @@ ALLOWED_DIRECT_REPOSITORIES="${CODEX_AUDIT_SERVICE_ALLOWED_DIRECT_REPOSITORIES:-
 ALLOWED_SOURCE_REPOSITORIES="${CODEX_AUDIT_SERVICE_ALLOWED_SOURCE_REPOSITORIES:-QuantStrategyLab/AIAuditBridge,QuantStrategyLab/BinancePlatform,QuantStrategyLab/CharlesSchwabPlatform,QuantStrategyLab/CnEquitySnapshotPipelines,QuantStrategyLab/CnEquityStrategies,QuantStrategyLab/CryptoLivePoolPipelines,QuantStrategyLab/CryptoStrategies,QuantStrategyLab/FirstradePlatform,QuantStrategyLab/HkEquitySnapshotPipelines,QuantStrategyLab/HkEquityStrategies,QuantStrategyLab/IBKRGatewayManager,QuantStrategyLab/InteractiveBrokersPlatform,QuantStrategyLab/LongBridgePlatform,QuantStrategyLab/MarketSignalSources,QuantStrategyLab/PoliticalEventTrackingResearch,QuantStrategyLab/QmtPlatform,QuantStrategyLab/QuantAdvisorResearch,QuantStrategyLab/QuantPlatformKit,QuantStrategyLab/QuantRuntimeSettings,QuantStrategyLab/QuantStrategyPlugins,QuantStrategyLab/ResearchSignalContextPipelines,QuantStrategyLab/SchwabTokenAutoRefresher,QuantStrategyLab/UsEquitySnapshotPipelines,QuantStrategyLab/UsEquityStrategies}"
 JOB_DIR="${CODEX_AUDIT_SERVICE_JOB_DIR:-/var/lib/codex-audit-bridge/jobs}"
 ADMIN_ENV_FILE="${CODEX_AUDIT_SERVICE_ADMIN_ENV_FILE:-/etc/codex-audit-bridge/admin.env}"
+TELEGRAM_ENV_FILE="${CODEX_AUDIT_SERVICE_TELEGRAM_ENV_FILE:-/etc/codex-audit-bridge/telegram.env}"
 EXECUTION_POLICY_FILE="${CODEX_AUDIT_SERVICE_EXECUTION_POLICY_PATH:-/etc/codex-audit-bridge-policy/execution_policy.json}"
 AUDIT_MODEL="${CODEX_AUDIT_SERVICE_MODEL:-}"
 AUDIT_REASONING_EFFORT="${CODEX_AUDIT_SERVICE_REASONING_EFFORT:-}"
@@ -385,6 +386,7 @@ Environment=CODEX_AUDIT_SERVICE_OPENAI_USAGE_WINDOW_DAYS=${OPENAI_USAGE_WINDOW_D
 Environment=CODEX_AUDIT_SERVICE_ANTHROPIC_USAGE_WINDOW_DAYS=${ANTHROPIC_USAGE_WINDOW_DAYS}
 Environment=CODEX_AUDIT_SERVICE_SANDBOX=read-only
 EnvironmentFile=-${ADMIN_ENV_FILE}
+EnvironmentFile=-${TELEGRAM_ENV_FILE}
 ${audit_model_line}
 ${audit_reasoning_effort_line}
 ${audit_token_line}

scripts/emit_pr_review_event.py

@@ -0,0 +1,209 @@
+#!/usr/bin/env python3
+"""Emit one bounded, non-sensitive PR review completion event."""
+
+from __future__ import annotations
+
+import json
+import os
+from pathlib import Path
+import re
+import sys
+import time
+from typing import Any, Mapping
+import urllib.error
+import urllib.parse
+import urllib.request
+
+
+EVENT_SCHEMA = "qsl.pr_review_event.v1"
+EVENT_TASK = "pr_review_completed"
+DEFAULT_AUDIENCE = "quant-codex-audit"
+MAX_INPUT_BYTES = 128 * 1024
+MAX_RESPONSE_BYTES = 128 * 1024
+REPOSITORY_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]{0,99}/[A-Za-z0-9][A-Za-z0-9._-]{0,99}")
+HEAD_SHA_RE = re.compile(r"[0-9a-f]{40}")
+RUN_ID_RE = re.compile(r"[1-9][0-9]{0,19}")
+REVIEW_OUTCOMES = frozenset({"success", "failure", "cancelled", "skipped"})
+
+
+class ReviewEventEmissionError(ValueError):
+    """Raised when the workflow event cannot be safely emitted."""
+
+
+def _required_string(env: Mapping[str, str], name: str) -> str:
+    value = env.get(name)
+    if type(value) is not str or not value.strip():
+        raise ReviewEventEmissionError(f"{name} is required")
+    return value.strip()
+
+
+def _read_json_object(path: Path) -> dict[str, Any]:
+    try:
+        raw = path.read_bytes()
+    except OSError as exc:
+        raise ReviewEventEmissionError(f"cannot read {path.name}") from exc
+    if len(raw) > MAX_INPUT_BYTES:
+        raise ReviewEventEmissionError(f"{path.name} exceeds size limit")
+    try:
+        value = json.loads(raw)
+    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
+        raise ReviewEventEmissionError(f"{path.name} is not valid JSON") from exc
+    if type(value) is not dict:
+        raise ReviewEventEmissionError(f"{path.name} must contain an object")
+    return value
+
+
+def _decision_flags(path_value: str) -> tuple[bool | None, bool | None]:
+    if not path_value:
+        return None, None
+    path = Path(path_value)
+    if not path.is_file():
+        return None, None
+    try:
+        decision = _read_json_object(path)
+    except ReviewEventEmissionError:
+        return None, None
+    blocked = decision.get("blocked")
+    conflict = decision.get("contract_conflict")
+    return (
+        blocked if type(blocked) is bool else None,
+        conflict if type(conflict) is bool else None,
+    )
+
+
+def build_payload(env: Mapping[str, str]) -> dict[str, Any]:
+    repository = _required_string(env, "GITHUB_REPOSITORY")
+    if REPOSITORY_RE.fullmatch(repository) is None:
+        raise ReviewEventEmissionError("GITHUB_REPOSITORY is invalid")
+    workflow_run_id = _required_string(env, "GITHUB_RUN_ID")
+    if RUN_ID_RE.fullmatch(workflow_run_id) is None:
+        raise ReviewEventEmissionError("GITHUB_RUN_ID is invalid")
+    review_outcome = _required_string(env, "CODEX_REVIEW_STEP_OUTCOME").lower()
+    if review_outcome not in REVIEW_OUTCOMES:
+        raise ReviewEventEmissionError("CODEX_REVIEW_STEP_OUTCOME is invalid")
+
+    pr_number_value = _required_string(env, "CODEX_REVIEW_PR_NUMBER")
+    if not pr_number_value.isascii() or not pr_number_value.isdigit():
+        raise ReviewEventEmissionError("CODEX_REVIEW_PR_NUMBER is invalid")
+    pr_number = int(pr_number_value)
+    if pr_number <= 0 or pr_number > 2**53 - 1:
+        raise ReviewEventEmissionError("CODEX_REVIEW_PR_NUMBER is invalid")
+    head_sha = _required_string(env, "CODEX_REVIEW_HEAD_SHA")
+    if HEAD_SHA_RE.fullmatch(head_sha) is None:
+        raise ReviewEventEmissionError("CODEX_REVIEW_HEAD_SHA is invalid")
+
+    blocked, contract_conflict = _decision_flags(str(env.get("CODEX_REVIEW_DECISION_PATH") or ""))
+    metadata = {
+        "schema": EVENT_SCHEMA,
+        "repository": repository,
+        "pr_number": pr_number,
+        "head_sha": head_sha,
+        "workflow_run_id": workflow_run_id,
+        "review_outcome": review_outcome,
+        "blocked": blocked,
+        "contract_conflict": contract_conflict,
+    }
+    return {
+        "run_id": f"review:{repository}:{pr_number}:{head_sha}:{workflow_run_id}",
+        "task": EVENT_TASK,
+        "task_state": "completed",
+        "mode": "review_only",
+        "source_repository": repository,
+        "metadata": metadata,
+    }
+
+
+def normalize_service_url(value: str) -> str:
+    raw = str(value or "").strip().rstrip("/")
+    parsed = urllib.parse.urlparse(raw)
+    if (
+        parsed.scheme not in {"http", "https"}
+        or not parsed.netloc
+        or parsed.username is not None
+        or parsed.password is not None
+        or parsed.query
+        or parsed.fragment
+    ):
+        raise ReviewEventEmissionError("CODEX_AUDIT_SERVICE_URL is invalid")
+    if parsed.scheme != "https" and parsed.hostname not in {"127.0.0.1", "localhost", "::1"}:
+        raise ReviewEventEmissionError("CODEX_AUDIT_SERVICE_URL must use HTTPS")
+    return raw
+
+
+def request_github_oidc_token(env: Mapping[str, str], audience: str) -> str:
+    request_url = _required_string(env, "ACTIONS_ID_TOKEN_REQUEST_URL")
+    request_token = _required_string(env, "ACTIONS_ID_TOKEN_REQUEST_TOKEN")
+    separator = "&" if "?" in request_url else "?"
+    url = f"{request_url}{separator}audience={urllib.parse.quote(audience)}"
+    request = urllib.request.Request(
+        url,
+        method="GET",
+        headers={
+            "Authorization": f"Bearer {request_token}",
+            "Accept": "application/json",
+            "User-Agent": "qsl-pr-review-event-bridge",
+        },
+    )
+    with urllib.request.urlopen(request, timeout=20) as response:
+        raw = response.read(MAX_RESPONSE_BYTES + 1)
+    if len(raw) > MAX_RESPONSE_BYTES:
+        raise ReviewEventEmissionError("GitHub OIDC response exceeds size limit")
+    payload = json.loads(raw.decode("utf-8"))
+    token = payload.get("value") if type(payload) is dict else None
+    if type(token) is not str or not token:
+        raise ReviewEventEmissionError("GitHub OIDC response is invalid")
+    return token
+
+
+def post_event(service_url: str, token: str, payload: dict[str, Any]) -> None:
+    body = json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8")
+    request = urllib.request.Request(
+        f"{service_url}/v1/ai/automation/runs",
+        data=body,
+        method="POST",
+        headers={
+            "Authorization": f"Bearer {token}",
+            "Content-Type": "application/json",
+            "Accept": "application/json",
+            "User-Agent": "qsl-pr-review-event-bridge",
+        },
+    )
+    with urllib.request.urlopen(request, timeout=20) as response:
+        raw = response.read(MAX_RESPONSE_BYTES + 1)
+    if len(raw) > MAX_RESPONSE_BYTES:
+        raise ReviewEventEmissionError("event service response exceeds size limit")
+    response_body = json.loads(raw.decode("utf-8"))
+    if type(response_body) is not dict or response_body.get("status") != "ok":
+        raise ReviewEventEmissionError("event service response is invalid")
+    notification = response_body.get("notification")
+    if type(notification) is not dict or notification.get("status") not in {"sent", "deduplicated"}:
+        raise ReviewEventEmissionError("review notification was not delivered")
+
+
+def main() -> int:
+    service_url_value = str(os.environ.get("CODEX_AUDIT_SERVICE_URL") or "").strip()
+    if not service_url_value:
+        print("::notice::Review event bridge skipped: CODEX_AUDIT_SERVICE_URL is not configured")
+        return 0
+    try:
+        service_url = normalize_service_url(service_url_value)
+        payload = build_payload(os.environ)
+        audience = str(os.environ.get("CODEX_AUDIT_SERVICE_AUDIENCE") or DEFAULT_AUDIENCE).strip()
+        token = request_github_oidc_token(os.environ, audience)
+        for attempt in range(1, 4):
+            try:
+                post_event(service_url, token, payload)
+                print(f"Review completion event recorded for {payload['source_repository']}")
+                return 0
+            except (OSError, urllib.error.URLError, ReviewEventEmissionError, json.JSONDecodeError):
+                if attempt == 3:
+                    raise
+                time.sleep(attempt)
+    except (OSError, urllib.error.URLError, ReviewEventEmissionError, json.JSONDecodeError) as exc:
+        print(f"::warning::Review event bridge delivery failed: {type(exc).__name__}", file=sys.stderr)
+        return 1
+    return 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())