evaluate_state.py

#!/usr/bin/env python3
"""Evaluate one GitHub Codex PR-loop snapshot without performing mutations.

The caller supplies already-fetched GitHub and Automation state as JSON. This
module validates the complete snapshot, applies the bounded lifecycle policy,
and returns one permitted next action. It has no third-party dependencies and
does not access Git, GitHub, the network, or Codex Automations itself.
"""

from __future__ import annotations

import argparse
import hashlib
import json
import re
import sys
import unicodedata
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any


DEFAULT_POLICY_PATH = Path(__file__).resolve().parents[1] / "references" / "loop-policy.json"
FULL_SHA = re.compile(r"^[0-9a-f]{40}$")
SHA256 = re.compile(r"^[0-9a-f]{64}$")
RUN_ID = re.compile(r"^[0-9a-f]{32}$")
REPOSITORY = re.compile(r"^[a-z0-9][a-z0-9-]{0,38}/[a-z0-9._-]{1,100}$")
REQUEST_KEY = re.compile(r"^[1-9][0-9]*:[0-9a-f]{40}$")
WINDOWS_DEVICE_STEM = re.compile(
    r"^(?:con|prn|aux|nul|clock\$|conin\$|conout\$|com[1-9]|lpt[1-9])$",
    re.IGNORECASE,
)
WINDOWS_GIT_ALIAS = re.compile(r"^\.?git~[1-9][0-9]*$", re.IGNORECASE)
AUTOMATION_STATUSES = {"ACTIVE", "PAUSED"}
AUTOMATION_PROMPT_PLACEHOLDERS = (
    "AUTOMATION_KEY",
    "BOOTSTRAP_CHECKPOINT_JSON",
)
AUTOMATION_PROMPT_TEMPLATE = "\n".join(
    (
        "Continue $github-codex-pr-loop from the exact checkpoint below. "
        "Re-fetch all external state and follow only the installed evaluator's emitted action.",
        "Loop key: {{AUTOMATION_KEY}}",
        "Bootstrap checkpoint JSON: {{BOOTSTRAP_CHECKPOINT_JSON}}",
    )
)
POLICY_V1 = {
    "schema_version": 1,
    "supported_host": "github.com",
    "same_repository_pull_requests_only": True,
    "minimum_python_version": "3.10",
    "manifest_hash": "sha256-canonical-json-v1",
    "git_execution_profile": "sanitized-gh-https-v1",
    "run_id_bits": 128,
    "trusted_codex_bot_database_id": 199175422,
    "heartbeat_interval_minutes": 2,
    "heartbeat_rrule": "FREQ=MINUTELY;INTERVAL=2",
    "no_response_timeout_minutes": 60,
    "max_fix_rounds": 5,
    "default_mode": "autonomous",
    "modes": ["autonomous", "approval-gated", "report-only"],
    "gate_states": ["pass", "fail", "pending", "unknown"],
    "terminal_message_template": (
        "No remaining actionable Codex findings for SHA `<full-sha>`."
    ),
}
RESPONSE_SOURCES = {"issue-comment", "review", "review-comment"}
RESPONSE_FIELDS = {
    "source",
    "id",
    "bot_type",
    "bot_id",
    "body_sha256",
    "content_version_at",
    "created_at",
    "commit_sha",
    "trigger_request_comment_id",
    "zero_findings_signal",
}
GATE_STATES = {"pass", "fail", "pending", "unknown"}
REQUEST_STATES = {
    "pending",
    "awaiting-remediation-approval",
    "awaiting-finalization-approval",
    "processing",
}
REQUIRED_STATE_FIELDS = {
    "actionable_codex_threads",
    "actionable_thread_snapshots",
    "approval_gate",
    "approved_finalization_plan_hash",
    "approved_remediation_plan_hash",
    "approved_request_key",
    "authenticated_user_id",
    "automation_key",
    "automations",
    "branch_protection_gate",
    "base_branch",
    "check_gate",
    "claimed_finalization_plan_hash",
    "claimed_finalization_request_key",
    "completed_request_keys",
    "current_head_sha",
    "current_base_sha",
    "eligible_codex_threads",
    "eligible_thread_snapshots",
    "finalization_plan_hash",
    "finalization_approved_request_key",
    "fix_rounds",
    "head_branch_exists",
    "head_branch",
    "host",
    "intervening_review_requests",
    "mergeability_gate",
    "mode",
    "non_codex_rooted_threads",
    "now",
    "owned_automation_id",
    "policy_schema_version",
    "post_finalization",
    "processing_claim_token",
    "pr_number",
    "pr_base_branch",
    "pr_head_branch",
    "pr_is_draft",
    "pr_state",
    "finalization_manifest",
    "execution_boundary",
    "remediation_manifest",
    "remediation_plan_hash",
    "remote_name",
    "repository",
    "request_author_id",
    "request_comment_id",
    "request_created_at",
    "request_base_sha",
    "request_head_sha",
    "request_state",
    "responses",
    "run_id",
    "same_repository_pr",
    "terminal_error",
}


def _timestamp(value: Any, field: str) -> datetime:
    if not isinstance(value, str) or not value:
        raise ValueError(f"{field} must be a non-empty ISO-8601 timestamp")
    normalized = value[:-1] + "+00:00" if value.endswith("Z") else value
    try:
        parsed = datetime.fromisoformat(normalized)
    except ValueError as exc:
        raise ValueError(f"{field} must be a valid ISO-8601 timestamp") from exc
    if parsed.tzinfo is None:
        raise ValueError(f"{field} must include a timezone")
    return parsed.astimezone(timezone.utc)


def _sha(value: Any, field: str) -> str:
    if not isinstance(value, str) or not FULL_SHA.fullmatch(value):
        raise ValueError(f"{field} must be a lowercase 40-character Git SHA")
    return value


def _nonnegative_int(value: Any, field: str) -> int:
    if isinstance(value, bool) or not isinstance(value, int) or value < 0:
        raise ValueError(f"{field} must be a non-negative integer")
    return value


def _positive_int(value: Any, field: str) -> int:
    result = _nonnegative_int(value, field)
    if result == 0:
        raise ValueError(f"{field} must be positive")
    return result


def _boolean(value: Any, field: str) -> bool:
    if not isinstance(value, bool):
        raise ValueError(f"{field} must be a boolean")
    return value


def _optional_request_key(value: Any, field: str) -> str | None:
    if value is None:
        return None
    if not isinstance(value, str) or not REQUEST_KEY.fullmatch(value):
        raise ValueError(f"{field} must be a canonical request key or null")
    return value


def _optional_sha256(value: Any, field: str) -> str | None:
    if value is None:
        return None
    if not isinstance(value, str) or not SHA256.fullmatch(value):
        raise ValueError(f"{field} must be a lowercase SHA-256 digest or null")
    return value


def _gate(value: Any, field: str) -> str:
    if value not in GATE_STATES:
        raise ValueError(f"{field} must be one of: {', '.join(sorted(GATE_STATES))}")
    return value


def _manifest_digest(value: Any, field: str) -> str | None:
    if value is None:
        return None
    if not isinstance(value, dict):
        raise ValueError(f"{field} must be an object or null")

    def validate(node: Any, path: str) -> None:
        if node is None or isinstance(node, (bool, int, str)):
            return
        if isinstance(node, float):
            raise ValueError(f"{path} must not contain floating-point values")
        if isinstance(node, list):
            for index, item in enumerate(node):
                validate(item, f"{path}[{index}]")
            return
        if isinstance(node, dict):
            if not all(isinstance(key, str) for key in node):
                raise ValueError(f"{path} object keys must be strings")
            for key, item in node.items():
                validate(item, f"{path}.{key}")
            return
        raise ValueError(f"{path} contains an unsupported JSON value")

    validate(value, field)
    canonical = json.dumps(value, ensure_ascii=False, separators=(",", ":"), sort_keys=True)
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()


def _exact_fields(value: Any, fields: set[str], field: str) -> dict[str, Any]:
    if not isinstance(value, dict):
        raise ValueError(f"{field} must be an object")
    actual = set(value)
    if actual != fields:
        missing = sorted(fields - actual)
        extra = sorted(actual - fields)
        details = []
        if missing:
            details.append(f"missing {', '.join(missing)}")
        if extra:
            details.append(f"unexpected {', '.join(extra)}")
        raise ValueError(f"{field} has invalid fields: {'; '.join(details)}")
    return value


def _ref_name(value: Any, field: str) -> str:
    if not isinstance(value, str) or not value:
        raise ValueError(f"{field} must be a non-empty Git ref name")
    forbidden = set(" ~^:?*[\\")
    if (
        value.startswith("-")
        or value.startswith("/")
        or value.endswith("/")
        or value.endswith(".")
        or "//" in value
        or ".." in value
        or "@{" in value
        or any(ord(char) < 0x20 or ord(char) == 0x7F or char in forbidden for char in value)
        or any(
            not component
            or component.startswith(".")
            or component.endswith(".lock")
            for component in value.split("/")
        )
    ):
        raise ValueError(f"{field} must be a normalized Git ref name")
    return value


