#!/usr/bin/env python3
"""Deterministic PRD Plugin workflow engine.

Workflow JSON selects only code-owned actions. The engine owns validation,
planning, transitions, persistence, receipts, retries, postconditions, and the
boundary around external judgment. It never calls a model and never treats a
requested judgment as completed work.
"""

from __future__ import annotations

import argparse
import copy
import hashlib
import json
import os
import re
import shutil
import subprocess
import sys
import tempfile
import time
import uuid
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Callable


SCHEMA_VERSION = "1.0"
TERMINAL = {"completed", "failed", "blocked", "cancelled"}
RUN_STATUSES = {"pending", "running", "waiting_judgment", *TERMINAL}
ID_RE = re.compile(r"^WFR-(\d+)$")
REF_RE = re.compile(r"^\$\{(inputs|steps)(?:\.([A-Za-z0-9_-]+))?(?:\.(.+))?\}$")


class WorkflowError(ValueError):
    pass


class CatalogError(WorkflowError):
    pass


class TransitionError(WorkflowError):
    pass


class JudgmentError(WorkflowError):
    pass


class ActionError(WorkflowError):
    pass


@dataclass(frozen=True)
class ActionSpec:
    name: str
    handler: Callable[[Path, dict[str, Any], dict[str, Any]], dict[str, Any]] | None
    determinism: str
    mutation: str
    idempotent: bool
    description: str


def _canonical(value: Any) -> str:
    return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True)


def _hash(value: Any) -> str:
    return hashlib.sha256(_canonical(value).encode("utf-8")).hexdigest()


def _now() -> str:
    return datetime.now(timezone.utc).isoformat(timespec="seconds").replace("+00:00", "Z")


def _read_json(path: Path, default: Any = None) -> Any:
    try:
        return json.loads(path.read_text(encoding="utf-8-sig"))
    except FileNotFoundError:
        if default is not None:
            return copy.deepcopy(default)
        raise WorkflowError(f"required JSON file does not exist: {path}")
    except (OSError, json.JSONDecodeError) as exc:
        raise WorkflowError(f"cannot read valid JSON from {path}: {exc}") from exc


def _atomic_json(path: Path, value: Any) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    fd, raw = tempfile.mkstemp(prefix=path.name + ".", suffix=".tmp", dir=path.parent)
    tmp = Path(raw)
    try:
        with os.fdopen(fd, "w", encoding="utf-8", newline="\n") as handle:
            json.dump(value, handle, indent=2, ensure_ascii=False)
            handle.write("\n")
            handle.flush()
            os.fsync(handle.fileno())
        os.replace(tmp, path)
    finally:
        tmp.unlink(missing_ok=True)


class _StateLock:
    """Cross-language lock compatible with mcp/server.cjs."""

    def __init__(self, root: Path, wait_seconds: float = 12, stale_seconds: float = 10):
        self.path = root / ".prd_plugin" / "local" / "mcp-state.lock"
        self.owner = f"{os.getpid()}-{uuid.uuid4().hex}"
        self.wait_seconds = wait_seconds
        self.stale_seconds = stale_seconds

    def __enter__(self) -> "_StateLock":
        self.path.parent.mkdir(parents=True, exist_ok=True)
        deadline = time.monotonic() + self.wait_seconds
        while True:
            try:
                self.path.mkdir()
                (self.path / "owner").write_text(self.owner, encoding="utf-8")
                return self
            except OSError as exc:
                try:
                    if time.time() - self.path.stat().st_mtime > self.stale_seconds:
                        grave = self.path.with_name(self.path.name + f".stale-{uuid.uuid4().hex}")
                        self.path.rename(grave)
                        shutil.rmtree(grave, ignore_errors=True)
                        continue
                except (FileNotFoundError, OSError):
                    pass
                if time.monotonic() >= deadline:
                    raise WorkflowError("timed out waiting for .prd_plugin/local/mcp-state.lock") from exc
                time.sleep(0.02)

    def __exit__(self, *_: Any) -> None:
        try:
            if (self.path / "owner").read_text(encoding="utf-8") == self.owner:
                shutil.rmtree(self.path)
        except (FileNotFoundError, OSError):
            pass


def _action_noop(_root: Path, _args: dict[str, Any], _context: dict[str, Any]) -> dict[str, Any]:
    return {"ok": True}


def _action_echo(_root: Path, args: dict[str, Any], _context: dict[str, Any]) -> dict[str, Any]:
    return {"value": args.get("value")}


def _action_assert(_root: Path, args: dict[str, Any], _context: dict[str, Any]) -> dict[str, Any]:
    actual = args.get("actual")
    expected = args.get("equals")
    if actual != expected:
        raise ActionError(f"assertion failed: expected {expected!r}, got {actual!r}")
    return {"actual": actual, "equals": expected, "passed": True}


def _action_config_audit(root: Path, _args: dict[str, Any], _context: dict[str, Any]) -> dict[str, Any]:
    import prd_config
    result = prd_config.audit_config(root)
    if result.get("status") not in {"ok", "pass"}:
        raise ActionError(f"config audit failed: {result.get('findings', result)}")
    return result


def _action_state_consistency(root: Path, args: dict[str, Any], _context: dict[str, Any]) -> dict[str, Any]:
    import state_consistency_check
    result = state_consistency_check.build_consistency_report(root, args.get("now"))
    if result.get("status") != "ok":
        raise ActionError(f"state consistency failed with {len(result.get('findings', []))} finding(s)")
    return {key: value for key, value in result.items() if key != "canonical_ids"}


def _action_requests_pull(root: Path, _args: dict[str, Any], _context: dict[str, Any]) -> dict[str, Any]:
    """Apply hub resolutions delivered into .prd_plugin/mailboxes/ (REQ-125).

    Runs at session start so an answered upstream request stops warning without
    a human relay. Fail-open: transport problems never block a session."""
    import request_pull
    import request_reply
    result = {"applied": 0, "resolved_ids": [], "unmatched": [], "mailboxes": []}
    try:
        result = request_pull.apply_pending_mailboxes(root)
    except (OSError, ValueError) as exc:
        result["error"] = f"{type(exc).__name__}: {exc}"
    # Outbound half: a reply written but never sent is the recurring failure
    # (REQ-133). Session start flushes anything still pending so a reply
    # cannot sit indefinitely.
    try:
        result["outbound"] = request_reply.flush_pending_replies(root)
    except (OSError, ValueError) as exc:
        result["outbound"] = {"delivered": 0, "undeliverable": [],
                              "error": f"{type(exc).__name__}: {exc}"}
    # Same guarantee for requests addressed to a peer (REQ-136): a request
    # aimed at another repo must not sit undelivered just because nobody ran
    # the send by hand.
    try:
        import request_routing
        result["addressed"] = request_routing.flush_addressed_requests(root)
    except (OSError, ValueError) as exc:
        result["addressed"] = {"delivered": 0, "undelivered": [],
                               "error": f"{type(exc).__name__}: {exc}"}
    return result


def _action_gate_check(root: Path, _args: dict[str, Any], _context: dict[str, Any]) -> dict[str, Any]:
    import prd_gate
    result = prd_gate.run_checks(root)
    if result.get("status") == "fail":
        raise ActionError(f"PRD gate failed with {result.get('summary', {}).get('errors', 0)} error(s)")
    return result


def _action_graph_build(root: Path, _args: dict[str, Any], _context: dict[str, Any]) -> dict[str, Any]:
    import prd_graph
    return prd_graph.build_graph(root)


