#!/usr/bin/env python3
"""Strict, zero-dependency primitives for ACK worker routing.

This module deliberately validates structured data rather than accepting a
shell command.  It does not invoke Orca, resolve executables, inspect Git, or
write ``tasks.yaml``.  A launcher can use the validated profile and the exact
argv renderer below, then persist its externally observed facts as a receipt.
"""

from __future__ import annotations

import hashlib
import json
import os
import re
from datetime import datetime
from pathlib import Path
from typing import Any


PROFILE_VERSION = 1
RECEIPT_VERSION = 1
LAUNCH_PROTOCOL_VERSION = 1
MAX_ROUNDS = 3

ROLES = frozenset({"developer", "test"})
CLIS = frozenset({"codex", "cursor-agent", "grok", "omp"})
TIERS = frozenset({"standard", "strong"})
REASONING_EFFORTS = frozenset({"low", "medium", "high", "xhigh"})
PERMISSION_MODES = frozenset({"read-only", "workspace-write"})
ORCHESTRATION_MODES = frozenset({"orca", "manual"})
DEFAULT_KEYS = frozenset({"developer", "test", "developerUpgraded"})
CLI_REQUIRES_REASONING_EFFORT = frozenset({"codex", "grok", "omp"})
CLI_REQUIRES_NULL_REASONING_EFFORT = frozenset({"cursor-agent"})


def _cli_choice_text() -> str:
    return "/".join(sorted(CLIS))


def executable_basename_matches_cli(executable: str, cli: str) -> bool:
    """Return whether a resolved executable basename is valid for ``cli``."""

    name = Path(executable).name
    if name == cli:
        return True
    return cli == "grok" and GROK_EXECUTABLE_NAME_RE.fullmatch(name) is not None

# allowedWorktrees 已废弃（v0.19 起）：新任务板不再配置，旧任务板遗留该字段仍被容忍。
# ORCHESTRATION_FIELDS 是“允许出现”的字段全集（含废弃字段），
# REQUIRED_ORCHESTRATION_FIELDS 是“必须存在”的子集（不含废弃字段）。
ORCHESTRATION_FIELDS = frozenset(
    {
        "profileVersion",
        "mode",
        "allowedWorktrees",
        "modelAllowlist",
        "profiles",
        "defaults",
    }
)
REQUIRED_ORCHESTRATION_FIELDS = frozenset(
    {
        "profileVersion",
        "mode",
        "modelAllowlist",
        "profiles",
        "defaults",
    }
)
PROFILE_FIELDS = frozenset(
    {
        "role",
        "cli",
        "tier",
        "model",
        "reasoningEffort",
        "permissionMode",
    }
)
RECEIPT_FIELDS = frozenset(
    {
        "receiptVersion",
        "id",
        "launchId",
        "profileId",
        "profileHash",
        "launchFingerprint",
        "slot",
        "createdFor",
        "worktree",
        "requested",
        "binding",
        "createdAt",
        "receiptHash",
    }
)
RECEIPT_CONTEXT_FIELDS = frozenset({"projectRoot", "boardHash"})
CREATED_FOR_FIELDS = frozenset({"taskId", "attemptId", "role"})
WORKTREE_FIELDS = frozenset(
    {
        "path",
        "device",
        "inode",
        "gitCommonDir",
        "gitCommonDevice",
        "gitCommonInode",
    }
)
REQUESTED_FIELDS = frozenset(
    {
        "cli",
        "tier",
        "model",
        "reasoningEffort",
        "permissionMode",
        "executable",
        "executableDevice",
        "executableInode",
        "cliVersion",
        "argv",
        "argvHash",
        "environmentPolicy",
    }
)
BINDING_FIELDS = frozenset(
    {
        "orchestrator",
        "runtimeId",
        "handle",
        "incarnationId",
        "observedWorktreePath",
        "connected",
        "writable",
        "boundAt",
    }
)
ROLE_DISPATCH_FIELDS = frozenset(
    {"profileId", "receiptId", "attemptId", "taskId", "dispatchId"}
)

PROFILE_ID_RE = re.compile(r"^[a-z][a-z0-9-]{1,63}$")
MODEL_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:/+@-]{0,127}$")
GROK_EXECUTABLE_NAME_RE = re.compile(
    r"^grok(?:-(?:linux|darwin|windows)-(?:x86_64|aarch64|arm64))?$"
)
RECEIPT_ID_RE = re.compile(r"^WR-[0-9a-f]{64}$")
LAUNCH_ID_RE = re.compile(r"^[0-9a-f]{64}$")
TASK_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$")
ATTEMPT_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*-A[1-3]$")
SHA256_RE = re.compile(r"^sha256:[0-9a-f]{64}$")


def _canonical_json(value: Any) -> str:
    """Return the one JSON representation used by all hashes in this module."""

    try:
        return json.dumps(
            value,
            ensure_ascii=False,
            sort_keys=True,
            separators=(",", ":"),
            allow_nan=False,
        )
    except (RecursionError, TypeError, ValueError) as exc:
        raise ValueError(f"value is not canonical JSON data: {exc}") from exc


