269 lines
11 KiB
Python
269 lines
11 KiB
Python
#!/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()
|
||
|
||
|
||
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())
|