def _action_staleness_audit(root: Path, args: dict[str, Any], _context: dict[str, Any]) -> dict[str, Any]:
    import staleness_audit
    return staleness_audit.audit(root, args.get("now"))


def _action_reflection_list(root: Path, args: dict[str, Any], _context: dict[str, Any]) -> dict[str, Any]:
    import prd_reflections
    return prd_reflections.list_reflections(root, entity=args.get("entity", "all"),
                                            category_id=args.get("category"), enabled=args.get("enabled", True))


def _action_verification_plan(root: Path, args: dict[str, Any], _context: dict[str, Any]) -> dict[str, Any]:
    import prd_test_scope
    return prd_test_scope.build_plan(root, changed_files=args.get("changed_files"),
                                     base_ref=args.get("base_ref", ""), release=bool(args.get("release", False)))


def _action_verification_execute(root: Path, args: dict[str, Any], _context: dict[str, Any]) -> dict[str, Any]:
    import prd_test_scope

    plan = args.get("plan")
    if not isinstance(plan, dict):
        raise ActionError("verification.execute requires a verification plan")
    result = prd_test_scope.execute_plan(root, plan)
    if result.get("status") != "passed":
        raise ActionError("verification execution failed")
    return result


def _action_reporting_bundle(root: Path, args: dict[str, Any], _context: dict[str, Any]) -> dict[str, Any]:
    import prd_reporting
    return prd_reporting.build_bundle(root, task=args.get("task", "report_summary"))


def _action_reporting_delegate(root: Path, args: dict[str, Any], _context: dict[str, Any]) -> dict[str, Any]:
    import prd_reporting
    import prd_runtime_worker
    import prd_substrate

    bundle = args.get("bundle")
    if not isinstance(bundle, dict) or not isinstance(bundle.get("task"), str):
        raise ActionError("reporting.delegate requires a reporting bundle object")
    substrate = prd_substrate.effective_policy(root)
    available = (
        substrate["active"]
        and substrate["effective_mode"] == "coordinate"
        and "delegated_reporting" in substrate["effective_capabilities"]
        and substrate["configured"]["automation"]["reporting_dispatch"]
    )
    decision = prd_reporting.delegation_decision(root, bundle["task"], executor_available=available)
    if decision["action"] != "delegate":
        return {"status": "fallback", "action": decision["action"], "reason": decision["reason"]}
    return prd_runtime_worker.dispatch_reporting(root, bundle)


def _action_substrate_snapshot(root: Path, args: dict[str, Any], _context: dict[str, Any]) -> dict[str, Any]:
    import prd_substrate
    return prd_substrate.build_snapshot(root, repo_id=args.get("repo_id", root.name))


def _action_services_audit(root: Path, _args: dict[str, Any], _context: dict[str, Any]) -> dict[str, Any]:
    import prd_services

    report = prd_services.audit_manifest(root)
    if report.get("status") != "ok":
        raise ActionError(f"repository service manifest has {sum(report.get('summary', {}).values())} finding(s)")
    return report


def _action_substrate_preflight(root: Path, _args: dict[str, Any], _context: dict[str, Any]) -> dict[str, Any]:
    import prd_services
    import prd_substrate

    services = prd_services.audit_manifest(root)
    policy = prd_substrate.effective_policy(root)
    if not policy["active"] or not policy["configured"]["automation"]["discovery_on_session_start"]:
        return {"status": "disabled", "services": services, "policy": policy}
    handshake = prd_substrate.build_handshake(root, contact_runtime=True)
    diagnosis = prd_substrate.diagnose_runtime(root)
    available = []
    if handshake.get("runtime", {}).get("status") == "ok":
        available.append({"service": "ai-collab.substrate", "provider": "ai-collab-v3"})
    reconciliation = prd_services.reconcile_services(root, available=available)
    return {
        "status": "error" if reconciliation["status"] == "error" else "ok",
        "services": services,
        "reconciliation": reconciliation,
        "handshake": handshake,
        "diagnosis": diagnosis,
    }


def _action_substrate_enrich(root: Path, args: dict[str, Any], _context: dict[str, Any]) -> dict[str, Any]:
    import prd_substrate

    query = args.get("query")
    refs = args.get("source_refs", [])
    if not isinstance(query, str) or not query.strip():
        raise ActionError("substrate.enrich requires a non-empty query")
    if not isinstance(refs, list) or any(not isinstance(ref, str) or not ref for ref in refs):
        raise ActionError("substrate.enrich source_refs must be non-empty strings")
    policy = prd_substrate.effective_policy(root)
    if not policy["active"] or policy["effective_mode"] != "coordinate":
        return {"status": "disabled", "results": [], "local_source_refs": refs}
    configured = policy["configured"]["automation"]
    requests = []
    if configured["knowledge_recall"] and "knowledge" in policy["effective_capabilities"]:
        requests.append(("knowledge", "knowledge_search", {"query": query, "k": 8}))
    if configured["memory_recall"] and "memory" in policy["effective_capabilities"]:
        requests.append(("memory", "memory_recall", {"query": query, "k": 5}))
    if configured["context_enrichment"] and "context" in policy["effective_capabilities"]:
        requests.append(("context", "context_pack", {"query": query, "repos": [root.name], "k": 8, "tokenBudget": 8000}))
    results = []
    for capability, tool, arguments in requests:
        try:
            outcome = prd_substrate.execute_tool(
                root, tool=tool, arguments=arguments, capability=capability,
                source_refs=refs, idempotency_key=f"prd-enrich:{_hash([query, capability, refs])}",
            )
            results.append({"capability": capability, "tool": tool, **outcome})
        except prd_substrate.ContractError as exc:
            results.append({"capability": capability, "tool": tool, "status": "degraded", "reason": str(exc), "fallback": "local"})
    return {"status": "ok" if all(row["status"] in {"completed", "degraded"} for row in results) else "error", "results": results, "local_source_refs": refs}


def _action_substrate_notices(root: Path, _args: dict[str, Any], _context: dict[str, Any]) -> dict[str, Any]:
    import prd_substrate

    policy = prd_substrate.effective_policy(root)
    if (
        not policy["active"] or policy["effective_mode"] != "coordinate"
        or "watches" not in policy["effective_capabilities"]
        or not policy["configured"]["automation"]["notices_on_session_start"]
    ):
        return {"status": "disabled"}
    return prd_substrate.execute_tool(
        root, tool="notices", arguments={"ack": False, "limit": 50}, capability="watches",
        source_refs=[".prd_plugin/services.json"], idempotency_key="prd-session-notices",
    )


def _action_substrate_telemetry(root: Path, _args: dict[str, Any], _context: dict[str, Any]) -> dict[str, Any]:
    import prd_substrate

    policy = prd_substrate.effective_policy(root)
    if (
        not policy["active"] or policy["effective_mode"] != "coordinate"
        or "telemetry" not in policy["effective_capabilities"]
        or not policy["configured"]["automation"]["telemetry_on_maintenance"]
    ):
        return {"status": "disabled"}
    return prd_substrate.execute_tool(
        root, tool="toolusage_stats", arguments={}, capability="telemetry",
        source_refs=[".prd_plugin/services.json"], idempotency_key="prd-maintenance-telemetry",
    )


