paperclip-harness/bin/lib/harness.py

"""Shared helpers for the harness commands.

Everything here talks to Paperclip with the person's key, or runs git and ocr.
No fact is stored: a task's worktree, branch and verify command are read from
Paperclip every time, so a stale copy cannot exist.
"""

import json
import os
import re
import shutil
import subprocess
import sys
import urllib.error
import urllib.request

# A program started from a TUIOS pane gets a smaller PATH than an interactive
# shell, so the places a person's tools actually live are added before anything
# is looked for. Without this, ocr is "not installed" inside a merge.
_HOME = os.path.expanduser("~")
_USER = os.environ.get("USER", "")
for _p in (
    os.path.join(_HOME, ".local", "bin"),
    os.path.join(_HOME, ".nix-profile", "bin"),
    "/run/current-system/sw/bin",
    "/etc/profiles/per-user/" + _USER + "/bin",
):
    if _p not in os.environ.get("PATH", ""):
        os.environ["PATH"] = _p + ":" + os.environ.get("PATH", "")

API = os.environ.get("PAPERCLIP_API_URL", "http://127.0.0.1:3100").rstrip("/")
COMPANY = os.environ.get("PAPERCLIP_COMPANY_ID", "d4921013-e5da-4d88-8016-57d5746e2d03")
KEY = os.environ.get("PAPERCLIP_BOARD_KEY", "")

BLOCKING_SEVERITIES = {"critical", "high"}


class Refused(Exception):
    """A check that did not pass, with the reason a person needs."""

    def __init__(self, code, message):
        super().__init__(message)
        self.code = code


def api(method, path, body=None, timeout=30):
    if not KEY:
        raise Refused(2, "PAPERCLIP_BOARD_KEY is not set")
    data = json.dumps(body).encode() if body is not None else None
    req = urllib.request.Request(
        API + path,
        data=data,
        method=method,
        headers={"Authorization": "Bearer " + KEY, "Content-Type": "application/json"},
    )
    try:
        with urllib.request.urlopen(req, timeout=timeout) as f:
            raw = f.read().decode()
            return json.loads(raw) if raw else {}
    except urllib.error.HTTPError as e:
        raise Refused(2, "%s %s → %s %s" % (method, path, e.code, e.read().decode()[:300]))


def find_issue(ref):
    """Find one task by id or identifier (AUD-163), then read it in full.

    The company list is a summary: it carries no worktree and no blockers. The
    single-task read does, so it is always a second call.
    """
    issues = api("GET", "/api/companies/%s/issues" % COMPANY)
    for i in issues:
        if i.get("identifier") == ref or i.get("id") == ref:
            return api("GET", "/api/issues/%s" % i["id"])
    raise Refused(2, "no task matches %r" % ref)


def me():
    """The person the board key acts as."""
    d = api("GET", "/api/cli-auth/me")
    uid = d.get("userId")
    if not uid:
        raise Refused(2, "the board key has no user id")
    return uid


def hand_to_person(issue):
    """Take the task off the agent before writing on it.

    A comment on a task assigned to an agent wakes that agent. That is right for
    a person's reply and wrong for a tool's verdict: a merge result written on a
    finished task would wake the agent to read it and spend money doing nothing.
    Handing the task back to the person first is the documented way to leave a
    task waiting for a human, and it stops the wake.
    """
    if not issue.get("assigneeAgentId"):
        return
    api(
        "PATCH",
        "/api/issues/%s" % issue["id"],
        {"assigneeAgentId": None, "assigneeUserId": me()},
    )
    issue["assigneeAgentId"] = None


def comment(issue_id, body):
    return api("POST", "/api/issues/%s/comments" % issue_id, {"body": body})


def verdict(issue, body):
    """Write a tool's verdict on a task without waking its agent."""
    hand_to_person(issue)
    return comment(issue["id"], body)


VERIFY_RE = re.compile(
    r"Done when this command succeeds:\s*\n+\s*(?P<cmd>[^\n]+)", re.IGNORECASE
)


def verify_command(issue, worktree):
    """The command that decides the task is finished.

    It comes from the task's own description, then the project convention
    .harness/verify.sh. A task with neither says so out loud.
    """
    m = VERIFY_RE.search(issue.get("description") or "")
    if m:
        return m.group("cmd").strip(), "from the task description"
    # The project's own check. check.sh is the current name; verify.sh is what
    # the earlier project wrote, and is still honoured.
    for name in (".harness/check.sh", ".harness/verify.sh"):
        if os.path.isfile(os.path.join(worktree, name)):
            return name, "the project's " + name
    return None, "none"


def worktree_of(issue):
    ws = issue.get("currentExecutionWorkspace") or {}
    cwd = ws.get("cwd") or ""
    if not cwd or not os.path.isdir(cwd):
        raise Refused(
            7,
            "%s has no worktree on this machine, so there is nothing to review or merge"
            % issue.get("identifier"),
        )
    return cwd, ws.get("branchName") or "", ws.get("baseRef") or "main"


def run(cmd, cwd=None, timeout=900):
    p = subprocess.run(cmd, cwd=cwd, capture_output=True, text=True, timeout=timeout)
    return p.returncode, (p.stdout or "") + (p.stderr or "")


def need(binary):
    if not shutil.which(binary):
        raise Refused(2, "%s is not installed" % binary)


def ocr_review(repo, base, head):
    """Run ocr over one range and return its JSON report."""
    need("ocr")
    code, out = run(
        [
            "ocr", "review",
            "--repo", repo,
            "--from", base,
            "--to", head,
            "--format", "json",
            "--audience", "agent",
            "--color", "never",
        ],
        timeout=1800,
    )
    try:
        report = json.loads(out[out.index("{") :])
    except ValueError:
        raise Refused(5, "ocr did not return a report (exit %d): %s" % (code, out[-400:]))
    return report


def blocking_findings(report):
    out = []
    for c in report.get("comments") or []:
        sev = (c.get("severity") or "").lower()
        if sev in BLOCKING_SEVERITIES:
            out.append(c)
    return out


def describe(c):
    where = c.get("path") or "?"
    line = c.get("start_line")
    if line:
        where += ":%s" % line
    body = (c.get("content") or "").strip().replace("\n", " ")
    if len(body) > 300:
        body = body[:300] + "…"
    return "- `%s` **%s/%s** — %s" % (where, c.get("severity"), c.get("category"), body)


def say(msg=""):
    print(msg, flush=True)


def main_guard(fn):
    try:
        sys.exit(fn())
    except Refused as e:
        say("refused: %s" % e)
        sys.exit(e.code)