def canonical_sha256(value: Any) -> str:
    """Hash canonical UTF-8 JSON and return ``sha256:<lowercase hex>``."""

    digest = hashlib.sha256(_canonical_json(value).encode("utf-8")).hexdigest()
    return f"sha256:{digest}"


def _unknown_fields(value: dict[str, Any], allowed: frozenset[str], where: str) -> list[str]:
    unknown = set(value) - allowed
    return [
        f"{where}: unknown field {field!r}"
        for field in sorted(unknown, key=repr)
    ]


def _missing_fields(value: dict[str, Any], required: frozenset[str], where: str) -> list[str]:
    return [f"{where}: missing field {field!r}" for field in sorted(required - set(value))]


def _is_nonempty_text(value: Any, *, max_length: int = 1024) -> bool:
    return (
        isinstance(value, str)
        and bool(value.strip())
        and len(value) <= max_length
        and all(ord(character) >= 32 and ord(character) != 127 for character in value)
    )


def _is_absolute_safe_path(value: Any) -> bool:
    if not _is_nonempty_text(value, max_length=4096):
        return False
    path = Path(value)
    return (
        path.is_absolute()
        and value != os.path.sep
        and not value.startswith("//")
        and ".." not in path.parts
        and os.path.normpath(value) == value
    )


def _is_nonnegative_int(value: Any) -> bool:
    return isinstance(value, int) and not isinstance(value, bool) and value >= 0


def _is_positive_int(value: Any) -> bool:
    return isinstance(value, int) and not isinstance(value, bool) and value > 0


def _is_timestamp(value: Any) -> bool:
    if not isinstance(value, str):
        return False
    try:
        parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
    except ValueError:
        return False
    return parsed.tzinfo is not None


def validate_profile(profile: Any, *, where: str = "profile") -> list[str]:
    """Validate one strict worker profile without consulting its allowlist."""

    if not isinstance(profile, dict):
        return [f"{where}: must be an object"]

    errors = _unknown_fields(profile, PROFILE_FIELDS, where)
    errors.extend(_missing_fields(profile, PROFILE_FIELDS, where))

    role = profile.get("role")
    cli = profile.get("cli")
    tier = profile.get("tier")
    model = profile.get("model")
    effort = profile.get("reasoningEffort")
    permission = profile.get("permissionMode")

    if not isinstance(role, str) or role not in ROLES:
        errors.append(f"{where}.role: must be developer/test")
    if not isinstance(cli, str) or cli not in CLIS:
        errors.append(f"{where}.cli: must be {_cli_choice_text()}")
    if not isinstance(tier, str) or tier not in TIERS:
        errors.append(f"{where}.tier: must be standard/strong")
    if not isinstance(model, str) or MODEL_ID_RE.fullmatch(model) is None:
        errors.append(f"{where}.model: must be a safe model ID")
    if not isinstance(permission, str) or permission not in PERMISSION_MODES:
        errors.append(
            f"{where}.permissionMode: must be read-only/workspace-write"
        )
    if isinstance(cli, str) and cli in CLI_REQUIRES_REASONING_EFFORT:
        if not isinstance(effort, str) or effort not in REASONING_EFFORTS:
            label = "Codex" if cli == "codex" else "Grok"
            errors.append(
                f"{where}.reasoningEffort: {label} requires low/medium/high/xhigh"
            )
    elif (
        isinstance(cli, str)
        and cli in CLI_REQUIRES_NULL_REASONING_EFFORT
        and effort is not None
    ):
        errors.append(f"{where}.reasoningEffort: Cursor requires null")

    if role == "test" and tier != "standard":
        errors.append(f"{where}.tier: Test may only use standard")
    if tier == "strong" and role != "developer":
        errors.append(f"{where}.tier: strong may only be used by Developer")

    return errors


def _validate_model_allowlist(value: Any, where: str) -> list[str]:
    if not isinstance(value, dict):
        return [f"{where}: must be an object"]

    errors: list[str] = []
    for cli in sorted(set(value) - CLIS, key=repr):
        errors.append(f"{where}: unknown CLI {cli!r}")
    for cli, roles in value.items():
        cli_where = f"{where}.{cli}"
        if cli not in CLIS:
            continue
        if not isinstance(roles, dict):
            errors.append(f"{cli_where}: must be an object")
            continue
        if not roles:
            errors.append(f"{cli_where}: must not be empty")
        for role in sorted(set(roles) - ROLES, key=repr):
            errors.append(f"{cli_where}: unknown role {role!r}")
        for role, tiers in roles.items():
            role_where = f"{cli_where}.{role}"
            if role not in ROLES:
                continue
            if not isinstance(tiers, dict):
                errors.append(f"{role_where}: must be an object")
                continue
            if not tiers:
                errors.append(f"{role_where}: must not be empty")
            for tier in sorted(set(tiers) - TIERS, key=repr):
                errors.append(f"{role_where}: unknown tier {tier!r}")
            if role == "test" and "strong" in tiers:
                errors.append(f"{role_where}: Test cannot define a strong allowlist")
            for tier, models in tiers.items():
                tier_where = f"{role_where}.{tier}"
                if tier not in TIERS:
                    continue
                if not isinstance(models, list) or not models:
                    errors.append(f"{tier_where}: must be a non-empty model list")
                    continue
                seen: set[str] = set()
                for index, model in enumerate(models):
                    item_where = f"{tier_where}[{index}]"
                    if not isinstance(model, str) or MODEL_ID_RE.fullmatch(model) is None:
                        errors.append(f"{item_where}: must be a safe model ID")
                    elif model in seen:
                        errors.append(f"{item_where}: duplicate model {model!r}")
                    else:
                        seen.add(model)
    return errors