def _runtime_id(value: Any, keys: tuple[str, ...]) -> str:
    if isinstance(value, dict):
        for key in keys:
            candidate = value.get(key)
            if isinstance(candidate, str) and candidate:
                return candidate
        for child in value.values():
            found = _runtime_id(child, keys)
            if found:
                return found
    if isinstance(value, list):
        for child in value:
            found = _runtime_id(child, keys)
            if found:
                return found
    return ""


def _action_substrate_goals(root: Path, _args: dict[str, Any], _context: dict[str, Any]) -> dict[str, Any]:
    """Create one-way goal bindings and observe remote state without importing it."""
    import prd_substrate
    import prd_substrate_links

    policy = prd_substrate.effective_policy(root)
    if (
        not policy["active"] or policy["effective_mode"] != "coordinate"
        or "goals" not in policy["effective_capabilities"]
        or not policy["configured"]["automation"]["goals_sync"]
    ):
        return {"status": "disabled", "created": 0, "observed": 0, "results": []}
    state = _read_json(root / ".prd_plugin" / "state" / "tracking.json")
    records = state.get("records", []) if isinstance(state, dict) else []
    goals = [row for row in records if isinstance(row, dict) and row.get("type") == "goal" and isinstance(row.get("id"), str)]
    results = []
    created = observed = 0
    for goal in goals:
        local_id = goal["id"]
        link = prd_substrate_links.get_link(root, "goal", local_id)
        try:
            if link is None:
                outcome = prd_substrate.execute_tool(
                    root, tool="goal_create", capability="goals",
                    arguments={
                        "title": str(goal.get("summary", local_id))[:300],
                        "outcome": str(goal.get("summary", local_id))[:1000],
                        "acceptance": [f"PRD Plugin reports {local_id} complete with linked verification evidence"],
                        "source_ref": local_id,
                        "repo_goals": [{"repo_id": root.name, "owner_agent": str(goal.get("owner_agent", "prd-plugin")), "title": str(goal.get("summary", local_id))[:300], "outcome": str(goal.get("summary", local_id))[:1000]}],
                    },
                    source_refs=[local_id], idempotency_key=f"prd-goal:{local_id}",
                )
                remote_id = _runtime_id(outcome.get("receipt", {}).get("result"), ("goal_id", "goalId", "id"))
                if not remote_id:
                    raise ActionError(f"runtime returned no goal id for {local_id}")
                prd_substrate_links.upsert_link(
                    root, kind="goal", local_id=local_id, remote_id=remote_id,
                    receipt_hash=str(outcome.get("receipt", {}).get("receipt_hash", "")),
                    status=str(goal.get("status", "active")),
                )
                created += 1
                results.append({"local_id": local_id, "remote_id": remote_id, "action": "created", "receipt_hash": outcome.get("receipt", {}).get("receipt_hash", "")})
            else:
                outcome = prd_substrate.execute_tool(
                    root, tool="goal_tree", capability="goals", arguments={"goal_id": link["remote_id"]},
                    source_refs=[local_id], idempotency_key=f"prd-goal-observe:{local_id}:{link['remote_id']}",
                )
                observed += 1
                results.append({"local_id": local_id, "remote_id": link["remote_id"], "action": "observed", "remote": outcome})
        except (prd_substrate.ContractError, prd_substrate_links.LinkError, ActionError) as exc:
            results.append({"local_id": local_id, "action": "degraded", "reason": str(exc)})
    return {"status": "ok", "created": created, "observed": observed, "results": results,
            "authority": {"project_truth": "prd-plugin", "remote_state_import": False, "remote_pause_stop_requires_explicit_action": True}}


def _git(root: Path, *args: str) -> subprocess.CompletedProcess[str]:
    return subprocess.run(["git", *args], cwd=root, text=True, encoding="utf-8", errors="replace",
                          capture_output=True, check=False)


def _action_git_status(root: Path, _args: dict[str, Any], _context: dict[str, Any]) -> dict[str, Any]:
    completed = _git(root, "status", "--short")
    if completed.returncode:
        raise ActionError(completed.stderr.strip() or "git status failed")
    files = [line[3:].replace("\\", "/") for line in completed.stdout.splitlines() if len(line) > 3]
    return {"clean": not files, "changed_files": files}


def _action_git_diff_check(root: Path, _args: dict[str, Any], _context: dict[str, Any]) -> dict[str, Any]:
    completed = _git(root, "diff", "--check")
    if completed.returncode:
        raise ActionError(completed.stdout.strip() or completed.stderr.strip() or "git diff --check failed")
    return {"passed": True}


MCP_QUERY_TOOLS = {"prd_status", "prd_find", "prd_get", "prd_config_list", "prd_config_get", "prd_validate",
                   "prd_reflection_list"}
MCP_MUTATION_TOOLS = {"prd_create", "prd_update", "prd_link", "prd_open_goal", "prd_update_goal", "prd_close_goal",
                      "prd_file_request", "prd_import_request", "prd_record_evidence", "prd_record_decision", "prd_log_change",
                      "prd_config_set", "prd_config_profile", "prd_reflection_create", "prd_reflection_update",
                      "prd_reflection_delete"}


def _mcp_call(root: Path, tool: str, arguments: dict[str, Any]) -> dict[str, Any]:
    if not isinstance(arguments, dict):
        raise ActionError("MCP arguments must be an object")
    server = root / ".prd_plugin" / "mcp" / "server.cjs"
    if not server.is_file():
        server = root / "mcp" / "server.cjs"
    if not server.is_file():
        raise ActionError("PRD MCP server is not installed")
    helper = "const m=require(process.argv[1]);const r=m.callTool(process.cwd(),process.argv[2],JSON.parse(process.argv[3]));process.stdout.write(JSON.stringify(r));"
    completed = subprocess.run(["node", "-e", helper, str(server), tool, json.dumps(arguments)], cwd=root,
                               text=True, encoding="utf-8", errors="replace", capture_output=True, check=False)
    if completed.returncode:
        raise ActionError(completed.stderr.strip() or f"MCP tool {tool} failed")
    try:
        return json.loads(completed.stdout)
    except json.JSONDecodeError as exc:
        raise ActionError(f"MCP tool {tool} returned invalid JSON") from exc


def _action_mcp_query(root: Path, args: dict[str, Any], _context: dict[str, Any]) -> dict[str, Any]:
    tool = args.get("tool")
    if tool not in MCP_QUERY_TOOLS:
        raise ActionError(f"MCP query tool is not allowlisted: {tool!r}")
    return _mcp_call(root, tool, args.get("arguments", {}))


def _action_mcp_mutate(root: Path, args: dict[str, Any], _context: dict[str, Any]) -> dict[str, Any]:
    if not _config(root)["allow_state_mutations"]:
        raise ActionError("canonical workflow mutations are disabled by workflows.allow_state_mutations")
    tool = args.get("tool")
    if tool not in MCP_MUTATION_TOOLS:
        raise ActionError(f"MCP mutation tool is not allowlisted: {tool!r}")
    return _mcp_call(root, tool, args.get("arguments", {}))


FIXED_OPERATIONS: dict[str, list[str]] = {
    "local_workflow_check": ["{python}", "scripts/local_workflow_check.py"],
    "release_check": ["{python}", "scripts/release_check.py"],
    "gap_audit": ["{python}", "scripts/gap_audit.py"],
    "focused_python_tests": ["{python}", "-m", "unittest", "discover", "-s", "tests", "-p", "test_prd_workflows.py", "-q"],
    "full_python_tests": ["{python}", "-m", "unittest", "discover", "-s", "tests", "-q"],
    "node_tests": ["npm", "test", "--", "--runInBand"],
    "package_check": ["npm", "pack", "--dry-run"],
    "workflow_chml_audit": ["{python}", "scripts/workflow_chml_audit.py", "--repo-root", ".", "--format", "json"],
    "substrate_capability_audit": ["{python}", "scripts/prd_substrate_catalog.py", "--repo-root", ".", "audit"],
}


