pad/scripts/decision-eval/eval.py

#!/usr/bin/env python3
"""Jev triage eval against docapp's human-labelled items.

Reads TYPESAFE_API_KEY from the environment (never printed). Input: the
bugs/tasks/ideas JSON dumps in this directory. Output: per-item JSONL and a
summary. Two tasks:

  A. field recovery — for items where a human set the field, ask a Choice
     over the schema's options and measure agreement vs confidence.
  B. collection classification — hide the collection, ask bug/task/idea.

Usage: eval.py [--n-per-collection N] [--model jev-latest] [--dry-run]
"""
import argparse, json, os, random, sys, time, urllib.request, urllib.error
from concurrent.futures import ThreadPoolExecutor, as_completed
from collections import Counter, defaultdict

HERE = os.path.dirname(os.path.abspath(__file__))
API = "https://api.typesafe.ai/v1/systemone"
MAX_BODY = 6000  # chars of content; keep state focused (docs: irrelevant state degrades)

COMPONENTS = ["web", "server", "cli", "store", "mcp", "backend", "frontend", "docs", "ci", "other"]

FIELD_QUESTIONS = {
    "bugs": {
        "severity": ("How severe is this bug for users of a self-hosted project tracker used by developers and AI agents?",
                     {"low": "cosmetic, or a workaround is obvious and cheap",
                      "medium": "a real defect a user will hit in normal use; workaround exists or impact is contained",
                      "high": "breaks a core flow, data shown wrong, or no reasonable workaround",
                      "critical": "data loss or corruption, security exposure, or the product is unusable"}),
        "component": ("Which part of the codebase does the fix for this bug primarily live in?",
                      {"web": "the SvelteKit web UI (components, pages, editor)",
                       "server": "the Go HTTP API handlers and server wiring",
                       "cli": "the `pad` command-line tool and its client",
                       "store": "the database layer, migrations, SQL",
                       "mcp": "the MCP tool catalog and dispatch for agents",
                       "backend": "Go code not clearly server, store or mcp",
                       "frontend": "web code described as frontend rather than a specific component",
                       "docs": "documentation, README, instructions text",
                       "ci": "GitHub Actions, build, release tooling",
                       "other": "none of the above"}),
    },
    "tasks": {
        "priority": ("How urgent is this task relative to other work on the same product?",
                     {"low": "nice to have; no one is waiting on it",
                      "medium": "should be done in the normal course of work",
                      "high": "blocks other work, a user, or a release; do soon",
                      "critical": "drop everything; production or data is at risk"}),
        "effort": ("How much implementation effort does this task take for an experienced engineer or agent?",
                   {"xs": "minutes: a one-line or one-file change",
                    "s": "an hour or two: a small contained change with a test",
                    "m": "half a day to a day: several files, tests, some design",
                    "l": "several days: cross-cutting, migrations or new surfaces",
                    "xl": "a week or more: a project, not a task"}),
    },
    "ideas": {
        "impact": ("If built, how much would this idea change the product for its users?",
                   {"low": "marginal polish or a niche convenience",
                    "medium": "a clear improvement many users would notice",
                    "high": "changes what the product can do or who it is for"}),
    },
}

COLLECTION_Q = ("Which kind of work item is this?",
                {"bug": "a defect: something that exists and behaves wrongly",
                 "task": "a unit of work to do, with a defined outcome, not primarily a defect report",
                 "idea": "a proposal or possibility to explore; not yet committed work"})


def load(coll):
    with open(os.path.join(HERE, f"{coll}.json")) as f:
        items = json.load(f)
    out = []
    for it in items:
        fields = it.get("fields") or {}
        if isinstance(fields, str):
            try:
                fields = json.loads(fields)
            except Exception:
                fields = {}
        out.append({"ref": it["ref"], "title": it["title"], "content": (it.get("content") or "")[:MAX_BODY],
                    "fields": fields, "collection": coll})
    return out


def call(key, model, state, questions, retries=5):
    body = json.dumps({"model": model, "state": state, "questions": questions}).encode()
    req = urllib.request.Request(API, data=body, method="POST", headers={
        "Authorization": f"Bearer {key}", "Content-Type": "application/json"})
    delay = 1.0
    for attempt in range(retries):
        t0 = time.time()
        try:
            with urllib.request.urlopen(req, timeout=60) as r:
                data = json.load(r)
                data["_latency_ms"] = int((time.time() - t0) * 1000)
                return data
        except urllib.error.HTTPError as e:
            if e.code in (429, 529) and attempt < retries - 1:
                time.sleep(delay); delay *= 2; continue
            detail = e.read().decode(errors="replace")[:300]
            raise RuntimeError(f"HTTP {e.code}: {detail}")
    raise RuntimeError("retries exhausted")