def _allowed_models(
    allowlist: Any,
    cli: Any,
    role: Any,
    tier: Any,
) -> list[str] | None:
    if not isinstance(allowlist, dict):
        return None
    roles = allowlist.get(cli)
    if not isinstance(roles, dict):
        return None
    tiers = roles.get(role)
    if not isinstance(tiers, dict):
        return None
    models = tiers.get(tier)
    return models if isinstance(models, list) else None


def validate_orchestration(
    orchestration: Any,
    *,
    where: str = "project.orchestration",
) -> list[str]:
    """Validate the complete strict routing configuration."""

    if not isinstance(orchestration, dict):
        return [f"{where}: must be an object"]

    errors = _unknown_fields(orchestration, ORCHESTRATION_FIELDS, where)
    errors.extend(_missing_fields(orchestration, REQUIRED_ORCHESTRATION_FIELDS, where))

    profile_version = orchestration.get("profileVersion")
    mode = orchestration.get("mode")
    allowlist = orchestration.get("modelAllowlist")
    profiles = orchestration.get("profiles")
    defaults = orchestration.get("defaults")

    if profile_version != PROFILE_VERSION or isinstance(profile_version, bool):
        errors.append(f"{where}.profileVersion: must be {PROFILE_VERSION}")
    if not isinstance(mode, str) or mode not in ORCHESTRATION_MODES:
        errors.append(f"{where}.mode: must be orca/manual")

    # allowedWorktrees 已废弃：worker 默认在 --project-root 工作，
    # 其它 worktree 由 launcher 按“同 git 仓库且已注册”约束（capture_worktree_identity）。
    # 旧任务板中遗留的该字段被容忍，不再校验。

    errors.extend(_validate_model_allowlist(allowlist, f"{where}.modelAllowlist"))

    valid_profiles: dict[str, dict[str, Any]] = {}
    if not isinstance(profiles, dict):
        errors.append(f"{where}.profiles: must be an object")
    else:
        if mode == "orca" and not profiles:
            errors.append(f"{where}.profiles: Orca mode requires profiles")
        for profile_id, profile in profiles.items():
            profile_where = f"{where}.profiles.{profile_id}"
            if not isinstance(profile_id, str) or PROFILE_ID_RE.fullmatch(profile_id) is None:
                errors.append(f"{where}.profiles: invalid profile ID {profile_id!r}")
                continue
            profile_errors = validate_profile(profile, where=profile_where)
            errors.extend(profile_errors)
            if profile_errors or not isinstance(profile, dict):
                continue
            valid_profiles[profile_id] = profile
            models = _allowed_models(
                allowlist,
                profile.get("cli"),
                profile.get("role"),
                profile.get("tier"),
            )
            if profile.get("model") not in (models or []):
                errors.append(
                    f"{profile_where}.model: {profile.get('model')!r} is not allowed "
                    "for its cli/role/tier"
                )

    if not isinstance(defaults, dict):
        errors.append(f"{where}.defaults: must be an object")
    else:
        for default_key in sorted(set(defaults) - DEFAULT_KEYS, key=repr):
            errors.append(f"{where}.defaults: unknown key {default_key!r}")
        if mode == "orca":
            for role in sorted(ROLES - set(defaults)):
                errors.append(f"{where}.defaults: missing role {role!r}")
        for default_key, profile_id in defaults.items():
            default_where = f"{where}.defaults.{default_key}"
            if default_key not in DEFAULT_KEYS:
                continue
            if not isinstance(profile_id, str) or PROFILE_ID_RE.fullmatch(profile_id) is None:
                errors.append(f"{default_where}: must be a profile ID")
                continue
            profile = valid_profiles.get(profile_id)
            if profile is None:
                errors.append(f"{default_where}: unknown or invalid profile {profile_id!r}")
                continue
            expected_role = "developer" if default_key == "developerUpgraded" else default_key
            expected_tier = "strong" if default_key == "developerUpgraded" else "standard"
            if profile.get("role") != expected_role:
                errors.append(f"{default_where}: profile role must be {expected_role}")
            if profile.get("tier") != expected_tier:
                errors.append(
                    f"{default_where}: default profile must use {expected_tier} tier"
                )
            if profile.get("permissionMode") not in PERMISSION_MODES:
                errors.append(f"{default_where}: default profile has unsafe permissions")

    return errors