def _action_fixed_process(root: Path, args: dict[str, Any], _context: dict[str, Any]) -> dict[str, Any]:
    operation = args.get("operation")
    if operation not in FIXED_OPERATIONS:
        raise ActionError(f"fixed operation is not allowlisted: {operation!r}")
    receipt_command = ["python" if part == "{python}" else part for part in FIXED_OPERATIONS[operation]]
    command = [sys.executable if part == "{python}" else part for part in FIXED_OPERATIONS[operation]]
    if operation == "gap_audit":
        package = _read_json(root / "package.json")
        version = package.get("version")
        if not isinstance(version, str) or not version.strip():
            raise ActionError("gap_audit requires package.json version")
        extra = [
            "--target-version", version,
            "--config", ".prd_plugin/config.json",
            "--releases", ".prd_plugin/state/releases.json",
        ]
        command.extend(extra)
        receipt_command.extend(extra)
    resolved_executable = shutil.which(command[0])
    if resolved_executable:
        command[0] = resolved_executable
    completed = subprocess.run(command, cwd=root, text=True, encoding="utf-8", errors="replace",
                               capture_output=True, check=False)
    stdout = completed.stdout
    stderr = completed.stderr
    for machine_executable in {str(sys.executable), str(command[0])}:
        if machine_executable and machine_executable != receipt_command[0]:
            stdout = stdout.replace(machine_executable, receipt_command[0])
            stderr = stderr.replace(machine_executable, receipt_command[0])
    result = {"operation": operation, "command": receipt_command, "exit_code": completed.returncode,
              "stdout": stdout[-20000:], "stderr": stderr[-20000:]}
    if completed.returncode:
        raise ActionError(f"fixed operation {operation} failed ({completed.returncode}): {stderr[-1000:].strip()}")
    return result


ACTIONS: dict[str, ActionSpec] = {
    "core.noop": ActionSpec("core.noop", _action_noop, "deterministic", "none", True, "Return success."),
    "core.echo": ActionSpec("core.echo", _action_echo, "deterministic", "none", True, "Return a resolved value."),
    "core.assert": ActionSpec("core.assert", _action_assert, "deterministic", "none", True, "Require exact equality."),
    "config.audit": ActionSpec("config.audit", _action_config_audit, "deterministic", "none", True, "Validate unified configuration."),
    "state.consistency": ActionSpec("state.consistency", _action_state_consistency, "deterministic", "none", True, "Validate canonical state references."),
    "state.gate": ActionSpec("state.gate", _action_gate_check, "deterministic", "none", True, "Run the PRD state gate."),
    "state.query": ActionSpec("state.query", _action_mcp_query, "deterministic", "none", True, "Call an allowlisted read-only state tool."),
    "state.mutate": ActionSpec("state.mutate", _action_mcp_mutate, "deterministic", "canonical_state", False, "Call an allowlisted canonical state mutation tool."),
    "requests.pull": ActionSpec("requests.pull", _action_requests_pull, "deterministic", "canonical_state", True, "Apply delivered resolutions and flush replies still pending delivery."),
    "graph.build": ActionSpec("graph.build", _action_graph_build, "deterministic", "none", True, "Build the traceability graph."),
    "staleness.audit": ActionSpec("staleness.audit", _action_staleness_audit, "deterministic", "none", True, "Audit stale records."),
    "reflection.list": ActionSpec("reflection.list", _action_reflection_list, "deterministic", "none", True, "Select configured reflection questions."),
    "verification.plan": ActionSpec("verification.plan", _action_verification_plan, "deterministic", "none", True, "Build a conservative verification plan."),
    "verification.execute": ActionSpec("verification.execute", _action_verification_execute, "external_deterministic", "process", True, "Execute a fingerprint-bound focused or widened verification plan."),
    "reporting.bundle": ActionSpec("reporting.bundle", _action_reporting_bundle, "deterministic", "none", True, "Build a source-backed reporting bundle."),
    "reporting.delegate": ActionSpec("reporting.delegate", _action_reporting_delegate, "external_deterministic", "runtime", True, "Dispatch a hash-bound reporting bundle through the configured runtime."),
    "substrate.snapshot": ActionSpec("substrate.snapshot", _action_substrate_snapshot, "deterministic", "none", True, "Project state through the substrate contract."),
    "services.audit": ActionSpec("services.audit", _action_services_audit, "deterministic", "none", True, "Validate repository service consume/provide declarations."),
    "substrate.preflight": ActionSpec("substrate.preflight", _action_substrate_preflight, "external_deterministic", "none", True, "Negotiate runtime identity, health, and declared services when enabled."),
    "substrate.enrich": ActionSpec("substrate.enrich", _action_substrate_enrich, "external_deterministic", "none", True, "Recall optional knowledge, memory, and context with local fallback."),
    "substrate.notices": ActionSpec("substrate.notices", _action_substrate_notices, "external_deterministic", "none", True, "Read unacknowledged runtime notices when explicitly enabled."),
    "substrate.telemetry": ActionSpec("substrate.telemetry", _action_substrate_telemetry, "external_deterministic", "none", True, "Read runtime telemetry without changing configuration."),
    "substrate.goals": ActionSpec("substrate.goals", _action_substrate_goals, "external_deterministic", "runtime_link_state", True, "Bind PRD goals once and observe remote state without importing it."),
    "git.status": ActionSpec("git.status", _action_git_status, "external_deterministic", "none", True, "Read exact working-tree status."),
    "git.diff_check": ActionSpec("git.diff_check", _action_git_diff_check, "external_deterministic", "none", True, "Check whitespace errors."),
    "process.fixed": ActionSpec("process.fixed", _action_fixed_process, "external_deterministic", "none", True, "Run a code-owned fixed operation."),
    "judgment.request": ActionSpec("judgment.request", None, "judgment", "none", True, "Pause for a validated external judgment."),
}


def register_action(spec: ActionSpec) -> None:
    if spec.name in ACTIONS:
        raise CatalogError(f"action already registered: {spec.name}")
    if spec.determinism not in {"deterministic", "external_deterministic", "judgment"}:
        raise CatalogError(f"invalid action determinism: {spec.determinism}")
    ACTIONS[spec.name] = spec


def action_inventory() -> dict[str, Any]:
    return {"schema_version": SCHEMA_VERSION, "actions": [
        {"name": item.name, "determinism": item.determinism, "mutation": item.mutation,
         "idempotent": item.idempotent, "description": item.description}
        for item in sorted(ACTIONS.values(), key=lambda item: item.name)
    ]}


