From dfbcb72e9426c621ad8b77588431d996e76e4fa6 Mon Sep 17 00:00:00 2001 From: Erich Blume Date: Sat, 18 Apr 2026 16:39:26 -0700 Subject: [PATCH] Add runner-logs mise task for Forgejo Actions Authenticated runner-logs script that lists workflow runs and fetches job logs. Uses Forgejo API token from 1Password and auto-detects the repo from git remote. Co-Authored-By: Claude Opus 4.6 (1M context) --- mise-tasks/runner-logs | 254 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 254 insertions(+) create mode 100755 mise-tasks/runner-logs diff --git a/mise-tasks/runner-logs b/mise-tasks/runner-logs new file mode 100755 index 0000000..6ee1531 --- /dev/null +++ b/mise-tasks/runner-logs @@ -0,0 +1,254 @@ +#!/usr/bin/env -S uv run --script +# /// script +# requires-python = ">=3.12" +# dependencies = ["httpx>=0.28.1", "rich>=14.0.0", "typer>=0.24.0"] +# /// +#MISE description="List recent Forgejo Actions runs or fetch logs for a specific job" +#USAGE arg "[run_number]" help="Run number to show jobs for (omit to list recent runs)" +#USAGE flag "--job -j " help="Job index (0-based) to fetch logs for" +#USAGE flag "--repo " help="Forge repo (owner/name), default: detected from git remote" +#USAGE flag "--limit -n " help="Max runs to display (0 for all)" +#USAGE flag "--token " help="Forgejo API token (default: read from 1Password)" +"""List recent Forgejo Actions runs and fetch job logs. + +Usage: + mise run runner-logs # list recent runs (default 15) + mise run runner-logs -n 0 # list ALL runs + mise run runner-logs --repo eblume/hermes # list runs for a different repo + mise run runner-logs 474 # show jobs in run 474 + mise run runner-logs 474 -j 1 # fetch logs for job 1 of run 474 +""" + +import os +import re +import subprocess +import sys +from typing import Annotated + +import httpx +import typer +from rich.console import Console +from rich.table import Table + +FORGE_URL = "https://forge.ops.eblu.me" +FORGE_API = f"{FORGE_URL}/api/v1" +OP_TOKEN_REF = "op://vg6xf6vvfmoh5hqjjhlhbeoaie/w3663ffnvkewbftncqxtcpeavy/api-token" + +app = typer.Typer(add_completion=False) + + +def resolve_token(explicit_token: str | None, console: Console) -> str: + """Resolve Forgejo API token: explicit flag > FORGEJO_TOKEN env > 1Password.""" + if explicit_token: + return explicit_token + env_token = os.environ.get("FORGEJO_TOKEN", "").strip() + if env_token: + return env_token + console.print("[dim]Reading Forgejo API token from 1Password...[/dim]") + result = subprocess.run( + ["op", "read", OP_TOKEN_REF], + capture_output=True, + text=True, + check=True, + ) + return result.stdout.strip() + + +def detect_repo_from_git() -> str | None: + """Sniff owner/repo from the git remote 'origin' URL.""" + try: + result = subprocess.run( + ["git", "remote", "get-url", "origin"], + capture_output=True, + text=True, + check=True, + ) + except (subprocess.CalledProcessError, FileNotFoundError): + return None + url = result.stdout.strip() + # Match SSH (git@host:owner/repo.git) or HTTPS (https://host/owner/repo.git) + m = re.search(r"[/:]([^/]+/[^/]+?)(?:\.git)?$", url) + if not m: + return None + candidate = m.group(1) + # Only use it if the remote points at our forge + if "forge.ops.eblu.me" in url or "forge.eblu.me" in url: + return candidate + return None + + +def auth_headers(token: str) -> dict[str, str]: + return {"Authorization": f"token {token}"} + + +def fetch_tasks(repo: str, token: str) -> list[dict]: + """Fetch all tasks from the Forgejo API, paginating if needed.""" + tasks: list[dict] = [] + page = 1 + while True: + resp = httpx.get( + f"{FORGE_API}/repos/{repo}/actions/tasks", + params={"page": page, "limit": 50}, + headers=auth_headers(token), + timeout=15, + ) + resp.raise_for_status() + batch = resp.json().get("workflow_runs", []) + if not batch: + break + tasks.extend(batch) + page += 1 + return tasks + + +def list_runs(repo: str, limit: int, token: str, console: Console) -> None: + """List recent workflow runs, grouped by run number.""" + tasks = fetch_tasks(repo, token) + + # Group tasks by run_number + runs: dict[int, list[dict]] = {} + for t in tasks: + rn = t["run_number"] + runs.setdefault(rn, []).append(t) + + table = Table(title=f"Recent runs — {repo}") + table.add_column("Run #", style="cyan", no_wrap=True) + table.add_column("Status") + table.add_column("Jobs") + table.add_column("Title") + table.add_column("Event") + + shown = 0 + for rn in sorted(runs, reverse=True): + if limit > 0 and shown >= limit: + break + + jobs = sorted(runs[rn], key=lambda x: x["id"]) + + # Aggregate status: worst status wins + statuses = [j.get("status", "") for j in jobs] + if "failure" in statuses: + status, style = "failure", "red" + elif "running" in statuses or "waiting" in statuses: + status, style = "running", "yellow" + elif all(s == "success" for s in statuses): + status, style = "success", "green" + else: + status, style = statuses[0], "yellow" + + job_names = ", ".join(j.get("name", "?")[:30] for j in jobs) + title = (jobs[0].get("display_title") or "")[:40] + event = jobs[0].get("event", "") + + table.add_row( + str(rn), + f"[{style}]{status}[/{style}]", + job_names, + title, + event, + ) + shown += 1 + + console.print(table) + console.print("\n[dim]Use: mise run runner-logs to see jobs in a run[/dim]") + console.print("[dim] mise run runner-logs -j N to fetch logs for job N[/dim]") + + +def show_jobs(run_number: int, repo: str, token: str, console: Console) -> None: + """Show the jobs within a specific run.""" + tasks = fetch_tasks(repo, token) + + jobs = sorted( + [t for t in tasks if t["run_number"] == run_number], + key=lambda x: x["id"], + ) + if not jobs: + typer.echo(f"Error: No jobs found for run #{run_number}", err=True) + raise typer.Exit(1) + + table = Table(title=f"Jobs in run #{run_number} — {repo}") + table.add_column("Job #", style="cyan", no_wrap=True) + table.add_column("Status") + table.add_column("Name") + table.add_column("Created") + + for i, job in enumerate(jobs): + status = job.get("status", "") + style = "green" if status == "success" else "red" if status == "failure" else "yellow" + table.add_row( + str(i), + f"[{style}]{status}[/{style}]", + job.get("name", ""), + job.get("created_at", ""), + ) + + console.print(table) + console.print(f"\n[dim]Use: mise run runner-logs {run_number} -j N to fetch logs for job N[/dim]") + + +def fetch_log(run_number: int, job_index: int, repo: str, token: str) -> None: + """Fetch logs for a specific job via the Forgejo web endpoint.""" + url = f"{FORGE_URL}/{repo}/actions/runs/{run_number}/jobs/{job_index}/attempt/1/logs" + resp = httpx.get(url, headers=auth_headers(token), timeout=30, follow_redirects=True) + if resp.status_code == 404: + typer.echo( + f"Error: No logs found for run #{run_number} job {job_index}", + err=True, + ) + typer.echo(f"URL: {url}", err=True) + raise typer.Exit(1) + resp.raise_for_status() + sys.stdout.write(resp.text) + + +@app.command() +def main( + run_number: Annotated[ + int | None, + typer.Argument(help="Run number to show jobs for (omit to list recent runs)"), + ] = None, + job: Annotated[ + int | None, + typer.Option("--job", "-j", help="Job index (0-based) to fetch logs for"), + ] = None, + repo: Annotated[ + str | None, + typer.Option("--repo", help="Forge repo (owner/name), default: detected from git remote"), + ] = None, + limit: Annotated[ + int, + typer.Option("--limit", "-n", help="Max runs to display (0 for all)"), + ] = 15, + token: Annotated[ + str | None, + typer.Option("--token", help="Forgejo API token (default: read from 1Password)"), + ] = None, +) -> None: + """List recent Forgejo Actions runs or fetch logs for a specific job.""" + console = Console() + + if repo is None: + repo = detect_repo_from_git() + if repo is None: + typer.echo( + "Error: could not detect repo from git remote; use --repo owner/name", + err=True, + ) + raise typer.Exit(1) + console.print(f"[dim]Detected repo: {repo}[/dim]") + + resolved_token = resolve_token(token, console) + + if run_number is None: + if job is not None: + typer.echo("Error: --job requires a run number", err=True) + raise typer.Exit(1) + list_runs(repo, limit, resolved_token, console) + elif job is None: + show_jobs(run_number, repo, resolved_token, console) + else: + fetch_log(run_number, job, repo, resolved_token) + + +if __name__ == "__main__": + app()