Blame for scripts/ship.py

#!/usr/bin/env python3
# SPDX-FileCopyrightText: 2026 Shamiur Rashid Sunny
# SPDX-License-Identifier: AGPL-3.0-only
"""Part 157: one-shot ship helper for the Sunnyware project.

Usage:
  python scripts/ship.py <part_number> "<short description>" [options]

Options:
  --body TEXT        Detailed body for HANDOFF.md (defaults to desc)
  --wait N           Seconds to wait for HF rebuild (default: 150)
  --verify-role R    Role to use for live verify (default: general)
  --verify-input T   Input for live verify (default: Reply with PONG.)
  --skip STEPS       Comma-separated steps to skip.
                     Valid: smokes,gate,docs,commit,gh,hf,verify
  --dry-run          Show what would happen without touching git
  --no-color         Disable ANSI colors

Steps performed:
  1. smokes    - run every scripts/smoke_test_part*.py
  2. gate      - scripts/pre_push_check.py
  3. docs      - append HANDOFF + WORKFLOW + bump AGENTS footer
  4. commit    - git add + commit with the assembled message
  5. gh        - git push origin main
  6. hf        - ./push_hf.sh
  7. verify    - wait for HF, GET /health/ready, fire live request

Never pushes broken code: if any smoke or the gate fails, we abort
before touching git.
"""

import argparse
import datetime
import json
import os
import re
import subprocess
import sys
import time
import urllib.error
import urllib.request
from pathlib import Path

# Part 157b: force UTF-8 stdout on Windows (cp1252 can't encode
# box-drawing chars, emoji, or the ANSI color codes we use).
try:
    sys.stdout.reconfigure(encoding="utf-8")
    sys.stderr.reconfigure(encoding="utf-8")
except Exception:
    pass

ROOT = Path.cwd()
SCRIPTS = ROOT / "scripts"
SRC = ROOT / "src" / "sunnyware"
BASE_URL = "https://shamiur-sunnyware.hf.space"

# Part 157c: legacy smokes that had hard-coded counts before
# newer parts changed them (e.g. roles went 9 -> 10 when the
# auditor role was added in Part 142a). ship.py warns on these
# but does not abort. A future cleanup part will update them.
LEGACY_FAILING_SMOKES = set()  # Part 157b: legacy archived

USE_COLOR = True


def c(txt, code):
    if not USE_COLOR:
        return txt
    return f"\033[{code}m{txt}\033[0m"


def info(msg):
    print(f"  {msg}")


def ok(step, msg):
    print(f"{c('[OK]', '32')} [{step}] {msg}")


def fail(step, msg, exit_code=1):
    print(f"{c('[FAIL]', '31')} [{step}] {msg}")
    sys.exit(exit_code)


def hdr(title):
    line = "=" * 70
    print("")
    print(c(line, "36"))
    print(c(f"  {title}", "36"))
    print(c(line, "36"))


# ─────────────────────────────────────────────────────────────
# STEP 1 — run smokes
# ─────────────────────────────────────────────────────────────
def run_smokes():
    hdr("STEP 1 — smokes")
    files = sorted(SCRIPTS.glob("smoke_test_part*.py"))

    def num_key(p):
        m = re.search(r"part(\d+)", p.name)
        return int(m.group(1)) if m else 0

    files.sort(key=num_key)
    failed = []
    for f in files:
        try:
            # Part 157b: force UTF-8 so smokes that print non-cp1252
            # chars (em-dash, box-drawing) don't spuriously fail on
            # Windows when ship.py captures their output.
            _env = os.environ.copy()
            _env["PYTHONIOENCODING"] = "utf-8"
            _env["PYTHONUTF8"] = "1"
            r = subprocess.run(
                [sys.executable, str(f)],
                capture_output=True, text=True, timeout=180,
                env=_env, encoding="utf-8", errors="replace",
            )
        except subprocess.TimeoutExpired:
            info(f"{c('[FAIL]', '31')} {f.name:36s} TIMEOUT")
            failed.append(f.name)
            continue
        lines = [ln for ln in (r.stdout or "").strip().splitlines() if ln.strip()]
        tail = lines[-1] if lines else "(no output)"
        passed = r.returncode == 0
        is_legacy = f.name in LEGACY_FAILING_SMOKES
        if passed:
            marker = c("[OK]", "32")
        elif is_legacy:
            marker = c("[WARN]", "33")   # yellow
            tail = "(legacy allowlist) " + tail
        else:
            marker = c("[FAIL]", "31")
        info(f"{marker} {f.name:36s} {tail}")
        if not passed and not is_legacy:
            failed.append(f.name)

    if failed:
        fail("1", f"{len(failed)} smoke(s) failed: {', '.join(failed)}")
    ok("1", f"{len(files)} smoke(s) passed")