def _config(root: Path) -> dict[str, Any]:
    config = _read_json(root / ".prd_plugin" / "config.json", {})
    raw = config.get("workflows", {}) if isinstance(config, dict) else {}
    if not isinstance(raw, dict):
        raise WorkflowError("workflows config must be an object")
    judgment = raw.get("judgment", {})
    if not isinstance(judgment, dict):
        raise WorkflowError("workflows.judgment must be an object")
    result = {
        "enabled": raw.get("enabled", True),
        "catalog_path": raw.get("catalog_path", ".prd_plugin/workflows.json"),
        "run_state_path": raw.get("run_state_path", ".prd_plugin/state/workflow-runs.json"),
        "allow_custom_definitions": raw.get("allow_custom_definitions", False),
        "allow_state_mutations": raw.get("allow_state_mutations", True),
        "enabled_ids": raw.get("enabled_ids", []),
        "max_attempts": raw.get("max_attempts", 3),
        "max_output_chars": raw.get("max_output_chars", 50000),
        "judgment": {
            "executor": judgment.get("executor", "ai-collab"),
            "profile": judgment.get("profile", "fast-capable"),
            "require_source_refs": judgment.get("require_source_refs", True),
            "fallback": judgment.get("fallback", "fail"),
        },
    }
    if not isinstance(result["enabled"], bool):
        raise WorkflowError("workflows.enabled must be boolean")
    if not result["enabled"]:
        raise WorkflowError("deterministic workflows are disabled by configuration")
    for key in ("catalog_path", "run_state_path"):
        value = result[key]
        if not isinstance(value, str) or not value or Path(value).is_absolute() or ".." in Path(value).parts:
            raise WorkflowError(f"workflows.{key} must be a safe repo-relative path")
    if not isinstance(result["allow_custom_definitions"], bool) or not isinstance(result["allow_state_mutations"], bool):
        raise WorkflowError("workflow allow flags must be boolean")
    if not isinstance(result["enabled_ids"], list) or any(not isinstance(item, str) for item in result["enabled_ids"]):
        raise WorkflowError("workflows.enabled_ids must be an array of strings")
    if isinstance(result["max_attempts"], bool) or not isinstance(result["max_attempts"], int) or not 1 <= result["max_attempts"] <= 20:
        raise WorkflowError("workflows.max_attempts must be 1..20")
    if isinstance(result["max_output_chars"], bool) or not isinstance(result["max_output_chars"], int) or not 1000 <= result["max_output_chars"] <= 1000000:
        raise WorkflowError("workflows.max_output_chars must be 1000..1000000")
    if result["judgment"]["fallback"] not in {"fail", "main"}:
        raise WorkflowError("workflows.judgment.fallback must be fail or main")
    return result


def _catalog_path(root: Path) -> Path:
    return root / _config(root)["catalog_path"]


def _run_path(root: Path) -> Path:
    return root / _config(root)["run_state_path"]


def _validate_input_spec(spec: Any, where: str, findings: list[dict[str, str]]) -> None:
    if not isinstance(spec, dict) or spec.get("type") not in {"string", "integer", "number", "boolean", "array", "object"}:
        findings.append({"severity": "high", "location": where, "message": "input must declare a supported type"})


def audit_catalog(repo_root: str | Path = ".") -> dict[str, Any]:
    root = Path(repo_root).resolve()
    findings: list[dict[str, str]] = []
    try:
        catalog = _read_json(_catalog_path(root))
    except WorkflowError as exc:
        return {"status": "error", "findings": [{"severity": "high", "location": "catalog", "message": str(exc)}]}
    if not isinstance(catalog, dict) or catalog.get("schema_version") != SCHEMA_VERSION:
        findings.append({"severity": "high", "location": "catalog", "message": f"schema_version must be {SCHEMA_VERSION}"})
    rows = catalog.get("workflows") if isinstance(catalog, dict) else None
    if not isinstance(rows, list):
        findings.append({"severity": "high", "location": "catalog", "message": "workflows must be an array"})
        rows = []
    workflow_ids: set[str] = set()
    config = _config(root)
    for wi, workflow in enumerate(rows):
        where = f"workflows[{wi}]"
        if not isinstance(workflow, dict):
            findings.append({"severity": "high", "location": where, "message": "workflow must be an object"})
            continue
        wid = workflow.get("id")
        if not isinstance(wid, str) or not re.fullmatch(r"[a-z][a-z0-9_.-]*", wid):
            findings.append({"severity": "high", "location": where, "message": "workflow id is invalid"})
        elif wid in workflow_ids:
            findings.append({"severity": "high", "location": where, "message": f"duplicate workflow id {wid}"})
        else:
            workflow_ids.add(wid)
        if workflow.get("origin", "shipped") == "custom" and not config["allow_custom_definitions"]:
            findings.append({"severity": "high", "location": where, "message": "custom workflow definitions are disabled"})
        if not isinstance(workflow.get("version"), int) or workflow.get("version", 0) < 1:
            findings.append({"severity": "high", "location": where, "message": "version must be a positive integer"})
        if not isinstance(workflow.get("description"), str) or not workflow.get("description", "").strip():
            findings.append({"severity": "medium", "location": where, "message": "description is required"})
        inputs = workflow.get("inputs", {})
        if not isinstance(inputs, dict):
            findings.append({"severity": "high", "location": where, "message": "inputs must be an object"})
        else:
            for name, spec in inputs.items():
                _validate_input_spec(spec, f"{where}.inputs.{name}", findings)
        steps = workflow.get("steps")
        if not isinstance(steps, list):
            findings.append({"severity": "high", "location": where, "message": "steps must be an array"})
            steps = []
        step_ids: set[str] = set()
        for si, step in enumerate(steps):
            sw = f"{where}.steps[{si}]"
            if not isinstance(step, dict):
                findings.append({"severity": "high", "location": sw, "message": "step must be an object"})
                continue
            sid = step.get("id")
            if not isinstance(sid, str) or not re.fullmatch(r"[a-z][a-z0-9_-]*", sid):
                findings.append({"severity": "high", "location": sw, "message": "step id is invalid"})
            elif sid in step_ids:
                findings.append({"severity": "high", "location": sw, "message": f"duplicate step id {sid}"})
            else:
                step_ids.add(sid)
            if step.get("action") not in ACTIONS:
                findings.append({"severity": "high", "location": sw, "message": f"unknown action {step.get('action')!r}"})
            if "with" in step and not isinstance(step["with"], dict):
                findings.append({"severity": "high", "location": sw, "message": "with must be an object"})
        postconditions = workflow.get("postconditions", [])
        if not isinstance(postconditions, list):
            findings.append({"severity": "high", "location": where, "message": "postconditions must be an array"})
        else:
            for pi, cond in enumerate(postconditions):
                if not isinstance(cond, dict) or cond.get("type") != "step_status" or cond.get("step") not in step_ids or cond.get("equals") != "completed":
                    findings.append({"severity": "high", "location": f"{where}.postconditions[{pi}]", "message": "unsupported or invalid postcondition"})
    counts = {level: sum(1 for row in findings if row["severity"] == level) for level in ("critical", "high", "medium", "low")}
    return {"status": "ok" if not findings else "error", "workflow_count": len(rows), "action_count": len(ACTIONS), "summary": counts, "findings": findings}


def load_catalog(repo_root: str | Path = ".") -> dict[str, Any]:
    root = Path(repo_root).resolve()
    report = audit_catalog(root)
    if report["status"] != "ok":
        raise CatalogError("; ".join(row["message"] for row in report["findings"]))
    return _read_json(_catalog_path(root))


def list_workflows(repo_root: str | Path = ".") -> dict[str, Any]:
    catalog = load_catalog(repo_root)
    enabled = _config(Path(repo_root).resolve())["enabled_ids"]
    return {"schema_version": SCHEMA_VERSION, "catalog_version": catalog.get("catalog_version"), "workflows": [
        {"id": row["id"], "version": row["version"], "description": row["description"],
         "enabled": not enabled or row["id"] in enabled,
         "step_count": len(row["steps"]), "judgment_steps": sum(1 for step in row["steps"] if step["action"] == "judgment.request")}
        for row in catalog["workflows"]
    ]}


