#!/usr/bin/env python3 """agent_dod.py: check AI agent "done" claims against Agent DoD v1.0. Spec: https://connorgallic.com/definition-of-done Usage: python3 agent_dod.py check receipts.jsonl # offline, reads what the receipts say python3 agent_dod.py check receipts.jsonl --live # also re-reads http_contains proofs itself python3 agent_dod.py check receipts.jsonl --json # machine-readable report python3 agent_dod.py check - < receipts.jsonl # read stdin Input: JSON Lines or a JSON array of Agent DoD receipts. Loose logs work too: a record with just {"status": "done"} is scored P0, and {"status": "done", "proof": "some text"} is scored P1 (asserted). Exit codes: 0 = every "done" claim meets the bar, 1 = at least one does not, 3 = invalid input. Offline by default: no network, no credentials. A pass is a verdict on the receipts, not an approval to send anything. Stdlib only. MIT license. Version 1.0.0. """ from __future__ import annotations import argparse import json import os import re import sys import urllib.request VERSION = "1.0.0" SPEC = "agent-dod-1.0" LEVELS = ["P0", "P1", "P2", "P3", "P4"] LEVEL_NAMES = { "P0": "none (status only)", "P1": "asserted (the agent says so)", "P2": "artifact (cited thing that resolves)", "P3": "read-back (independent check of the real after-state)", "P4": "reconciled (read-back tied to a business outcome)", } # Proof kinds that read the authoritative system where the result lives. READBACK_KINDS = { "http_contains", "provider_object", "sent_log", "call_log", "payment", "deployment", "record", } # Proof kinds that show something was produced, not that it landed. ARTIFACT_KINDS = {"commit", "pull_request", "file", "test_run", "screenshot", "log"} INDEPENDENT = {"runner", "independent", "human", "verifier", "agent_dod"} DONE_WORDS = {"done", "complete", "completed", "success", "succeeded", "shipped", "finished", "closed"} CONTRADICTIONS = re.compile( r"\b(blocked|not done|incomplete|could not|couldn't|cannot|can't|unable to|" r"waiting on|waiting for|failed to|did not|didn't|todo|placeholder|not verified)\b", re.I, ) RULES = { "DOD-2": "The actor cannot issue its own verdict.", "DOD-3": "A zero exit code is not proof.", "DOD-4": "Done needs a read-back from the authoritative system (P3).", "DOD-5": "Cited proof must resolve when checked.", "DOD-7": "The agent's own report says it is not done.", "DOD-8": "A live target was only proven with build artifacts (scaffold).", "DOD-9": "A blocked task must name the exact external condition.", "GOV-5": "Done while the required approval is not granted.", "GOV-8": "No cost recorded for the attempt.", } def load(path: str) -> list[dict]: raw = sys.stdin.read() if path == "-" else open(path, encoding="utf-8").read() raw = raw.strip() if not raw: return [] if raw.startswith("["): data = json.loads(raw) else: data = [json.loads(line) for line in raw.splitlines() if line.strip()] if not all(isinstance(r, dict) for r in data): raise ValueError("every record must be a JSON object") return data def claimed(r: dict) -> str: return str(r.get("claimed_status") or r.get("status") or r.get("verdict") or "").lower() def is_done_claim(r: dict) -> bool: return claimed(r) in DONE_WORDS def proofs(r: dict) -> list[dict]: p = r.get("proof") if p is None: return [] if isinstance(p, (str, int, float)): return [{"kind": "assertion", "ref": str(p), "checked_by": "actor", "result": "not_run"}] if isinstance(p, dict): p = [p] return [x if isinstance(x, dict) else {"kind": "assertion", "ref": str(x), "checked_by": "actor"} for x in p] def live_check(p: dict, timeout: float) -> tuple[str, str]: """Re-read an http_contains proof ourselves. Returns (result, note).""" url, expect = p.get("ref", ""), p.get("expect", "") if not str(url).startswith(("http://", "https://")): return "fail", "ref is not an http(s) URL" try: req = urllib.request.Request(url, headers={"User-Agent": f"agent-dod/{VERSION}"}) with urllib.request.urlopen(req, timeout=timeout) as resp: body = resp.read(2_000_000).decode("utf-8", "replace") code = resp.status except Exception as exc: # network errors are a failed read-back, not a crash code = getattr(exc, "code", None) return "fail", f"HTTP {code}" if code else f"fetch failed: {type(exc).__name__}" if code >= 400: return "fail", f"HTTP {code}" if expect and expect not in body: return "fail", f"HTTP {code} but expected text not found" return "pass", f"HTTP {code}" + (", expected text found" if expect else "") def score(r: dict, live: bool, check_files: bool, timeout: float) -> dict: owner = str(r.get("owner") or "").lower() verdict_by = str(r.get("verdict_by") or "").lower() ps = proofs(r) violations: list[str] = [] notes: list[str] = [] passing_artifact = passing_readback = False any_asserted = False for p in ps: kind = str(p.get("kind", "assertion")).lower() by = str(p.get("checked_by", "actor")).lower() result = str(p.get("result", "not_run")).lower() if live and kind == "http_contains": result, note = live_check(p, timeout) by = "agent_dod" notes.append(f"live {p.get('ref')}: {note}") if check_files and kind == "file": exists = os.path.exists(str(p.get("ref", ""))) result = "pass" if exists else "fail" notes.append(f"file {p.get('ref')}: {'exists' if exists else 'missing'}") if result == "fail": violations.append("DOD-5") continue if kind in READBACK_KINDS and result == "pass" and by in INDEPENDENT and by != owner: passing_readback = True elif kind in ARTIFACT_KINDS and result in ("pass", "not_run"): passing_artifact = True else: any_asserted = True bo = r.get("business_outcome") or {} if passing_readback and isinstance(bo, dict) and bo.get("verified") is True: level = "P4" elif passing_readback: level = "P3" elif passing_artifact: level = "P2" elif ps or any_asserted: level = "P1" else: level = "P0" done = is_done_claim(r) status = claimed(r) if done: if verdict_by and (verdict_by in ("actor", "self", "agent") or (owner and verdict_by == owner)): violations.append("DOD-2") if not ps and r.get("exit_code") == 0: violations.append("DOD-3") if LEVELS.index(level) < LEVELS.index("P3"): violations.append("DOD-4") if CONTRADICTIONS.search(str(r.get("report") or "")): violations.append("DOD-7") target = str(r.get("target") or "") if target.startswith(("http://", "https://")) and passing_artifact and not passing_readback: violations.append("DOD-8") appr = r.get("approval") or {} if isinstance(appr, dict) and appr.get("required") and str(appr.get("status", "")).lower() != "approved": violations.append("GOV-5") if status == "blocked" and not str(r.get("blocker") or r.get("report") or "").strip(): violations.append("DOD-9") if "cost" not in r: notes.append(RULES["GOV-8"]) return { "task_id": r.get("task_id") or r.get("id") or r.get("slug"), "claimed": status or None, "level": level, "violations": sorted(set(violations)), "notes": notes, "usd": (r.get("cost") or {}).get("usd") if isinstance(r.get("cost"), dict) else None, } def summarize(results: list[dict], min_level: str) -> dict: done = [x for x in results if x["claimed"] in DONE_WORDS] bar = LEVELS.index(min_level) meets = [x for x in done if LEVELS.index(x["level"]) >= bar and not (set(x["violations"]) - {"DOD-4"})] by_level = {lvl: sum(1 for x in done if x["level"] == lvl) for lvl in LEVELS} statuses: dict[str, int] = {} for x in results: statuses[x["claimed"] or "unknown"] = statuses.get(x["claimed"] or "unknown", 0) + 1 usd = [x["usd"] for x in results if isinstance(x["usd"], (int, float))] return { "spec": SPEC, "checker": VERSION, "records": len(results), "statuses": statuses, "done_claimed": len(done), "done_meeting_bar": len(meets), "bar": min_level, "proof_rate": round(len(meets) / len(done), 4) if done else None, "done_by_level": by_level, "held": statuses.get("held", 0), "total_usd": round(sum(usd), 4) if usd else None, "usd_per_verified_done": round(sum(usd) / len(meets), 4) if usd and meets else None, } def main(argv: list[str] | None = None) -> int: ap = argparse.ArgumentParser(prog="agent_dod", description="Check AI agent done claims against Agent DoD v1.0.") sub = ap.add_subparsers(dest="cmd", required=True) c = sub.add_parser("check", help="score receipts") c.add_argument("input", help="receipts file (.jsonl or .json array), or - for stdin") c.add_argument("--min-level", default="P3", choices=LEVELS, help="bar for done (default P3)") c.add_argument("--live", action="store_true", help="re-read http_contains proofs over the network") c.add_argument("--check-files", action="store_true", help="confirm cited file proofs exist on disk") c.add_argument("--timeout", type=float, default=10.0) c.add_argument("--json", action="store_true", help="print the full report as JSON") sub.add_parser("version") a = ap.parse_args(argv) if a.cmd == "version": print(f"agent_dod {VERSION} ({SPEC})") return 0 try: records = load(a.input) except (OSError, ValueError) as exc: print(f"invalid input: {exc}", file=sys.stderr) return 3 results = [score(r, a.live, a.check_files, a.timeout) for r in records] summary = summarize(results, a.min_level) failing = [x for x in results if x["claimed"] in DONE_WORDS and LEVELS.index(x["level"]) < LEVELS.index(a.min_level) or (set(x["violations"]) - {"DOD-4"})] if a.json: print(json.dumps({"summary": summary, "results": results}, indent=2)) else: s = summary print(f"Agent DoD check ({SPEC}, checker {VERSION})") print(f" records: {s['records']} statuses: {s['statuses']}") rate = "n/a" if s["proof_rate"] is None else f"{s['proof_rate']:.1%}" print(f" done claimed: {s['done_claimed']} meeting {s['bar']}: {s['done_meeting_bar']} proof rate: {rate}") print(" done by proof level: " + ", ".join(f"{k} {v}" for k, v in s["done_by_level"].items())) if s["total_usd"] is not None: per = "n/a" if s["usd_per_verified_done"] is None else f"${s['usd_per_verified_done']}" print(f" cost: ${s['total_usd']} total, {per} per verified done") for x in results: if x["violations"]: rules = "; ".join(f"{v} {RULES[v]}" for v in x["violations"]) print(f" - {x['task_id']}: claimed {x['claimed']}, level {x['level']}: {rules}") for n in x["notes"]: if n.startswith(("live", "file")): print(f" {x['task_id']}: {n}") print(" PASS" if not failing else f" FAIL: {len(failing)} done claim(s) below the bar") return 0 if not failing else 1 if __name__ == "__main__": sys.exit(main())