# ─────────────────────────────────────────────────────────────
# STEP 2 — pre-push gate
# ─────────────────────────────────────────────────────────────
def run_gate():
    hdr("STEP 2 — pre-push gate")
    r = subprocess.run(
        [sys.executable, str(SCRIPTS / "pre_push_check.py")],
        capture_output=True, text=True,
    )
    for line in (r.stdout or "").splitlines():
        if "SUMMARY" in line or "RESULT" in line:
            info(line.strip())
    if r.returncode != 0:
        fail("2", "pre-push gate blocked")
    ok("2", "gate passed")


# ─────────────────────────────────────────────────────────────
# STEP 3 — docs sync
# ─────────────────────────────────────────────────────────────
def update_docs(part, desc, body, dry_run=False):
    hdr("STEP 3 — docs sync")
    # HANDOFF.md
    p = ROOT / "HANDOFF.md"
    t = p.read_text(encoding="utf-8")
    marker = f"### Part {part} - {desc}"
    if marker in t:
        ok("3", f"HANDOFF already has Part {part}")
    else:
        when = datetime.date.today().isoformat()
        entry = f"\n\n{marker}\n{body}\n"
        if not dry_run:
            p.write_text(t.rstrip() + entry, encoding="utf-8")
        ok("3", f"HANDOFF extended (Part {part})")

    # WORKFLOW.md
    p = ROOT / "WORKFLOW.md"
    t = p.read_text(encoding="utf-8")
    if f"| {part} " in t:
        ok("3", f"WORKFLOW already has Part {part}")
    else:
        when = datetime.date.today().isoformat()
        block = (
            f"\n\n## {when} - Part {part}\n\n"
            f"| Part | Feature |\n"
            f"|------|---------|\n"
            f"| {part} | {desc} |\n\n"
            f"{body}\n"
        )
        if not dry_run:
            p.write_text(t.rstrip() + block, encoding="utf-8")
        ok("3", f"WORKFLOW extended (Part {part})")

    # AGENTS.md footer
    p = ROOT / "AGENTS.md"
    t = p.read_text(encoding="utf-8")
    lines = t.splitlines()
    footer_idx = None
    for i, ln in enumerate(lines):
        if ln.startswith("**Last update:**"):
            footer_idx = i
            break
    new_footer = f"**Last update:** {datetime.date.today().isoformat()} (Part {part} - {desc})"
    if footer_idx is None:
        if not dry_run:
            p.write_text(t.rstrip() + "\n\n" + new_footer + "\n", encoding="utf-8")
        ok("3", "AGENTS footer appended")
    elif lines[footer_idx] == new_footer:
        ok("3", "AGENTS footer already current")
    else:
        lines[footer_idx] = new_footer
        if not dry_run:
            p.write_text("\n".join(lines) + "\n", encoding="utf-8")
        ok("3", "AGENTS footer bumped")