def _workflow(root: Path, workflow_id: str) -> dict[str, Any]:
    enabled = _config(root)["enabled_ids"]
    if enabled and workflow_id not in enabled:
        raise CatalogError(f"workflow {workflow_id!r} is disabled by workflows.enabled_ids")
    for row in load_catalog(root)["workflows"]:
        if row["id"] == workflow_id:
            return copy.deepcopy(row)
    raise CatalogError(f"unknown workflow {workflow_id!r}")


def _check_type(value: Any, kind: str) -> bool:
    return {"string": lambda: isinstance(value, str), "integer": lambda: isinstance(value, int) and not isinstance(value, bool),
            "number": lambda: isinstance(value, (int, float)) and not isinstance(value, bool), "boolean": lambda: isinstance(value, bool),
            "array": lambda: isinstance(value, list), "object": lambda: isinstance(value, dict)}[kind]()


def _inputs(workflow: dict[str, Any], supplied: dict[str, Any]) -> dict[str, Any]:
    if not isinstance(supplied, dict):
        raise WorkflowError("inputs must be an object")
    specs = workflow.get("inputs", {})
    unknown = sorted(set(supplied) - set(specs))
    if unknown:
        raise WorkflowError(f"unknown input(s): {unknown}")
    result = {}
    for name, spec in specs.items():
        if name in supplied:
            value = supplied[name]
        elif "default" in spec:
            value = copy.deepcopy(spec["default"])
        elif spec.get("required"):
            raise WorkflowError(f"missing required input {name!r}")
        else:
            continue
        if not _check_type(value, spec["type"]):
            raise WorkflowError(f"input {name!r} must be {spec['type']}")
        if "allowed" in spec and value not in spec["allowed"]:
            raise WorkflowError(f"input {name!r} must be one of {spec['allowed']}")
        result[name] = value
    return result


def _dig(value: Any, path: str | None) -> Any:
    if not path:
        return value
    for part in path.split("."):
        if not isinstance(value, dict) or part not in value:
            raise WorkflowError(f"reference path does not exist: {path}")
        value = value[part]
    return value


def _resolve(value: Any, inputs: dict[str, Any], steps: dict[str, Any]) -> Any:
    if isinstance(value, str):
        match = REF_RE.fullmatch(value)
        if not match:
            return value
        scope, name, tail = match.groups()
        if scope == "inputs":
            return copy.deepcopy(_dig(inputs, ".".join(p for p in (name, tail) if p)))
        if not name or name not in steps:
            raise WorkflowError(f"step reference does not exist: {value}")
        return copy.deepcopy(_dig(steps[name], tail))
    if isinstance(value, list):
        return [_resolve(item, inputs, steps) for item in value]
    if isinstance(value, dict):
        return {key: _resolve(item, inputs, steps) for key, item in value.items()}
    return value


def plan_workflow(repo_root: str | Path, workflow_id: str, inputs: dict[str, Any]) -> dict[str, Any]:
    root = Path(repo_root).resolve()
    workflow = _workflow(root, workflow_id)
    normalized = _inputs(workflow, inputs)
    known: dict[str, Any] = {}
    planned = []
    for step in workflow["steps"]:
        try:
            args = _resolve(step.get("with", {}), normalized, known)
        except WorkflowError:
            args = copy.deepcopy(step.get("with", {}))
        planned.append({"id": step["id"], "action": step["action"], "with": args,
                        "determinism": ACTIONS[step["action"]].determinism,
                        "mutation": ACTIONS[step["action"]].mutation})
        known[step["id"]] = {"status": "planned", "output": {}}
    plan = {"schema_version": SCHEMA_VERSION, "workflow_id": workflow_id, "workflow_version": workflow["version"],
            "definition_hash": _hash(workflow), "inputs": normalized, "input_hash": _hash(normalized),
            "steps": planned, "postconditions": workflow.get("postconditions", [])}
    plan["plan_hash"] = _hash(plan)
    return plan


def load_runs(repo_root: str | Path = ".") -> dict[str, Any]:
    root = Path(repo_root).resolve()
    data = _read_json(_run_path(root), {"schema_version": SCHEMA_VERSION, "runs": []})
    if not isinstance(data, dict) or data.get("schema_version") != SCHEMA_VERSION or not isinstance(data.get("runs"), list):
        raise WorkflowError("workflow run state must contain schema_version 1.0 and a runs array")
    ids = [run.get("id") for run in data["runs"] if isinstance(run, dict)]
    if len(ids) != len(set(ids)):
        raise WorkflowError("workflow run state contains duplicate IDs")
    return data


def _allocate(root: Path, state: dict[str, Any]) -> str:
    registry_path = root / ".prd_plugin" / "ids" / "registry.json"
    registry = _read_json(registry_path, {"version": 1, "next": {}})
    if not isinstance(registry.get("next"), dict):
        raise WorkflowError("ID registry has no next object")
    used = {run.get("id") for run in state["runs"]}
    number = registry["next"].get("WFR", 1)
    if not isinstance(number, int) or number < 1:
        raise WorkflowError("ID registry next.WFR must be a positive integer")
    while f"WFR-{number:03d}" in used:
        number += 1
    result = f"WFR-{number:03d}"
    registry["next"]["WFR"] = number + 1
    _atomic_json(registry_path, registry)
    return result


def get_run(repo_root: str | Path, run_id: str) -> dict[str, Any]:
    for run in load_runs(repo_root)["runs"]:
        if run.get("id") == run_id:
            return copy.deepcopy(run)
    raise WorkflowError(f"unknown workflow run {run_id}")


def bind_runtime_dispatch(
    repo_root: str | Path,
    run_id: str,
    dispatch: dict[str, Any],
) -> dict[str, Any]:
    """Bind one external capsule to a waiting judgment without exposing its payload.

    The binding is idempotent and deliberately stores only stable identifiers and
    hashes. The external worker must return through ``resume_run``; it never gains
    direct canonical-state authority.
    """
    root = Path(repo_root).resolve()
    allowed = ("capsule_id", "operation_id", "receipt_hash", "request_hash", "dispatched_at")
    normalized = {key: dispatch[key] for key in allowed if isinstance(dispatch.get(key), str) and dispatch[key]}
    if not normalized.get("capsule_id"):
        raise WorkflowError("runtime dispatch binding requires capsule_id")
    with _StateLock(root):
        run = get_run(root, run_id)
        if run.get("status") != "waiting_judgment":
            raise TransitionError(f"cannot bind runtime dispatch to {run.get('status')} run")
        current = run.get("runtime_dispatch")
        if isinstance(current, dict):
            if current.get("capsule_id") != normalized["capsule_id"]:
                raise TransitionError("workflow run is already bound to a different runtime capsule")
            return copy.deepcopy(run)
        pending_hash = run.get("pending_judgment", {}).get("request_hash")
        if normalized.get("request_hash") and normalized["request_hash"] != pending_hash:
            raise JudgmentError("runtime dispatch request_hash does not match pending judgment")
        normalized["request_hash"] = str(pending_hash or "")
        normalized.setdefault("dispatched_at", _now())
        run["runtime_dispatch"] = normalized
        run["updated_at"] = _now()
        _save_run_unlocked(root, run)
        return copy.deepcopy(run)


