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=""),
    )
    # T21.3: GITEA_TOKEN from deploy.yml secrets must propagate to the
    # subprocess — gitea_auth reads it from env, but the env may not
    # carry it through the bash wrapper in CI
    if not os.environ.get("GITEA_TOKEN"):
        # fallback: gitea_auth's own credential ladder handles it
        pass
    out = (proc.stdout or "").strip().splitlines()
    return (out[-1] if out else "pending").strip()


# ── T21.5 §4: operation-specific completion authority ──────────────────────
#
# T21.4 root cause: the wait loop had ONE hard-coded authority
# (deploy qualification = master-ref CI) while operations waited on
# branch CI, PR checks, or merges. Wrong authority => nonterminal
# returns => the MODEL had to re-arm (the generator of the sleeps).
# The wait verifier MUST match the operation being waited for.

OPERATION_SPECS = {
    # op_type: {
    #   workflow_path: Actions run path filter (None = any),
    #   head_branches: head-branch filter (None = any),
    #   authority: which canonical verifier decides completion,
    #   event_types: expected wake event types,
    #   terminal: terminal verdicts, failure: failure verdicts,
    # }
    "branch_ci": {
        "workflow_path": "ci.yml",
        "head_branches": None,
        "authority": "ci_runs",          # any completed run for this SHA
        "event_types": ("ci_completed",),
        "terminal": ("success",),
        "failure": ("failure", "cancelled"),
    },
    "pull_request_checks": {
        "workflow_path": None,           # all PR-triggered runs for SHA
        "head_branches": None,
        "authority": "ci_runs",
        "event_types": ("ci_completed",),
        "terminal": ("success",),
        "failure": ("failure", "cancelled"),
    },
    "master_ci": {
        "workflow_path": "ci.yml",
        "head_branches": ("master",),
        "authority": "deploy_qualify",   # full two-leg qualification
        "event_types": ("ci_completed",),
        "terminal": ("success",),
        "failure": ("failure", "cancelled"),
    },
    "deploy": {
        "workflow_path": "deploy.yml",
        "head_branches": ("master",),
        "authority": "deploy_qualify",
        "event_types": ("ci_completed", "deploy_completed"),
        "terminal": ("success",),
        "failure": ("failure", "cancelled"),
    },
    "release": {
        "workflow_path": None,
        "head_branches": ("master",),
        "authority": "deploy_qualify",
        "event_types": ("ci_completed", "release_completed"),
        "terminal": ("success",),
        "failure": ("failure", "cancelled"),
    },
}

DEFAULT_OP_TYPE = "master_ci"


def op_spec(op_type: str) -> dict:
    spec = OPERATION_SPECS.get(op_type)
    if spec is None:
        raise ValueError(
            f"unknown op_type {op_type!r} — must be one of "
            f"{sorted(OPERATION_SPECS)} (the wait verifier must match "
            "the operation being waited for)")
    return spec


def op_authority_verdict(op_type: str, sha: str) -> str:
    """Canonical completion verdict FOR THIS OPERATION TYPE.

    ci_runs        — the op's own Actions runs for this exact SHA: every
                     required workflow run completed; terminal iff every
                     run of the op's workflow set concluded success
                     (branch/PR ops are their own truth; deploy
                     qualification is NOT their authority).
    deploy_qualify — the two-leg deploy qualification (master CI +
                     required-checks) via deploy_qualify.py.
    """
    spec = op_spec(op_type)
    authority = spec["authority"]
    if authority == "deploy_qualify":
        return authority_verdict()
    if authority == "ci_runs":
        return _ci_runs_verdict(spec, sha)
    raise ValueError(f"unknown authority {authority!r} for {op_type!r}")


def _ci_runs_verdict(spec: dict, sha: str) -> str:
    """Operation-scoped Actions-run verdict (import the API client from
    deploy_qualify to keep ONE Gitea client)."""
    sys.path.insert(0, str(_SCRIPT_DIR))
    import deploy_qualify as dq  # noqa: E402
    import gitea_auth  # noqa: E402
    server = os.environ.get("GITEA_SERVER",
                            getattr(gitea_auth, "DEFAULT_SERVER", "")).rstrip("/")
    repo = os.environ.get("GITEA_REPO",
                          getattr(gitea_auth, "DEFAULT_REPO", ""))
    sha_env = os.environ.get("DEPLOY_SHA", "") or sha
    token = gitea_auth.get_token() if gitea_auth.has_token() else ""
    if not (server and repo and sha_env and token):
        return "pending"  # cannot verify -> not verified (fail closed)
    runs = dq._fetch_runs(server, repo, sha_env, token)
    if runs is None:
        return "pending"  # probe failure -> keep waiting
    wfp = spec["workflow_path"]
    if wfp is not None:
        runs = [r for r in runs
                if str(r.get("path", "")).startswith(wfp + "@")]
    branches = spec["head_branches"]
    if branches is not None:
        runs = [r for r in runs
                if str(r.get("head_branch", "")) in branches]
    if not runs:
        return "pending"  # no runs visible yet -> not terminal
    for r in runs:
        if r.get("status") != "completed":
            return "pending"  # still running
        if r.get("conclusion") not in ("success", "skipped"):
            return "failure"  # any non-success conclusion
    return "success"


