#!/usr/bin/env python3 """Dispatch hash-bound PRD judgment/reporting work through AI-Collab. This module is the executable consumer missing from the original adapter. It dispatches only validated PRD-owned envelopes. The selected runtime worker must return through a PRD MCP validation tool, so it cannot write canonical state. """ from __future__ import annotations import argparse import datetime as dt import hashlib import json import sys from pathlib import Path from typing import Any import prd_reporting import prd_substrate import prd_workflows CONTRACT_VERSION = 1 class WorkerError(ValueError): """A runtime dispatch could not be constructed or safely bound.""" def _hash(value: Any) -> str: raw = json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True) return hashlib.sha256(raw.encode("utf-8")).hexdigest() def _now() -> str: return dt.datetime.now(dt.timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z") def _find_capsule_id(value: Any) -> str: if isinstance(value, dict): for key in ("capsule_id", "cap_id", "capsuleId", "capId"): candidate = value.get(key) if isinstance(candidate, str) and candidate.strip(): return candidate.strip() for child in value.values(): found = _find_capsule_id(child) if found: return found if isinstance(value, list): for child in value: found = _find_capsule_id(child) if found: return found return "" def _completed_dispatch(result: dict[str, Any]) -> tuple[str, dict[str, Any], dict[str, Any]]: if result.get("status") != "completed": raise WorkerError(f"runtime dispatch did not complete: {result.get('reason', result.get('status'))}") request = result.get("request") if isinstance(result.get("request"), dict) else {} receipt = result.get("receipt") if isinstance(result.get("receipt"), dict) else {} capsule_id = _find_capsule_id(receipt.get("result")) if not capsule_id: raise WorkerError("runtime dispatch returned no capsule identifier") return capsule_id, request, receipt def _judgment_body(request: dict[str, Any]) -> dict[str, Any]: return { "kind": "prd.workflow_judgment.v1", "request": request, "completion": { "tool": "prd_workflow_resume", "arguments": {"id": request["run_id"], "result": ""}, "requirements": [ "Echo request_hash, executor, and profile exactly.", "Return JSON satisfying result_schema.", "Use only declared source_refs.", "Do not write project files or canonical state.", ], }, } def dispatch_judgment(repo_root: str | Path, run_id: str) -> dict[str, Any]: root = Path(repo_root).resolve() run = prd_workflows.get_run(root, run_id) if run.get("status") != "waiting_judgment": raise WorkerError(f"workflow {run_id} is not waiting for judgment") current = run.get("runtime_dispatch") if isinstance(current, dict) and current.get("capsule_id"): return {"status": "already_dispatched", **current} request = run.get("pending_judgment") if not isinstance(request, dict) or not request.get("request_hash"): raise WorkerError(f"workflow {run_id} has no valid pending judgment") body = _judgment_body(request) result = prd_substrate.execute_tool( root, tool="dispatch", arguments={ "required_cap": "workflow_judgment", "body": json.dumps(body, sort_keys=True, separators=(",", ":")), "query": f"{request.get('task', 'workflow judgment')} using profile {request.get('profile', '')}", "root_id": run_id, "kind": "prd_workflow_judgment", "priority": "normal", }, capability="workflow_judgment", source_refs=list(request.get("source_refs", [])), idempotency_key=f"prd-judgment:{request['request_hash']}", required=request.get("constraints", {}).get("fallback") == "fail", ) capsule_id, operation, receipt = _completed_dispatch(result) link = { "capsule_id": capsule_id, "operation_id": str(operation.get("operation_id", "")), "receipt_hash": str(receipt.get("receipt_hash", "")), "request_hash": request["request_hash"], "dispatched_at": _now(), } prd_workflows.bind_runtime_dispatch(root, run_id, link) return {"status": "dispatched", **link} def dispatch_pending(repo_root: str | Path = ".") -> dict[str, Any]: root = Path(repo_root).resolve() rows = [] for run in prd_workflows.load_runs(root)["runs"]: if run.get("status") == "waiting_judgment": rows.append(dispatch_judgment(root, run["id"])) return {"status": "ok", "dispatched": sum(row["status"] == "dispatched" for row in rows), "runs": rows} def dispatch_reporting(repo_root: str | Path, bundle: dict[str, Any]) -> dict[str, Any]: if not isinstance(bundle, dict): raise WorkerError("reporting bundle must be an object") root = Path(repo_root).resolve() bundle_hash = _hash(bundle) body = { "kind": "prd.delegated_reporting.v1", "bundle_hash": bundle_hash, "bundle": bundle, "completion": { "tool": "prd_reporting_validate", "arguments": {"bundle": bundle, "result": ""}, "requirements": [ "Return the configured executor and profile exactly.", "Keep every notable item and action source-backed.", "Do not write project files or canonical state.", ], }, } policy = bundle.get("policy", {}) result = prd_substrate.execute_tool( root, tool="dispatch", arguments={ "required_cap": "delegated_reporting", "body": json.dumps(body, sort_keys=True, separators=(",", ":")), "query": f"{bundle.get('task', 'reporting')} using profile {policy.get('profile', '')}", "root_id": f"report:{bundle_hash[:24]}", "kind": "prd_delegated_reporting", "priority": "normal", }, capability="delegated_reporting", source_refs=list(bundle.get("source_refs", [])), idempotency_key=f"prd-report:{bundle_hash}", ) capsule_id, operation, receipt = _completed_dispatch(result) return { "status": "dispatched", "capsule_id": capsule_id, "bundle_hash": bundle_hash, "operation_id": operation.get("operation_id", ""), "receipt_hash": receipt.get("receipt_hash", ""), } def dispatch_reporting_task(repo_root: str | Path, task: str) -> dict[str, Any]: root = Path(repo_root).resolve() decision = prd_reporting.delegation_decision(root, task, executor_available=True) if decision["action"] != "delegate": return {"status": "fallback", "action": decision["action"], "reason": decision["reason"]} return dispatch_reporting(root, prd_reporting.build_bundle(root, task)) def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--repo-root", default=".") sub = parser.add_subparsers(dest="command", required=True) one = sub.add_parser("judgment"); one.add_argument("run_id") sub.add_parser("drain") report = sub.add_parser("report"); report.add_argument("--task", default="report_summary", choices=prd_reporting.prd_config.REPORTING_TASKS) args = parser.parse_args(argv) try: if args.command == "judgment": result = dispatch_judgment(args.repo_root, args.run_id) elif args.command == "drain": result = dispatch_pending(args.repo_root) else: result = dispatch_reporting_task(args.repo_root, args.task) except (WorkerError, prd_substrate.ContractError, prd_workflows.WorkflowError, prd_reporting.PolicyError) as exc: print(json.dumps({"status": "error", "error": str(exc)}, indent=2)) return 2 print(json.dumps(result, indent=2)) return 0 if __name__ == "__main__": sys.exit(main())