mirror of
https://github.com/kevinveenbirkenbach/computer-playbook.git
synced 2026-11-06 02:54:26 +00:00
kevinveenbirkenbachandClaude Opus 5
2aca0f62d0
i8cifail died twice on the same run, each failure hiding the next. The retrigger recovers a source run's dispatch inputs from a job log, because the title truncates at 512 bytes and REST answers inputs: null for a finished workflow_dispatch. That reader picked the first depth-1 job outright. A skipped job never uploads a log blob, so its log endpoint answers BlobNotFound as a bare HTTP 404 -- and a force-cancelled run can leave every depth-1 job skipped, which is why deeper jobs are now read too rather than assumed inputs-free. Candidates are taken shallowest first, so the orchestrator's own inputs still win over what a caller forwarded. gh 2.97 also refuses to emit a log holding terminal escape sequences and exits 1; the log downloader read that as no log and silently wrote none. A missed fetch is now tolerated instead of aborting, so the guard in dispatched_priority reports an expired log in words rather than the command dying on gh: HTTP 404. Behind that sat the resolver: it decoded whole priority entries, but an entry may pin axes (web-app-zammad#1@compose+tor), which the display-name codec does not parse. Every pinned entry aborted as corrupt. split_axes -- already the seam decode_list uses -- separates name from suffix, so only the name is resolved and the truncation guard stays sharp. The pin rides along into the retrigger: the entry never ran, so the run holds no evidence its axes were wrong. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
107 lines
3.4 KiB
Python
107 lines
3.4 KiB
Python
"""Git and ``gh`` plumbing for the CI deploy-run status/trigger commands."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
|
|
|
|
def run_id_from_url(url: str) -> str:
|
|
match = re.search(r"/runs/(\d+)", url)
|
|
if not match:
|
|
raise ValueError(f"no run id found in URL: {url}")
|
|
return match.group(1)
|
|
|
|
|
|
def slug_from_url(url: str) -> str:
|
|
"""Extract the ``owner/repo`` slug from a github.com URL (https or ssh)."""
|
|
match = re.search(r"github\.com[/:]([^/]+)/([^/]+?)(?:\.git)?(?:/|$)", url)
|
|
if not match:
|
|
raise ValueError(f"no owner/repo found in URL: {url}")
|
|
return f"{match.group(1)}/{match.group(2)}"
|
|
|
|
|
|
def _run(args: list[str]) -> str:
|
|
return subprocess.run(
|
|
args, check=True, capture_output=True, text=True
|
|
).stdout.strip()
|
|
|
|
|
|
def _branch_remote() -> str:
|
|
"""The remote the current branch tracks (e.g. a fork), falling back to
|
|
``remote.pushDefault``, then 'origin' or the only configured remote."""
|
|
try:
|
|
upstream = _run(
|
|
["git", "rev-parse", "--abbrev-ref", "--symbolic-full-name", "@{upstream}"]
|
|
)
|
|
if "/" in upstream:
|
|
return upstream.split("/", 1)[0]
|
|
except subprocess.CalledProcessError:
|
|
pass
|
|
remotes = _run(["git", "remote"]).split()
|
|
try:
|
|
push_default = _run(["git", "config", "--get", "remote.pushDefault"])
|
|
if push_default in remotes:
|
|
return push_default
|
|
except subprocess.CalledProcessError:
|
|
pass
|
|
if "origin" in remotes:
|
|
return "origin"
|
|
if not remotes:
|
|
raise RuntimeError("no git remote configured")
|
|
return remotes[0]
|
|
|
|
|
|
def resolve_repo() -> str:
|
|
"""The ``owner/repo`` the current branch lives on, derived from its
|
|
tracking remote (not gh's default repo, which may be the upstream)."""
|
|
return slug_from_url(_run(["git", "remote", "get-url", _branch_remote()]))
|
|
|
|
|
|
def _gh(args: list[str], repo: str | None = None, check: bool = True) -> str:
|
|
"""Run ``gh`` and return its stdout.
|
|
|
|
Args:
|
|
args: the ``gh`` arguments, without the ``gh`` itself.
|
|
repo: appended as ``--repo`` when given.
|
|
check: abort the command on a non-zero exit. Pass ``False`` for calls
|
|
whose failure the caller handles, which then yields ``""``.
|
|
"""
|
|
cmd = ["gh", *args]
|
|
if repo:
|
|
cmd += ["--repo", repo]
|
|
proc = subprocess.run(cmd, check=False, capture_output=True, text=True)
|
|
if proc.returncode != 0:
|
|
if not check:
|
|
return ""
|
|
sys.stderr.write(
|
|
proc.stderr or f"gh exited {proc.returncode}: {' '.join(cmd)}\n"
|
|
)
|
|
raise SystemExit(proc.returncode)
|
|
return proc.stdout
|
|
|
|
|
|
def current_branch() -> str:
|
|
return _run(["git", "rev-parse", "--abbrev-ref", "HEAD"])
|
|
|
|
|
|
def fetch_jobs(run_id: str, repo: str | None = None) -> list[dict]:
|
|
return json.loads(_gh(["run", "view", run_id, "--json", "jobs"], repo=repo)).get(
|
|
"jobs", []
|
|
)
|
|
|
|
|
|
def fetch_run(run_id: str, repo: str | None = None) -> dict:
|
|
"""Jobs plus the run title, in one ``gh`` call.
|
|
|
|
The REST API answers ``inputs: null`` for a finished workflow_dispatch, so
|
|
the inputs are recovered from the run itself: short ones from the title
|
|
(``config_from_title``), the long ones from a called job's log
|
|
(``inputs_from_jobs``), which the title truncates.
|
|
"""
|
|
return json.loads(
|
|
_gh(["run", "view", run_id, "--json", "jobs,displayTitle"], repo=repo)
|
|
)
|