def _manifest_paths(value: Any, field: str, *, allow_empty: bool) -> list[str]:
    if not isinstance(value, list) or (not allow_empty and not value):
        qualifier = "a list" if allow_empty else "a non-empty list"
        raise ValueError(f"{field} must be {qualifier}")
    if not all(isinstance(path, str) and path for path in value):
        raise ValueError(f"{field} must contain non-empty strings")
    if value != sorted(set(value)):
        raise ValueError(f"{field} must be sorted and unique")
    for path in value:
        parts = path.split("/")
        windows_parts = [unicodedata.normalize("NFKC", part) for part in parts]
        windows_stems = [part.split(".", 1)[0].rstrip(" .") for part in windows_parts]
        normalized_first = windows_parts[0].rstrip(" .")
        if (
            path.startswith("/")
            or "\\" in path
            or "\x00" in path
            or ":" in path
            or any(part in {"", ".", ".."} for part in parts)
            or any(any(ord(char) < 0x20 or ord(char) == 0x7F for char in part) for part in parts)
            or any(part.endswith((" ", ".")) for part in parts)
            or any(WINDOWS_DEVICE_STEM.fullmatch(stem) for stem in windows_stems)
            or normalized_first.casefold() == ".git"
            or WINDOWS_GIT_ALIAS.fullmatch(normalized_first) is not None
        ):
            raise ValueError(f"{field} must be normalized repository-relative paths")
    return value


def _manifest_commands(value: Any, field: str) -> list[dict[str, Any]]:
    if not isinstance(value, list):
        raise ValueError(f"{field} must be a list")
    for index, command in enumerate(value):
        command = _exact_fields(
            command, {"argv", "definition_sha256"}, f"{field}[{index}]"
        )
        argv = command["argv"]
        if (
            not isinstance(argv, list)
            or not argv
            or not all(isinstance(argument, str) for argument in argv)
            or not argv[0]
        ):
            raise ValueError(f"{field}[{index}].argv must be a non-empty string list")
        digest = command["definition_sha256"]
        if not isinstance(digest, str) or not SHA256.fullmatch(digest):
            raise ValueError(f"{field}[{index}].definition_sha256 must be a lowercase SHA-256")
    return value


def _path_records(value: Any, paths: list[str], field: str) -> list[dict[str, Any]]:
    if not isinstance(value, list):
        raise ValueError(f"{field} must be a list")
    records: list[dict[str, Any]] = []
    record_fields = {
        "path",
        "lstat_chain_sha256",
        "resolved_target_sha256",
        "inside_worktree",
        "outside_git_metadata",
    }
    for index, item in enumerate(value):
        record = _exact_fields(item, record_fields, f"{field}[{index}]")
        if not isinstance(record["path"], str) or not record["path"]:
            raise ValueError(f"{field}[{index}].path must be a non-empty string")
        for name in ("lstat_chain_sha256", "resolved_target_sha256"):
            digest = record[name]
            if not isinstance(digest, str) or not SHA256.fullmatch(digest):
                raise ValueError(f"{field}[{index}].{name} must be a lowercase SHA-256")
        if _boolean(record["inside_worktree"], f"{field}[{index}].inside_worktree") is not True:
            raise ValueError(f"{field}[{index}] must resolve inside the worktree")
        if (
            _boolean(
                record["outside_git_metadata"],
                f"{field}[{index}].outside_git_metadata",
            )
            is not True
        ):
            raise ValueError(f"{field}[{index}] must resolve outside Git metadata")
        records.append(record)
    if [record["path"] for record in records] != paths:
        raise ValueError(f"{field} must be sorted and exactly match approved paths")
    return records


def _automation_prompt_template(value: Any, field: str) -> str:
    if not isinstance(value, str) or not value or "\x00" in value:
        raise ValueError(f"{field} must be a non-empty NUL-free string")
    found = re.findall(r"\{\{([A-Z0-9_]+)\}\}", value)
    if sorted(found) != sorted(AUTOMATION_PROMPT_PLACEHOLDERS):
        raise ValueError(
            f"{field} must contain each supported typed placeholder exactly once"
        )
    stripped = re.sub(r"\{\{[A-Z0-9_]+\}\}", "", value)
    if "{{" in stripped or "}}" in stripped:
        raise ValueError(f"{field} contains an unsupported placeholder")
    if value != AUTOMATION_PROMPT_TEMPLATE:
        raise ValueError(f"{field} must equal the evaluator-owned v1 template")
    return value


def _review_request_template(value: Any, field: str, run_id: str, round_number: int) -> str:
    if not isinstance(value, str) or not value or "\x00" in value:
        raise ValueError(f"{field} must be a non-empty NUL-free string")
    placeholder = "{{FINAL_HEAD_SHA}}"
    marker = (
        "<!-- github-codex-pr-loop-request "
        f"run={run_id} round={round_number} sha={placeholder} -->"
    )
    if value.count(placeholder) != 1 or marker not in value or "@codex" not in value:
        raise ValueError(f"{field} must contain the exact Codex request marker and placeholder")
    stripped = value.replace(placeholder, "")
    if "{{" in stripped or "}}" in stripped:
        raise ValueError(f"{field} contains an unsupported placeholder")
    return value


def _thread_snapshots(value: Any, field: str, now: datetime) -> list[dict[str, Any]]:
    if not isinstance(value, list):
        raise ValueError(f"{field} must be a list")
    snapshots: list[dict[str, Any]] = []
    seen_ids: set[str] = set()
    fields = {
        "id",
        "root_comment_id",
        "body_sha256",
        "position_sha256",
        "content_version_at",
    }
    for index, item in enumerate(value):
        snapshot = _exact_fields(item, fields, f"{field}[{index}]")
        for name in ("id", "root_comment_id"):
            if not isinstance(snapshot[name], str) or not snapshot[name]:
                raise ValueError(f"{field}[{index}].{name} must be a non-empty string")
        if snapshot["id"] in seen_ids:
            raise ValueError(f"{field} IDs must be unique")
        seen_ids.add(snapshot["id"])
        for name in ("body_sha256", "position_sha256"):
            if not isinstance(snapshot[name], str) or not SHA256.fullmatch(snapshot[name]):
                raise ValueError(f"{field}[{index}].{name} must be a lowercase SHA-256")
        if _timestamp(
            snapshot["content_version_at"], f"{field}[{index}].content_version_at"
        ) > now:
            raise ValueError(f"{field}[{index}].content_version_at is after snapshot time")
        snapshots.append(snapshot)
    if [item["id"] for item in snapshots] != sorted(item["id"] for item in snapshots):
        raise ValueError(f"{field} must be sorted by id")
    return snapshots


def _response_manifest_view(response: dict[str, Any]) -> dict[str, Any]:
    return {field: response[field] for field in sorted(RESPONSE_FIELDS)}


def _validate_manifest_identity(
    manifest: dict[str, Any],
    *,
    kind: str,
    policy_schema_version: int,
    run_id: str,
    repository: str,
    pr_number: int,
    remote_name: str,
    base_branch: str,
    head_branch: str,
    base_sha: str,
    request_key: str,
    request_sha: str,
) -> None:
    expected = {
        "kind": kind,
        "version": 1,
        "policy_schema_version": policy_schema_version,
        "run_id": run_id,
        "repository": repository,
        "pr_number": pr_number,
        "remote_name": remote_name,
        "base_branch": base_branch,
        "head_branch": head_branch,
        "base_sha": base_sha,
        "request_key": request_key,
        "current_head_sha": request_sha,
    }
    if any(
        type(manifest.get(key)) is not type(value) or manifest.get(key) != value
        for key, value in expected.items()
    ):
        raise ValueError(f"{kind}_manifest identity does not match current state")


def _validate_response_binding(
    value: Any, expected: list[dict[str, Any]], field: str
) -> None:
    if not isinstance(value, list):
        raise ValueError(f"{field} must be a list")
    actual_json = json.dumps(value, ensure_ascii=False, separators=(",", ":"), sort_keys=True)
    expected_json = json.dumps(
        expected, ensure_ascii=False, separators=(",", ":"), sort_keys=True
    )
    if actual_json != expected_json:
        raise ValueError(f"{field} does not match qualifying response state")