def profile_hash(
    profile: dict[str, Any],
    *,
    profile_version: int = PROFILE_VERSION,
) -> str:
    """Return the canonical hash of a valid strict profile."""

    errors = validate_profile(profile)
    if errors:
        raise ValueError("invalid profile: " + "; ".join(errors))
    if (
        not isinstance(profile_version, int)
        or isinstance(profile_version, bool)
        or profile_version < 1
    ):
        raise ValueError("profile_version must be a positive integer")
    return canonical_sha256(
        {
            "profileVersion": profile_version,
            "profile": profile,
        }
    )


def render_worker_argv(
    profile: dict[str, Any],
    executable: str,
    worktree: str,
) -> list[str]:
    """Render the only argv shapes allowed by routing profile version 1."""

    errors = validate_profile(profile)
    if errors:
        raise ValueError("invalid profile: " + "; ".join(errors))
    if not _is_absolute_safe_path(executable):
        raise ValueError("executable must be a safe absolute path other than root")
    if not executable_basename_matches_cli(executable, profile["cli"]):
        raise ValueError("executable basename must match profile.cli")
    if not _is_absolute_safe_path(worktree):
        raise ValueError("worktree must be a safe absolute path other than root")

    model = profile["model"]
    permission = profile["permissionMode"]
    cli = profile["cli"]
    if cli == "codex":
        return [
            executable,
            "--strict-config",
            "--model",
            model,
            "--config",
            f"model_reasoning_effort={profile['reasoningEffort']}",
            "--sandbox",
            permission,
            "--ask-for-approval",
            "never",
            "--cd",
            worktree,
        ]

    if cli == "cursor-agent":
        argv = [executable, "--model", model]
        if permission == "read-only":
            argv.extend(["--mode", "plan"])
        else:
            argv.append("--auto-review")
        argv.extend(["--sandbox", "enabled", "--workspace", worktree])
        return argv

    if cli == "omp":
        # OMP workspace-write workers default to yolo approval (rules allow it);
        # read-only workers always use always-ask.
        omp_approval = "always-ask" if permission == "read-only" else "yolo"
        return [
            executable,
            "--model",
            model,
            "--thinking",
            profile["reasoningEffort"],
            "--approval-mode",
            omp_approval,
            "--cwd",
            worktree,
            "--no-session",
        ]

    if cli != "grok":
        raise ValueError(f"unsupported cli: {cli}")
    grok_permission = "plan" if permission == "read-only" else "acceptEdits"
    grok_sandbox = "read-only" if permission == "read-only" else "workspace"
    return [
        executable,
        "--model",
        model,
        "--reasoning-effort",
        profile["reasoningEffort"],
        "--permission-mode",
        grok_permission,
        "--always-approve",
        "--sandbox",
        grok_sandbox,
        "--cwd",
        worktree,
    ]


def receipt_hash(receipt: dict[str, Any]) -> str:
    """Hash every receipt field except the self-referential ``receiptHash``."""

    if not isinstance(receipt, dict):
        raise ValueError("receipt must be an object")
    payload = {key: value for key, value in receipt.items() if key != "receiptHash"}
    return canonical_sha256(payload)


def _validate_created_for(value: Any, where: str) -> list[str]:
    if not isinstance(value, dict):
        return [f"{where}: must be an object"]
    errors = _unknown_fields(value, CREATED_FOR_FIELDS, where)
    errors.extend(_missing_fields(value, CREATED_FOR_FIELDS, where))
    task_id = value.get("taskId")
    attempt_id = value.get("attemptId")
    if not isinstance(task_id, str) or TASK_ID_RE.fullmatch(task_id) is None:
        errors.append(f"{where}.taskId: must be a safe task ID")
    if not isinstance(attempt_id, str) or ATTEMPT_ID_RE.fullmatch(attempt_id) is None:
        errors.append(f"{where}.attemptId: must use <task-id>-A1..A3")
    elif isinstance(task_id, str) and not attempt_id.startswith(f"{task_id}-A"):
        errors.append(f"{where}.attemptId: must belong to taskId")
    role = value.get("role")
    if not isinstance(role, str) or role not in ROLES:
        errors.append(f"{where}.role: must be developer/test")
    return errors


def _validate_worktree_snapshot(value: Any, where: str) -> list[str]:
    if not isinstance(value, dict):
        return [f"{where}: must be an object"]
    errors = _unknown_fields(value, WORKTREE_FIELDS, where)
    errors.extend(_missing_fields(value, WORKTREE_FIELDS, where))
    for field in ("path", "gitCommonDir"):
        if not _is_absolute_safe_path(value.get(field)):
            errors.append(f"{where}.{field}: must be a safe absolute path other than root")
    for field in ("device", "gitCommonDevice"):
        if not _is_nonnegative_int(value.get(field)):
            errors.append(f"{where}.{field}: must be a non-negative integer")
    for field in ("inode", "gitCommonInode"):
        if not _is_positive_int(value.get(field)):
            errors.append(f"{where}.{field}: must be a positive integer")
    return errors


