the-custodian/scripts/monitor_dependency_triage.py

#!/usr/bin/env python3
"""Read-only CUST-WP-0075 snapshot. Requires PyYAML for the file scanner.

Writes evidence only to --output; never changes owner records or contacts owners.
"""
import argparse
from collections import Counter
from datetime import datetime, timezone
import json
import os
from pathlib import Path
import subprocess
import sys
from urllib.parse import urlencode
from urllib.request import urlopen

ROOT = Path(__file__).resolve().parents[1]
SCANNER = ROOT / "docs/evidence/2026-10-03-dependency-graph-triage/scan_workplan_files.py"
OPEN = {"proposed", "ready", "active", "blocked", "backlog"}
HANDOFFS = {
    "state-hub": "b4825539-1425-4028-a2e9-9243785a344b",
    "railiance-platform": "f84637fa-2355-4971-a2dc-7f1fd01710a8",
    "ops-warden": "79549660-4fc8-44d5-989e-074f6c7b6f07",
}
RECEIPTS = ROOT / "docs/evidence/2026-10-03-dependency-graph-triage/owner-handoff-receipts.json"


def get_rows(base, route, **filters):
    # Only /tasks has offset pagination. /messages uses a caller-selected
    # limit; /repos returns the complete registry without paging parameters.
    if route != "tasks":
        query = urlencode(dict(filters, limit=5000)) if route == "messages" else urlencode(filters)
        with urlopen(f"{base.rstrip('/')}/{route}/?{query}", timeout=30) as response:
            rows = json.load(response)
        if isinstance(rows, dict):
            rows = rows["items"]
        if route == "messages" and len(rows) >= 5000:
            raise ValueError("message limit reached; narrow the owner filter before scanning")
        return rows
    rows = []
    while True:
        query = urlencode(dict(filters, limit=200, offset=len(rows)))
        with urlopen(f"{base.rstrip('/')}/{route}/?{query}", timeout=30) as response:
            page = json.load(response)
        if isinstance(page, dict):
            page = page["items"]
        if not isinstance(page, list):
            raise ValueError(f"{route}: expected a list")
        rows.extend(page)
        if len(page) < 200:
            return rows


def audit(records, hub_tasks):
    plans, tasks = records["WP"], records["T"]
    def terminal(token):
        if token in tasks:
            return tasks[token]["status"] in {"done", "cancel"}
        return records["WP"].get(token, {}).get("status") in {"finished", "archived"}
    plans = {k: v for k, v in plans.items() if v["status"] in OPEN and v["flavor"] != "residual"}
    waits = {k: v for k, v in tasks.items() if v["wp"] in plans and v["status"] == "wait"}
    by_id = {t["id"]: t for t in hub_tasks}
    drift = []
    for tid, task in sorted(waits.items()):
        hub = by_id.get(task["hub_id"])
        if hub is None:
            drift.append({"task": tid, "field": "projection", "file": task["hub_id"], "hub": None})
            continue
        fields = {"status": "status", "needs_human": "needs_human", "reason": "blocking_reason",
                  "decision_id": "decision_id", "blocked_on": "blocked_on", "wait_until": "wait_until"}
        for file_key, hub_key in fields.items():
            expected = task[file_key]
            actual = hub.get(hub_key)
            if file_key == "needs_human":
                actual = bool(actual)
            elif file_key != "status":
                expected, actual = expected or None, actual or None
                if file_key == "wait_until":
                    expected = str(expected)[:10] if expected else None
                    actual = str(actual)[:10] if actual else None
            if expected != actual:
                drift.append({"task": tid, "field": hub_key, "file": expected, "hub": actual})
    return {
        "open_non_residual_workplans": len(plans),
        "workplan_statuses": dict(sorted(Counter(p["status"] for p in plans.values()).items())),
        "wait_tasks": len(waits),
        "human_gates": sum(t["needs_human"] for t in waits.values()),
        "unrouted_human_gates": sorted(k for k, t in waits.items() if t["needs_human"] and not t["decision_id"]),
        "unqualified_waits": sorted(k for k, t in waits.items() if not any(
            t[field] for field in ("depends_on", "needs_human", "blocked_on", "wait_until"))),
        "owner_return_waits": sum(str(t["blocked_on"] or "").startswith("message-from:") for t in waits.values()),
        "terminal_dependency_waits": [
            {"task": k, "depends_on": t["depends_on"], "needs_human": t["needs_human"],
             "blocked_on": t["blocked_on"], "wait_until": t["wait_until"]}
            for k, t in sorted(waits.items()) if t["depends_on"] and all(terminal(x) for x in t["depends_on"])],
        "qualifier_drift": drift,
    }