def _validate_execution_boundary(
    value: Any, field: str, safe_profile: str
) -> dict[str, Any]:
    execution = _exact_fields(
        value,
        {
            "filesystem_read_scope_sha256",
            "filesystem_write_scope_sha256",
            "isolation",
            "risk_acceptance_sha256",
            "git",
        },
        field,
    )
    for name in ("filesystem_read_scope_sha256", "filesystem_write_scope_sha256"):
        digest = execution[name]
        if not isinstance(digest, str) or not SHA256.fullmatch(digest):
            raise ValueError(f"{field}.{name} must be a lowercase SHA-256")
    isolation = execution["isolation"]
    risk_digest = execution["risk_acceptance_sha256"]
    if isolation == "no-network-minimal-env-checkout-only":
        if risk_digest is not None:
            raise ValueError(f"{field} isolated execution cannot carry a risk-acceptance hash")
    elif isolation == "explicit-risk-acceptance":
        if not isinstance(risk_digest, str) or not SHA256.fullmatch(risk_digest):
            raise ValueError(f"{field} explicit risk acceptance requires a lowercase SHA-256")
    else:
        raise ValueError(f"{field}.isolation is unsupported")

    git_fields = {
        "active_clean_filters",
        "attributes_snapshot_sha256",
        "config_snapshot_sha256",
        "credential_strategy",
        "external_diff_disabled",
        "fsmonitor_disabled",
        "git_metadata_snapshot_sha256",
        "hooks_disabled",
        "invocation_sha256",
        "profile",
        "protect_hfs_enabled",
        "protect_ntfs_enabled",
        "protocol_restricted",
        "signing_disabled",
        "textconv_disabled",
        "transport_url_sha256",
    }
    git_state = _exact_fields(execution["git"], git_fields, f"{field}.git")
    for name in (
        "attributes_snapshot_sha256",
        "config_snapshot_sha256",
        "git_metadata_snapshot_sha256",
        "invocation_sha256",
        "transport_url_sha256",
    ):
        digest = git_state[name]
        if not isinstance(digest, str) or not SHA256.fullmatch(digest):
            raise ValueError(f"{field}.git.{name} must be a lowercase SHA-256")
    for name in (
        "external_diff_disabled",
        "fsmonitor_disabled",
        "hooks_disabled",
        "protect_hfs_enabled",
        "protect_ntfs_enabled",
        "protocol_restricted",
        "signing_disabled",
        "textconv_disabled",
    ):
        _boolean(git_state[name], f"{field}.git.{name}")
    filters = git_state["active_clean_filters"]
    if (
        not isinstance(filters, list)
        or not all(isinstance(item, str) and item for item in filters)
        or filters != sorted(set(filters))
    ):
        raise ValueError(f"{field}.git.active_clean_filters must be sorted unique strings")
    credential_strategy = git_state["credential_strategy"]
    if credential_strategy not in {"gh-cli-https", "explicitly-accepted"}:
        raise ValueError(f"{field}.git.credential_strategy is unsupported")
    profile = git_state["profile"]
    if profile == safe_profile:
        if credential_strategy != "gh-cli-https":
            raise ValueError(f"{field} safe Git profile requires the GitHub CLI HTTPS helper")
        if filters:
            raise ValueError(f"{field} safe Git profile forbids active clean filters")
        for name in (
            "external_diff_disabled",
            "fsmonitor_disabled",
            "hooks_disabled",
            "protect_hfs_enabled",
            "protect_ntfs_enabled",
            "protocol_restricted",
            "signing_disabled",
            "textconv_disabled",
        ):
            if git_state[name] is not True:
                raise ValueError(f"{field} safe Git profile requires {name}")
    elif profile == "explicit-risk-acceptance-v1":
        if isolation != "explicit-risk-acceptance":
            raise ValueError(f"{field} unsafe Git profile requires explicit risk acceptance")
    else:
        raise ValueError(f"{field}.git.profile is unsupported")
    return execution


def _validate_remediation_manifest(
    manifest: dict[str, Any],
    *,
    policy_schema_version: int,
    run_id: str,
    repository: str,
    pr_number: int,
    remote_name: str,
    base_branch: str,
    head_branch: str,
    base_sha: str,
    request_key: str,
    request_sha: str,
    expected_responses: list[dict[str, Any]],
    expected_actionable_threads: list[dict[str, Any]],
    expected_execution: dict[str, Any],
    safe_git_profile: str,
    fix_rounds: int,
    now: datetime,
) -> None:
    fields = {
        "kind",
        "version",
        "policy_schema_version",
        "run_id",
        "repository",
        "pr_number",
        "remote_name",
        "base_branch",
        "head_branch",
        "base_sha",
        "request_key",
        "current_head_sha",
        "responses",
        "actionable_threads",
        "paths",
        "path_records",
        "commands",
        "execution",
        "review_request_body_template",
        "review_request_body_template_sha256",
        "operations",
    }
    _exact_fields(manifest, fields, "remediation_manifest")
    _validate_manifest_identity(
        manifest,
        kind="remediation",
        policy_schema_version=policy_schema_version,
        run_id=run_id,
        repository=repository,
        pr_number=pr_number,
        remote_name=remote_name,
        base_branch=base_branch,
        head_branch=head_branch,
        base_sha=base_sha,
        request_key=request_key,
        request_sha=request_sha,
    )
    _validate_response_binding(
        manifest["responses"], expected_responses, "remediation_manifest.responses"
    )
    threads = _thread_snapshots(
        manifest["actionable_threads"], "remediation_manifest.actionable_threads", now
    )
    if json.dumps(
        threads, ensure_ascii=False, separators=(",", ":"), sort_keys=True
    ) != json.dumps(
        expected_actionable_threads,
        ensure_ascii=False,
        separators=(",", ":"),
        sort_keys=True,
    ):
        raise ValueError(
            "remediation_manifest.actionable_threads does not match current thread snapshots"
        )

    paths = _manifest_paths(manifest["paths"], "remediation_manifest.paths", allow_empty=False)
    _path_records(manifest["path_records"], paths, "remediation_manifest.path_records")
    commands = _manifest_commands(manifest["commands"], "remediation_manifest.commands")
    request_template = _review_request_template(
        manifest["review_request_body_template"],
        "remediation_manifest.review_request_body_template",
        run_id,
        fix_rounds + 1,
    )
    if manifest["review_request_body_template_sha256"] != hashlib.sha256(
        request_template.encode("utf-8")
    ).hexdigest():
        raise ValueError(
            "remediation_manifest.review_request_body_template_sha256 does not match the template"
        )

    execution = _validate_execution_boundary(
        manifest["execution"], "remediation_manifest.execution", safe_git_profile
    )
    if json.dumps(
        execution, ensure_ascii=False, separators=(",", ":"), sort_keys=True
    ) != json.dumps(
        expected_execution, ensure_ascii=False, separators=(",", ":"), sort_keys=True
    ):
        raise ValueError("remediation_manifest.execution does not match current execution state")

    expected_operations = ["edit-files"]
    if commands:
        expected_operations.append("run-validation")
    expected_operations.extend(
        [
            "commit",
            "push",
            "reply-to-codex-threads",
            "request-codex-review",
            "update-automation",
        ]
    )
    if manifest["operations"] != expected_operations:
        raise ValueError("remediation_manifest.operations does not match the bounded write batch")


def _validate_finalization_manifest(
    manifest: dict[str, Any],
    *,
    policy_schema_version: int,
    run_id: str,
    repository: str,
    pr_number: int,
    remote_name: str,
    base_branch: str,
    head_branch: str,
    base_sha: str,
    request_key: str,
    request_sha: str,
    expected_responses: list[dict[str, Any]],
    expected_eligible_threads: list[dict[str, Any]],
    non_codex_threads: int,
    pr_is_draft: bool,
    gates: dict[str, str],
    now: datetime,
) -> None:
    fields = {
        "kind",
        "version",
        "policy_schema_version",
        "run_id",
        "repository",
        "pr_number",
        "remote_name",
        "base_branch",
        "head_branch",
        "base_sha",
        "request_key",
        "current_head_sha",
        "responses",
        "eligible_threads",
        "non_codex_rooted_threads",
        "pr_is_draft",
        "gates",
        "operations",
    }
    _exact_fields(manifest, fields, "finalization_manifest")
    _validate_manifest_identity(
        manifest,
        kind="finalization",
        policy_schema_version=policy_schema_version,
        run_id=run_id,
        repository=repository,
        pr_number=pr_number,
        remote_name=remote_name,
        base_branch=base_branch,
        head_branch=head_branch,
        base_sha=base_sha,
        request_key=request_key,
        request_sha=request_sha,
    )
    _validate_response_binding(
        manifest["responses"], expected_responses, "finalization_manifest.responses"
    )
    threads = _thread_snapshots(
        manifest["eligible_threads"], "finalization_manifest.eligible_threads", now
    )
    if json.dumps(
        threads, ensure_ascii=False, separators=(",", ":"), sort_keys=True
    ) != json.dumps(
        expected_eligible_threads,
        ensure_ascii=False,
        separators=(",", ":"),
        sort_keys=True,
    ):
        raise ValueError(
            "finalization_manifest.eligible_threads does not match current thread snapshots"
        )
    manifest_non_codex_threads = _nonnegative_int(
        manifest["non_codex_rooted_threads"],
        "finalization_manifest.non_codex_rooted_threads",
    )
    if manifest_non_codex_threads != non_codex_threads:
        raise ValueError(
            "finalization_manifest.non_codex_rooted_threads does not match current state"
        )
    if manifest["pr_is_draft"] is not pr_is_draft:
        raise ValueError("finalization_manifest.pr_is_draft does not match current state")
    _exact_fields(manifest["gates"], set(gates), "finalization_manifest.gates")
    if manifest["gates"] != gates:
        raise ValueError("finalization_manifest.gates does not match current gate state")

    expected_operations = []
    if pr_is_draft:
        expected_operations.append("mark-draft-ready")
    if expected_eligible_threads:
        expected_operations.append("resolve-codex-threads")
    expected_operations.append("delete-owned-automation")
    if manifest["operations"] != expected_operations:
        raise ValueError("finalization_manifest.operations does not match the bounded write batch")