def _validate_requested(value: Any, where: str) -> list[str]:
    if not isinstance(value, dict):
        return [f"{where}: must be an object"]
    errors = _unknown_fields(value, REQUESTED_FIELDS, where)
    errors.extend(_missing_fields(value, REQUESTED_FIELDS, where))
    cli = value.get("cli")
    tier = value.get("tier")
    permission = value.get("permissionMode")
    if not isinstance(cli, str) or cli not in CLIS:
        errors.append(f"{where}.cli: must be {_cli_choice_text()}")
    if not isinstance(tier, str) or tier not in TIERS:
        errors.append(f"{where}.tier: must be standard/strong")
    model = value.get("model")
    if not isinstance(model, str) or MODEL_ID_RE.fullmatch(model) is None:
        errors.append(f"{where}.model: must be a safe model ID")
    effort = value.get("reasoningEffort")
    if isinstance(cli, str) and cli in CLI_REQUIRES_REASONING_EFFORT and (
        not isinstance(effort, str) or effort not in REASONING_EFFORTS
    ):
        label = "Codex" if cli == "codex" else "Grok"
        errors.append(f"{where}.reasoningEffort: invalid {label} effort")
    if (
        isinstance(cli, str)
        and cli in CLI_REQUIRES_NULL_REASONING_EFFORT
        and effort is not None
    ):
        errors.append(f"{where}.reasoningEffort: Cursor requires null")
    if not isinstance(permission, str) or permission not in PERMISSION_MODES:
        errors.append(f"{where}.permissionMode: must be read-only/workspace-write")
    executable = value.get("executable")
    if not _is_absolute_safe_path(executable):
        errors.append(f"{where}.executable: must be a safe absolute path")
    elif (
        isinstance(cli, str)
        and cli in CLIS
        and not executable_basename_matches_cli(str(executable), cli)
    ):
        errors.append(f"{where}.executable: basename must match cli")
    if not _is_nonnegative_int(value.get("executableDevice")):
        errors.append(f"{where}.executableDevice: must be a non-negative integer")
    if not _is_positive_int(value.get("executableInode")):
        errors.append(f"{where}.executableInode: must be a positive integer")
    if not _is_nonempty_text(value.get("cliVersion"), max_length=256):
        errors.append(f"{where}.cliVersion: must be non-empty single-line text")
    argv = value.get("argv")
    if (
        not isinstance(argv, list)
        or len(argv) < 2
        or any(not isinstance(arg, str) for arg in argv)
    ):
        errors.append(f"{where}.argv: must be a string array with at least 2 items")
    argv_hash = value.get("argvHash")
    if not isinstance(argv_hash, str) or SHA256_RE.fullmatch(argv_hash) is None:
        errors.append(f"{where}.argvHash: must be a canonical sha256 hex digest")
    elif isinstance(argv, list) and all(isinstance(arg, str) for arg in argv):
        if argv_hash != canonical_sha256(argv):
            errors.append(f"{where}.argvHash: does not match argv")
    if value.get("environmentPolicy") != "per-cli-allowlist-v1":
        errors.append(
            f"{where}.environmentPolicy: must be 'per-cli-allowlist-v1'"
        )
    return errors


def _validate_binding(value: Any, where: str) -> list[str]:
    if not isinstance(value, dict):
        return [f"{where}: must be an object"]
    errors = _unknown_fields(value, BINDING_FIELDS, where)
    errors.extend(_missing_fields(value, BINDING_FIELDS, where))
    if value.get("orchestrator") != "orca":
        errors.append(f"{where}.orchestrator: must be orca")
    for field in ("runtimeId", "handle", "incarnationId"):
        if not _is_nonempty_text(value.get(field), max_length=512):
            errors.append(f"{where}.{field}: must be non-empty single-line text")
    if not _is_absolute_safe_path(value.get("observedWorktreePath")):
        errors.append(f"{where}.observedWorktreePath: must be a safe absolute path")
    for field in ("connected", "writable"):
        if value.get(field) is not True:
            errors.append(f"{where}.{field}: must be true")
    if not _is_timestamp(value.get("boundAt")):
        errors.append(f"{where}.boundAt: must be a timezone-aware ISO 8601 timestamp")
    return errors