def _save_run_unlocked(root: Path, run: dict[str, Any]) -> None:
    state = load_runs(root)
    for index, existing in enumerate(state["runs"]):
        if existing.get("id") == run["id"]:
            state["runs"][index] = run
            break
    else:
        state["runs"].append(run)
    _atomic_json(_run_path(root), state)


def _save_run(root: Path, run: dict[str, Any]) -> None:
    with _StateLock(root):
        _save_run_unlocked(root, run)


def _schema_validate(value: Any, schema: dict[str, Any], where: str = "result") -> None:
    kind = schema.get("type")
    if kind and not _check_type(value, kind):
        raise JudgmentError(f"{where} must be {kind}")
    if kind == "object":
        for name in schema.get("required", []):
            if name not in value:
                raise JudgmentError(f"{where}.{name} is required")
        props = schema.get("properties", {})
        for name, child in props.items():
            if name in value:
                _schema_validate(value[name], child, f"{where}.{name}")
    if kind == "array" and "items" in schema:
        for index, item in enumerate(value):
            _schema_validate(item, schema["items"], f"{where}[{index}]")


def _judgment_request(root: Path, run: dict[str, Any], step: dict[str, Any], args: dict[str, Any]) -> dict[str, Any]:
    policy = _config(root)["judgment"]
    refs = args.get("source_refs", [])
    if not isinstance(refs, list) or any(not isinstance(ref, str) or not ref for ref in refs):
        raise ActionError("judgment source_refs must be an array of non-empty strings")
    if policy["require_source_refs"] and not refs:
        raise ActionError("judgment requires source_refs")
    schema = args.get("result_schema")
    if not isinstance(schema, dict) or schema.get("type") != "object":
        raise ActionError("judgment result_schema must describe an object")
    request = {"contract_version": 1, "run_id": run["id"], "step_id": step["id"],
               "task": args.get("task"), "instructions": args.get("instructions"),
               "source_refs": refs, "source_hashes": {}, "result_schema": schema,
               "executor": policy["executor"], "profile": policy["profile"],
               "constraints": {"no_project_state_writes": True, "require_source_refs": policy["require_source_refs"]}}
    for ref in refs:
        path = root / ref
        if path.is_file() and path.resolve().is_relative_to(root):
            request["source_hashes"][ref] = hashlib.sha256(path.read_bytes()).hexdigest()
    request["request_hash"] = _hash(request)
    return request


def _auto_dispatch_judgment(root: Path, run_id: str) -> dict[str, Any]:
    """Dispatch a waiting judgment when the explicit Substrate policy allows it.

    Runtime unavailability never fabricates completion. The run remains waiting
    with a bounded error so it can be retried or fulfilled by the configured
    fallback executor.
    """
    import prd_substrate

    policy = prd_substrate.effective_policy(root)
    if (
        not policy["active"]
        or policy["effective_mode"] != "coordinate"
        or "workflow_judgment" not in policy["effective_capabilities"]
        or not policy["configured"]["automation"]["judgment_dispatch"]
    ):
        return get_run(root, run_id)
    try:
        import prd_runtime_worker

        prd_runtime_worker.dispatch_judgment(root, run_id)
    except Exception as exc:
        with _StateLock(root):
            persisted = get_run(root, run_id)
            persisted["runtime_dispatch_error"] = {
                "type": exc.__class__.__name__, "message": str(exc), "at": _now()
            }
            persisted["updated_at"] = _now()
            _save_run_unlocked(root, persisted)
    return get_run(root, run_id)


def _postconditions(run: dict[str, Any], workflow: dict[str, Any]) -> None:
    steps = {step["id"]: step for step in run["steps"]}
    for condition in workflow.get("postconditions", []):
        actual = steps[condition["step"]]["status"]
        if actual != condition["equals"]:
            raise ActionError(f"postcondition failed for {condition['step']}: expected {condition['equals']}, got {actual}")


def _bounded_output(root: Path, output: dict[str, Any]) -> tuple[dict[str, Any], str]:
    full_hash = _hash(output)
    encoded = _canonical(output)
    limit = _config(root)["max_output_chars"]
    if len(encoded) <= limit:
        return output, full_hash
    summary = output.get("summary") if isinstance(output.get("summary"), (str, int, float, bool, dict, list)) else None
    return {"omitted": True, "reason": "output_exceeds_configured_limit", "full_output_hash": full_hash,
            "full_output_chars": len(encoded), "keys": sorted(output), "summary": summary}, full_hash


def _execute(root: Path, run: dict[str, Any]) -> dict[str, Any]:
    workflow = _workflow(root, run["workflow_id"])
    if _hash(workflow) != run["definition_hash"]:
        run["status"] = "blocked"
        run["error"] = {"type": "definition_changed", "message": "workflow definition changed after run creation"}
        run["updated_at"] = _now()
        _save_run(root, run)
        return run
    run["status"] = "running"
    run["updated_at"] = _now()
    _save_run(root, run)
    context_steps = {step["id"]: step for step in run["steps"]}
    try:
        for index, definition in enumerate(workflow["steps"]):
            step = run["steps"][index]
            if step["status"] == "completed":
                continue
            args = _resolve(definition.get("with", {}), run["inputs"], context_steps)
            step["resolved_input_hash"] = _hash(args)
            step["started_at"] = _now()
            if definition["action"] == "judgment.request":
                request = _judgment_request(root, run, step, args)
                step["status"] = "waiting_judgment"
                step["judgment_request"] = request
                run["status"] = "waiting_judgment"
                run["pending_judgment"] = request
                run["updated_at"] = _now()
                _save_run(root, run)
                return _auto_dispatch_judgment(root, run["id"])
            spec = ACTIONS[definition["action"]]
            if spec.handler is None:
                raise ActionError(f"action has no handler: {spec.name}")
            if not spec.idempotent:
                step["status"] = "executing_non_idempotent"
                step["execution_token"] = _hash({"run": run["id"], "step": step["id"], "attempt": run["attempt"], "input": args})
                run["updated_at"] = _now()
                _save_run(root, run)
            output = spec.handler(root, args, {"run": run, "step": step})
            if not isinstance(output, dict):
                raise ActionError(f"action {spec.name} returned a non-object")
            stored_output, full_output_hash = _bounded_output(root, output)
            step.update({"status": "completed", "output": stored_output, "output_hash": full_output_hash, "completed_at": _now()})
            run["receipts"].append({"step_id": step["id"], "action": step["action"],
                                    "input_hash": step["resolved_input_hash"], "output_hash": step["output_hash"],
                                    "completed_at": step["completed_at"]})
            run["updated_at"] = _now()
            _save_run(root, run)
        _postconditions(run, workflow)
        run["status"] = "completed"
        run["completed_at"] = _now()
        run.pop("pending_judgment", None)
        run.pop("error", None)
    except Exception as exc:
        if run["status"] not in {"blocked", "waiting_judgment"}:
            uncertain = any(step.get("status") == "executing_non_idempotent" for step in run["steps"])
            run["status"] = "blocked" if uncertain else "failed"
            run["error"] = {"type": exc.__class__.__name__, "message": str(exc)}
            for step in run["steps"]:
                if step.get("started_at") and step["status"] in {"pending", "executing_non_idempotent"}:
                    step["status"] = "outcome_unknown" if uncertain else "failed"
                    break
    run["updated_at"] = _now()
    _save_run(root, run)
    return run