def validate_initial_document(
    document: dict[str, Any], policy: dict[str, Any]
) -> dict[str, Any]:
    """Validate the exact initial write batch without performing a mutation."""

    outer = _exact_fields(
        document,
        {"initial_snapshot", "initial_manifest", "initial_plan_hash"},
        "initial_document",
    )
    snapshot_fields = {
        "policy_schema_version",
        "host",
        "repository",
        "authenticated_user_id",
        "base_branch",
        "base_sha",
        "head_branch",
        "current_head_sha",
        "remote_name",
        "existing_pr_number",
        "existing_pr_base_branch",
        "existing_pr_head_branch",
        "existing_pr_head_sha",
        "existing_pr_same_repository",
        "heartbeat_rrule",
        "create_branch_required",
        "commit_required",
        "push_required",
        "paths",
        "path_records",
        "diff_sha256",
        "worktree_status_sha256",
        "remote_refs_sha256",
        "open_prs_sha256",
        "commands",
        "execution_boundary",
    }
    snapshot = _exact_fields(outer["initial_snapshot"], snapshot_fields, "initial_snapshot")
    policy_schema_version = _positive_int(
        snapshot["policy_schema_version"], "initial_snapshot.policy_schema_version"
    )
    if policy_schema_version != policy["schema_version"]:
        raise ValueError("initial_snapshot policy version differs from installed policy")
    if snapshot["host"] != policy["supported_host"]:
        raise ValueError("initial_snapshot.host is outside the supported policy")
    repository = snapshot["repository"]
    if not isinstance(repository, str) or not REPOSITORY.fullmatch(repository):
        raise ValueError("initial_snapshot.repository must be canonical lowercase owner/repo")
    _positive_int(
        snapshot["authenticated_user_id"], "initial_snapshot.authenticated_user_id"
    )
    base_branch = _ref_name(snapshot["base_branch"], "initial_snapshot.base_branch")
    head_branch = _ref_name(snapshot["head_branch"], "initial_snapshot.head_branch")
    if base_branch == head_branch:
        raise ValueError("initial_snapshot base and head branches must differ")
    _sha(snapshot["base_sha"], "initial_snapshot.base_sha")
    _sha(snapshot["current_head_sha"], "initial_snapshot.current_head_sha")
    _ref_name(snapshot["remote_name"], "initial_snapshot.remote_name")

    existing_pr_number = snapshot["existing_pr_number"]
    existing_base = snapshot["existing_pr_base_branch"]
    existing_head = snapshot["existing_pr_head_branch"]
    existing_head_sha = snapshot["existing_pr_head_sha"]
    existing_same_repository = snapshot["existing_pr_same_repository"]
    if existing_pr_number is None:
        if any(
            value is not None
            for value in (
                existing_base,
                existing_head,
                existing_head_sha,
                existing_same_repository,
            )
        ):
            raise ValueError("initial_snapshot existing PR fields must be null without a PR")
    else:
        _positive_int(existing_pr_number, "initial_snapshot.existing_pr_number")
        if (
            _ref_name(existing_base, "initial_snapshot.existing_pr_base_branch")
            != base_branch
        ):
            raise ValueError("initial_snapshot existing PR base does not match base_branch")
        if (
            _ref_name(existing_head, "initial_snapshot.existing_pr_head_branch")
            != head_branch
        ):
            raise ValueError("initial_snapshot existing PR head does not match head_branch")
        if (
            _sha(existing_head_sha, "initial_snapshot.existing_pr_head_sha")
            != snapshot["current_head_sha"]
        ):
            raise ValueError("initial_snapshot existing PR head SHA does not match current_head_sha")
        if _boolean(
            existing_same_repository, "initial_snapshot.existing_pr_same_repository"
        ) is not True:
            raise ValueError("initial_snapshot existing PR must use same-repository topology")
    if snapshot["heartbeat_rrule"] != policy["heartbeat_rrule"]:
        raise ValueError("initial_snapshot.heartbeat_rrule differs from installed policy")

    create_branch_required = _boolean(
        snapshot["create_branch_required"], "initial_snapshot.create_branch_required"
    )
    commit_required = _boolean(snapshot["commit_required"], "initial_snapshot.commit_required")
    push_required = _boolean(snapshot["push_required"], "initial_snapshot.push_required")
    paths = _manifest_paths(snapshot["paths"], "initial_snapshot.paths", allow_empty=True)
    _path_records(snapshot["path_records"], paths, "initial_snapshot.path_records")
    if commit_required and not paths:
        raise ValueError("initial_snapshot commit_required needs at least one approved path")
    if (create_branch_required or commit_required) and not push_required:
        raise ValueError("initial_snapshot branch creation or commit requires a push")
    if create_branch_required and existing_pr_number is not None:
        raise ValueError("initial_snapshot cannot create a branch for an existing PR")
    for name in (
        "diff_sha256",
        "worktree_status_sha256",
        "remote_refs_sha256",
        "open_prs_sha256",
    ):
        digest = snapshot[name]
        if not isinstance(digest, str) or not SHA256.fullmatch(digest):
            raise ValueError(f"initial_snapshot.{name} must be a lowercase SHA-256")
    commands = _manifest_commands(snapshot["commands"], "initial_snapshot.commands")
    _validate_execution_boundary(
        snapshot["execution_boundary"],
        "initial_snapshot.execution_boundary",
        policy["git_execution_profile"],
    )

    snapshot_digest = _manifest_digest(snapshot, "initial_snapshot")
    manifest_fields = {
        "kind",
        "version",
        "run_id",
        "mode",
        "snapshot_sha256",
        "pr_title_sha256",
        "pr_body_sha256",
        "review_request_body_template",
        "review_request_body_template_sha256",
        "automation_prompt_template",
        "automation_prompt_template_sha256",
        "operations",
    }
    manifest = _exact_fields(outer["initial_manifest"], manifest_fields, "initial_manifest")
    if (
        manifest["kind"] != "initial"
        or isinstance(manifest["version"], bool)
        or manifest["version"] != 1
    ):
        raise ValueError("initial_manifest kind/version is unsupported")
    if not isinstance(manifest["run_id"], str) or not RUN_ID.fullmatch(manifest["run_id"]):
        raise ValueError("initial_manifest.run_id must be 32 lowercase hex characters")
    if manifest["mode"] not in {"autonomous", "approval-gated"}:
        raise ValueError("initial_manifest.mode must be autonomous or approval-gated")
    if manifest["snapshot_sha256"] != snapshot_digest:
        raise ValueError("initial_manifest.snapshot_sha256 does not match initial_snapshot")
    for name in ("pr_title_sha256", "pr_body_sha256"):
        digest = manifest[name]
        if not isinstance(digest, str) or not SHA256.fullmatch(digest):
            raise ValueError(f"initial_manifest.{name} must be a lowercase SHA-256")
    request_template = _review_request_template(
        manifest["review_request_body_template"],
        "initial_manifest.review_request_body_template",
        manifest["run_id"],
        0,
    )
    request_template_digest = hashlib.sha256(request_template.encode("utf-8")).hexdigest()
    if manifest["review_request_body_template_sha256"] != request_template_digest:
        raise ValueError(
            "initial_manifest.review_request_body_template_sha256 does not match the template"
        )
    prompt_template = _automation_prompt_template(
        manifest["automation_prompt_template"],
        "initial_manifest.automation_prompt_template",
    )
    prompt_template_digest = hashlib.sha256(prompt_template.encode("utf-8")).hexdigest()
    if manifest["automation_prompt_template_sha256"] != prompt_template_digest:
        raise ValueError(
            "initial_manifest.automation_prompt_template_sha256 does not match the template"
        )

    expected_operations = ["fetch-base"]
    if create_branch_required:
        expected_operations.append("create-branch")
    if commands:
        expected_operations.append("run-validation")
    if commit_required:
        expected_operations.extend(["stage-files", "commit"])
    if push_required:
        expected_operations.append("push")
    expected_operations.append("reuse-pr" if existing_pr_number is not None else "create-pr")
    expected_operations.extend(["request-codex-review", "bootstrap-automation"])
    if manifest["operations"] != expected_operations:
        raise ValueError("initial_manifest.operations does not match the bounded write batch")

    manifest_digest = _manifest_digest(manifest, "initial_manifest")
    plan_hash = outer["initial_plan_hash"]
    if not isinstance(plan_hash, str) or not SHA256.fullmatch(plan_hash):
        raise ValueError("initial_plan_hash must be a lowercase SHA-256")
    if plan_hash != manifest_digest:
        raise ValueError("initial_plan_hash does not match the canonical manifest")
    return {
        "action": "initial-plan-valid",
        "reason": "the exact initial snapshot and bounded write batch are valid",
        "initial_plan_hash": plan_hash,
        "snapshot_sha256": snapshot_digest,
        "operations": expected_operations,
        "may_mutate_repo": False,
        "may_mutate_github": False,
        "automation_action": "none",
    }