def validate_worker_receipt(
    receipt: Any,
    *,
    orchestration: dict[str, Any] | None = None,
    task_ids: set[str] | None = None,
    where: str = "workerReceipt",
) -> list[str]:
    """Validate one audit receipt's structure/checksums and routing references.

    This is not origin authentication: ``receiptHash`` is unkeyed, so callers
    must never use this result alone to authorize reuse of an old terminal.
    """

    if not isinstance(receipt, dict):
        return [f"{where}: must be an object"]
    errors = _unknown_fields(receipt, RECEIPT_FIELDS | RECEIPT_CONTEXT_FIELDS, where)
    errors.extend(_missing_fields(receipt, RECEIPT_FIELDS, where))
    project_root = receipt.get("projectRoot")
    board_hash = receipt.get("boardHash")
    if (project_root is None) != (board_hash is None):
        errors.append(f"{where}: projectRoot and boardHash must be present together")
    if project_root is not None and (
        not isinstance(project_root, str) or not project_root.startswith("/")
    ):
        errors.append(f"{where}.projectRoot: must be an absolute path")
    if board_hash is not None and (
        not isinstance(board_hash, str) or SHA256_RE.fullmatch(board_hash) is None
    ):
        errors.append(f"{where}.boardHash: must be a canonical sha256 hex digest")

    version = receipt.get("receiptVersion")
    if version != RECEIPT_VERSION or isinstance(version, bool):
        errors.append(f"{where}.receiptVersion: must be {RECEIPT_VERSION}")
    for field in ("id", "launchId", "profileId"):
        value = receipt.get(field)
        matcher = {
            "id": RECEIPT_ID_RE,
            "launchId": LAUNCH_ID_RE,
            "profileId": PROFILE_ID_RE,
        }[field]
        if not isinstance(value, str) or matcher.fullmatch(value) is None:
            errors.append(f"{where}.{field}: invalid identifier")
    for field in ("profileHash", "launchFingerprint", "receiptHash"):
        value = receipt.get(field)
        if not isinstance(value, str) or SHA256_RE.fullmatch(value) is None:
            errors.append(f"{where}.{field}: must be a canonical sha256 hex digest")
    slot = receipt.get("slot")
    if not isinstance(slot, int) or isinstance(slot, bool) or not 1 <= slot <= 99:
        errors.append(f"{where}.slot: must be an integer from 1 to 99")
    receipt_id = receipt.get("id")
    launch_id = receipt.get("launchId")
    if (
        isinstance(receipt_id, str)
        and RECEIPT_ID_RE.fullmatch(receipt_id)
        and isinstance(launch_id, str)
        and LAUNCH_ID_RE.fullmatch(launch_id)
        and receipt_id != f"WR-{launch_id}"
    ):
        errors.append(f"{where}.id: must equal 'WR-' + launchId")

    created_for = receipt.get("createdFor")
    worktree = receipt.get("worktree")
    requested = receipt.get("requested")
    binding = receipt.get("binding")
    errors.extend(_validate_created_for(created_for, f"{where}.createdFor"))
    errors.extend(_validate_worktree_snapshot(worktree, f"{where}.worktree"))
    errors.extend(_validate_requested(requested, f"{where}.requested"))
    errors.extend(_validate_binding(binding, f"{where}.binding"))
    if not _is_timestamp(receipt.get("createdAt")):
        errors.append(f"{where}.createdAt: must be a timezone-aware ISO 8601 timestamp")

    if isinstance(created_for, dict) and task_ids is not None:
        task_id = created_for.get("taskId")
        if isinstance(task_id, str) and task_id not in task_ids:
            errors.append(f"{where}.createdFor.taskId: unknown task {task_id!r}")

    profile: dict[str, Any] | None = None
    if orchestration is not None and not isinstance(orchestration, dict):
        errors.append(f"{where}: orchestration must be an object")
    elif isinstance(orchestration, dict):
        profiles = orchestration.get("profiles")
        profile_id = receipt.get("profileId")
        candidate = profiles.get(profile_id) if (
            isinstance(profiles, dict) and isinstance(profile_id, str)
        ) else None
        if isinstance(candidate, dict):
            profile = candidate
        else:
            errors.append(f"{where}.profileId: unknown profile {profile_id!r}")
    if profile is not None:
        try:
            expected_profile_hash = profile_hash(
                profile,
                profile_version=orchestration.get(
                    "profileVersion",
                    PROFILE_VERSION,
                ),
            )
        except ValueError:
            errors.append(f"{where}.profileId: referenced profile is invalid")
        else:
            if receipt.get("profileHash") != expected_profile_hash:
                errors.append(f"{where}.profileHash: does not match profile")

        if isinstance(created_for, dict) and created_for.get("role") != profile.get("role"):
            errors.append(f"{where}.createdFor.role: does not match profile")
        if isinstance(requested, dict):
            for field in (
                "cli",
                "tier",
                "model",
                "reasoningEffort",
                "permissionMode",
            ):
                if requested.get(field) != profile.get(field):
                    errors.append(f"{where}.requested.{field}: does not match profile")
            executable = requested.get("executable")
            worktree_path = worktree.get("path") if isinstance(worktree, dict) else None
            if isinstance(executable, str) and isinstance(worktree_path, str):
                try:
                    expected_argv = render_worker_argv(profile, executable, worktree_path)
                except ValueError:
                    errors.append(f"{where}.requested.argv: cannot render referenced profile")
                else:
                    if requested.get("argv") != expected_argv:
                        errors.append(f"{where}.requested.argv: does not match exact renderer")

    if isinstance(binding, dict) and isinstance(worktree, dict):
        if binding.get("observedWorktreePath") != worktree.get("path"):
            errors.append(f"{where}.binding.observedWorktreePath: does not match worktree.path")

    if (
        isinstance(created_for, dict)
        and isinstance(worktree, dict)
        and isinstance(requested, dict)
    ):
        try:
            facts = {
                "protocolVersion": LAUNCH_PROTOCOL_VERSION,
                "backend": "orca",
                "profileId": receipt.get("profileId"),
                "profileHash": receipt.get("profileHash"),
                "createdFor": created_for,
                "worktree": worktree,
                "requested": requested,
                "slot": slot,
            }
            if project_root is not None and board_hash is not None:
                facts["projectRoot"] = project_root
                facts["boardHash"] = board_hash
            expected_fingerprint = canonical_sha256(facts)
        except ValueError:
            errors.append(f"{where}.launchFingerprint: cannot hash launch facts")
        else:
            if receipt.get("launchFingerprint") != expected_fingerprint:
                errors.append(
                    f"{where}.launchFingerprint: does not match launch facts"
                )

    try:
        expected_receipt_hash = receipt_hash(receipt)
    except ValueError:
        errors.append(f"{where}: must contain canonical JSON data")
    else:
        if receipt.get("receiptHash") != expected_receipt_hash:
            errors.append(f"{where}.receiptHash: does not match receipt")
    return errors