# ─────────────────────────────────────────────────────────────
# STEP 4 — commit
# ─────────────────────────────────────────────────────────────
def do_commit(part, desc, body, dry_run=False):
    hdr("STEP 4 — commit")

    # files to add: source tree + scripts + docs (safe glob)
    to_add = [
        "src/",
        "scripts/",
        "HANDOFF.md",
        "WORKFLOW.md",
        "AGENTS.md",
        "RESUME.md",
    ]
    if dry_run:
        ok("4", "dry-run: skipped git add")
    else:
        subprocess.run(["git", "add"] + to_add, check=False)

    status = subprocess.run(
        ["git", "status", "--porcelain"],
        capture_output=True, text=True,
    )
    dirty = [ln for ln in (status.stdout or "").splitlines() if ln.strip()]
    if not dirty:
        ok("4", "nothing to commit (tree clean)")
        return None

    msg = f"Part {part}: {desc}\n\n{body}\n"
    mf = ROOT / f".commit_msg_part{part}.txt"
    mf.write_text(msg, encoding="utf-8")

    if dry_run:
        ok("4", f"dry-run: would commit {len(dirty)} file(s)")
        mf.unlink()
        return None

    r = subprocess.run(
        ["git", "commit", "-F", str(mf)],
        capture_output=True, text=True,
    )
    try:
        mf.unlink()
    except Exception:
        pass

    if r.returncode != 0:
        info((r.stderr or r.stdout).strip()[-400:])
        fail("4", "git commit failed")

    head_line = (r.stdout or "").strip().splitlines()[0]
    info(head_line)
    # extract short hash
    m = re.search(r"\[main ([0-9a-f]+)\]", head_line)
    head_hash = m.group(1) if m else ""
    ok("4", f"commit created ({head_hash})")
    return head_hash


# ─────────────────────────────────────────────────────────────
# STEP 5 — push GitHub
# ─────────────────────────────────────────────────────────────
def push_github(dry_run=False):
    hdr("STEP 5 — push GitHub")
    if dry_run:
        ok("5", "dry-run: skipped")
        return
    r = subprocess.run(
        ["git", "push", "origin", "main"],
        capture_output=True, text=True,
    )
    if r.returncode != 0:
        info((r.stderr or "")[-300:])
        fail("5", "git push origin failed")
    for line in (r.stdout or "").strip().splitlines()[-2:]:
        info(line)
    ok("5", "GitHub push OK")


# ─────────────────────────────────────────────────────────────
# STEP 6 — push HF
# ─────────────────────────────────────────────────────────────
def push_hf(dry_run=False):
    hdr("STEP 6 — push HF")
    if dry_run:
        ok("6", "dry-run: skipped")
        return
    r = subprocess.run(
        ["bash", "push_hf.sh"],
        capture_output=True, text=True, timeout=200,
    )
    if r.returncode != 0:
        info((r.stderr or "")[-300:])
        fail("6", "push_hf.sh failed")
    for line in (r.stdout or "").strip().splitlines()[-2:]:
        info(line)
    ok("6", "HF push OK")


# ─────────────────────────────────────────────────────────────
# STEP 7 — wait + verify
# ─────────────────────────────────────────────────────────────
def http_get(path, timeout=20):
    try:
        with urllib.request.urlopen(BASE_URL + path, timeout=timeout) as r:
            return r.status, r.read()
    except urllib.error.HTTPError as e:
        return e.code, b""
    except Exception:
        return 0, b""


def http_post_json(path, payload, timeout=120):
    req = urllib.request.Request(
        BASE_URL + path,
        data=json.dumps(payload).encode(),
        headers={"Content-Type": "application/json"},
        method="POST",
    )
    try:
        with urllib.request.urlopen(req, timeout=timeout) as r:
            return r.status, json.loads(r.read().decode())
    except urllib.error.HTTPError as e:
        try:
            return e.code, json.loads(e.read().decode())
        except Exception:
            return e.code, {}
    except Exception as e:
        return 0, {"error": f"{type(e).__name__}: {e}"}


