scripts/project_process.py - main - yrostov/project-process

scripts/project_process.py
750 строк36 KB
urostov
feat: add fail-closed project process gate
15 июл 2026, 13:41
15 июл 2026, 13:41c919908
#!/usr/bin/env python3
"""Fail-closed validator for project-process manifests.

The validator intentionally uses only Python's standard library. Unknown input,
unsupported schema versions and environment errors are fatal.
"""

from __future__ import annotations

import argparse
import fnmatch
import json
import os
import re
import subprocess
import sys
import tempfile
import xml.etree.ElementTree as ET
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Iterable


FULL_SHA = re.compile(r"^[0-9a-f]{40}$")
REQ_ID = re.compile(r"^REQ-[A-Z0-9]+(?:-[A-Z0-9]+)+$")
TASK_ID = re.compile(r"^[A-Z][A-Z0-9]+-[0-9]+$")
PLACEHOLDER = re.compile(r"(?:\bTBD\b|\bTODO\b|Нужно заполнить)", re.IGNORECASE)
MINIMUM_PROTECTED_PATHS = (
    ".process/**",
    ".github/workflows/project-process.yml",
    ".github/CODEOWNERS",
    "scripts/project_process.py",
)


class ValidationFailure(Exception):
    """Raised for a deterministic process validation failure."""