def validate_prompt_document(
    document: dict[str, Any], policy: dict[str, Any]
) -> dict[str, Any]:
    """Validate a concrete paused checkpoint before activating its heartbeat."""

    prompt_document = _exact_fields(
        document,
        {
            "initial_document",
            "bootstrap_snapshot",
            "bootstrap_snapshot_sha256",
            "rendered_prompt",
            "rendered_prompt_sha256",
        },
        "prompt_document",
    )
    initial_result = validate_initial_document(prompt_document["initial_document"], policy)
    initial_document = prompt_document["initial_document"]
    initial_snapshot = initial_document["initial_snapshot"]
    manifest = initial_document["initial_manifest"]
    bootstrap = _exact_fields(
        prompt_document["bootstrap_snapshot"],
        {
            "initial_plan_hash",
            "post_write_receipt",
            "pr_url",
            "request_deadline",
            "state",
        },
        "bootstrap_snapshot",
    )
    bootstrap_digest = _manifest_digest(bootstrap, "bootstrap_snapshot")
    if prompt_document["bootstrap_snapshot_sha256"] != bootstrap_digest:
        raise ValueError("bootstrap_snapshot_sha256 does not match bootstrap_snapshot")
    if bootstrap["initial_plan_hash"] != initial_result["initial_plan_hash"]:
        raise ValueError("bootstrap_snapshot initial plan hash does not match")

    receipt = _exact_fields(
        bootstrap["post_write_receipt"],
        {
            "base_sha",
            "final_head_sha",
            "head_descends_from_initial",
            "initial_head_sha",
            "paths",
            "pr_head_sha",
            "pushed_diff_sha256",
            "remote_head_sha",
            "review_request_body",
            "review_request_body_sha256",
            "worktree_clean",
        },
        "bootstrap_snapshot.post_write_receipt",
    )
    initial_head_sha = initial_snapshot["current_head_sha"]
    final_head_sha = _sha(
        receipt["final_head_sha"], "bootstrap_snapshot.post_write_receipt.final_head_sha"
    )
    for name, expected in (
        ("base_sha", initial_snapshot["base_sha"]),
        ("initial_head_sha", initial_head_sha),
        ("remote_head_sha", final_head_sha),
        ("pr_head_sha", final_head_sha),
        ("pushed_diff_sha256", initial_snapshot["diff_sha256"]),
        ("paths", initial_snapshot["paths"]),
    ):
        if type(receipt[name]) is not type(expected) or receipt[name] != expected:
            raise ValueError(f"bootstrap_snapshot post-write receipt {name} does not match")
    if _boolean(
        receipt["head_descends_from_initial"],
        "bootstrap_snapshot.post_write_receipt.head_descends_from_initial",
    ) is not True:
        raise ValueError("bootstrap_snapshot final head must descend from the initial head")
    if _boolean(
        receipt["worktree_clean"],
        "bootstrap_snapshot.post_write_receipt.worktree_clean",
    ) is not True:
        raise ValueError("bootstrap_snapshot post-write worktree must be clean")
    if initial_snapshot["commit_required"]:
        if final_head_sha == initial_head_sha:
            raise ValueError("bootstrap_snapshot required commit did not create a new head")
    elif final_head_sha != initial_head_sha:
        raise ValueError("bootstrap_snapshot changed head without an approved commit")
    request_body = receipt["review_request_body"]
    if not isinstance(request_body, str) or "\x00" in request_body:
        raise ValueError("bootstrap_snapshot review_request_body must be a NUL-free string")
    expected_request_body = manifest["review_request_body_template"].replace(
        "{{FINAL_HEAD_SHA}}", final_head_sha
    )
    if request_body != expected_request_body:
        raise ValueError("bootstrap_snapshot review-request body is not the approved rendering")
    request_body_digest = hashlib.sha256(request_body.encode("utf-8")).hexdigest()
    if receipt["review_request_body_sha256"] != request_body_digest:
        raise ValueError("bootstrap_snapshot review-request body hash does not match")

    state = bootstrap["state"]
    bootstrap_decision = evaluate(state, policy)
    if (
        bootstrap_decision["action"] != "halt-state-inconsistency"
        or bootstrap_decision["reason"]
        != "request state pending requires automation status ACTIVE"
        or bootstrap_decision["automation_action"] != "none"
    ):
        raise ValueError("bootstrap_snapshot.state is not the exact paused pending checkpoint")

    expected_bindings = {
        "policy_schema_version": initial_snapshot["policy_schema_version"],
        "host": initial_snapshot["host"],
        "repository": initial_snapshot["repository"],
        "remote_name": initial_snapshot["remote_name"],
        "authenticated_user_id": initial_snapshot["authenticated_user_id"],
        "request_author_id": initial_snapshot["authenticated_user_id"],
        "base_branch": initial_snapshot["base_branch"],
        "head_branch": initial_snapshot["head_branch"],
        "pr_base_branch": initial_snapshot["base_branch"],
        "pr_head_branch": initial_snapshot["head_branch"],
        "request_base_sha": initial_snapshot["base_sha"],
        "current_base_sha": initial_snapshot["base_sha"],
        "request_head_sha": final_head_sha,
        "current_head_sha": final_head_sha,
        "mode": manifest["mode"],
        "run_id": manifest["run_id"],
        "same_repository_pr": True,
        "request_state": "pending",
        "fix_rounds": 0,
        "completed_request_keys": [],
        "intervening_review_requests": 0,
        "processing_claim_token": None,
        "post_finalization": False,
        "remediation_manifest": None,
        "execution_boundary": None,
        "remediation_plan_hash": None,
        "approved_request_key": None,
        "approved_remediation_plan_hash": None,
        "finalization_manifest": None,
        "finalization_plan_hash": None,
        "finalization_approved_request_key": None,
        "approved_finalization_plan_hash": None,
        "claimed_finalization_request_key": None,
        "claimed_finalization_plan_hash": None,
        "terminal_error": None,
        "pr_state": "open",
        "head_branch_exists": True,
    }
    if any(
        type(state.get(key)) is not type(value) or state.get(key) != value
        for key, value in expected_bindings.items()
    ):
        raise ValueError("bootstrap_snapshot.state does not match the approved initial identity")

    pr_number = _positive_int(state["pr_number"], "bootstrap_snapshot.state.pr_number")
    existing_pr_number = initial_snapshot["existing_pr_number"]
    if existing_pr_number is not None and pr_number != existing_pr_number:
        raise ValueError("bootstrap_snapshot PR number differs from the reused PR")
    expected_url = f"https://github.com/{initial_snapshot['repository']}/pull/{pr_number}"
    if bootstrap["pr_url"] != expected_url:
        raise ValueError("bootstrap_snapshot.pr_url is not the canonical pull-request URL")
    created = _timestamp(state["request_created_at"], "bootstrap_snapshot request_created_at")
    deadline = _timestamp(bootstrap["request_deadline"], "bootstrap_snapshot.request_deadline")
    if deadline != created + timedelta(minutes=policy["no_response_timeout_minutes"]):
        raise ValueError("bootstrap_snapshot.request_deadline is not the policy deadline")
    if _timestamp(state["now"], "bootstrap_snapshot state.now") >= deadline:
        raise ValueError("bootstrap_snapshot is already expired")

    automations = state["automations"]
    owned_automation_id = state["owned_automation_id"]
    if (
        len(automations) != 1
        or automations[0]["id"] != owned_automation_id
        or automations[0]["key"] != state["automation_key"]
        or automations[0]["status"] != "PAUSED"
        or automations[0]["schedule"] != policy["heartbeat_rrule"]
    ):
        raise ValueError("bootstrap_snapshot must bind one exact paused owned automation")

    checkpoint_json = json.dumps(
        bootstrap, ensure_ascii=False, separators=(",", ":"), sort_keys=True
    )
    rendered = manifest["automation_prompt_template"].replace(
        "{{AUTOMATION_KEY}}", state["automation_key"]
    ).replace("{{BOOTSTRAP_CHECKPOINT_JSON}}", checkpoint_json)
    if prompt_document["rendered_prompt"] != rendered:
        raise ValueError("rendered_prompt does not exactly match the canonical bootstrap rendering")
    rendered_digest = hashlib.sha256(rendered.encode("utf-8")).hexdigest()
    if prompt_document["rendered_prompt_sha256"] != rendered_digest:
        raise ValueError("rendered_prompt_sha256 does not match rendered_prompt")
    return {
        "action": "prompt-ready",
        "reason": "the derived heartbeat prompt is exactly bound to the initial plan",
        "initial_plan_hash": initial_result["initial_plan_hash"],
        "bootstrap_snapshot_sha256": bootstrap_digest,
        "rendered_prompt_sha256": rendered_digest,
        "automation_id": owned_automation_id,
        "automation_schedule": policy["heartbeat_rrule"],
        "may_mutate_repo": False,
        "may_mutate_github": False,
        "automation_action": "none",
    }


def load_policy(path: Path = DEFAULT_POLICY_PATH) -> dict[str, Any]:
    with path.open(encoding="utf-8") as handle:
        policy = json.load(handle)
    _exact_fields(policy, set(POLICY_V1), "policy")
    mismatched = [
        field
        for field, expected in POLICY_V1.items()
        if type(policy[field]) is not type(expected) or policy[field] != expected
    ]
    if mismatched:
        raise ValueError(
            "policy differs from pinned schema-v1 constants: " + ", ".join(mismatched)
        )
    return policy