def verify(role, user_input, wait_s, dry_run=False):
    hdr("STEP 7 — wait + live verify")
    if dry_run:
        ok("7", "dry-run: skipped")
        return
    if wait_s > 0:
        info(f"waiting {wait_s}s for HF rebuild...")
        time.sleep(wait_s)

    # poll with GET (not HEAD — /health/ready only accepts GET)
    up = False
    for i in range(1, 16):
        code, _ = http_get("/health/ready", timeout=15)
        if code == 200:
            info(f"HF up at attempt {i}")
            up = True
            break
        info(f"attempt {i}: HTTP {code}, waiting 15s")
        time.sleep(15)
    if not up:
        fail("7", "HF did not come back up in ~3 min")

    info("")
    info(f"-- live {role} test --")
    code, d = http_post_json("/api/agent/run", {
        "input": user_input,
        "session_id": f"ship-part-{int(time.time())}",
        "role": role,
    }, timeout=240)

    if code != 200:
        info(f"  HTTP {code}: {json.dumps(d)[:300]}")
        fail("7", "live verify HTTP not 200")

    r = d.get("result") or {}
    content = (r.get("content") or "").strip()
    err = d.get("detail") or d.get("error")
    steps = r.get("steps") or []
    tools_used = [s.get("tool") for s in steps]
    info(f"  provider: {r.get('provider')}")
    info(f"  steps:    {len(steps)}")
    info(f"  tools:    {tools_used}")
    info(f"  error:    {err!r}")
    info(f"  content:  {content[:200]!r}")
    if err:
        fail("7", "live verify returned an error")
    ok("7", "live verify OK")


# ─────────────────────────────────────────────────────────────
# main
# ─────────────────────────────────────────────────────────────
def main():
    global USE_COLOR
    ap = argparse.ArgumentParser(
        prog="ship.py",
        description="One-shot ship helper for Sunnyware.",
    )
    ap.add_argument("part", help="Part number, e.g. 157 or 157b")
    ap.add_argument("desc", help="Short description (one line)")
    ap.add_argument("--body", default=None,
                    help="Detailed body for HANDOFF.md (defaults to desc)")
    ap.add_argument("--wait", type=int, default=150,
                    help="Seconds to wait for HF rebuild (default: 150)")
    ap.add_argument("--verify-role", default="general",
                    help="Role for live verify (default: general)")
    ap.add_argument("--verify-input", default="Reply with the single word PONG.",
                    help="Input for live verify")
    ap.add_argument("--skip", default="",
                    help="Comma-separated steps to skip: "
                         "smokes,gate,docs,commit,gh,hf,verify")
    ap.add_argument("--dry-run", action="store_true",
                    help="Do everything except modify git / HF")
    ap.add_argument("--no-color", action="store_true")
    args = ap.parse_args()

    if args.no_color:
        USE_COLOR = False

    skip = {s.strip().lower() for s in args.skip.split(",") if s.strip()}
    body = args.body if args.body is not None else args.desc

    print("=" * 70)
    print(f"ship.py — Part {args.part}: {args.desc}")
    print(f"skip={sorted(skip) if skip else '(none)'}  dry_run={args.dry_run}")
    print("=" * 70)

    # 1. smokes (before touching git)
    if "smokes" in skip:
        info("[1] skipped")
    else:
        run_smokes()

    # 2. gate
    if "gate" in skip:
        info("[2] skipped")
    else:
        run_gate()

    # 3. docs
    if "docs" in skip:
        info("[3] skipped")
    else:
        update_docs(args.part, args.desc, body, dry_run=args.dry_run)

    # 4. commit
    head_hash = None
    if "commit" in skip:
        info("[4] skipped")
    else:
        head_hash = do_commit(args.part, args.desc, body, dry_run=args.dry_run)

    # 5. gh push
    if "gh" in skip:
        info("[5] skipped")
    else:
        push_github(dry_run=args.dry_run)

    # 6. hf push
    if "hf" in skip:
        info("[6] skipped")
    else:
        push_hf(dry_run=args.dry_run)

    # 7. verify
    if "verify" in skip:
        info("[7] skipped")
    else:
        verify(args.verify_role, args.verify_input, args.wait,
               dry_run=args.dry_run)

    print("")
    print("=" * 70)
    print(f"[DONE] Part {args.part} shipped"
          + (f"  (HEAD {head_hash})" if head_hash else ""))
    print("=" * 70)


if __name__ == "__main__":
    main()