rune/scripts/forge_board.py

#!/usr/bin/env python3
"""forge_board.py — #204: the forge as the single pipeline state machine.

Reads the forge's OWN state (open issues, labels, CI status on each head)
and derives the next action per issue. No agent memory involved: any agent
(or a crashed-and-replaced session) runs this and knows what happens next.

Derivation rules (deterministic, zero tokens):
  1. issue carries `blocked`        -> BLOCKED (say on what, from the body)
  2. head has failing CI            -> FIX FIRST (fault-watch contract)
  3. head has pending CI            -> WAITING (subject + ETA shape)
  4. issue carries `awaiting-review`-> REVIEW (reviewer gates decide)
  5. no linked work (no contract)   -> OPEN CONTRACT (scaffolding is absolute)
  6. otherwise                      -> IMPLEMENT per the body's acceptance

Exit codes: 0 = board read; 1 = forge unreachable; 2 = no open issues
(the factory's only other legal stop).
"""

from __future__ import annotations

import argparse
import json
import sys
from typing import Any, Callable, Dict, List, Optional

Fetch = Callable[[str], Optional[Dict[str, Any]]]

BLOCKED_LABELS = {"blocked", "waiting-on-user", "on-hold"}
REVIEW_LABELS = {"awaiting-review", "in-review", "review-required"}
CI_PENDING = {"pending", "running", "unknown"}
CI_FAILED = {"failure", "cancelled"}


def derive_next_action(issue: Dict[str, Any], ci_state: Optional[Dict[str, Any]]) -> Dict[str, Any]:
    """Pure derivation: forge state in, next action out. No network, no memory."""
    labels = {l.get("name", "") for l in issue.get("labels", [])}
    num = issue.get("number")
    title = issue.get("title", "")

    if labels & BLOCKED_LABELS:
        return {
            "issue": num,
            "title": title,
            "action": "BLOCKED",
            "why": "carries a blocking label; unblock before the queue touches it",
        }

    if ci_state:
        verdict = ci_state.get("verdict")
        if verdict in CI_FAILED:
            return {
                "issue": num,
                "title": title,
                "action": "FIX FIRST",
                "why": f"CI on head {ci_state.get('sha', '?')[:8]} is {verdict}; the fault-watch contract fixes the submit before the queue resumes",
            }
        if verdict in CI_PENDING:
            return {
                "issue": num,
                "title": title,
                "action": "WAITING",
                "why": f"CI on head {ci_state.get('sha', '?')[:8]} is {verdict}; factory multiplexes the next batch meanwhile",
            }

    if labels & REVIEW_LABELS:
        return {
            "issue": num,
            "title": title,
            "action": "REVIEW",
            "why": "awaiting reviewer gates (spec-compliance + code-quality)",
        }

    body = issue.get("body") or ""
    if "BUILDER-" in body:
        return {
            "issue": num,
            "title": title,
            "action": "CONTINUE CONTRACT",
            "why": "a construction contract is already linked in the body",
        }
    return {
        "issue": num,
        "title": title,
        "action": "OPEN CONTRACT",
        "why": "no linked work in the body; scaffolding is absolute — contract first",
    }


def read_board(
    fetch: Fetch, owner: str = "conductor", repo: str = "rune", base: str = "http://127.0.0.1:3000/api/v1"
) -> Dict[str, Any]:
    """Pull open issues + CI state from the forge and derive the board."""
    raw = fetch(f"{base}/repos/{owner}/{repo}/issues?state=open&limit=50")
    if raw is None:
        raise RuntimeError("forge unreachable")
    issues = raw if isinstance(raw, list) else raw.get("data", [])

    board: List[Dict[str, Any]] = []
    for issue in issues:
        sha = _head_sha_from_labels(issue) or _head_sha_from_body(issue)
        ci_state = None
        if sha:
            status = fetch(f"{base}/repos/{owner}/{repo}/commits/{sha}/status")
            if status:
                ci_state = {"sha": sha, "verdict": status.get("state", "unknown")}
        board.append(derive_next_action(issue, ci_state))

    order = {"FIX FIRST": 0, "CONTINUE CONTRACT": 1, "OPEN CONTRACT": 2, "REVIEW": 3, "WAITING": 4, "BLOCKED": 5}
    board.sort(key=lambda r: (order.get(r["action"], 9), r["issue"] or 0))
    return {"repo": f"{owner}/{repo}", "open": len(board), "board": board}


def _head_sha_from_labels(issue: Dict[str, Any]) -> Optional[str]:
    for l in issue.get("labels", []):
        name = l.get("name", "")
        if name.startswith("head:"):
            return name.split("head:", 1)[1].strip()
    return None


def _head_sha_from_body(issue: Dict[str, Any]) -> Optional[str]:
    import re

    m = re.search(r"\b([0-9a-f]{40})\b", issue.get("body") or "")
    return m.group(1) if m else None


def _default_fetch(url: str) -> Optional[Any]:
    try:
        import urllib.request

        sys.path.insert(0, "/home/nick/.config/opencode/shared")
        from gitea_auth import load_gitea_token

        req = urllib.request.Request(url, headers={"Authorization": f"token {load_gitea_token()}"})
        with urllib.request.urlopen(req, timeout=10) as r:
            return json.load(r)
    except Exception:
        return None


def main() -> int:
    ap = argparse.ArgumentParser(description="#204 forge board: the next action, from forge state alone")
    ap.add_argument("--json", action="store_true", help="machine-readable output")
    args = ap.parse_args()
    try:
        board = read_board(_default_fetch)
    except RuntimeError as e:
        print(f"forge_board: {e}", file=sys.stderr)
        return 1
    if args.json:
        print(json.dumps(board, indent=2))
    else:
        print(f"== FORGE BOARD {board['repo']} — {board['open']} open ==")
        for row in board["board"]:
            print(f"  #{row['issue']} [{row['action']}] {row['title'][:60]} — {row['why']}")
    if board["open"] == 0:
        return 2
    return 0


if __name__ == "__main__":
    raise SystemExit(main())