def evaluate(state: dict[str, Any], policy: dict[str, Any]) -> dict[str, Any]:
    """Return the one permitted next action for a complete normalized snapshot."""

    if not isinstance(state, dict):
        raise ValueError("state must be a JSON object")
    missing = sorted(REQUIRED_STATE_FIELDS - state.keys())
    if missing:
        raise ValueError(f"state is missing required fields: {', '.join(missing)}")
    extra = sorted(state.keys() - REQUIRED_STATE_FIELDS)
    if extra:
        raise ValueError(f"state contains unsupported fields: {', '.join(extra)}")

    mode = state["mode"]
    if mode not in policy["modes"]:
        raise ValueError(f"mode must be one of: {', '.join(policy['modes'])}")

    repository = state["repository"]
    if not isinstance(repository, str) or not REPOSITORY.fullmatch(repository):
        raise ValueError("repository must be canonical lowercase owner/repo")
    pr_number = _positive_int(state["pr_number"], "pr_number")
    expected_loop_key = f"github-codex-pr-loop:{repository}#{pr_number}"
    if state["automation_key"] != expected_loop_key:
        raise ValueError("automation_key does not match canonical repository and PR fields")
    if mode == "report-only":
        if state["remote_name"] is not None:
            raise ValueError("report-only mode requires remote_name to be null")
        remote_name = None
    else:
        remote_name = _ref_name(state["remote_name"], "remote_name")
    run_id = state["run_id"]
    if not isinstance(run_id, str) or not RUN_ID.fullmatch(run_id):
        raise ValueError("run_id must be 32 lowercase hexadecimal characters")

    host = state["host"]
    if not isinstance(host, str) or not host:
        raise ValueError("host must be a non-empty string")
    same_repository_pr = _boolean(state["same_repository_pr"], "same_repository_pr")
    policy_schema_version = _positive_int(
        state["policy_schema_version"], "policy_schema_version"
    )
    base_branch = _ref_name(state["base_branch"], "base_branch")
    head_branch = _ref_name(state["head_branch"], "head_branch")
    pr_base_branch = _ref_name(state["pr_base_branch"], "pr_base_branch")
    pr_head_branch = _ref_name(state["pr_head_branch"], "pr_head_branch")
    if base_branch == head_branch:
        raise ValueError("base_branch and head_branch must differ")

    request_sha = _sha(state["request_head_sha"], "request_head_sha")
    current_sha = _sha(state["current_head_sha"], "current_head_sha")
    request_base_sha = _sha(state["request_base_sha"], "request_base_sha")
    current_base_sha = _sha(state["current_base_sha"], "current_base_sha")
    comment_id = state["request_comment_id"]
    if isinstance(comment_id, bool) or not isinstance(comment_id, (str, int)):
        raise ValueError("request_comment_id must be a positive integer or digit string")
    if not str(comment_id).isdigit() or int(comment_id) < 1:
        raise ValueError("request_comment_id must be a positive integer or digit string")
    canonical_comment_id = str(int(comment_id))
    if isinstance(comment_id, str) and comment_id != canonical_comment_id:
        raise ValueError("request_comment_id digit string must not contain leading zeroes")
    request_key = f"{canonical_comment_id}:{request_sha}"
    request_author_id = _positive_int(state["request_author_id"], "request_author_id")
    authenticated_user_id = _positive_int(
        state["authenticated_user_id"], "authenticated_user_id"
    )
    intervening_review_requests = _nonnegative_int(
        state["intervening_review_requests"], "intervening_review_requests"
    )
    request_created = _timestamp(state["request_created_at"], "request_created_at")
    now = _timestamp(state["now"], "now")
    if now < request_created:
        raise ValueError("now must not be earlier than request_created_at")
    try:
        deadline = request_created + timedelta(minutes=policy["no_response_timeout_minutes"])
    except OverflowError as exc:
        raise ValueError("request_created_at is outside the supported timestamp range") from exc

    request_state = state["request_state"]
    if request_state not in REQUEST_STATES:
        raise ValueError(f"request_state must be one of: {', '.join(sorted(REQUEST_STATES))}")
    processing_claim_token = state["processing_claim_token"]
    post_finalization = _boolean(state["post_finalization"], "post_finalization")
    if processing_claim_token is not None and (
        not isinstance(processing_claim_token, str) or not processing_claim_token
    ):
        raise ValueError("processing_claim_token must be a non-empty string or null")
    if request_state == "processing" and processing_claim_token is None:
        raise ValueError("processing request state requires a claim token")
    if request_state != "processing" and processing_claim_token is not None:
        raise ValueError("only processing request state may carry a claim token")
    if post_finalization and request_state != "processing":
        raise ValueError("post_finalization requires processing request state")
    completed_keys = state["completed_request_keys"]
    if not isinstance(completed_keys, list) or not all(isinstance(item, str) for item in completed_keys):
        raise ValueError("completed_request_keys must be a list of strings")
    if any(not REQUEST_KEY.fullmatch(item) for item in completed_keys):
        raise ValueError("completed_request_keys contains a non-canonical request key")
    if len(completed_keys) != len(set(completed_keys)):
        raise ValueError("completed_request_keys must not contain duplicates")

    approved_request_key = _optional_request_key(
        state["approved_request_key"], "approved_request_key"
    )
    finalization_approved_request_key = _optional_request_key(
        state["finalization_approved_request_key"], "finalization_approved_request_key"
    )
    claimed_finalization_request_key = _optional_request_key(
        state["claimed_finalization_request_key"], "claimed_finalization_request_key"
    )
    remediation_plan_hash = _optional_sha256(
        state["remediation_plan_hash"], "remediation_plan_hash"
    )
    approved_remediation_plan_hash = _optional_sha256(
        state["approved_remediation_plan_hash"], "approved_remediation_plan_hash"
    )
    finalization_plan_hash = _optional_sha256(
        state["finalization_plan_hash"], "finalization_plan_hash"
    )
    approved_finalization_plan_hash = _optional_sha256(
        state["approved_finalization_plan_hash"], "approved_finalization_plan_hash"
    )
    claimed_finalization_plan_hash = _optional_sha256(
        state["claimed_finalization_plan_hash"], "claimed_finalization_plan_hash"
    )
    remediation_manifest = state["remediation_manifest"]
    finalization_manifest = state["finalization_manifest"]
    execution_boundary = state["execution_boundary"]
    remediation_manifest_digest = _manifest_digest(
        remediation_manifest, "remediation_manifest"
    )
    finalization_manifest_digest = _manifest_digest(
        finalization_manifest, "finalization_manifest"
    )
    if (remediation_manifest is None) != (remediation_plan_hash is None):
        raise ValueError("remediation manifest and plan hash must both be null or both be set")
    if (execution_boundary is None) != (remediation_manifest is None):
        raise ValueError(
            "execution_boundary and remediation manifest must both be null or both be set"
        )
    if (finalization_manifest is None) != (finalization_plan_hash is None):
        raise ValueError("finalization manifest and plan hash must both be null or both be set")
    if remediation_manifest_digest != remediation_plan_hash:
        raise ValueError("remediation_plan_hash does not match the canonical manifest")
    if finalization_manifest_digest != finalization_plan_hash:
        raise ValueError("finalization_plan_hash does not match the canonical manifest")
    if execution_boundary is not None:
        execution_boundary = _validate_execution_boundary(
            execution_boundary,
            "execution_boundary",
            policy["git_execution_profile"],
        )
    if mode == "report-only" and any(
        value is not None
        for value in (
            remediation_manifest,
            remediation_plan_hash,
            execution_boundary,
            finalization_manifest,
            finalization_plan_hash,
        )
    ):
        raise ValueError("report-only mode cannot carry write plans or execution state")
    if (approved_request_key is None) != (approved_remediation_plan_hash is None):
        raise ValueError("remediation approval key and plan hash must both be null or both be set")
    if (finalization_approved_request_key is None) != (
        approved_finalization_plan_hash is None
    ):
        raise ValueError("finalization approval key and plan hash must both be null or both be set")
    if (claimed_finalization_request_key is None) != (
        claimed_finalization_plan_hash is None
    ):
        raise ValueError("claimed finalization key and plan hash must both be null or both be set")
    if post_finalization:
        if claimed_finalization_request_key != request_key:
            raise ValueError("post-finalization claim does not match the current request key")
        if mode == "approval-gated" and (
            finalization_approved_request_key != claimed_finalization_request_key
            or approved_finalization_plan_hash != claimed_finalization_plan_hash
        ):
            raise ValueError("post-finalization claim does not match the approved finalization plan")
    elif claimed_finalization_request_key is not None:
        raise ValueError("only post_finalization state may carry a claimed finalization plan")
    if mode != "approval-gated" and any(
        value is not None
        for value in (
            approved_request_key,
            approved_remediation_plan_hash,
            finalization_approved_request_key,
            approved_finalization_plan_hash,
        )
    ):
        raise ValueError("only approval-gated mode may carry approval bindings")
    if mode == "approval-gated" and request_state == "pending" and any(
        value is not None
        for value in (
            approved_request_key,
            approved_remediation_plan_hash,
            finalization_approved_request_key,
            approved_finalization_plan_hash,
        )
    ):
        raise ValueError("pending request state must not carry approval bindings")
    actionable_threads = _nonnegative_int(
        state["actionable_codex_threads"], "actionable_codex_threads"
    )
    eligible_codex_threads = _nonnegative_int(
        state["eligible_codex_threads"], "eligible_codex_threads"
    )
    actionable_thread_snapshots = _thread_snapshots(
        state["actionable_thread_snapshots"], "actionable_thread_snapshots", now
    )
    eligible_thread_snapshots = _thread_snapshots(
        state["eligible_thread_snapshots"], "eligible_thread_snapshots", now
    )
    if len(actionable_thread_snapshots) != actionable_threads:
        raise ValueError(
            "actionable_thread_snapshots length must equal actionable_codex_threads"
        )
    if len(eligible_thread_snapshots) != eligible_codex_threads:
        raise ValueError(
            "eligible_thread_snapshots length must equal eligible_codex_threads"
        )
    non_codex_threads = _nonnegative_int(
        state["non_codex_rooted_threads"], "non_codex_rooted_threads"
    )
    fix_rounds = _nonnegative_int(state["fix_rounds"], "fix_rounds")
    if fix_rounds > policy["max_fix_rounds"]:
        raise ValueError("fix_rounds exceeds the policy maximum")
    if len(completed_keys) != fix_rounds:
        raise ValueError("completed_request_keys length must equal fix_rounds")
    head_branch_exists = _boolean(state["head_branch_exists"], "head_branch_exists")
    check_gate = _gate(state["check_gate"], "check_gate")
    approval_gate = _gate(state["approval_gate"], "approval_gate")
    mergeability_gate = _gate(state["mergeability_gate"], "mergeability_gate")
    branch_protection_gate = _gate(state["branch_protection_gate"], "branch_protection_gate")
    pr_is_draft = _boolean(state["pr_is_draft"], "pr_is_draft")
    pr_state = state["pr_state"]
    if not isinstance(pr_state, str) or not pr_state:
        raise ValueError("pr_state must be a non-empty string")
    terminal_error = state["terminal_error"]
    if terminal_error is not None and (not isinstance(terminal_error, str) or not terminal_error):
        raise ValueError("terminal_error must be a non-empty string or null")

    owned_automation_id = state["owned_automation_id"]
    if owned_automation_id is not None and (
        not isinstance(owned_automation_id, str) or not owned_automation_id
    ):
        raise ValueError("owned_automation_id must be a non-empty string or null")
    if mode == "report-only" and owned_automation_id is not None:
        raise ValueError("report-only mode cannot claim ownership of an automation")

    automations = state["automations"]
    if not isinstance(automations, list):
        raise ValueError("automations must be a list")
    if mode == "report-only" and automations:
        raise ValueError("report-only mode requires automations to be empty")
    seen_automation_ids: set[str] = set()
    semantic_matches: list[dict[str, str]] = []
    for index, automation in enumerate(automations):
        if not isinstance(automation, dict):
            raise ValueError(f"automations[{index}] must be an object")
        if set(automation) != {"id", "key", "status", "schedule"}:
            raise ValueError(
                f"automations[{index}] must contain exactly id, key, status, and schedule"
            )
        automation_id = automation["id"]
        key = automation["key"]
        status = automation["status"]
        schedule = automation["schedule"]
        if not isinstance(automation_id, str) or not automation_id:
            raise ValueError(f"automations[{index}].id must be a non-empty string")
        if automation_id in seen_automation_ids:
            raise ValueError("automation IDs must be unique")
        seen_automation_ids.add(automation_id)
        if not isinstance(key, str) or not key:
            raise ValueError(f"automations[{index}].key must be a non-empty string")
        if status not in AUTOMATION_STATUSES:
            raise ValueError(f"automations[{index}].status must be ACTIVE or PAUSED")
        if not isinstance(schedule, str) or not schedule:
            raise ValueError(f"automations[{index}].schedule must be a non-empty string")
        if key.lower() == expected_loop_key:
            semantic_matches.append(automation)

    responses = state["responses"]
    if not isinstance(responses, list):
        raise ValueError("responses must be a list")
    normalized_responses: list[dict[str, Any]] = []
    seen_response_ids: set[tuple[str, str]] = set()
    for index, response in enumerate(responses):
        if not isinstance(response, dict):
            raise ValueError(f"responses[{index}] must be an object")
        if set(response) != RESPONSE_FIELDS:
            raise ValueError(f"responses[{index}] must contain the complete response schema")
        source = response["source"]
        if source not in RESPONSE_SOURCES:
            raise ValueError(f"responses[{index}].source is unsupported")
        response_id = response["id"]
        if isinstance(response_id, bool) or not isinstance(response_id, (str, int)):
            raise ValueError(f"responses[{index}].id must be a string or integer")
        if isinstance(response_id, int):
            normalized_response_id = str(_positive_int(response_id, f"responses[{index}].id"))
        elif response_id.isdigit():
            normalized_response_id = str(_positive_int(int(response_id), f"responses[{index}].id"))
            if response_id != normalized_response_id:
                raise ValueError(f"responses[{index}].id decimal string must be canonical")
        else:
            raise ValueError(
                f"responses[{index}].id must be a positive integer or canonical decimal string"
            )
        response_identity = (source, normalized_response_id)
        if response_identity in seen_response_ids:
            raise ValueError("response source/id pairs must be unique")
        seen_response_ids.add(response_identity)
        if not isinstance(response["bot_type"], str):
            raise ValueError(f"responses[{index}].bot_type must be a string")
        if isinstance(response["bot_id"], bool) or not isinstance(response["bot_id"], int):
            raise ValueError(f"responses[{index}].bot_id must be an integer")
        _boolean(response["zero_findings_signal"], f"responses[{index}].zero_findings_signal")
        created_at = _timestamp(response["created_at"], f"responses[{index}].created_at")
        content_version_at = _timestamp(
            response["content_version_at"], f"responses[{index}].content_version_at"
        )
        if content_version_at < created_at:
            raise ValueError(f"responses[{index}].content_version_at predates creation")
        body_sha256 = response["body_sha256"]
        if not isinstance(body_sha256, str) or not SHA256.fullmatch(body_sha256):
            raise ValueError(f"responses[{index}].body_sha256 must be a lowercase SHA-256")
        commit_sha = response["commit_sha"]
        if commit_sha is not None:
            commit_sha = _sha(commit_sha, f"responses[{index}].commit_sha")
        trigger_request_comment_id = response["trigger_request_comment_id"]
        if trigger_request_comment_id is not None:
            trigger_request_comment_id = _positive_int(
                trigger_request_comment_id,
                f"responses[{index}].trigger_request_comment_id",
            )
        normalized_responses.append(
            {
                **response,
                "id": normalized_response_id,
                "created_at_parsed": created_at,
                "content_version_at_parsed": content_version_at,
                "trigger_request_comment_id": trigger_request_comment_id,
            }
        )

    trusted_responses = [
        response
        for response in normalized_responses
        if response["bot_type"] == "Bot"
        and response["bot_id"] == policy["trusted_codex_bot_database_id"]
        and request_created < response["created_at_parsed"]
        and response["content_version_at_parsed"] <= min(now, deadline)
        and (response["commit_sha"] is None or response["commit_sha"] == request_sha)
        and (
            response["trigger_request_comment_id"] == int(canonical_comment_id)
            if response["trigger_request_comment_id"] is not None
            else intervening_review_requests == 0 and response["commit_sha"] == request_sha
        )
    ]
    expected_manifest_responses = sorted(
        (_response_manifest_view(response) for response in trusted_responses),
        key=lambda response: (response["source"], response["id"]),
    )
    if remediation_manifest is not None:
        _validate_remediation_manifest(
            remediation_manifest,
            policy_schema_version=policy_schema_version,
            run_id=run_id,
            repository=repository,
            pr_number=pr_number,
            remote_name=remote_name,
            base_branch=base_branch,
            head_branch=head_branch,
            base_sha=request_base_sha,
            request_key=request_key,
            request_sha=request_sha,
            expected_responses=expected_manifest_responses,
            expected_actionable_threads=actionable_thread_snapshots,
            expected_execution=execution_boundary,
            safe_git_profile=policy["git_execution_profile"],
            fix_rounds=fix_rounds,
            now=now,
        )
    if finalization_manifest is not None:
        _validate_finalization_manifest(
            finalization_manifest,
            policy_schema_version=policy_schema_version,
            run_id=run_id,
            repository=repository,
            pr_number=pr_number,
            remote_name=remote_name,
            base_branch=base_branch,
            head_branch=head_branch,
            base_sha=request_base_sha,
            request_key=request_key,
            request_sha=request_sha,
            expected_responses=expected_manifest_responses,
            expected_eligible_threads=eligible_thread_snapshots,
            non_codex_threads=non_codex_threads,
            pr_is_draft=pr_is_draft,
            gates={
                "approvals": approval_gate,
                "branch_protection": branch_protection_gate,
                "checks": check_gate,
                "mergeability": mergeability_gate,
            },
            now=now,
        )

    decision: dict[str, Any] = {
        "action": "halt",
        "reason": "unclassified state",
        "request_key": request_key,
        "owned_automation_id": None,
        "may_mutate_repo": False,
        "may_mutate_github": False,
        "requires_approval": False,
        "claim_required": False,
        "delete_owned_automation": False,
        "delete_after_success": False,
        "automation_action": "none",
        "next_request_state": request_state,
        "terminal_message": None,
        "merge_ready": False,
        "post_finalization_required": False,
        "claim_request_key": None,
        "claim_plan_hash": None,
    }

    exact_owned: dict[str, str] | None = None
    if (
        len(semantic_matches) == 1
        and semantic_matches[0]["key"] == expected_loop_key
        and semantic_matches[0]["id"] == owned_automation_id
    ):
        exact_owned = semantic_matches[0]
        decision["owned_automation_id"] = owned_automation_id

    paused_checkpoint_states = {
        "processing",
        "awaiting-remediation-approval",
        "awaiting-finalization-approval",
    }
    expected_automation_status = "ACTIVE" if request_state == "pending" else "PAUSED"

    def stop(action: str, reason: str, *, delete: bool = True) -> dict[str, Any]:
        decision["action"] = action
        decision["reason"] = reason
        status_inconsistent = (
            mode != "report-only"
            and exact_owned is not None
            and exact_owned["status"] != expected_automation_status
        )
        if (
            delete
            and mode != "report-only"
            and request_state not in paused_checkpoint_states
            and request_author_id == authenticated_user_id
            and not status_inconsistent
            and exact_owned is not None
            and exact_owned["schedule"] == policy["heartbeat_rrule"]
        ):
            decision["delete_owned_automation"] = True
            decision["automation_action"] = "delete-owned"
        return decision

    if len(semantic_matches) > 1:
        return stop(
            "halt-duplicate-automations",
            f"found {len(semantic_matches)} non-deleted automations for the canonical loop key",
            delete=False,
        )
    if semantic_matches and semantic_matches[0]["key"] != expected_loop_key:
        return stop(
            "halt-noncanonical-automation",
            "an automation uses a case-variant of the canonical loop key",
            delete=False,
        )
    if mode != "report-only" and not semantic_matches:
        return stop(
            "halt-missing-automation",
            "no non-deleted automation matches the canonical loop key",
            delete=False,
        )
    if mode != "report-only" and exact_owned is None:
        return stop(
            "halt-automation-ownership",
            "recorded automation ID does not match the sole canonical loop automation",
            delete=False,
        )
    if (
        pr_base_branch != base_branch
        or pr_head_branch != head_branch
        or current_base_sha != request_base_sha
    ):
        return stop(
            "halt-pr-ref-drift",
            "pull-request base/head identity or base tip differs from the request checkpoint",
            delete=False,
        )
    if policy_schema_version != policy["schema_version"]:
        return stop("halt-policy-version", "recorded and installed policy versions differ")
    if host != policy["supported_host"] or (
        policy["same_repository_pull_requests_only"] and not same_repository_pr
    ):
        return stop(
            "halt-unsupported-target",
            "target host or pull-request topology is outside the supported policy",
        )
    if terminal_error:
        return stop("halt-terminal", terminal_error)
    if pr_state != "open":
        return stop("halt-terminal", "pull request is no longer open")
    if not head_branch_exists:
        return stop("halt-terminal", "head branch no longer exists")
    if mode != "report-only" and request_author_id != authenticated_user_id:
        return stop(
            "halt-auth-changed",
            "authenticated GitHub user differs from the canonical request author",
            delete=False,
        )
    awaiting_states = {
        "awaiting-remediation-approval",
        "awaiting-finalization-approval",
    }
    if mode == "autonomous" and request_state in awaiting_states:
        return stop(
            "halt-state-inconsistency",
            "autonomous mode cannot carry an approval-wait request state",
            delete=False,
        )
    if mode != "report-only":
        if exact_owned is None or exact_owned["status"] != expected_automation_status:
            return stop(
                "halt-state-inconsistency",
                f"request state {request_state} requires automation status {expected_automation_status}",
                delete=False,
            )
    if exact_owned is not None and exact_owned["schedule"] != policy["heartbeat_rrule"]:
        decision["action"] = "halt-automation-schedule"
        decision["reason"] = "owned automation schedule differs from the canonical heartbeat"
        if mode != "report-only" and exact_owned["status"] == "ACTIVE":
            decision["automation_action"] = "pause"
        return decision
    if request_state == "processing" and not post_finalization:
        return stop(
            "halt-processing-claim",
            "a paused processing claim needs explicit manual audit; refusing an automatic retry",
            delete=False,
        )
    if request_key in completed_keys:
        return stop(
            "halt-replayed-request",
            "request key is already completed; refusing a repeated mutation",
        )
    if current_sha != request_sha:
        return stop("halt-head-changed", "current PR head differs from the requested full SHA")

    if not trusted_responses:
        if now >= deadline:
            return stop(
                "expired",
                "no qualifying Codex response arrived within "
                f"{policy['no_response_timeout_minutes']} minutes",
            )
        decision["action"] = "wait"
        decision["reason"] = "no qualifying Codex response yet"
        if mode != "report-only" and exact_owned is not None and exact_owned["status"] == "ACTIVE":
            decision["automation_action"] = "keep"
        return decision

    latest_time = max(response["content_version_at_parsed"] for response in trusted_responses)
    latest_responses = [
        response
        for response in trusted_responses
        if response["content_version_at_parsed"] == latest_time
    ]
    latest_signals = {response["zero_findings_signal"] for response in latest_responses}
    if len(latest_signals) != 1:
        return stop(
            "halt-ambiguous-response",
            "latest qualifying Codex responses at the same timestamp conflict",
        )
    latest_zero_findings = latest_responses[0]["zero_findings_signal"]

    if actionable_threads > 0:
        if post_finalization:
            return stop(
                "halt-post-finalization-drift",
                "actionable Codex feedback appeared during finalization",
                delete=False,
            )
        if mode == "report-only":
            decision["action"] = "report-feedback"
            decision["reason"] = "actionable Codex feedback is present; report-only mode forbids writes"
            return decision
        if remediation_plan_hash is None:
            raise ValueError("actionable feedback requires remediation_plan_hash")
        if finalization_plan_hash is not None:
            raise ValueError("actionable feedback cannot carry finalization_plan_hash")
        if fix_rounds >= policy["max_fix_rounds"]:
            return stop(
                "max-rounds",
                f"{policy['max_fix_rounds']} successfully pushed feedback-fix rounds have completed",
            )
        remediation_approved = (
            request_state == "awaiting-remediation-approval"
            and approved_request_key == request_key
            and approved_remediation_plan_hash == remediation_plan_hash
        )
        if mode == "approval-gated" and not remediation_approved:
            decision["action"] = "await-remediation-approval"
            decision["reason"] = "this exact request key and remediation plan lack matching approval"
            decision["requires_approval"] = True
            decision["automation_action"] = "pause"
            decision["next_request_state"] = "awaiting-remediation-approval"
            return decision
        decision["action"] = "apply-fix"
        decision["reason"] = "latest trusted Codex state has actionable unresolved threads"
        decision["may_mutate_repo"] = True
        decision["may_mutate_github"] = True
        decision["claim_required"] = True
        decision["automation_action"] = "pause-and-claim"
        decision["next_request_state"] = "processing"
        return decision

    if remediation_plan_hash is not None:
        raise ValueError("state without actionable feedback cannot carry remediation_plan_hash")

    if latest_zero_findings:
        terminal_message = policy["terminal_message_template"].replace("<full-sha>", request_sha)
        fresh_merge_ready = (
            not pr_is_draft
            and check_gate == "pass"
            and approval_gate == "pass"
            and mergeability_gate == "pass"
            and branch_protection_gate == "pass"
            and non_codex_threads == 0
        )
        if mode == "report-only":
            decision["terminal_message"] = terminal_message
            decision["merge_ready"] = fresh_merge_ready
            decision["action"] = "report-no-findings"
            decision["reason"] = "latest trusted Codex state reports no findings for the request SHA"
            return decision
        if finalization_plan_hash is None:
            raise ValueError("no-findings state requires finalization_plan_hash")
        if post_finalization:
            if pr_is_draft or eligible_codex_threads != 0:
                return stop(
                    "halt-finalization-incomplete",
                    "draft or eligible Codex-thread finalization operations remain unapplied",
                    delete=False,
                )
            decision["action"] = "finish-finalization"
            decision["reason"] = (
                "fresh post-mutation state still reports no Codex findings for the request SHA"
            )
            decision["merge_ready"] = fresh_merge_ready
            decision["terminal_message"] = terminal_message
            decision["delete_owned_automation"] = True
            decision["automation_action"] = "delete-owned"
            return decision
        finalization_approved = (
            request_state == "awaiting-finalization-approval"
            and finalization_approved_request_key == request_key
            and approved_finalization_plan_hash == finalization_plan_hash
        )
        if mode == "approval-gated" and not finalization_approved:
            decision["action"] = "await-finalization-approval"
            decision["reason"] = "this exact request key and finalization plan lack matching approval"
            decision["requires_approval"] = True
            decision["automation_action"] = "pause"
            decision["next_request_state"] = "awaiting-finalization-approval"
            return decision
        decision["action"] = "complete-no-findings"
        decision["reason"] = "latest trusted Codex state reports no findings for the request SHA"
        decision["may_mutate_github"] = True
        decision["claim_required"] = True
        decision["automation_action"] = "pause-and-claim"
        decision["next_request_state"] = "processing"
        decision["delete_after_success"] = False
        decision["post_finalization_required"] = True
        decision["claim_request_key"] = request_key
        decision["claim_plan_hash"] = finalization_plan_hash
        return decision

    if finalization_plan_hash is not None:
        raise ValueError("inconclusive state cannot carry finalization_plan_hash")

    return stop(
        "halt-inconclusive-response",
        "latest trusted Codex state has neither actionable threads nor a no-findings signal",
    )


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--policy", type=Path, default=DEFAULT_POLICY_PATH)
    source = parser.add_mutually_exclusive_group()
    source.add_argument(
        "--state",
        type=Path,
        help="JSON state file; omit to read the state object from standard input",
    )
    source.add_argument(
        "--initial",
        type=Path,
        help="JSON initial-plan document to validate before any initial side effect",
    )
    source.add_argument(
        "--prompt",
        type=Path,
        help="JSON typed prompt-substitution document to validate before heartbeat activation",
    )
    args = parser.parse_args()
    try:
        input_path = args.initial or args.prompt or args.state
        if input_path:
            with input_path.open(encoding="utf-8") as handle:
                document = json.load(handle)
        else:
            document = json.load(sys.stdin)
        policy = load_policy(args.policy)
        if args.initial:
            result = validate_initial_document(document, policy)
        elif args.prompt:
            result = validate_prompt_document(document, policy)
        else:
            result = evaluate(document, policy)
    except (OSError, OverflowError, json.JSONDecodeError, KeyError, TypeError, ValueError) as exc:
        print(f"state evaluation failed: {exc}", file=sys.stderr)
        return 2
    print(json.dumps(result, indent=2, sort_keys=True))
    return 0


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