def validate_worker_receipts(
    receipts: Any,
    orchestration: dict[str, Any] | None = None,
    *,
    task_ids: set[str] | None = None,
    where: str = "workerReceipts",
) -> list[str]:
    """Validate the top-level receipt list and reject duplicate identities."""

    if not isinstance(receipts, list):
        return [f"{where}: must be a list"]
    errors: list[str] = []
    seen_ids: set[str] = set()
    seen_launch_ids: set[str] = set()
    for index, receipt in enumerate(receipts):
        receipt_where = f"{where}[{index}]"
        errors.extend(
            validate_worker_receipt(
                receipt,
                orchestration=orchestration,
                task_ids=task_ids,
                where=receipt_where,
            )
        )
        if not isinstance(receipt, dict):
            continue
        receipt_id = receipt.get("id")
        if isinstance(receipt_id, str):
            if receipt_id in seen_ids:
                errors.append(f"{receipt_where}.id: duplicate receipt ID {receipt_id!r}")
            seen_ids.add(receipt_id)
        launch_id = receipt.get("launchId")
        if isinstance(launch_id, str):
            if launch_id in seen_launch_ids:
                errors.append(f"{receipt_where}.launchId: duplicate launch ID {launch_id!r}")
            seen_launch_ids.add(launch_id)
    return errors