def _reject_duplicate_keys(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
    result: dict[str, Any] = {}
    for key, value in pairs:
        if key in result:
            raise ValidationFailure(f"duplicate JSON key: {key}")
        result[key] = value
    return result


def load_json_bytes(raw: bytes, source: str) -> dict[str, Any]:
    try:
        value = json.loads(raw.decode("utf-8"), object_pairs_hook=_reject_duplicate_keys)
    except (UnicodeDecodeError, json.JSONDecodeError, ValidationFailure) as exc:
        raise ValidationFailure(f"{source}: invalid JSON: {exc}") from exc
    if not isinstance(value, dict):
        raise ValidationFailure(f"{source}: top-level value must be an object")
    return value


def load_json(path: Path) -> dict[str, Any]:
    try:
        return load_json_bytes(path.read_bytes(), str(path))
    except OSError as exc:
        raise ValidationFailure(f"{path}: cannot read: {exc}") from exc


def run_git(root: Path, args: list[str], *, text: bool = True) -> str | bytes:
    try:
        completed = subprocess.run(
            ["git", "-C", str(root), *args],
            check=False,
            capture_output=True,
            text=text,
            timeout=30,
        )
    except (OSError, subprocess.TimeoutExpired) as exc:
        raise ValidationFailure(f"git {' '.join(args)} failed: {exc}") from exc
    if completed.returncode != 0:
        stderr = completed.stderr if text else completed.stderr.decode("utf-8", "replace")
        raise ValidationFailure(f"git {' '.join(args)} failed: {stderr.strip()}")
    return completed.stdout


def require_string(value: Any, field: str, errors: list[str]) -> None:
    if not isinstance(value, str) or not value.strip():
        errors.append(f"{field}: must be a non-empty string")
    elif PLACEHOLDER.search(value):
        errors.append(f"{field}: placeholder is forbidden")


def require_string_list(value: Any, field: str, errors: list[str], *, nonempty: bool = True) -> None:
    if not isinstance(value, list) or (nonempty and not value):
        errors.append(f"{field}: must be {'a non-empty ' if nonempty else 'an '}array")
        return
    for index, item in enumerate(value):
        require_string(item, f"{field}[{index}]", errors)


def matches(path: str, patterns: Iterable[str]) -> bool:
    return any(fnmatch.fnmatchcase(path, pattern) for pattern in patterns)


def is_safe_relative(value: str) -> bool:
    path = Path(value)
    return bool(value) and not path.is_absolute() and ".." not in path.parts


def iter_manifest_files(root: Path, directories: list[str]) -> Iterable[Path]:
    for directory in directories:
        base = root / directory
        if not base.exists():
            continue
        if not base.is_dir():
            raise ValidationFailure(f"{directory}: manifest location is not a directory")
        yield from sorted(base.rglob("*.json"))


@dataclass
class Model:
    config: dict[str, Any]
    requirements: dict[str, dict[str, Any]]
    requirement_paths: dict[str, str]
    tasks: dict[str, dict[str, Any]]
    task_paths: dict[str, str]
    errors: list[str]


def validate_config(config: dict[str, Any], source: str) -> list[str]:
    errors: list[str] = []
    allowed = {
        "$schema", "schema_version", "required_documents", "requirement_directories",
        "task_directories", "protected_paths", "ignored_paths", "planning_paths",
        "impact_rules", "checks",
    }
    unknown = sorted(set(config) - allowed)
    if unknown:
        errors.append(f"{source}: unknown fields: {', '.join(unknown)}")
    if config.get("schema_version") != 1:
        errors.append(f"{source}: unsupported schema_version")
    for field in ("required_documents", "requirement_directories", "task_directories"):
        require_string_list(config.get(field), f"{source}:{field}", errors)
    for field in ("protected_paths", "ignored_paths"):
        require_string_list(config.get(field), f"{source}:{field}", errors, nonempty=False)
    require_string_list(config.get("planning_paths"), f"{source}:planning_paths", errors)
    for field in (
        "required_documents", "requirement_directories", "task_directories",
        "protected_paths", "ignored_paths", "planning_paths",
    ):
        for value in config.get(field, []):
            if isinstance(value, str) and not is_safe_relative(value):
                errors.append(f"{source}:{field}: unsafe path {value}")
    protected = config.get("protected_paths", [])
    ignored = config.get("ignored_paths", [])
    overlap = sorted(set(protected) & set(ignored))
    if overlap:
        errors.append(f"{source}: protected_paths and ignored_paths overlap: {', '.join(overlap)}")
    for protected_sample in (".process/config.json", ".github/workflows/project-process.yml", ".github/CODEOWNERS", "scripts/project_process.py"):
        if matches(protected_sample, ignored):
            errors.append(f"{source}: ignored_paths hides protected path {protected_sample}")

    rules = config.get("impact_rules")
    if not isinstance(rules, list) or not rules:
        errors.append(f"{source}:impact_rules: must be a non-empty array")
    else:
        for index, rule in enumerate(rules):
            prefix = f"{source}:impact_rules[{index}]"
            if not isinstance(rule, dict) or set(rule) != {"paths", "require_changed_documents"}:
                errors.append(f"{prefix}: requires exactly paths and require_changed_documents")
                continue
            require_string_list(rule.get("paths"), f"{prefix}:paths", errors)
            require_string_list(
                rule.get("require_changed_documents"),
                f"{prefix}:require_changed_documents", errors, nonempty=False,
            )
            for field in ("paths", "require_changed_documents"):
                for value in rule.get(field, []):
                    if isinstance(value, str) and not is_safe_relative(value):
                        errors.append(f"{prefix}:{field}: unsafe path {value}")

    checks = config.get("checks")
    seen_checks: set[str] = set()
    if not isinstance(checks, list) or not checks:
        errors.append(f"{source}:checks: must be a non-empty array")
    else:
        for index, check in enumerate(checks):
            prefix = f"{source}:checks[{index}]"
            if not isinstance(check, dict) or set(check) != {"id", "command", "result"}:
                errors.append(f"{prefix}: requires exactly id, command and result")
                continue
            require_string(check.get("id"), f"{prefix}:id", errors)
            require_string_list(check.get("command"), f"{prefix}:command", errors)
            result = check.get("result")
            if not isinstance(result, dict) or set(result) != {"format", "path", "minimum_executed"}:
                errors.append(f"{prefix}:result requires exactly format, path and minimum_executed")
            else:
                if result.get("format") != "junit":
                    errors.append(f"{prefix}:result: only junit is supported")
                require_string(result.get("path"), f"{prefix}:result:path", errors)
                if isinstance(result.get("path"), str) and not is_safe_relative(result["path"]):
                    errors.append(f"{prefix}:result:path must stay inside the external evidence directory")
                if not isinstance(result.get("minimum_executed"), int) or result["minimum_executed"] < 1:
                    errors.append(f"{prefix}:result:minimum_executed must be an integer >= 1")
            check_id = check.get("id")
            if isinstance(check_id, str):
                normalized = check_id.casefold()
                if normalized in seen_checks:
                    errors.append(f"{prefix}: duplicate check ID {check_id}")
                seen_checks.add(normalized)
    return errors


def load_policy(root: Path, policy_ref: str | None) -> tuple[dict[str, Any], str]:
    relative = ".process/config.json"
    if policy_ref:
        raw = run_git(root, ["show", f"{policy_ref}:{relative}"], text=False)
        assert isinstance(raw, bytes)
        return load_json_bytes(raw, f"{policy_ref}:{relative}"), f"{policy_ref}:{relative}"
    path = root / relative
    return load_json(path), relative


def build_model(root: Path, policy_ref: str | None = None) -> Model:
    config, config_source = load_policy(root, policy_ref)
    errors = validate_config(config, config_source)

    for document in config.get("required_documents", []):
        path = root / document
        if not path.is_file():
            errors.append(f"required document missing: {document}")
            continue
        try:
            content = path.read_text(encoding="utf-8")
        except (OSError, UnicodeDecodeError) as exc:
            errors.append(f"required document unreadable: {document}: {exc}")
            continue
        if not content.strip():
            errors.append(f"required document empty: {document}")
        if PLACEHOLDER.search(content):
            errors.append(f"required document contains placeholder: {document}")

    requirements: dict[str, dict[str, Any]] = {}
    requirement_paths: dict[str, str] = {}
    id_casefold: dict[str, str] = {}
    for path in iter_manifest_files(root, config.get("requirement_directories", [])):
        relative = path.relative_to(root).as_posix()
        try:
            item = load_json(path)
        except ValidationFailure as exc:
            errors.append(str(exc))
            continue
        allowed = {"$schema", "schema_version", "id", "title", "status", "statement"}
        unknown = sorted(set(item) - allowed)
        if unknown:
            errors.append(f"{relative}: unknown fields: {', '.join(unknown)}")
        if item.get("schema_version") != 1:
            errors.append(f"{relative}: unsupported schema_version")
        req_id = item.get("id")
        if not isinstance(req_id, str) or not REQ_ID.fullmatch(req_id):
            errors.append(f"{relative}: invalid requirement ID")
            continue
        normalized = req_id.casefold()
        if normalized in id_casefold:
            errors.append(f"duplicate ID ignoring case: {req_id} and {id_casefold[normalized]}")
            continue
        id_casefold[normalized] = req_id
        if path.stem != req_id:
            errors.append(f"{relative}: filename must equal requirement ID {req_id}.json")
        if item.get("status") not in {"active", "deprecated"}:
            errors.append(f"{relative}: invalid requirement status")
        require_string(item.get("title"), f"{relative}:title", errors)
        require_string(item.get("statement"), f"{relative}:statement", errors)
        requirements[req_id] = item
        requirement_paths[req_id] = relative

    checks = {check["id"]: check for check in config.get("checks", []) if isinstance(check, dict) and isinstance(check.get("id"), str)}
    tasks: dict[str, dict[str, Any]] = {}
    task_paths: dict[str, str] = {}
    for path in iter_manifest_files(root, config.get("task_directories", [])):
        relative = path.relative_to(root).as_posix()
        try:
            item = load_json(path)
        except ValidationFailure as exc:
            errors.append(str(exc))
            continue
        allowed = {
            "$schema", "schema_version", "id", "title", "type", "priority", "status",
            "source_of_truth", "goal", "out_of_scope", "affected_paths", "requirement_ids",
            "acceptance_criteria", "verifications", "documents_to_update", "merged_commit_sha",
        }
        unknown = sorted(set(item) - allowed)
        if unknown:
            errors.append(f"{relative}: unknown fields: {', '.join(unknown)}")
        if item.get("schema_version") != 1:
            errors.append(f"{relative}: unsupported schema_version")
        task_id = item.get("id")
        if not isinstance(task_id, str) or not TASK_ID.fullmatch(task_id):
            errors.append(f"{relative}: invalid task ID")
            continue
        normalized = task_id.casefold()
        if normalized in id_casefold:
            errors.append(f"duplicate ID ignoring case: {task_id} and {id_casefold[normalized]}")
            continue
        id_casefold[normalized] = task_id
        require_string(item.get("title"), f"{relative}:title", errors)
        require_string(item.get("goal"), f"{relative}:goal", errors)
        for field in ("source_of_truth", "out_of_scope", "affected_paths", "requirement_ids", "documents_to_update"):
            require_string_list(item.get(field), f"{relative}:{field}", errors)
        for source_path in item.get("source_of_truth", []):
            if not isinstance(source_path, str) or not is_safe_relative(source_path):
                errors.append(f"{relative}: unsafe source_of_truth path {source_path}")
            elif not (root / source_path).is_file():
                errors.append(f"{relative}: source_of_truth does not exist: {source_path}")
            elif not (root / source_path).resolve().is_relative_to(root.resolve()):
                errors.append(f"{relative}: source_of_truth escapes repository: {source_path}")
        for affected_path in item.get("affected_paths", []):
            if isinstance(affected_path, str) and not is_safe_relative(affected_path):
                errors.append(f"{relative}: unsafe affected_paths pattern {affected_path}")
        for document_path in item.get("documents_to_update", []):
            if isinstance(document_path, str) and not is_safe_relative(document_path):
                errors.append(f"{relative}: unsafe documents_to_update path {document_path}")
        if item.get("type") not in {"story", "task", "bug"}:
            errors.append(f"{relative}: invalid type")
        if item.get("priority") not in {"critical", "high", "medium", "low"}:
            errors.append(f"{relative}: invalid priority")
        status = item.get("status")
        if status not in {"open", "in_progress", "ready_for_verification", "archived"}:
            errors.append(f"{relative}: invalid status")
        if status == "archived" and not FULL_SHA.fullmatch(str(item.get("merged_commit_sha", ""))):
            errors.append(f"{relative}: archived task requires full merged_commit_sha")
        if status != "archived" and "merged_commit_sha" in item:
            errors.append(f"{relative}: merged_commit_sha is allowed only for archived tasks")

        for req_id in item.get("requirement_ids", []):
            if req_id not in requirements:
                errors.append(f"{relative}: dangling requirement reference {req_id}")

        verifications = item.get("verifications")
        verification_ids: set[str] = set()
        if not isinstance(verifications, list) or not verifications:
            errors.append(f"{relative}:verifications must be a non-empty array")
            verifications = []
        for index, verification in enumerate(verifications):
            prefix = f"{relative}:verifications[{index}]"
            if not isinstance(verification, dict) or set(verification) != {"id", "kind", "check_id"}:
                errors.append(f"{prefix}: requires exactly id, kind and check_id")
                continue
            verification_id = verification.get("id")
            expected_prefix = f"VERIFY-{task_id}-"
            if not isinstance(verification_id, str) or not verification_id.startswith(expected_prefix):
                errors.append(f"{prefix}: invalid verification ID")
                continue
            normalized_verification = verification_id.casefold()
            if normalized_verification in id_casefold:
                errors.append(f"duplicate ID ignoring case: {verification_id} and {id_casefold[normalized_verification]}")
            else:
                id_casefold[normalized_verification] = verification_id
            verification_ids.add(verification_id)
            if verification.get("kind") != "automated":
                errors.append(f"{prefix}: strict MVP supports automated verification only")
            if verification.get("check_id") not in checks:
                errors.append(f"{prefix}: unknown check_id {verification.get('check_id')}")

        criteria = item.get("acceptance_criteria")
        if not isinstance(criteria, list) or not criteria:
            errors.append(f"{relative}:acceptance_criteria must be a non-empty array")
            criteria = []
        used_verifications: set[str] = set()
        for index, criterion in enumerate(criteria):
            prefix = f"{relative}:acceptance_criteria[{index}]"
            if not isinstance(criterion, dict) or set(criterion) != {"id", "text", "verification_ids"}:
                errors.append(f"{prefix}: requires exactly id, text and verification_ids")
                continue
            criterion_id = criterion.get("id")
            expected_prefix = f"AC-{task_id}-"
            if not isinstance(criterion_id, str) or not criterion_id.startswith(expected_prefix):
                errors.append(f"{prefix}: invalid acceptance criterion ID")
            else:
                normalized_criterion = criterion_id.casefold()
                if normalized_criterion in id_casefold:
                    errors.append(f"duplicate ID ignoring case: {criterion_id} and {id_casefold[normalized_criterion]}")
                else:
                    id_casefold[normalized_criterion] = criterion_id
            require_string(criterion.get("text"), f"{prefix}:text", errors)
            require_string_list(criterion.get("verification_ids"), f"{prefix}:verification_ids", errors)
            for verification_id in criterion.get("verification_ids", []):
                if verification_id not in verification_ids:
                    errors.append(f"{prefix}: dangling verification reference {verification_id}")
                used_verifications.add(verification_id)
        for orphan in sorted(verification_ids - used_verifications):
            errors.append(f"{relative}: orphan verification {orphan}")
        tasks[task_id] = item
        task_paths[task_id] = relative

    return Model(config, requirements, requirement_paths, tasks, task_paths, errors)


def load_ref_manifests(root: Path, ref: str, directories: list[str]) -> tuple[dict[str, dict[str, Any]], dict[str, str]]:
    output = run_git(root, ["ls-tree", "-r", "--name-only", ref, "--", *directories])
    assert isinstance(output, str)
    items: dict[str, dict[str, Any]] = {}
    paths: dict[str, str] = {}
    for relative in sorted(line for line in output.splitlines() if line.endswith(".json")):
        raw = run_git(root, ["show", f"{ref}:{relative}"], text=False)
        assert isinstance(raw, bytes)
        item = load_json_bytes(raw, f"{ref}:{relative}")
        item_id = item.get("id")
        if not isinstance(item_id, str):
            raise ValidationFailure(f"{ref}:{relative}: manifest has no string ID")
        normalized = item_id.casefold()
        if normalized in {value.casefold() for value in items}:
            raise ValidationFailure(f"{ref}: duplicate manifest ID ignoring case: {item_id}")
        items[item_id] = item
        paths[item_id] = relative
    return items, paths


def validate_archives(
    root: Path,
    tasks: dict[str, dict[str, Any]],
    ancestor: str,
    task_directories: list[str],
) -> list[str]:
    errors: list[str] = []
    for task_id, task in tasks.items():
        if task.get("status") != "archived":
            continue
        sha = task.get("merged_commit_sha")
        if not isinstance(sha, str) or not FULL_SHA.fullmatch(sha):
            continue
        try:
            verify_revision(root, sha, f"{task_id}.merged_commit_sha")
            run_git(root, ["merge-base", "--is-ancestor", sha, ancestor])
        except ValidationFailure:
            errors.append(f"{task_id}: merged_commit_sha must exist and be an ancestor of {ancestor}")
            continue
        try:
            merged_tasks, _ = load_ref_manifests(root, sha, task_directories)
        except ValidationFailure as exc:
            errors.append(f"{task_id}: cannot inspect merged task revision: {exc}")
            continue
        merged_task = merged_tasks.get(task_id)
        if not merged_task or merged_task.get("status") != "ready_for_verification":
            errors.append(f"{task_id}: merged_commit_sha does not contain this task in ready_for_verification")
            continue
        archived_definition = {key: value for key, value in task.items() if key not in {"status", "merged_commit_sha"}}
        merged_definition = {key: value for key, value in merged_task.items() if key not in {"status", "merged_commit_sha"}}
        if archived_definition != merged_definition:
            errors.append(f"{task_id}: archived task definition differs from merged ready revision")
    return errors


def verify_revision(root: Path, sha: str, label: str) -> None:
    if not FULL_SHA.fullmatch(sha):
        raise ValidationFailure(f"{label} must be a full lowercase 40-character Git SHA")
    resolved = run_git(root, ["rev-parse", f"{sha}^{{commit}}"])
    assert isinstance(resolved, str)
    if resolved.strip() != sha:
        raise ValidationFailure(f"{label} does not resolve exactly to {sha}")


def changed_paths(root: Path, base: str, head: str) -> list[str]:
    verify_revision(root, base, "base")
    verify_revision(root, head, "head")
    try:
        run_git(root, ["merge-base", "--is-ancestor", base, head])
    except ValidationFailure as exc:
        raise ValidationFailure("base must be an ancestor of head") from exc
    raw = run_git(root, ["diff", "--name-status", "-z", base, head], text=False)
    assert isinstance(raw, bytes)
    parts = raw.decode("utf-8", "strict").split("\0")
    if parts and parts[-1] == "":
        parts.pop()
    paths: list[str] = []
    index = 0
    while index < len(parts):
        status = parts[index]
        index += 1
        if not status:
            raise ValidationFailure("malformed git diff status")
        if status.startswith(("R", "C")):
            if index + 1 >= len(parts):
                raise ValidationFailure("malformed rename/copy in git diff")
            paths.extend([parts[index], parts[index + 1]])
            index += 2
        else:
            if index >= len(parts):
                raise ValidationFailure("malformed path in git diff")
            paths.append(parts[index])
            index += 1
    return sorted(set(paths))


def verify_checkout(root: Path, head: str) -> None:
    current = run_git(root, ["rev-parse", "HEAD"])
    assert isinstance(current, str)
    if current.strip() != head:
        raise ValidationFailure("checked-out HEAD does not exactly equal --head")
    status = run_git(root, ["status", "--porcelain=v1", "-z"], text=False)
    assert isinstance(status, bytes)
    if status:
        raise ValidationFailure("working tree must be clean before gate execution")


def validate_diff(root: Path, model: Model, base: str, head: str) -> tuple[list[str], set[str]]:
    errors: list[str] = []
    paths = changed_paths(root, base, head)
    ignored = model.config.get("ignored_paths", [])
    protected = sorted(set(MINIMUM_PROTECTED_PATHS) | set(model.config.get("protected_paths", [])))
    relevant_paths = [path for path in paths if not matches(path, ignored)]
    for path in relevant_paths:
        candidate = root / path
        if candidate.is_symlink():
            try:
                target = candidate.resolve(strict=True)
            except OSError as exc:
                errors.append(f"changed symlink has invalid target: {path}: {exc}")
                continue
            if not target.is_relative_to(root.resolve()):
                errors.append(f"changed symlink escapes repository: {path}")
        if matches(path, protected):
            errors.append(f"protected process path changed in ordinary PR: {path}")

    rules = model.config.get("impact_rules", [])
    required_changed_documents: set[str] = set()
    for path in relevant_paths:
        matching_rules = [rule for rule in rules if isinstance(rule, dict) and matches(path, rule.get("paths", []))]
        if not matching_rules and not matches(path, protected):
            errors.append(f"unknown changed path has no impact rule: {path}")
            continue
        for rule in matching_rules:
            required_changed_documents.update(rule.get("require_changed_documents", []))
    for document in sorted(required_changed_documents):
        if document not in relevant_paths:
            errors.append(f"impact rule requires changed document: {document}")

    base_tasks, base_task_paths = load_ref_manifests(root, base, model.config.get("task_directories", []))
    base_requirements, base_requirement_paths = load_ref_manifests(root, base, model.config.get("requirement_directories", []))
    changed_set = set(relevant_paths)
    changed_task_ids = {
        task_id
        for task_id in set(model.tasks) | set(base_tasks)
        if model.task_paths.get(task_id) in changed_set or base_task_paths.get(task_id) in changed_set
    }
    changed_ready_task = any(
        model.tasks.get(task_id, {}).get("status") == "ready_for_verification"
        for task_id in changed_task_ids
    )
    planning_only = (
        bool(relevant_paths)
        and not changed_ready_task
        and all(matches(path, model.config.get("planning_paths", [])) for path in relevant_paths)
    )
    eligible_tasks: dict[str, dict[str, Any]] = {}
    if planning_only:
        for task_id in changed_task_ids:
            task = model.tasks.get(task_id)
            base_task = base_tasks.get(task_id)
            if not task:
                continue
            status = task.get("status")
            base_status = base_task.get("status") if base_task else None
            if status in {"open", "in_progress"} and base_status in {None, "open", "in_progress"}:
                eligible_tasks[task_id] = task
            elif status == "archived" and base_status == "ready_for_verification":
                eligible_tasks[task_id] = task
            else:
                errors.append(f"invalid planning/archive transition for {task_id}: {base_status} -> {status}")
    else:
        for task_id in changed_task_ids:
            task = model.tasks.get(task_id)
            base_task = base_tasks.get(task_id)
            if not task or task.get("status") != "ready_for_verification":
                continue
            base_status = base_task.get("status") if base_task else None
            if base_status != "in_progress":
                errors.append(f"implementation task {task_id} must transition in_progress -> ready_for_verification")
                continue
            eligible_tasks[task_id] = task
        if relevant_paths and not eligible_tasks:
            errors.append("implementation diff has no changed task transitioning in_progress -> ready_for_verification")

    for path in relevant_paths:
        if matches(path, protected):
            continue
        if not any(matches(path, task.get("affected_paths", [])) for task in eligible_tasks.values()):
            errors.append(f"changed path is outside every eligible task scope: {path}")

    if not planning_only:
        for task_id, task in eligible_tasks.items():
            for document in task.get("documents_to_update", []):
                if document not in changed_set:
                    errors.append(f"task {task_id} requires changed document: {document}")

    for req_id in sorted(set(base_requirements) - set(model.requirements)):
        errors.append(f"requirement {req_id} was deleted; mark it deprecated instead")
    changed_requirements = {
        req_id for req_id in set(model.requirements) | set(base_requirements)
        if model.requirement_paths.get(req_id) in changed_set or base_requirement_paths.get(req_id) in changed_set
    }
    referenced_requirements = {
        req_id for task in eligible_tasks.values() for req_id in task.get("requirement_ids", [])
    }
    for req_id in sorted(changed_requirements - referenced_requirements):
        errors.append(f"changed requirement is not referenced by a ready task: {req_id}")
    errors.extend(validate_archives(root, model.tasks, base, model.config.get("task_directories", [])))
    return errors, {task_id for task_id, task in eligible_tasks.items() if task.get("status") == "ready_for_verification"}


def run_checks(root: Path, model: Model, ready_task_ids: set[str], expected_head: str) -> list[str]:
    errors: list[str] = []
    check_ids = {
        verification["check_id"]
        for task_id in ready_task_ids
        for verification in model.tasks[task_id].get("verifications", [])
        if isinstance(verification, dict) and isinstance(verification.get("check_id"), str)
    }
    checks = {check["id"]: check for check in model.config.get("checks", []) if isinstance(check, dict) and "id" in check}
    with tempfile.TemporaryDirectory(prefix="project-process-evidence-") as evidence_dir:
        evidence_root = Path(evidence_dir).resolve()
        for check_id in sorted(check_ids):
            check = checks.get(check_id)
            if not check:
                errors.append(f"required check not registered: {check_id}")
                continue
            command = [part.replace("{evidence_dir}", str(evidence_root)) for part in check["command"]]
            report_path = (evidence_root / check["result"]["path"]).resolve()
            if not report_path.is_relative_to(evidence_root):
                errors.append(f"check {check_id} report escapes external evidence directory")
                continue
            report_path.parent.mkdir(parents=True, exist_ok=True)
            try:
                completed = subprocess.run(
                    command, cwd=root, check=False, capture_output=True, text=True, timeout=600,
                    env={**os.environ, "PROJECT_PROCESS_EVIDENCE_DIR": str(evidence_root)},
                )
            except (OSError, subprocess.TimeoutExpired) as exc:
                errors.append(f"check {check_id} could not run: {exc}")
                continue
            if completed.returncode != 0:
                errors.append(f"check {check_id} failed with exit code {completed.returncode}")
                continue
            if not report_path.is_file():
                errors.append(f"check {check_id} produced no fresh JUnit report")
                continue
            try:
                xml_root = ET.parse(report_path).getroot()
                suites = [xml_root] if xml_root.tag == "testsuite" else list(xml_root.findall(".//testsuite"))
                cases = list(xml_root.iter("testcase"))
                if not suites or not cases:
                    raise ValueError("report must contain testsuite and testcase elements")
                for suite in suites:
                    direct_cases = suite.findall("testcase")
                    if direct_cases and int(suite.attrib.get("tests", "-1")) != len(direct_cases):
                        raise ValueError("testsuite counter does not match testcase count")
                if xml_root.tag == "testsuites" and "tests" in xml_root.attrib and int(xml_root.attrib["tests"]) != len(cases):
                    raise ValueError("testsuites counter does not match testcase count")
                failures = sum(1 for case in cases if case.find("failure") is not None)
                errors_count = sum(1 for case in cases if case.find("error") is not None)
                skipped = sum(1 for case in cases if case.find("skipped") is not None)
            except (OSError, ET.ParseError, ValueError) as exc:
                errors.append(f"check {check_id} produced invalid JUnit report: {exc}")
                continue
            executed = len(cases) - skipped
            minimum = check["result"]["minimum_executed"]
            if executed < minimum:
                errors.append(f"check {check_id} executed {executed} tests; minimum is {minimum}")
            if failures or errors_count:
                errors.append(f"check {check_id} report contains {failures} failures and {errors_count} errors")
    try:
        verify_checkout(root, expected_head)
    except ValidationFailure as exc:
        errors.append(f"check command changed the repository: {exc}")
    return errors


def emit(errors: list[str]) -> int:
    if errors:
        for error in errors:
            print(f"ERROR: {error}", file=sys.stderr)
        print(f"FAIL: {len(errors)} process violation(s)", file=sys.stderr)
        return 1
    print("PASS: all formal project-process invariants hold")
    return 0


def command_lint(args: argparse.Namespace) -> int:
    root = Path(args.root).resolve()
    try:
        model = build_model(root, args.policy_ref)
    except ValidationFailure as exc:
        return emit([str(exc)])
    errors = list(model.errors)
    if any(task.get("status") == "archived" for task in model.tasks.values()):
        try:
            head = run_git(root, ["rev-parse", "HEAD"])
            assert isinstance(head, str)
            errors.extend(
                validate_archives(
                    root, model.tasks, head.strip(), model.config.get("task_directories", [])
                )
            )
        except ValidationFailure as exc:
            errors.append(str(exc))
    return emit(errors)


def command_gate(args: argparse.Namespace) -> int:
    root = Path(args.root).resolve()
    try:
        verify_revision(root, args.base, "base")
        verify_revision(root, args.head, "head")
        if not args.policy_ref:
            raise ValidationFailure("gate requires --policy-ref equal to the trusted base SHA")
        if args.policy_ref != args.base:
            raise ValidationFailure("--policy-ref must exactly equal --base")
        verify_checkout(root, args.head)
        model = build_model(root, args.policy_ref)
        diff_errors, ready_tasks = validate_diff(root, model, args.base, args.head)
        errors = [*model.errors, *diff_errors]
        if not errors:
            errors.extend(run_checks(root, model, ready_tasks, args.head))
    except (ValidationFailure, UnicodeDecodeError) as exc:
        return emit([str(exc)])
    except Exception as exc:  # Fail closed for unexpected environment/tool failures.
        return emit([f"unexpected validator failure: {type(exc).__name__}: {exc}"])
    return emit(errors)


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description=__doc__)
    subparsers = parser.add_subparsers(dest="command", required=True)
    lint = subparsers.add_parser("lint", help="validate schemas and traceability graph")
    lint.add_argument("--root", default=".")
    lint.add_argument("--policy-ref", help="load .process/config.json from trusted Git revision")
    lint.set_defaults(func=command_lint)
    gate = subparsers.add_parser("gate", help="validate manifests, exact diff, scope and checks")
    gate.add_argument("--root", default=".")
    gate.add_argument("--base", required=True)
    gate.add_argument("--head", required=True)
    gate.add_argument("--policy-ref", required=True, help="trusted policy revision; must equal --base")
    gate.set_defaults(func=command_gate)
    return parser


def main(argv: list[str] | None = None) -> int:
    args = build_parser().parse_args(argv)
    return args.func(args)


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