rune/scripts/ci_event_wait.py

#!/usr/bin/env python3
"""ci_event_wait.py — T16 §16C/§16D: event-driven CI qualification wait.

Replaces the deploy.yml sleep-30 → query → repeat polling loop with the
EXISTING W/R event bridge (src/rune/event_wait.py — the SAME primitive the
builder adapter uses; §16E: one bridge, adapters differ only in identity
and authority probe).

Flow (mandate §16C):

    push / CI workflow starts
        ↓
    register correlation  (provider="gitea", event_type="ci_completed",
                           run_ref="ci-<sha>", sha=<sha>)
        ↓
    WAITING_FOR_EVENT     (await_resume_event: select() on FIFO, no floor)
        ↓
    ci.yml final job POSTs /process-event  (the WAKE)
        ↓
    daemon matches → resume_cmd fires → FIFO token → controller resumes
        ↓
    canonical Gitea state INDEPENDENTLY re-read (deploy_qualify.py verdict)
        ↓
    verdict

The event is a WAKE, never truth. The final verdict always comes from
deploy_qualify.py (required-checks context + exact-SHA ci.yml run, both
legs). Watchdog deadline is a FAILURE DETECTOR, never the scheduler:

  - event arrives → authority re-read; pending → bounded re-arm until the
    overall deadline (ack-window semantics, same as the daemon's wake sweep).
  - event lost/suppressed → watchdog expires → ONE-SHOT authority query:
    terminal verdict → recovered truthfully (no invented wait);
    CI still running (non-terminal Gitea runs for the SHA) → re-arm;
    nothing alive and no verdict → truthful failure (deployment refused).

Exit codes: 0 = qualified (success), 1 = refused (failure/timeout).
Output: one JSON line with the §16D latency-proof timestamps.
"""
from __future__ import annotations

import json
import os
import subprocess
import sys
import time
from pathlib import Path

_SCRIPT_DIR = Path(__file__).resolve().parent
sys.path.insert(0, str(_SCRIPT_DIR.parent / "src"))

from rune import event_wait as ew  # noqa: E402

REPO_ROOT = _SCRIPT_DIR.parent
QUALIFY_SCRIPT = _SCRIPT_DIR / "deploy_qualify.py"


def authority_verdict() -> str:
    """ONE-SHOT canonical Gitea verdict via deploy_qualify.py — the only
    completion truth. Never parses the event payload."""
    proc = subprocess.run(
        [sys.executable, str(QUALIFY_SCRIPT)],
        capture_output=True, text=True, timeout=120,
        env=dict(os.environ, PYTHONPATH=""),
    )
    out = (proc.stdout or "").strip().splitlines()
    return (out[-1] if out else "pending").strip()


def authority_after_wake(max_tries: int = 3, gap_secs: int = 4) -> str:
    """Post-wake authority read. The daemon's own wake sweep uses a
    bounded event-triggered retry (4s ×3) for the job-finished-microseconds-
    ago ack window; a verdict read here can hit the same window. This is a
    bounded schedule, not a polling loop — 3 reads max, then back to the
    event wait / watchdog."""
    verdict = "pending"
    for attempt in range(max_tries):
        verdict = authority_verdict()
        if verdict != "pending":
            return verdict
        if attempt < max_tries - 1:
            time.sleep(gap_secs)
    return verdict


def ci_alive() -> bool:
    """Liveness probe for THIS adapter: any non-terminal Actions run for
    the SHA in canonical Gitea (one-shot on watchdog expiry — not a poll).
    Import the probe from deploy_qualify to keep ONE API client."""
    sys.path.insert(0, str(_SCRIPT_DIR))
    import deploy_qualify as dq  # noqa: E402  (spec-loaded scripts pattern)
    import gitea_auth
    server = os.environ.get("GITEA_SERVER", "").rstrip("/")
    repo = os.environ.get("GITEA_REPO", "")
    sha = os.environ.get("DEPLOY_SHA", "")
    if not (server and repo and sha):
        return False
    token = gitea_auth.get_token() if gitea_auth.has_token() else ""
    if not token:
        return False
    runs = dq._fetch_runs(server, repo, sha, token)
    if not runs:
        return False
    terminal = {"success", "failure", "cancelled", "skipped"}
    return any(str(r.get("status", "")).lower() not in terminal
               for r in runs)