def _validate_dispatch_links(
    tasks: Any,
    receipts: Any,
    orchestration: Any,
) -> list[str]:
    """Cross-link role dispatch records to persisted audit receipts."""

    if not isinstance(tasks, list):
        return []
    receipt_by_id: dict[str, dict[str, Any]] = {}
    if isinstance(receipts, list):
        for receipt in receipts:
            if not isinstance(receipt, dict):
                continue
            receipt_id = receipt.get("id")
            if isinstance(receipt_id, str) and receipt_id not in receipt_by_id:
                receipt_by_id[receipt_id] = receipt
    profiles = orchestration.get("profiles") if isinstance(orchestration, dict) else None

    errors: list[str] = []
    for task_index, task in enumerate(tasks):
        if not isinstance(task, dict):
            continue
        ack_task_id = task.get("id")
        dispatch = task.get("dispatch")
        if not isinstance(dispatch, dict):
            continue
        for role in sorted(ROLES):
            if role not in dispatch:
                continue
            role_dispatch = dispatch.get(role)
            where = f"tasks[{task_index}].dispatch.{role}"
            if not isinstance(role_dispatch, dict):
                errors.append(f"{where}: must be an object")
                continue
            errors.extend(_unknown_fields(role_dispatch, ROLE_DISPATCH_FIELDS, where))
            errors.extend(_missing_fields(role_dispatch, ROLE_DISPATCH_FIELDS, where))

            profile_id = role_dispatch.get("profileId")
            if profile_id is not None:
                if not isinstance(profile_id, str) or PROFILE_ID_RE.fullmatch(profile_id) is None:
                    errors.append(f"{where}.profileId: must be null or a profile ID")
                else:
                    profile = profiles.get(profile_id) if isinstance(profiles, dict) else None
                    if not isinstance(profile, dict):
                        errors.append(f"{where}.profileId: unknown profile {profile_id!r}")
                    elif profile.get("role") != role:
                        errors.append(f"{where}.profileId: profile role must be {role}")

            runtime_values: list[Any] = []
            for field in ("taskId", "dispatchId"):
                runtime_value = role_dispatch.get(field)
                runtime_values.append(runtime_value)
                if runtime_value is not None and not _is_nonempty_text(
                    runtime_value,
                    max_length=512,
                ):
                    errors.append(
                        f"{where}.{field}: must be null or non-empty single-line text"
                    )
            if (runtime_values[0] is None) != (runtime_values[1] is None):
                errors.append(
                    f"{where}: taskId and dispatchId must both be null or both be set"
                )

            receipt_id = role_dispatch.get("receiptId")
            attempt_id = role_dispatch.get("attemptId")
            current_task_attempts = {
                f"{ack_task_id}-A{round_number}" for round_number in range(1, 4)
            }
            if attempt_id is not None:
                if (
                    not isinstance(attempt_id, str)
                    or ATTEMPT_ID_RE.fullmatch(attempt_id) is None
                ):
                    errors.append(
                        f"{where}.attemptId: must be null or use <task-id>-A1..A3"
                    )
                elif (
                    isinstance(ack_task_id, str)
                    and attempt_id not in current_task_attempts
                ):
                    errors.append(
                        f"{where}.attemptId: must belong to current ACK task {ack_task_id!r}"
                    )
            if receipt_id is None:
                if attempt_id is not None:
                    errors.append(
                        f"{where}.attemptId: must be null when receiptId is null"
                    )
                if any(value is not None for value in runtime_values):
                    errors.append(
                        f"{where}.receiptId: is required before runtime dispatch IDs"
                    )
                continue
            if attempt_id is None:
                errors.append(
                    f"{where}.attemptId: is required when receiptId is set"
                )
            if not isinstance(receipt_id, str) or RECEIPT_ID_RE.fullmatch(receipt_id) is None:
                errors.append(f"{where}.receiptId: must be null or a receipt ID")
                continue
            receipt = receipt_by_id.get(receipt_id)
            if receipt is None:
                errors.append(f"{where}.receiptId: unknown receipt {receipt_id!r}")
                continue
            if receipt.get("profileId") != profile_id:
                errors.append(f"{where}.profileId: does not match referenced receipt")
            created_for = receipt.get("createdFor")
            receipt_task_id = (
                created_for.get("taskId") if isinstance(created_for, dict) else None
            )
            receipt_role = created_for.get("role") if isinstance(created_for, dict) else None
            receipt_attempt_id = (
                created_for.get("attemptId") if isinstance(created_for, dict) else None
            )
            if receipt_task_id != ack_task_id:
                errors.append(
                    f"{where}.receiptId: referenced receipt task must be current "
                    f"ACK task {ack_task_id!r}"
                )
            if receipt_role != role:
                errors.append(f"{where}.receiptId: referenced receipt role must be {role}")
            if receipt_attempt_id != attempt_id:
                errors.append(
                    f"{where}.attemptId: does not match referenced receipt"
                )
    return errors


def validate_routing_document(data: Any) -> list[str]:
    """Validate routing and receipts embedded in an ACK tasks document."""

    if not isinstance(data, dict):
        return ["<root>: must be an object"]
    project = data.get("project")
    if not isinstance(project, dict):
        return ["project: must be an object"]
    if "orchestration" not in project:
        return ["project.orchestration: is required"]
    orchestration = project.get("orchestration")
    errors = validate_orchestration(orchestration)

    task_ids: set[str] = set()
    tasks = data.get("tasks")
    if isinstance(tasks, list):
        for task in tasks:
            if isinstance(task, dict) and isinstance(task.get("id"), str):
                task_ids.add(task["id"])

    receipts = data.get("workerReceipts")
    if "workerReceipts" not in data:
        errors.append("workerReceipts: is required")
    else:
        errors.extend(
            validate_worker_receipts(
                receipts,
                orchestration if isinstance(orchestration, dict) else None,
                task_ids=task_ids,
            )
        )
    errors.extend(_validate_dispatch_links(tasks, receipts, orchestration))
    if isinstance(orchestration, dict) and orchestration.get("mode") == "manual":
        receipts = data.get("workerReceipts")
        if isinstance(receipts, list) and receipts:
            errors.append("workerReceipts: manual orchestration requires an empty list")
    return errors


# Short aliases for callers that treat this module as a builtin validator.
validate_routing = validate_orchestration
validate_receipts = validate_worker_receipts
validate_builtin = validate_routing_document


__all__ = [
    "PROFILE_VERSION",
    "RECEIPT_VERSION",
    "LAUNCH_PROTOCOL_VERSION",
    "MAX_ROUNDS",
    "canonical_sha256",
    "profile_hash",
    "receipt_hash",
    "render_worker_argv",
    "executable_basename_matches_cli",
    "validate_profile",
    "validate_orchestration",
    "validate_worker_receipt",
    "validate_worker_receipts",
    "validate_routing_document",
    "validate_routing",
    "validate_receipts",
    "validate_builtin",
]