#!/usr/bin/env python3 """Versioned PRD Plugin bridge to the AI-Collab Substrate (REQ-099). PRD Plugin owns policy, canonical state, IDs, graph semantics, and validation. AI-Collab owns runtime execution, indexing, routing, models, hosted applications, and its credentials. This stdlib-only module discovers exact UTCP call templates, executes allowlisted requests, validates bounded results, and never grants the runtime direct authority to write canonical PRD state. """ from __future__ import annotations import argparse import datetime as dt import hashlib import json import os import re import sys import urllib.error import urllib.parse import urllib.request from pathlib import Path from typing import Any import prd_config import prd_graph import prd_services CONTRACT_VERSION = 2 SCHEMA_VERSION = "1.0" DEFAULT_MAX_RECORDS = 10_000 DEFAULT_MAX_EVENTS = 10_000 OBSERVE_CAPABILITIES = {"records", "graph", "events"} ID_RE = re.compile(r"^[A-Z][A-Z0-9]*(?:-[A-Z0-9]+)*-\d+$") class ContractError(ValueError): """The adapter policy or requested export is invalid.""" def _read_json(path: Path, default: Any) -> Any: try: return json.loads(path.read_text(encoding="utf-8-sig")) except (OSError, json.JSONDecodeError): return default def _canonical_hash(value: Any) -> str: payload = json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True) return hashlib.sha256(payload.encode("utf-8")).hexdigest() def _now() -> str: return dt.datetime.now(dt.timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z") def _config_value(root: Path, key: str, fallback: Any) -> Any: try: return prd_config.get(root, key) except (KeyError, ValueError, TypeError): return fallback def _safe_url(value: str, *, label: str) -> str: try: parsed = urllib.parse.urlsplit(str(value)) except ValueError as exc: raise ContractError(f"invalid {label}: {exc}") from exc if parsed.scheme not in {"http", "https"} or not parsed.hostname or parsed.username or parsed.password or parsed.fragment: raise ContractError(f"invalid {label}: expected an http(s) URL without credentials or fragment") if parsed.scheme == "http" and parsed.hostname not in {"127.0.0.1", "localhost", "::1"}: raise ContractError(f"insecure remote {label}: use https outside loopback") return urllib.parse.urlunsplit(parsed) def effective_policy(repo_root: str | Path = ".") -> dict[str, Any]: """Return configured and effective policy with strict master precedence.""" root = Path(repo_root).resolve() configured = { "enabled": bool(prd_config.get(root, "integrations.substrate.enabled")), "mode": prd_config.get(root, "integrations.substrate.mode"), "capabilities": list(prd_config.get(root, "integrations.substrate.capabilities")), "contract_version": prd_config.get(root, "integrations.substrate.contract_version"), "require_source_refs": bool( prd_config.get(root, "integrations.substrate.require_source_refs") ), "endpoint": _safe_url(_config_value(root, "integrations.substrate.endpoint", "http://127.0.0.1:47124"), label="substrate endpoint"), "manual_url": _safe_url(_config_value(root, "integrations.substrate.manual_url", "http://127.0.0.1:47124/utcp.json"), label="substrate manual URL"), "knowledge_browse_url": _safe_url(_config_value(root, "integrations.substrate.knowledge_browse_url", "http://127.0.0.1:5276"), label="knowledge browse URL"), "auth_env": str(_config_value(root, "integrations.substrate.auth_env", "AI_COLLAB_API_KEY")), "timeout_seconds": int(_config_value(root, "integrations.substrate.timeout_seconds", 5)), "max_result_bytes": int(_config_value(root, "integrations.substrate.max_result_bytes", 1_048_576)), "cache_ttl_seconds": int(_config_value(root, "integrations.substrate.cache_ttl_seconds", 60)), "fallback": str(_config_value(root, "integrations.substrate.fallback", "local")), "automation": { name: bool(_config_value(root, f"integrations.substrate.automation.{name}", default)) for name, default in { "discovery_on_session_start": True, "knowledge_recall": True, "memory_recall": True, "context_enrichment": True, "notices_on_session_start": False, "goals_sync": False, "telemetry_on_maintenance": False, "reporting_dispatch": True, "judgment_dispatch": True, "verification_execution": True, }.items() }, } if configured["contract_version"] != CONTRACT_VERSION: raise ContractError( f"unsupported integrations.substrate.contract_version " f"{configured['contract_version']}; supported: {CONTRACT_VERSION}" ) active = configured["enabled"] and configured["mode"] != "off" mode = configured["mode"] if active else "off" allowed = set(configured["capabilities"]) if mode == "observe": allowed &= OBSERVE_CAPABILITIES elif mode != "coordinate": allowed.clear() effective = [name for name in prd_config.SUBSTRATE_CAPABILITIES if name in allowed] return { "configured": configured, "active": active, "effective_mode": mode, "effective_capabilities": effective, } def _headers(policy: dict[str, Any], *, idempotency_key: str = "") -> dict[str, str]: headers = {"Accept": "application/json", "Content-Type": "application/json", "User-Agent": "prd-plugin-substrate/2"} auth_env = policy["configured"]["auth_env"] secret = os.environ.get(auth_env, "") if auth_env else "" if secret: headers["x-api-key"] = secret if idempotency_key: headers["x-prd-idempotency-key"] = idempotency_key return headers def _http_json( url: str, *, policy: dict[str, Any], method: str = "GET", payload: dict[str, Any] | None = None, idempotency_key: str = "", ) -> Any: safe = _safe_url(url, label="runtime tool URL") body = None if payload is None else json.dumps(payload, separators=(",", ":"), ensure_ascii=True).encode("utf-8") request = urllib.request.Request(safe, data=body, method=method.upper(), headers=_headers(policy, idempotency_key=idempotency_key)) limit = policy["configured"]["max_result_bytes"] timeout = policy["configured"]["timeout_seconds"] try: with urllib.request.urlopen(request, timeout=timeout) as response: declared = response.headers.get("Content-Length") if declared and int(declared) > limit: raise ContractError(f"runtime response exceeds max_result_bytes={limit}") raw = response.read(limit + 1) except (urllib.error.URLError, TimeoutError, OSError, ValueError) as exc: raise ContractError(f"runtime request failed for {urllib.parse.urlsplit(safe).path}: {exc}") from exc if len(raw) > limit: raise ContractError(f"runtime response exceeds max_result_bytes={limit}") try: return json.loads(raw.decode("utf-8-sig")) except (UnicodeDecodeError, json.JSONDecodeError) as exc: raise ContractError(f"runtime returned invalid JSON: {exc}") from exc def _runtime_catalog(repo_root: str | Path = ".") -> tuple[dict[str, Any], dict[str, dict[str, Any]]]: root = Path(repo_root).resolve() policy = effective_policy(root) if not policy["active"]: raise ContractError("substrate integration is disabled") manual = _http_json(policy["configured"]["manual_url"], policy=policy) if not isinstance(manual, dict) or not isinstance(manual.get("tools"), list): raise ContractError("runtime UTCP manual must contain a tools array") catalog: dict[str, dict[str, Any]] = {} for row in manual["tools"]: if not isinstance(row, dict) or not isinstance(row.get("name"), str): continue template = row.get("tool_call_template") if not isinstance(template, dict) or template.get("call_template_type") != "http": continue url = _safe_url(str(template.get("url", "")), label=f"runtime tool {row['name']} URL") method = str(template.get("http_method", "POST")).upper() if method not in {"POST", "GET"}: continue catalog[row["name"]] = {"name": row["name"], "url": url, "method": method, "inputs": row.get("inputs", {})} if not catalog: raise ContractError("runtime UTCP manual exposes no supported HTTP tools") return manual, catalog def discover_runtime(repo_root: str | Path = ".") -> dict[str, Any]: manual, catalog = _runtime_catalog(repo_root) return { "status": "ok", "contract_version": CONTRACT_VERSION, "utcp_version": str(manual.get("utcp_version", "unknown")), "runtime_name": str(manual.get("name", manual.get("manual", {}).get("name", "ai-collab"))), "tools": sorted(catalog), "tool_count": len(catalog), "manual_hash": _canonical_hash(manual), } def _service_projection(root: Path) -> dict[str, Any]: try: return {"status": "ok", **prd_services.manifest_projection(root)} except prd_services.ManifestError as exc: return {"status": "unavailable", "reason": str(exc), "manifest_hash": "", "consumes": [], "provides": [], "service_count": 0} def execute_tool( repo_root: str | Path = ".", *, tool: str, arguments: dict[str, Any], capability: str, source_refs: list[str], idempotency_key: str = "", required: bool = False, ) -> dict[str, Any]: root = Path(repo_root).resolve() policy = effective_policy(root) if not policy["active"] or policy["effective_mode"] != "coordinate": raise ContractError("runtime execution requires substrate coordinate mode") if capability not in policy["effective_capabilities"]: raise ContractError(f"substrate capability is not enabled: {capability}") if not isinstance(arguments, dict): raise ContractError("tool arguments must be an object") if policy["configured"]["require_source_refs"] and not source_refs: raise ContractError("source_refs are required for runtime execution") if any(not isinstance(ref, str) or not ref.strip() for ref in source_refs): raise ContractError("source_refs must contain non-empty strings") try: manual, catalog = _runtime_catalog(root) if tool not in catalog: raise ContractError(f"runtime tool is not advertised: {tool}") services = _service_projection(root) stable_key = idempotency_key or _canonical_hash({"tool": tool, "arguments": arguments, "source_refs": sorted(set(source_refs))}) request = { "contract_version": CONTRACT_VERSION, "tool": tool, "capability": capability, "arguments_hash": _canonical_hash(arguments), "source_refs": sorted(set(source_refs)), "source_hash": _canonical_hash(sorted(set(source_refs))), "manifest_hash": services["manifest_hash"], "idempotency_key": stable_key, } request["operation_id"] = "OP-" + _canonical_hash(request)[:24] descriptor = catalog[tool] result = _http_json(descriptor["url"], policy=policy, method=descriptor["method"], payload=arguments, idempotency_key=stable_key) receipt = { "operation_id": request["operation_id"], "tool": tool, "capability": capability, "runtime_utcp_version": str(manual.get("utcp_version", "unknown")), "result": result, "result_hash": _canonical_hash(result), "completed_at": _now(), } receipt["receipt_hash"] = _canonical_hash(receipt) return {"status": "completed", "request": request, "receipt": receipt} except ContractError as exc: fallback = policy["configured"]["fallback"] if required or fallback == "fail" or "not advertised" in str(exc): raise return { "status": "degraded", "reason": str(exc), "fallback": fallback, "request": {"tool": tool, "capability": capability, "source_refs": sorted(set(source_refs)), "idempotency_key": idempotency_key}, } def build_handshake(repo_root: str | Path = ".", *, contact_runtime: bool = False) -> dict[str, Any]: """Describe the local contract and optionally negotiate live runtime identity.""" root = Path(repo_root).resolve() config = _read_json(root / ".prd_plugin" / "config.json", {}) plugin = config.get("plugin", {}) if isinstance(config, dict) else {} surface_paths = { "utcp": ".prd_plugin/scripts/prd_tools.py", "mcp": ".prd_plugin/mcp/server.cjs", "graph": ".prd_plugin/scripts/prd_graph.py", "reporting": ".prd_plugin/scripts/prd_reporting.py", "verification": ".prd_plugin/scripts/prd_test_scope.py", "workflows": ".prd_plugin/scripts/prd_workflows.py", "services": ".prd_plugin/scripts/prd_services.py", } result = { "schema_version": SCHEMA_VERSION, "contract_version": CONTRACT_VERSION, "plugin_version": str(plugin.get("installed_version", "unknown")), "policy": effective_policy(root), "services": _service_projection(root), "surfaces": { name: {"path": path, "available": (root / path).is_file()} for name, path in surface_paths.items() }, "invariants": { "canonical_state_owner": "prd-plugin", "runtime_owner": "ai-collab", "writes_via_mcp_only": True, "local_state_excluded": True, "secrets_excluded": True, "reflection_answers_excluded": True, "workflow_judgment_is_hash_and_schema_bound": True, "workflow_judgment_has_no_state_write_authority": True, "runtime_calls_are_allowlisted": True, "runtime_results_are_hash_bound": True, "repository_services_are_declared": True, }, "knowledge_browser": { "base_url": effective_policy(root)["configured"]["knowledge_browse_url"], "owner": "ai-collab", "canonical_source": "repository-markdown", "availability": "runtime-owned", }, } if contact_runtime: try: discovery = discover_runtime(root) identity_call = execute_tool( root, tool="whoami", arguments={}, capability="discovery", source_refs=[result["services"].get("manifest_hash") or "services:unavailable"], idempotency_key="prd-handshake-whoami", ) identity = identity_call.get("receipt", {}).get("result", {}) if identity_call.get("status") == "completed" else {} result["runtime"] = {**discovery, "identity": identity, "endpoint_identity": urllib.parse.urlsplit(result["policy"]["configured"]["endpoint"]).netloc} except ContractError as exc: result["runtime"] = {"status": "unavailable", "reason": str(exc), "endpoint_identity": urllib.parse.urlsplit(result["policy"]["configured"]["endpoint"]).netloc} return result def _as_records(payload: Any, keys: tuple[str, ...]) -> list[dict[str, Any]]: if isinstance(payload, list): return [row for row in payload if isinstance(row, dict)] if not isinstance(payload, dict): return [] for key in keys: rows = payload.get(key) if isinstance(rows, list): return [row for row in rows if isinstance(row, dict)] return [] def _links(row: dict[str, Any]) -> list[str]: values: list[str] = [] for key in ("linked_ids", "affected_ids", "source_ids", "source_refs", "graduated_to"): raw = row.get(key, []) raw = [raw] if isinstance(raw, str) else raw if isinstance(raw, list): values.extend(str(item) for item in raw if isinstance(item, str) and ID_RE.match(item)) return sorted(set(values)) def _project_record(repo_id: str, kind: str, row: dict[str, Any], source_ref: str) -> dict[str, Any] | None: record_id = row.get("id") or row.get("branch_id") if not isinstance(record_id, str) or not ID_RE.match(record_id): return None projected = { "repo_key": f"{repo_id}::{kind}::{record_id}", "kind": kind, "id": record_id, "status": str(row.get("status", row.get("state", "logged"))), "summary": str( row.get("summary", row.get("title", row.get("method", row.get("name", "")))) ).strip()[:1000], "linked_ids": _links(row), "source_session": str( row.get("source_session", row.get("created_from_session", row.get("session_id", ""))) ), "source_agent": str( row.get("source_agent", row.get("owner_agent", row.get("created_by_agent", ""))) ), "source_ref": source_ref.replace("\\", "/"), } projected["content_hash"] = _canonical_hash(projected) return projected def _collect_records(root: Path, repo_id: str) -> list[dict[str, Any]]: state = root / ".prd_plugin" / "state" specs = ( ("requests.json", "REQ", ("requests", "records")), ("tracking.json", "TRK", ("records", "tracking")), ("health.json", "HLT", ("findings", "records")), ("changelog.json", "CHG", ("changes", "records")), ("evidence.json", "EV", ("records", "evidence")), ("decisions.json", "DEC", ("decisions", "records")), ("workflow-runs.json", "WFR", ("runs",)), ) records: list[dict[str, Any]] = [] for filename, kind, keys in specs: path = state / filename for row in _as_records(_read_json(path, {}), keys): projected = _project_record(repo_id, kind, row, str(path.relative_to(root))) if projected: records.append(projected) artifacts = state / "artifacts" if artifacts.is_dir(): for path in sorted(artifacts.rglob("*.json")): payload = _read_json(path, {}) for row in _as_records(payload, ("validations",)): projected = _project_record(repo_id, "VAL", row, str(path.relative_to(root))) if projected: records.append(projected) branches = state / "tracking-branches" if branches.is_dir(): for path in sorted(branches.glob("*.json")): payload = _read_json(path, {}) if isinstance(payload, dict): projected = _project_record(repo_id, "DBR", payload, str(path.relative_to(root))) if projected: records.append(projected) return sorted(records, key=lambda row: row["repo_key"]) def build_snapshot( repo_root: str | Path = ".", *, repo_id: str, max_records: int = DEFAULT_MAX_RECORDS, ) -> dict[str, Any]: """Build a complete bounded snapshot for a registered repository.""" if not str(repo_id).strip(): raise ContractError("repo_id is required") if max_records < 1: raise ContractError("max_records must be positive") root = Path(repo_root).resolve() policy = effective_policy(root) active = policy["active"] and "records" in policy["effective_capabilities"] records = _collect_records(root, str(repo_id)) if active else [] if len(records) > max_records: raise ContractError( f"snapshot contains {len(records)} records, exceeding max_records={max_records}; " "refusing to return an incomplete projection" ) return { "schema_version": SCHEMA_VERSION, "contract_version": CONTRACT_VERSION, "repo_id": str(repo_id), "active": active, "record_count": len(records), "records": records, "snapshot_hash": _canonical_hash(records), } def build_graph_export(repo_root: str | Path = ".", *, repo_id: str) -> dict[str, Any]: """Build repo-qualified cause/effect edges for Substrate causal ingest.""" if not str(repo_id).strip(): raise ContractError("repo_id is required") root = Path(repo_root).resolve() policy = effective_policy(root) active = policy["active"] and "graph" in policy["effective_capabilities"] raw_edges = prd_graph.to_cause_effect(prd_graph.build_graph(root)) if active else [] edges = [ { "cause": f"{repo_id}::{edge['cause']}", "effect": f"{repo_id}::{edge['effect']}", "label": edge["label"], } for edge in raw_edges ] edges.sort(key=lambda edge: (edge["cause"], edge["effect"], edge["label"])) return { "schema_version": SCHEMA_VERSION, "contract_version": CONTRACT_VERSION, "repo_id": str(repo_id), "active": active, "edge_count": len(edges), "edges": edges, "graph_hash": _canonical_hash(edges), } def build_event_export( repo_root: str | Path = ".", *, repo_id: str, max_events: int = DEFAULT_MAX_EVENTS, ) -> dict[str, Any]: """Project bounded canonical change and workflow lifecycle events.""" if not str(repo_id).strip(): raise ContractError("repo_id is required") if max_events < 1: raise ContractError("max_events must be positive") root = Path(repo_root).resolve() policy = effective_policy(root) active = policy["active"] and "events" in policy["effective_capabilities"] events: list[dict[str, Any]] = [] if active: changes = _as_records(_read_json(root / ".prd_plugin" / "state" / "changelog.json", {}), ("changes", "records")) runs = _as_records(_read_json(root / ".prd_plugin" / "state" / "workflow-runs.json", {}), ("runs",)) for kind, rows in (("change", changes), ("workflow", runs)): for row in rows: record_id = str(row.get("id", "")) if not ID_RE.match(record_id): continue event = { "repo_id": str(repo_id), "kind": kind, "record_id": record_id, "status": str(row.get("status", "logged")), "timestamp": str(row.get("updated_at", row.get("timestamp", row.get("created_at", "")))), "summary": str(row.get("summary", row.get("workflow_id", "")))[:1000], "source_refs": _links(row), } event["event_id"] = f"{repo_id}::{kind}::{record_id}::{_canonical_hash(event)[:16]}" event["content_hash"] = _canonical_hash(event) events.append(event) events.sort(key=lambda row: (row["timestamp"], row["event_id"])) if len(events) > max_events: raise ContractError(f"event export contains {len(events)} events, exceeding max_events={max_events}; refusing incomplete projection") return { "schema_version": SCHEMA_VERSION, "contract_version": CONTRACT_VERSION, "repo_id": str(repo_id), "active": active, "event_count": len(events), "events": events, "events_hash": _canonical_hash(events), } def diagnose_runtime(repo_root: str | Path = ".") -> dict[str, Any]: root = Path(repo_root).resolve() services = _service_projection(root) try: discovery = discover_runtime(root) available = [{"service": "ai-collab.substrate", "provider": "ai-collab-v3"}] reconciliation = prd_services.reconcile_services(root, available=available) if services["status"] == "ok" else {"status": "error", "missing_required": [], "degraded_optional": []} policy = effective_policy(root) probes = [] if policy["effective_mode"] == "coordinate": specs = [ ("discovery", "whoami", {}), ("diagnostics", "substrate_health", {"consumer": "prd-plugin"}), ("diagnostics", "diagnostics_doctor", {}), ("impact", "index_status", {"repo": root.name}), ("knowledge", "knowledge_freshness", {"scope": "repo", "repoId": root.name}), ] for capability, tool, arguments in specs: if capability not in policy["effective_capabilities"]: continue try: probes.append({"capability": capability, "tool": tool, **execute_tool( root, tool=tool, arguments=arguments, capability=capability, source_refs=[".prd_plugin/services.json"], idempotency_key=f"prd-diagnose:{tool}", )}) except ContractError as exc: probes.append({"capability": capability, "tool": tool, "status": "degraded", "reason": str(exc)}) degraded = reconciliation["status"] == "degraded" or any(row.get("status") != "completed" for row in probes) status = "error" if reconciliation["status"] == "error" else ("degraded" if degraded else "ok") return {"status": status, "discovery": discovery, "services": reconciliation, "probes": probes} except (ContractError, prd_services.ManifestError) as exc: try: reconciliation = prd_services.reconcile_services(root, available=[]) except prd_services.ManifestError: reconciliation = {"status": "error", "missing_required": [], "degraded_optional": []} return {"status": "error" if reconciliation.get("missing_required") else "degraded", "reason": str(exc), "services": reconciliation} def main(argv=None) -> int: parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) parser.add_argument("action", choices=("handshake", "snapshot", "graph", "events", "discover", "call", "diagnose")) parser.add_argument("--repo-root", default=".") parser.add_argument("--repo-id", default="") parser.add_argument("--max-records", type=int, default=DEFAULT_MAX_RECORDS) parser.add_argument("--max-events", type=int, default=DEFAULT_MAX_EVENTS) parser.add_argument("--contact-runtime", action="store_true") parser.add_argument("--tool", default="") parser.add_argument("--arguments", default="{}") parser.add_argument("--capability", default="") parser.add_argument("--source-ref", action="append", default=[]) parser.add_argument("--idempotency-key", default="") parser.add_argument("--required", action="store_true") parser.add_argument("--format", choices=("json",), default="json") args = parser.parse_args(argv) try: if args.action == "handshake": result = build_handshake(args.repo_root, contact_runtime=args.contact_runtime) elif args.action == "snapshot": result = build_snapshot(args.repo_root, repo_id=args.repo_id, max_records=args.max_records) elif args.action == "graph": result = build_graph_export(args.repo_root, repo_id=args.repo_id) elif args.action == "events": result = build_event_export(args.repo_root, repo_id=args.repo_id, max_events=args.max_events) elif args.action == "discover": result = discover_runtime(args.repo_root) elif args.action == "diagnose": result = diagnose_runtime(args.repo_root) else: try: arguments = json.loads(args.arguments) except json.JSONDecodeError as exc: raise ContractError(f"invalid --arguments JSON: {exc}") from exc result = execute_tool(args.repo_root, tool=args.tool, arguments=arguments, capability=args.capability, source_refs=args.source_ref, idempotency_key=args.idempotency_key, required=args.required) print(json.dumps(result, indent=2)) return 0 except ContractError as exc: print(f"[PRD Plugin] substrate contract error: {exc}", file=sys.stderr) return 2 if __name__ == "__main__": raise SystemExit(main())