def run(*, sha: str, deadline_secs: int,
        watchdog_secs: int = 300,
        daemon_url: str = ew.DEFAULT_DAEMON_URL) -> int:
    op = f"ci-{sha}"
    t0 = time.time()
    # §16Y #10 race fix: the FIFO reader exists BEFORE registration, so a
    # wake arriving in the registration window can always be delivered.
    fifo_fd = ew.open_fifo_reader(op)
    # Event replay (T16 §16C live finding): on a serialized runner the
    # completion event usually arrives BEFORE the deploy job registers.
    # A matching durable event means the wake already happened — skip
    # the rendezvous entirely and go straight to the authority read.
    prior = ew.find_matching_event("gitea", "ci_completed", op)
    if prior is not None and prior.get("exit_code") is not None:
        rec = ew.register_process_wait(
            op, project="rune",
            objective=f"event-driven CI qualification for {sha}",
            watched_for="Gitea CI completion event (replayed from the "
                        "durable event log)",
            provider="gitea", event_type="ci_completed", sha=sha,
            watchdog_secs=min(watchdog_secs, deadline_secs))
        registered_at = rec["created_at"]
        event_at = time.strftime(
            "%Y-%m-%dT%H:%M:%SZ", time.gmtime())
        resumed_at = event_at
        # the event is a WAKE, never truth — canonical authority decides
        verdict = authority_after_wake()
        if verdict == "pending" and not ci_alive():
            verdict = authority_verdict()
        mode = ("event_replayed" if verdict != "pending"
                else "authority_unresolved")
        verdict_at = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
        out = {
            "sha": sha,
            "registered_at": registered_at,
            "event_received_at": event_at,
            "controller_resumed_at": resumed_at,
            "verdict_at": verdict_at,
            "mode": mode,
            "rearm_reason": "completion event replayed from durable log "
                            "(arrived before registration)",
            "verdict": verdict,
            "wall_secs": round(time.time() - t0, 3),
        }
        print(json.dumps(out))
        return 0 if verdict == "success" else 1
    rec = ew.register_process_wait(
        op, project="rune",
        objective=f"event-driven CI qualification for {sha}",
        watched_for="Gitea CI completion event (wake; verdict re-read "
                    "from canonical Gitea)",
        provider="gitea", event_type="ci_completed", sha=sha,
        watchdog_secs=min(watchdog_secs, deadline_secs))
    registered_at = rec["created_at"]
    deadline = time.monotonic() + deadline_secs
    verdict = "pending"
    mode = "event"
    event_at = ""
    resumed_at = ""
    rearm_reason = ""
    try:
        while time.monotonic() < deadline:
            remaining = max(1, int(deadline - time.monotonic()))
            result = ew.await_resume_event(
                op, watchdog_secs=min(remaining, watchdog_secs), pgid=None,
                fifo_fd=fifo_fd)
            if result["kind"] == "event":
                event_at = event_at or time.strftime(
                    "%Y-%m-%dT%H:%M:%SZ", time.gmtime())
                resumed_at = time.strftime("%Y-%m-%dT%H:%M:%SZ",
                                           time.gmtime())
                # WAKE received — canonical authority decides, never the
                # event payload
                verdict = authority_after_wake()
                if verdict != "pending":
                    mode = "event"
                    break
                rearm_reason = "event seen; authority pending"
                if not ci_alive():
                    # CI terminal in Gitea but aggregator lagging: one
                    # final truthful read, then stop
                    verdict = authority_verdict()
                    mode = ("event" if verdict != "pending"
                            else "authority_unresolved")
                    break
                continue
            if result["kind"] == "eof":
                # writer gone without a token: the DURABLE record decides
                rec = ew.operation_wait(op) or {}
                if rec.get("status") in ("COMPLETED", "FAILED",
                                         "READY_TO_RESUME", "RESUMING") or \
                        (rec.get("status") == "RESUME_UNVERIFIED"
                         and rec.get("process_event_evidence")):
                    event_at = event_at or time.strftime(
                        "%Y-%m-%dT%H:%M:%SZ", time.gmtime())
                    resumed_at = time.strftime("%Y-%m-%dT%H:%M:%SZ",
                                               time.gmtime())
                    verdict = authority_after_wake()
                    if verdict != "pending":
                        mode = "event_recovered_from_durable_record"
                        break
                rearm_reason = "eof without token; durable record pending"
                continue
            # watchdog expiry: adapter liveness (Gitea runs) decides
            # re-arm vs truthful recovery — failure detector, never
            # scheduler
            if ci_alive():
                rearm_reason = "watchdog expiry; CI runs non-terminal " \
                               "in Gitea — re-arm"
                continue
            verdict = authority_verdict()
            if verdict == "pending":
                mode = "watchdog_lost_event_no_verdict"
            else:
                mode = "watchdog_authority_recovered"
            break
    finally:
        os.close(fifo_fd)
    if verdict == "pending":
        if mode == "event":
            mode = "deadline_pending"
        verdict = authority_verdict()  # final one-shot truth
    verdict_at = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
    out = {
        "sha": sha,
        "registered_at": registered_at,
        "event_received_at": event_at,
        "controller_resumed_at": resumed_at,
        "verdict_at": verdict_at,
        "mode": mode,
        "rearm_reason": rearm_reason,
        "verdict": verdict,
        "wall_secs": round(time.time() - t0, 3),
    }
    print(json.dumps(out))
    return 0 if verdict == "success" else 1


def main() -> int:
    sha = os.environ.get("DEPLOY_SHA", "")
    args = sys.argv[1:]
    if "--sha" in args:
        sha = args[args.index("--sha") + 1]
    deadline = 900
    if "--deadline" in args:
        deadline = int(args[args.index("--deadline") + 1])
    if not sha:
        print(json.dumps({"error": "no sha (DEPLOY_SHA or --sha required)"}))
        return 1
    return run(sha=sha, deadline_secs=deadline)


if __name__ == "__main__":
    sys.exit(main())