def run_workflow(repo_root: str | Path, workflow_id: str, inputs: dict[str, Any], *, idempotency_key: str = "") -> dict[str, Any]:
    root = Path(repo_root).resolve()
    plan = plan_workflow(root, workflow_id, inputs)
    with _StateLock(root):
        state = load_runs(root)
        if idempotency_key:
            for existing in state["runs"]:
                if existing.get("workflow_id") == workflow_id and existing.get("idempotency_key") == idempotency_key:
                    if existing.get("input_hash") != plan["input_hash"]:
                        raise WorkflowError("idempotency key already belongs to different inputs")
                    replay = copy.deepcopy(existing)
                    replay["idempotent_replay"] = True
                    return replay
        run_id = _allocate(root, state)
        now = _now()
        run = {"id": run_id, "schema_version": SCHEMA_VERSION, "workflow_id": workflow_id,
               "workflow_version": plan["workflow_version"], "definition_hash": plan["definition_hash"],
               "plan_hash": plan["plan_hash"], "input_hash": plan["input_hash"], "inputs": plan["inputs"],
               "idempotency_key": idempotency_key or None, "status": "pending", "attempt": 1,
               "created_at": now, "updated_at": now, "receipts": [],
               "steps": [{"id": step["id"], "action": step["action"], "status": "pending"} for step in plan["steps"]]}
        state["runs"].append(run)
        _atomic_json(_run_path(root), state)
    return _execute(root, run)


def resume_run(repo_root: str | Path, run_id: str, result_envelope: dict[str, Any]) -> dict[str, Any]:
    root = Path(repo_root).resolve()
    with _StateLock(root):
        run = get_run(root, run_id)
        if run["status"] != "waiting_judgment":
            raise TransitionError(f"cannot resume {run['status']} run")
        request = run.get("pending_judgment", {})
        if result_envelope.get("request_hash") != request.get("request_hash"):
            raise JudgmentError("judgment result request_hash does not match pending request")
        if result_envelope.get("executor") != request.get("executor") or result_envelope.get("profile") != request.get("profile"):
            raise JudgmentError("judgment executor/profile does not match pending request")
        result = result_envelope.get("result")
        _schema_validate(result, request["result_schema"])
        refs = result.get("source_refs", []) if isinstance(result, dict) else []
        if refs and (not isinstance(refs, list) or any(ref not in request["source_refs"] for ref in refs)):
            raise JudgmentError("judgment result contains unknown source_refs")
        index = next(i for i, step in enumerate(run["steps"]) if step["id"] == request["step_id"])
        step = run["steps"][index]
        step.update({"status": "completed", "output": result, "output_hash": _hash(result), "completed_at": _now(),
                     "judgment_result": {"request_hash": request["request_hash"], "executor": request["executor"], "profile": request["profile"]}})
        run["receipts"].append({"step_id": step["id"], "action": step["action"], "input_hash": step["resolved_input_hash"],
                                "output_hash": step["output_hash"], "request_hash": request["request_hash"], "completed_at": step["completed_at"]})
        run.pop("pending_judgment", None)
        run["status"] = "running"
        run["updated_at"] = _now()
        _save_run_unlocked(root, run)
    return _execute(root, run)


def cancel_run(repo_root: str | Path, run_id: str, reason: str) -> dict[str, Any]:
    root = Path(repo_root).resolve()
    with _StateLock(root):
        run = get_run(root, run_id)
        if run["status"] in TERMINAL:
            raise TransitionError(f"cannot cancel {run['status']} run")
        run["status"] = "cancelled"
        run["cancelled_at"] = _now()
        run["cancel_reason"] = reason
        run["updated_at"] = _now()
        _save_run_unlocked(root, run)
        return run


def retry_run(repo_root: str | Path, run_id: str) -> dict[str, Any]:
    root = Path(repo_root).resolve()
    with _StateLock(root):
        run = get_run(root, run_id)
        if run["status"] not in {"failed", "blocked"}:
            raise TransitionError(f"cannot retry {run['status']} run")
        if any(step.get("status") == "outcome_unknown" for step in run["steps"]):
            raise TransitionError("cannot retry an outcome-unknown non-idempotent step; reconcile canonical state first")
        if run["attempt"] >= _config(root)["max_attempts"]:
            raise TransitionError("workflow retry limit reached")
        first_incomplete = next((i for i, step in enumerate(run["steps"]) if step["status"] != "completed"), len(run["steps"]))
        for step in run["steps"][first_incomplete:]:
            run["steps"][run["steps"].index(step)] = {"id": step["id"], "action": step["action"], "status": "pending"}
        run["attempt"] += 1
        run["status"] = "pending"
        run.pop("error", None)
        run["updated_at"] = _now()
        _save_run_unlocked(root, run)
    return _execute(root, run)


def _json_arg(value: str) -> dict[str, Any]:
    try:
        result = json.loads(value)
    except json.JSONDecodeError as exc:
        raise argparse.ArgumentTypeError(str(exc)) from exc
    if not isinstance(result, dict):
        raise argparse.ArgumentTypeError("value must be a JSON object")
    return result


def _parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--repo-root", default=".")
    sub = parser.add_subparsers(dest="command", required=True)
    for name in ("list", "actions", "audit"):
        item = sub.add_parser(name)
        item.add_argument("--json", action="store_true")
    plan = sub.add_parser("plan"); plan.add_argument("workflow"); plan.add_argument("--inputs", type=_json_arg, default={}); plan.add_argument("--json", action="store_true")
    run = sub.add_parser("run"); run.add_argument("workflow"); run.add_argument("--inputs", type=_json_arg, default={}); run.add_argument("--idempotency-key", default=""); run.add_argument("--json", action="store_true")
    status = sub.add_parser("status"); status.add_argument("run_id"); status.add_argument("--json", action="store_true")
    resume = sub.add_parser("resume"); resume.add_argument("run_id"); resume.add_argument("--result", type=_json_arg, required=True); resume.add_argument("--json", action="store_true")
    cancel = sub.add_parser("cancel"); cancel.add_argument("run_id"); cancel.add_argument("--reason", required=True); cancel.add_argument("--json", action="store_true")
    retry = sub.add_parser("retry"); retry.add_argument("run_id"); retry.add_argument("--json", action="store_true")
    return parser


def main(argv: list[str] | None = None) -> int:
    args = _parser().parse_args(argv)
    root = Path(args.repo_root)
    try:
        if args.command == "list": result = list_workflows(root)
        elif args.command == "actions": result = action_inventory()
        elif args.command == "audit": result = audit_catalog(root)
        elif args.command == "plan": result = plan_workflow(root, args.workflow, args.inputs)
        elif args.command == "run": result = run_workflow(root, args.workflow, args.inputs, idempotency_key=args.idempotency_key)
        elif args.command == "status": result = get_run(root, args.run_id)
        elif args.command == "resume": result = resume_run(root, args.run_id, args.result)
        elif args.command == "cancel": result = cancel_run(root, args.run_id, args.reason)
        else: result = retry_run(root, args.run_id)
        # Workflow receipts may contain arbitrary subprocess output. Escaping
        # non-ASCII keeps the JSON printable on Windows cp1252 consoles while
        # preserving the exact Unicode value for JSON consumers.
        print(json.dumps(result, indent=2, ensure_ascii=True))
        if args.command == "audit" and result["status"] != "ok": return 1
        return 0
    except WorkflowError as exc:
        print(json.dumps({"status": "error", "error": str(exc)}, indent=2), file=sys.stderr)
        return 2


if __name__ == "__main__":
    raise SystemExit(main())