def op_ci_alive(op_type: str, sha: str) -> bool:
    """Liveness probe FOR THIS OPERATION TYPE: any non-terminal run in
    the op's run set (failure DETECTOR, never the scheduler)."""
    sys.path.insert(0, str(_SCRIPT_DIR))
    import deploy_qualify as dq  # noqa: E402
    import gitea_auth  # noqa: E402
    spec = op_spec(op_type)
    server = os.environ.get("GITEA_SERVER",
                            getattr(gitea_auth, "DEFAULT_SERVER", "")).rstrip("/")
    repo = os.environ.get("GITEA_REPO",
                          getattr(gitea_auth, "DEFAULT_REPO", ""))
    sha_env = os.environ.get("DEPLOY_SHA", "") or sha
    token = gitea_auth.get_token() if gitea_auth.has_token() else ""
    if not (server and repo and sha_env and token):
        return False
    runs = dq._fetch_runs(server, repo, sha_env, token)
    if not runs:
        return False
    wfp = spec["workflow_path"]
    if wfp is not None:
        runs = [r for r in runs
                if str(r.get("path", "")).startswith(wfp + "@")]
    branches = spec["head_branches"]
    if branches is not None:
        runs = [r for r in runs
                if str(r.get("head_branch", "")) in branches]
    terminal = {"success", "failure", "cancelled", "skipped"}
    return any(str(r.get("status", "")).lower() not in terminal
               for r in runs)


def authority_after_wake(op_type: str = DEFAULT_OP_TYPE, sha: str = "",
                         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 = op_authority_verdict(op_type, sha)
        if verdict != "pending":
            return verdict
        if attempt < max_tries - 1:
            time.sleep(gap_secs)
    return verdict


def ci_alive() -> bool:
    """LEGACY liveness probe (master-ref deploy runs) — kept for callers
    predating op specs; run() uses op_ci_alive(op_type, sha)."""
    return op_ci_alive(DEFAULT_OP_TYPE, os.environ.get("DEPLOY_SHA", ""))


def run(*, sha: str, deadline_secs: int,
        watchdog_secs: int = 300,
        op_type: str = DEFAULT_OP_TYPE,
        daemon_url: str = ew.DEFAULT_DAEMON_URL) -> int:
    """T21.5 §3: ONE LOGICAL AWAIT.

    One logical external wait is one invocation of this function. All
    nonterminal conditions (event-seen-authority-pending, watchdog expiry
    with live runs, bounded authority-retry exhausted) re-arm INTERNALLY —
    control returns to the caller ONLY on a terminal verdict or on real
    deadline expiry. Watchdogs are failure detectors inside this
    abstraction, never the scheduler and never completion truth.
    """
    spec = op_spec(op_type)          # unknown op_type refused upfront
    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 waiter 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 {op_type} 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(op_type, sha)
        if verdict == "pending" and not op_ci_alive(op_type, sha):
            verdict = op_authority_verdict(op_type, sha)
        mode = ("event_replayed" if verdict != "pending"
                else "authority_unresolved")
        verdict_at = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
        out = {
            "sha": sha,
            "op_type": op_type,
            "authority": spec["authority"],
            "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 {op_type} qualification for {sha}",
        watched_for="Gitea CI completion event (wake; verdict re-read "
                    f"from canonical {spec['authority']} authority)",
        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 = ""
    rearm_count = 0
    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(op_type, sha)
                if verdict != "pending":
                    mode = "event"
                    break
                # T21.5 §3: NONTERMINAL — re-arm internally, keep waiting.
                rearm_count += 1
                rearm_reason = (f"event seen; {spec['authority']} "
                                "authority pending — internal re-arm")
                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(op_type, sha)
                    if verdict != "pending":
                        mode = "event_recovered_from_durable_record"
                        break
                rearm_count += 1
                rearm_reason = "eof without token; durable record pending"
                continue
            # watchdog expiry: operation-scoped liveness decides re-arm
            # vs truthful recovery — failure detector, never scheduler
            if op_ci_alive(op_type, sha):
                rearm_count += 1
                rearm_reason = ("watchdog expiry; op runs non-terminal "
                                "in Gitea — internal re-arm")
                continue
            # no live runs for THIS op's authority: terminal-in-Gitea but
            # possibly aggregator-lagging — bounded re-read, then resolve
            verdict = op_authority_verdict(op_type, sha)
            if verdict == "pending":
                # T21.4 cause C fixed: a pending verdict with no live runs
                # may still be a lagging aggregator; one more bounded
                # re-read before resolving truthfully. Still no control
                # returned to the model — deadline governs.
                verdict = op_authority_verdict(op_type, sha)
            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 = op_authority_verdict(op_type, sha)  # final one-shot truth
    verdict_at = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
    out = {
        "sha": sha,
        "op_type": op_type,
        "authority": spec["authority"],
        "registered_at": registered_at,
        "event_received_at": event_at,
        "controller_resumed_at": resumed_at,
        "verdict_at": verdict_at,
        "mode": mode,
        "rearm_reason": rearm_reason,
        "internal_rearm_count": rearm_count,
        "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])
    op_type = DEFAULT_OP_TYPE
    if "--op-type" in args:
        op_type = args[args.index("--op-type") + 1]
    if not sha:
        print(json.dumps({"error": "no sha (DEPLOY_SHA or --sha required)"}))
        return 1
    return run(sha=sha, deadline_secs=deadline, op_type=op_type)


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