def handoff_case(owner, original, incoming, records, receipt=None):
    message_id = original["id"]
    candidates = [m for m in incoming if m["from_agent"] == owner
                  and m["created_at"] > original["created_at"]]
    case = {
        "case_id": f"case:cross-owner:CUST-WP-0075-T06:{owner}",
        "scenario_type": "cross_owner_wait",
        "beneficiary_workplan_id": "CUST-WP-0075", "obligor_repo": owner,
        "obligor_workplan_id": None, "state": "waiting",
        "reason": "Owner acceptance or decline not yet recorded",
        "proposal_message_id": message_id, "proposal_read_at": original.get("read_at"),
        "reply_candidates": [{"id": m["id"], "subject": m["subject"], "thread_id": m.get("thread_id")}
                             for m in candidates],
    }
    if receipt is None:
        return case
    reply = next((m for m in candidates if m["id"] == receipt["reply_message_id"]), None)
    if reply is None or reply.get("thread_id") != message_id or reply.get("to_agent") != "the-custodian":
        raise ValueError(f"{owner}: reviewed reply is missing or does not answer this proposal")
    if receipt["disposition"] not in {"accepted", "declined", "partially_accepted"} or not receipt.get("reason"):
        raise ValueError(f"{owner}: disposition and reason required")
    plan_id = receipt.get("owner_workplan_id")
    if receipt["disposition"] != "declined":
        plan = records["WP"].get(plan_id)
        if not plan or plan["repo"] != owner or plan["flavor"] == "residual":
            raise ValueError(f"{owner}: accepted handoff needs a non-residual owner workplan")
        if not receipt.get("owner_task_ids") or any(
                records["T"].get(tid, {}).get("wp") != plan_id for tid in receipt["owner_task_ids"]):
            raise ValueError(f"{owner}: accepted handoff needs owner task records")
    case.update(state="satisfied", reason=receipt["reason"], disposition=receipt["disposition"],
                obligor_workplan_id=plan_id, owner_task_ids=receipt.get("owner_task_ids", []),
                reply_message_id=reply["id"],
                scope="Proposal handoff only; owner implementation and human gates retain their own status")
    return case


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--api-base", default="http://127.0.0.1:8000")
    parser.add_argument("--output", type=Path, required=True)
    parser.add_argument("--handoff-receipts", type=Path, default=RECEIPTS,
                        help="Reviewed owner dispositions; absence leaves handoffs waiting")
    parser.add_argument("--repo-path", action="append", default=[], metavar="SLUG=PATH",
                        help="Explicit local checkout override; preserves registry evidence")
    args = parser.parse_args()
    args.output.mkdir(parents=True, exist_ok=True)
    repos = get_rows(args.api_base, "repos")
    (args.output / "registry.json").write_text(json.dumps(repos))
    overrides = {}
    for value in args.repo_path:
        slug, separator, path = value.partition("=")
        if not separator or not Path(path).is_absolute() or not Path(path).is_dir():
            parser.error("--repo-path requires SLUG=existing-absolute-directory")
        repo = next((r for r in repos if r["slug"] == slug), None)
        if repo is None:
            parser.error(f"--repo-path names unregistered repo: {slug}")
        repo["local_path"] = path
        overrides[slug] = path
    (args.output / "repos.json").write_text(json.dumps(repos))
    result = subprocess.run([sys.executable, str(SCANNER)], capture_output=True, text=True,
                            env=dict(os.environ, TRIAGE_SCRATCH=str(args.output.resolve())), check=True)
    (args.output / "file-view.txt").write_text(result.stdout)
    records = json.loads((args.output / "files.json").read_text())
    report = audit(records, get_rows(args.api_base, "tasks"))
    report["parse_warnings"] = records.get("parse_warnings", [])
    report["captured_at"] = datetime.now(timezone.utc).isoformat()
    report["registered_repos"] = len(repos)
    report["checkout_overrides"] = overrides
    report["unavailable_checkouts"] = sorted(r["slug"] for r in repos
        if not r.get("local_path") or not Path(r["local_path"]).is_dir())
    report["handoffs"] = []
    receipts = json.loads(args.handoff_receipts.read_text()) if args.handoff_receipts.exists() else {}
    for owner, message_id in HANDOFFS.items():
        incoming = get_rows(args.api_base, "messages", to_agent="the-custodian", from_agent=owner)
        outgoing = get_rows(args.api_base, "messages", to_agent=owner, from_agent="the-custodian")
        original = next(m for m in outgoing if m["id"] == message_id)
        report["handoffs"].append(handoff_case(owner, original, incoming, records, receipts.get(message_id)))
    (args.output / "monitor.json").write_text(json.dumps(report, indent=2) + "\n")
    print(json.dumps(report, indent=2))


if __name__ == "__main__":
    main()