def choice_q(instructions, criteria):
    return {"type": "choice", "instructions": instructions, "criteria": criteria}


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--n-per-collection", type=int, default=120)
    ap.add_argument("--model", default="jev-latest")
    ap.add_argument("--workers", type=int, default=8)
    ap.add_argument("--seed", type=int, default=73)
    ap.add_argument("--dry-run", action="store_true")
    a = ap.parse_args()

    key = os.environ.get("TYPESAFE_API_KEY")
    if not key and not a.dry_run:
        print("TYPESAFE_API_KEY not in environment", file=sys.stderr); sys.exit(2)

    random.seed(a.seed)
    jobs = []
    for coll, fq in FIELD_QUESTIONS.items():
        items = load(coll)
        labelled = [it for it in items if any(it["fields"].get(k) for k in fq)]
        random.shuffle(labelled)
        for it in labelled[: a.n_per_collection]:
            questions = {k: choice_q(*fq[k]) for k in fq}
            questions["collection"] = choice_q(*COLLECTION_Q)
            state = {"title": it["title"], "body": it["content"]}
            jobs.append((it, questions, state))

    print(f"{len(jobs)} items queued ({', '.join(f'{c}:{sum(1 for j in jobs if j[0]['collection']==c)}' for c in FIELD_QUESTIONS)})")
    if a.dry_run:
        print(json.dumps({"model": a.model, "state": jobs[0][2], "questions": jobs[0][1]}, indent=1)[:1500]); return

    results, errors, usage_in, lat = [], [], 0, []
    out_path = os.path.join(HERE, f"results-{a.model}.jsonl")
    with open(out_path, "w") as out, ThreadPoolExecutor(a.workers) as ex:
        futs = {ex.submit(call, key, a.model, st, q): it for it, q, st in jobs}
        for n, f in enumerate(as_completed(futs), 1):
            it = futs[f]
            try:
                r = f.result()
            except Exception as e:
                errors.append((it["ref"], str(e))); continue
            usage_in += r.get("usage", {}).get("input_tokens", 0); lat.append(r["_latency_ms"])
            rec = {"ref": it["ref"], "collection": it["collection"], "human": it["fields"], "answers": r["answers"],
                   "latency_ms": r["_latency_ms"], "usage": r.get("usage")}
            results.append(rec); out.write(json.dumps(rec) + "\n")
            if n % 50 == 0: print(f"  {n}/{len(jobs)}", file=sys.stderr)

    # ---- summary ----
    print(f"\nmodel={results[0]['answers'] and results and (r.get('model') or a.model)} items={len(results)} errors={len(errors)} "
          f"input_tokens={usage_in} est_cost=${usage_in/1e6*0.042:.4f} "
          f"latency p50={sorted(lat)[len(lat)//2]}ms p90={sorted(lat)[int(len(lat)*0.9)]}ms")
    for e in errors[:5]: print("  error", e)

    print("\nTask B — collection classification (human collection is ground truth)")
    cm = Counter(); conf_ok, conf_bad = [], []
    for rec in results:
        truth = {"bugs": "bug", "tasks": "task", "ideas": "idea"}[rec["collection"]]
        ans = rec["answers"]["collection"]; cm[(truth, ans["choice"])] += 1
        (conf_ok if ans["choice"] == truth else conf_bad).append(ans["confidence"])
    total = sum(cm.values()); correct = sum(v for (t, p), v in cm.items() if t == p)
    print(f"  accuracy {correct}/{total} = {correct/total:.1%}   mean conf when right {sum(conf_ok)/max(1,len(conf_ok)):.2f}, when wrong {sum(conf_bad)/max(1,len(conf_bad)):.2f}")
    for t in ("bug", "task", "idea"):
        row = {p: cm[(t, p)] for p in ("bug", "task", "idea")}
        print(f"  truth={t:5s} -> {row}")

    print("\nTask A — field recovery (only items where a human set the field)")
    for coll, fq in FIELD_QUESTIONS.items():
        for k in fq:
            rows = [(rec["human"].get(k), rec["answers"][k]) for rec in results if rec["collection"] == coll and rec["human"].get(k)]
            if not rows: continue
            exact = sum(1 for h, ans in rows if ans["choice"] == h)
            # adjacent = off by one level on ordered scales
            order = list(fq[k][1].keys())
            adj = sum(1 for h, ans in rows if h in order and ans["choice"] in order and abs(order.index(h) - order.index(ans["choice"])) <= 1)
            hi = [(h, ans) for h, ans in rows if ans["confidence"] >= 0.5]
            hi_exact = sum(1 for h, ans in hi if ans["choice"] == h)
            vhi = [(h, ans) for h, ans in rows if ans["confidence"] >= 0.8]
            vhi_exact = sum(1 for h, ans in vhi if ans["choice"] == h)
            base = Counter(h for h, _ in rows).most_common(1)[0]
            print(f"  {coll}.{k:10s} n={len(rows):3d} exact={exact/len(rows):.1%} adjacent={adj/len(rows):.1%} | "
                  f"conf>=0.5: {len(hi)} items, exact={hi_exact/max(1,len(hi)):.1%} | conf>=0.8: {len(vhi)} items, exact={vhi_exact/max(1,len(vhi)):.1%} | "
                  f"majority-baseline={base[1]/len(rows):.1%} ({base[0]})")
            conf = Counter()
            for h, ans in rows: conf[(h, ans["choice"])] += 1
            top = sorted(((h, p, v) for (h, p), v in conf.items() if h != p), key=lambda x: -x[2])[:3]
            if top: print(f"      top confusions (human->jev): " + ", ".join(f"{h}->{p}:{v}" for h, p, v in top))
    print(f"\nper-item results: {out_path}")


if __name__ == "__main__":
    main()