#!/usr/bin/env python3
from __future__ import annotations

import argparse
import fcntl
import hashlib
import json
import os
import shutil
import socket
import sqlite3
import subprocess
import sys
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

VERSION = "0002.1-infrastructure-defer"

REPO = Path("/home/shire3d/ARMOR")
ACADEMY = REPO / "data/academy/design_mastery"
RUNTIME = ACADEMY / "runtime"

STATE = RUNTIME / "practice_state.json"
AUTOPILOT = RUNTIME / "autopilot_state.json"
SPRINT = RUNTIME / "commercial_design_sprint.json"
INDEX = RUNTIME / "academy_index.sqlite3"

PYTHON = REPO / "venv/bin/python"

QUALIFIER = (
    REPO
    / "tools"
    / "design_academy_generic_qualify.py"
)

PRACTICE_TOOL = (
    REPO
    / "tools"
    / "design_academy_practice.py"
)

SAFE_CADQUERY = Path(
    "/home/ray/.local/bin/"
    "shire-academy-safe-cadquery-evidence"
)

STATUS_ROOT = Path(
    "/home/ray/.local/state/"
    "shire-academy-safe-autonomous"
)

STATUS_PATH = STATUS_ROOT / "status.json"
LOCK_PATH = STATUS_ROOT / "scheduler.lock"

EVIDENCE_ROOT = Path(
    "/SHiREVault/SHiREAcademy/"
    "Autopilot/SafeScheduler"
)

BACKUP_ROOT = Path(
    "/SHiREVault/Backup/OSBackups"
)

OLD_WORKER = (
    "shire-academy-autonomous-practice.service"
)

CORE_SERVICES = (
    "shire-brain.service",
    "shire-academy-cadquery-fast.service",
    "shire-academy-ui.service",
)

DENYLIST = {
    # Never rerun the historical Boolean canary route.
    "geometry_foundations.boolean-operations",

    # This skill has older hard-coded certification history.
    "parametric_cad.parameter-architecture",
}

DOMAIN_PRIORITY = {
    "geometry_foundations": 0,
    "parametric_cad": 1,
    "validation_and_simulation": 2,
}


class PermissionDrift(RuntimeError):
    pass


def now() -> str:
    return datetime.now(
        timezone.utc
    ).isoformat()


def timestamp() -> str:
    return datetime.now(
        timezone.utc
    ).strftime("%Y%m%dT%H%M%S%fZ")


def load(
    path: Path,
    default: Any | None = None,
) -> Any:
    try:
        return json.loads(
            path.read_text(encoding="utf-8")
        )
    except (
        OSError,
        json.JSONDecodeError,
    ):
        if default is not None:
            return default
        raise


def atomic(
    path: Path,
    payload: Any,
) -> None:
    path.parent.mkdir(
        parents=True,
        exist_ok=True,
    )

    temporary = path.with_suffix(
        path.suffix + ".tmp"
    )

    temporary.write_text(
        json.dumps(
            payload,
            indent=2,
            sort_keys=True,
            ensure_ascii=False,
        ) + "\n",
        encoding="utf-8",
    )

    os.replace(
        temporary,
        path,
    )


def sha256(path: Path) -> str:
    return hashlib.sha256(
        path.read_bytes()
    ).hexdigest()


def service_state(
    service: str,
) -> str:
    process = subprocess.run(
        [
            "systemctl",
            "is-active",
            service,
        ],
        text=True,
        capture_output=True,
    )

    return (
        process.stdout.strip()
        or process.stderr.strip()
        or "unknown"
    )


def skill_paths(
    skill_id: str,
) -> dict[str, Path]:
    domain, name = skill_id.split(".", 1)

    root = (
        ACADEMY
        / "skills"
        / domain
        / name
    )

    return {
        "root": root,
        "certification":
            root / "certification.json",
        "manifest":
            root / "manifest.json",
        "tests":
            root / "tests.json",
    }


def index_record(
    skill_id: str,
) -> tuple[str, int] | None:
    with sqlite3.connect(INDEX) as connection:
        row = connection.execute(
            """
            SELECT status, forge_allowed
            FROM skills
            WHERE skill_id = ?
            """,
            (skill_id,),
        ).fetchone()

    if row is None:
        return None

    return str(row[0]), int(row[1])


def forge_locked(
    skill_id: str,
) -> tuple[bool, list[str]]:
    state = load(STATE)
    autopilot = load(AUTOPILOT)

    record = (
        state.get("skills") or {}
    ).get(skill_id)

    problems: list[str] = []

    if not isinstance(record, dict):
        problems.append(
            "practice_record_missing"
        )
    elif record.get("forge_allowed") is not False:
        problems.append(
            "practice_forge_open"
        )

    paths = skill_paths(skill_id)

    certification = load(
        paths["certification"],
        {},
    )

    manifest = load(
        paths["manifest"],
        {},
    )

    if certification.get("forge_allowed") is not False:
        problems.append(
            "certification_forge_open"
        )

    if manifest.get("forge_allowed") is not False:
        problems.append(
            "manifest_forge_open"
        )

    if autopilot.get("forge_unlock_allowed") is not False:
        problems.append(
            "autopilot_global_forge_open"
        )

    qualification = (
        autopilot.get("skills") or {}
    ).get(skill_id)

    if (
        isinstance(qualification, dict)
        and qualification.get("forge_allowed")
        is not False
    ):
        problems.append(
            "autopilot_skill_forge_open"
        )

    indexed = index_record(skill_id)

    if indexed is None:
        problems.append(
            "academy_index_record_missing"
        )
    elif indexed[1] != 0:
        problems.append(
            "academy_index_forge_open"
        )

    return not problems, problems


def ensure_environment() -> None:
    if socket.gethostname() != "Work":
        raise RuntimeError(
            "Scheduler is running on the wrong hostname."
        )

    if os.environ.get("HOME") != "/home/ray":
        raise RuntimeError(
            "Scheduler HOME is not /home/ray."
        )

    mount = subprocess.run(
        [
            "mountpoint",
            "-q",
            "/mnt/shirevault-network",
        ]
    )

    if mount.returncode != 0:
        raise RuntimeError(
            "SHiREVault network mount is unavailable."
        )

    for path in (
        STATE,
        AUTOPILOT,
        SPRINT,
        INDEX,
        PYTHON,
        QUALIFIER,
        PRACTICE_TOOL,
        SAFE_CADQUERY,
    ):
        if not path.is_file():
            raise RuntimeError(
                f"Required scheduler dependency is missing: {path}"
            )

    for service in CORE_SERVICES:
        if service_state(service) != "active":
            raise RuntimeError(
                f"Required core service is not active: {service}"
            )

    if service_state(OLD_WORKER) == "active":
        raise RuntimeError(
            "The retired autonomous worker is active."
        )

    EVIDENCE_ROOT.mkdir(
        parents=True,
        exist_ok=True,
    )

    probe = (
        EVIDENCE_ROOT
        / f".write-probe-{os.getpid()}"
    )

    probe.write_text(
        "safe scheduler write probe\n",
        encoding="utf-8",
    )

    probe.unlink()


def successful_rounds(
    qualification: dict[str, Any],
) -> set[int]:
    rounds: set[int] = set()

    for item in (
        qualification.get(
            "qualification_runs"
        )
        or []
    ):
        if not isinstance(item, dict):
            continue

        passed = (
            item.get("passed") is True
            and int(
                item.get("score_percent") or 0
            ) == 100
        )

        try:
            round_number = int(
                item.get("round")
            )
        except (
            TypeError,
            ValueError,
        ):
            continue

        if (
            passed
            and round_number in {1, 2, 3}
        ):
            rounds.add(round_number)

    return rounds


def candidate_safety(
    skill_id: str,
    record: dict[str, Any],
) -> tuple[bool, list[str]]:
    reasons: list[str] = []

    if skill_id in DENYLIST:
        reasons.append(
            "explicitly_denied"
        )

    if record.get("lane") != "tool_evidence":
        reasons.append(
            "not_tool_evidence_lane"
        )

    if record.get("required_tool") != "cadquery":
        reasons.append(
            "not_cadquery_required"
        )

    if record.get("cadquery_supported") is not True:
        reasons.append(
            "cadquery_not_supported"
        )

    if record.get("forge_allowed") is not False:
        reasons.append(
            "practice_forge_open"
        )

    if int(record.get("attempts") or 0) >= 5:
        reasons.append(
            "counted_attempt_limit_reached"
        )

    if int(
        record.get("tool_evidence_attempts") or 0
    ) >= 3:
        reasons.append(
            "tool_attempt_limit_reached"
        )

    paths = skill_paths(skill_id)

    for name, path in paths.items():
        if name == "root":
            continue

        if not path.is_file():
            reasons.append(
                f"{name}_missing"
            )

    manifest = load(
        paths["manifest"],
        {},
    )

    certification = load(
        paths["certification"],
        {},
    )

    if str(
        manifest.get("risk_level") or ""
    ).lower() != "low":
        reasons.append(
            "risk_not_low"
        )

    if (
        certification.get(
            "professional_review_required"
        )
        is True
    ):
        reasons.append(
            "professional_review_required"
        )

    locked, lock_problems = forge_locked(
        skill_id
    )

    if not locked:
        reasons.extend(lock_problems)

    return not reasons, reasons


def infrastructure_deferred_skill_ids() -> set[str]:
    autopilot = load(AUTOPILOT)

    entries = autopilot.get(
        "infrastructure_deferred_skills",
        [],
    )

    if not isinstance(entries, list):
        raise RuntimeError(
            "Infrastructure-deferred skills must be a list."
        )

    deferred: set[str] = set()

    for entry in entries:
        if isinstance(entry, str):
            skill_id = entry
        elif isinstance(entry, dict):
            skill_id = entry.get("skill_id")
        else:
            raise RuntimeError(
                "Invalid infrastructure-deferred entry."
            )

        if not isinstance(skill_id, str):
            raise RuntimeError(
                "Deferred entry has no valid skill_id."
            )

        deferred.add(skill_id)

    return deferred


def eligible_candidates() -> list[str]:
    state = load(STATE)
    deferred = infrastructure_deferred_skill_ids()

    candidates: list[str] = []

    for skill_id, record in (
        state.get("skills") or {}
    ).items():
        if not isinstance(record, dict):
            continue

        if skill_id in deferred:
            continue

        if (
            record.get(
                "cadquery_evidence_validated"
            )
            is True
        ):
            continue

        safe, _ = candidate_safety(
            skill_id,
            record,
        )

        if safe:
            candidates.append(skill_id)

    def key(skill_id: str) -> tuple[int, str]:
        domain = skill_id.split(".", 1)[0]

        return (
            DOMAIN_PRIORITY.get(domain, 99),
            skill_id,
        )

    return sorted(
        candidates,
        key=key,
    )


def write_status(
    payload: dict[str, Any],
) -> dict[str, Any]:
    payload = {
        "schema":
            "shire.safe_autonomous_scheduler.v1",
        "version": VERSION,
        "updated_at": now(),
        "forge_unlock_allowed": False,
        "old_worker_allowed": False,
        **payload,
    }

    atomic(
        STATUS_PATH,
        payload,
    )

    return payload


def live_status() -> dict[str, Any]:
    state = load(STATE)
    autopilot = load(AUTOPILOT)
    saved = load(
        STATUS_PATH,
        {},
    )

    skill_id = autopilot.get(
        "current_skill"
    )

    record = None
    qualification = None

    if isinstance(skill_id, str):
        record = (
            state.get("skills") or {}
        ).get(skill_id)

        qualification = (
            autopilot.get("skills") or {}
        ).get(skill_id)

    return {
        "saved_status": saved,
        "current_skill": skill_id,
        "current_plan":
            autopilot.get("current_plan"),
        "scheduler_enabled":
            autopilot.get("enabled"),
        "scheduler_paused":
            autopilot.get("paused"),
        "forge_unlock_allowed":
            autopilot.get(
                "forge_unlock_allowed"
            ),
        "practice_record": record,
        "qualification_record":
            qualification,
        "remaining_safe_candidates":
            eligible_candidates(),
        "remaining_safe_candidate_count":
            len(eligible_candidates()),
        "old_worker_state":
            service_state(OLD_WORKER),
        "core_services": {
            service: service_state(service)
            for service in CORE_SERVICES
        },
    }


def create_snapshot(
    skill_id: str,
    stage: str,
) -> Path:
    slug = skill_id.replace(".", "--")

    root = (
        BACKUP_ROOT
        / (
            "SHIRE-ACADEMY-SAFE-AUTO-TICK-"
            f"{timestamp()}-{slug}-{stage}"
        )
    )

    root.mkdir(
        parents=True,
        exist_ok=False,
    )

    shutil.copy2(
        STATE,
        root / "practice_state.json",
    )

    shutil.copy2(
        AUTOPILOT,
        root / "autopilot_state.json",
    )

    shutil.copy2(
        SPRINT,
        root / "commercial_design_sprint.json",
    )

    shutil.copy2(
        INDEX,
        root / "academy_index.sqlite3",
    )

    paths = skill_paths(skill_id)

    shutil.copy2(
        paths["certification"],
        root / "certification.json",
    )

    shutil.copy2(
        paths["manifest"],
        root / "manifest.json",
    )

    shutil.copy2(
        paths["tests"],
        root / "tests.json",
    )

    hashes = {}

    for path in root.iterdir():
        if path.is_file():
            hashes[path.name] = sha256(path)

    atomic(
        root / "snapshot.json",
        {
            "skill_id": skill_id,
            "stage": stage,
            "created_at": now(),
            "files": hashes,
        },
    )

    return root


def restore_snapshot(
    snapshot: Path,
    skill_id: str,
) -> None:
    shutil.copy2(
        snapshot / "practice_state.json",
        STATE,
    )

    shutil.copy2(
        snapshot / "autopilot_state.json",
        AUTOPILOT,
    )

    shutil.copy2(
        snapshot
        / "commercial_design_sprint.json",
        SPRINT,
    )

    shutil.copy2(
        snapshot / "academy_index.sqlite3",
        INDEX,
    )

    paths = skill_paths(skill_id)

    shutil.copy2(
        snapshot / "certification.json",
        paths["certification"],
    )

    shutil.copy2(
        snapshot / "manifest.json",
        paths["manifest"],
    )

    shutil.copy2(
        snapshot / "tests.json",
        paths["tests"],
    )


def journal_directory(
    skill_id: str,
    stage: str,
) -> Path:
    domain, name = skill_id.split(".", 1)

    root = (
        EVIDENCE_ROOT
        / domain
        / name
        / f"{timestamp()}-{stage}"
    )

    root.mkdir(
        parents=True,
        exist_ok=False,
    )

    return root


def parse_json_output(
    stdout: str,
) -> dict[str, Any] | None:
    text = stdout.strip()

    if not text:
        return None

    try:
        value = json.loads(text)
    except json.JSONDecodeError:
        return None

    return value if isinstance(value, dict) else None


def run_command(
    command: list[str],
    journal: Path,
    timeout_seconds: int = 700,
) -> tuple[int, dict[str, Any] | None]:
    atomic(
        journal / "command.json",
        {
            "argv": command,
            "started_at": now(),
        },
    )

    try:
        process = subprocess.run(
            command,
            cwd=str(REPO),
            text=True,
            capture_output=True,
            timeout=timeout_seconds,
            env={
                **os.environ,
                "HOME": "/home/ray",
                "PATH":
                    "/home/ray/.local/bin:"
                    "/usr/local/sbin:"
                    "/usr/local/bin:"
                    "/usr/sbin:"
                    "/usr/bin:"
                    "/sbin:"
                    "/bin",
            },
        )

        returncode = process.returncode
        stdout = process.stdout
        stderr = process.stderr

    except subprocess.TimeoutExpired as exc:
        returncode = 124
        stdout = (
            exc.stdout.decode()
            if isinstance(exc.stdout, bytes)
            else exc.stdout or ""
        )

        stderr = (
            exc.stderr.decode()
            if isinstance(exc.stderr, bytes)
            else exc.stderr or ""
        )

        stderr += (
            "\nScheduler command timed out.\n"
        )

    (
        journal / "stdout.txt"
    ).write_text(
        stdout,
        encoding="utf-8",
    )

    (
        journal / "stderr.txt"
    ).write_text(
        stderr,
        encoding="utf-8",
    )

    payload = parse_json_output(stdout)

    atomic(
        journal / "command-result.json",
        {
            "returncode": returncode,
            "payload": payload,
            "finished_at": now(),
        },
    )

    return returncode, payload


def set_safe_autopilot_fields(
    autopilot: dict[str, Any],
) -> None:
    autopilot["mode"] = (
        "safe_supported_tool_evidence_autonomy"
    )

    autopilot["scheduler_version"] = VERSION
    autopilot["counted_execution_enabled"] = False
    autopilot["forge_unlock_allowed"] = False


def enroll_candidate(
    skill_id: str,
) -> None:
    autopilot = load(AUTOPILOT)
    sprint = load(SPRINT)

    skills = autopilot.setdefault(
        "skills",
        {},
    )

    qualification = skills.setdefault(
        skill_id,
        {},
    )

    runs = qualification.get(
        "qualification_runs"
    )

    if not isinstance(runs, list):
        runs = []
        qualification[
            "qualification_runs"
        ] = runs

    rounds = successful_rounds(
        qualification
    )

    missing = sorted(
        {1, 2, 3} - rounds
    )

    if missing:
        next_action = (
            "run_generic_uncounted_round_"
            f"{missing[0]}"
        )

        status = (
            "autonomous_qualification_pending"
        )
    else:
        next_action = (
            "run_counted_practice_round_1"
        )

        status = (
            "qualification_complete_"
            "waiting_counted_practice"
        )

    qualification.update({
        "status": status,
        "successful_qualifications":
            len(rounds),
        "consecutive_qualification_failures":
            int(
                qualification.get(
                    "consecutive_qualification_failures"
                )
                or 0
            ),
        "consecutive_infrastructure_failures":
            int(
                qualification.get(
                    "consecutive_infrastructure_failures"
                )
                or 0
            ),
        "knowledge_certified": False,
        "tool_evidence_certified": False,
        "forge_allowed": False,
        "scheduler_managed": True,
        "updated_at": now(),
    })

    plan = {
        "skill_id": skill_id,
        "purpose":
            "Safe autonomous learning of supported "
            "CadQuery tool-evidence skills.",
        "next_action": next_action,
        "status": status,
        "knowledge_certified": False,
        "tool_evidence_validated": False,
        "forge_review_pending": False,
        "forge_allowed": False,
    }

    autopilot["current_skill"] = skill_id
    autopilot["current_plan"] = plan
    autopilot["enabled"] = True
    autopilot["paused"] = False
    autopilot["safety_stop"] = False
    autopilot["safety_stop_reason"] = None
    autopilot["review_required"] = False
    autopilot["updated_at"] = now()

    set_safe_autopilot_fields(
        autopilot
    )

    sprint["current_skill"] = skill_id
    sprint["updated_at"] = now()

    atomic(
        AUTOPILOT,
        autopilot,
    )

    atomic(
        SPRINT,
        sprint,
    )


def update_plan(
    skill_id: str,
    next_action: str,
    status: str,
    *,
    paused: bool = False,
    review_required: bool = False,
) -> None:
    state = load(STATE)
    autopilot = load(AUTOPILOT)

    record = state["skills"][skill_id]

    qualification = (
        autopilot.setdefault(
            "skills",
            {},
        )
        .setdefault(
            skill_id,
            {},
        )
    )

    qualification.update({
        "status": status,
        "counted_attempts":
            int(record.get("attempts") or 0),
        "successful_counted_rounds":
            int(
                record.get(
                    "successful_attempts"
                )
                or 0
            ),
        "knowledge_certified":
            record.get(
                "knowledge_certified"
            )
            is True,
        "tool_evidence_validated":
            record.get(
                "cadquery_evidence_validated"
            )
            is True,
        "forge_review_pending":
            record.get(
                "forge_review_pending"
            )
            is True,
        "forge_allowed": False,
        "scheduler_managed": True,
        "updated_at": now(),
    })

    plan = {
        "skill_id": skill_id,
        "purpose":
            "Safe autonomous learning of supported "
            "CadQuery tool-evidence skills.",
        "next_action": next_action,
        "status": status,
        "knowledge_certified":
            record.get(
                "knowledge_certified"
            )
            is True,
        "tool_evidence_validated":
            record.get(
                "cadquery_evidence_validated"
            )
            is True,
        "forge_review_pending":
            record.get(
                "forge_review_pending"
            )
            is True,
        "forge_allowed": False,
    }

    autopilot["current_skill"] = skill_id
    autopilot["current_plan"] = plan
    autopilot["enabled"] = not paused
    autopilot["paused"] = paused
    autopilot["review_required"] = (
        review_required
    )
    autopilot["safety_stop"] = False

    if not paused:
        autopilot[
            "safety_stop_reason"
        ] = None

    autopilot["updated_at"] = now()

    set_safe_autopilot_fields(
        autopilot
    )

    atomic(
        AUTOPILOT,
        autopilot,
    )


def safety_stop(
    skill_id: str | None,
    reason: str,
) -> None:
    autopilot = load(AUTOPILOT)

    autopilot["enabled"] = False
    autopilot["paused"] = True
    autopilot["safety_stop"] = True
    autopilot["review_required"] = True
    autopilot["safety_stop_reason"] = reason
    autopilot["updated_at"] = now()

    set_safe_autopilot_fields(
        autopilot
    )

    if isinstance(skill_id, str):
        qualification = (
            autopilot.setdefault(
                "skills",
                {},
            )
            .setdefault(
                skill_id,
                {},
            )
        )

        qualification.update({
            "status":
                "scheduler_safety_stopped",
            "forge_allowed": False,
            "updated_at": now(),
        })

        plan = dict(
            autopilot.get(
                "current_plan"
            )
            or {}
        )

        plan.update({
            "skill_id": skill_id,
            "next_action":
                "local_safety_review_required",
            "status":
                "scheduler_safety_stopped",
            "forge_allowed": False,
        })

        autopilot["current_plan"] = plan

    atomic(
        AUTOPILOT,
        autopilot,
    )


def complete_current(
    skill_id: str,
) -> None:
    autopilot = load(AUTOPILOT)
    sprint = load(SPRINT)

    qualification = (
        autopilot.setdefault(
            "skills",
            {},
        )
        .setdefault(
            skill_id,
            {},
        )
    )

    qualification.update({
        "status":
            "academy_learning_complete_"
            "waiting_forge_review",
        "knowledge_certified": True,
        "tool_evidence_validated": True,
        "tool_evidence_complete": True,
        "tool_evidence_certified": False,
        "forge_review_pending": True,
        "forge_allowed": False,
        "scheduler_managed": True,
        "completed_at": now(),
        "updated_at": now(),
    })

    completed = list(
        autopilot.get(
            "safe_supported_skills_completed"
        )
        or []
    )

    if skill_id not in completed:
        completed.append(skill_id)

    autopilot[
        "safe_supported_skills_completed"
    ] = completed

    autopilot["current_skill"] = None

    autopilot["current_plan"] = {
        "skill_id": skill_id,
        "next_action":
            "select_next_safe_candidate",
        "status":
            "academy_learning_complete_"
            "waiting_forge_review",
        "knowledge_certified": True,
        "tool_evidence_validated": True,
        "forge_review_pending": True,
        "forge_allowed": False,
    }

    autopilot["enabled"] = True
    autopilot["paused"] = False
    autopilot["review_required"] = False
    autopilot["safety_stop"] = False
    autopilot["updated_at"] = now()

    set_safe_autopilot_fields(
        autopilot
    )

    sprint["current_skill"] = None
    sprint["updated_at"] = now()

    atomic(
        AUTOPILOT,
        autopilot,
    )

    atomic(
        SPRINT,
        sprint,
    )


def qualification_tick(
    skill_id: str,
    round_number: int,
    journal: Path,
) -> dict[str, Any]:
    state_hash_before = sha256(STATE)

    returncode, payload = run_command(
        [
            str(PYTHON),
            str(QUALIFIER),
            skill_id,
            str(round_number),
        ],
        journal,
    )

    if sha256(STATE) != state_hash_before:
        raise PermissionDrift(
            "Uncounted qualification changed "
            "protected practice state."
        )

    locked, problems = forge_locked(
        skill_id
    )

    if not locked:
        raise PermissionDrift(
            "Forge permission drift after qualification: "
            + ", ".join(problems)
        )

    autopilot = load(AUTOPILOT)

    qualification = (
        autopilot.setdefault(
            "skills",
            {},
        )
        .setdefault(
            skill_id,
            {},
        )
    )

    runs = qualification.get(
        "qualification_runs"
    )

    if not isinstance(runs, list):
        runs = []
        qualification[
            "qualification_runs"
        ] = runs

    classification = ""

    if isinstance(payload, dict):
        classification = str(
            payload.get(
                "classification"
            )
            or ""
        )

        tests = (
            payload.get("tests")
            or []
        )
    else:
        tests = []

    passed = all([
        returncode == 0,
        isinstance(payload, dict),
        classification
            == "UNCOUNTED_QUALIFICATION_PASS",
        isinstance(payload, dict)
            and payload.get("ok") is True,
        isinstance(payload, dict)
            and int(
                payload.get("round") or 0
            )
            == int(round_number),
        isinstance(payload, dict)
            and int(
                payload.get(
                    "score_percent"
                )
                or 0
            )
            == 100,
        isinstance(payload, dict)
            and payload.get("skill_id")
            == skill_id,
        isinstance(payload, dict)
            and payload.get(
                "forge_allowed"
            )
            is False,
        isinstance(payload, dict)
            and payload.get(
                "counted_attempt_consumed"
            )
            is False,
        isinstance(payload, dict)
            and payload.get(
                "practice_state_unchanged"
            )
            is True,
        bool(tests),
        all(
            isinstance(item, dict)
            and item.get("pass") is True
            for item in tests
        ),
    ])

    infrastructure = (
        isinstance(payload, dict)
        and (
            payload.get(
                "infrastructure_failure"
            )
            is True
            or "INFRASTRUCTURE"
            in classification.upper()
        )
    )

    evidence = ""

    if isinstance(payload, dict):
        evidence = str(
            payload.get("evidence")
            or ""
        )

    duplicate = any(
        isinstance(item, dict)
        and int(
            item.get("round") or 0
        )
        == int(round_number)
        and (
            (
                bool(evidence)
                and str(
                    item.get("evidence")
                    or ""
                )
                == evidence
            )
            or (
                not evidence
                and str(
                    item.get("journal")
                    or ""
                )
                == str(journal)
            )
        )
        for item in runs
    )

    if not duplicate:
        runs.append({
            "recorded_at": now(),
            "round":
                int(round_number),
            "passed": passed,
            "score_percent":
                int(
                    payload.get(
                        "score_percent"
                    )
                    or 0
                )
                if isinstance(
                    payload,
                    dict,
                )
                else 0,
            "classification":
                classification or None,
            "evidence":
                evidence or None,
            "journal": str(journal),
            "returncode":
                int(returncode),
            "infrastructure_failure":
                infrastructure,
            "counted_attempt_consumed":
                False,
            "forge_allowed": False,
            "source":
                "safe_autonomous_scheduler_v2",
        })

    if passed:
        qualification[
            "consecutive_qualification_failures"
        ] = 0

        qualification[
            "consecutive_infrastructure_failures"
        ] = 0

    elif infrastructure:
        qualification[
            "consecutive_infrastructure_failures"
        ] = (
            int(
                qualification.get(
                    "consecutive_infrastructure_failures"
                )
                or 0
            )
            + 1
        )

        qualification[
            "consecutive_qualification_failures"
        ] = 0

    else:
        qualification[
            "consecutive_qualification_failures"
        ] = (
            int(
                qualification.get(
                    "consecutive_qualification_failures"
                )
                or 0
            )
            + 1
        )

        qualification[
            "consecutive_infrastructure_failures"
        ] = 0

    rounds = successful_rounds(
        qualification
    )

    qualification[
        "successful_qualifications"
    ] = len(rounds)

    qualification[
        "last_qualification_classification"
    ] = classification or None

    qualification[
        "latest_qualification_journal"
    ] = str(journal)

    qualification[
        "forge_allowed"
    ] = False

    qualification[
        "updated_at"
    ] = now()

    autopilot[
        "forge_unlock_allowed"
    ] = False

    autopilot[
        "counted_execution_enabled"
    ] = False

    autopilot["updated_at"] = now()

    atomic(
        AUTOPILOT,
        autopilot,
    )

    qualification_failures = int(
        qualification.get(
            "consecutive_qualification_failures"
        )
        or 0
    )

    infrastructure_failures = int(
        qualification.get(
            "consecutive_infrastructure_failures"
        )
        or 0
    )

    if passed:
        missing = sorted(
            {1, 2, 3} - rounds
        )

        if missing:
            next_action = (
                "run_generic_uncounted_round_"
                f"{missing[0]}"
            )

            status = (
                "autonomous_qualification_"
                "in_progress"
            )
        else:
            next_action = (
                "run_counted_practice_round_1"
            )

            status = (
                "qualification_complete_"
                "waiting_counted_practice"
            )

        update_plan(
            skill_id,
            next_action,
            status,
        )

        return {
            "ok": True,
            "action":
                "uncounted_qualification_pass",
            "skill_id": skill_id,
            "round": round_number,
            "successful_unique_rounds":
                sorted(rounds),
            "next_action": next_action,
            "journal": str(journal),
            "payload": payload,
        }

    if (
        qualification_failures >= 3
        or infrastructure_failures >= 3
    ):
        reason = (
            "Three consecutive autonomous "
            "qualification or infrastructure "
            "failures require local review."
        )

        safety_stop(
            skill_id,
            reason,
        )

        return {
            "ok": False,
            "action":
                "qualification_safety_stop",
            "skill_id": skill_id,
            "round": round_number,
            "returncode": returncode,
            "qualification_failures":
                qualification_failures,
            "infrastructure_failures":
                infrastructure_failures,
            "journal": str(journal),
            "reason": reason,
            "payload": payload,
        }

    update_plan(
        skill_id,
        (
            "retry_generic_uncounted_round_"
            f"{round_number}"
        ),
        (
            "qualification_infrastructure_"
            "retry_scheduled"
            if infrastructure
            else
            "qualification_retry_scheduled"
        ),
    )

    return {
        "ok": False,
        "action":
            (
                "qualification_infrastructure_"
                "retry_scheduled"
                if infrastructure
                else
                "qualification_retry_scheduled"
            ),
        "skill_id": skill_id,
        "round": round_number,
        "returncode": returncode,
        "qualification_failures":
            qualification_failures,
        "infrastructure_failures":
            infrastructure_failures,
        "journal": str(journal),
        "payload": payload,
    }


def counted_tick(
    skill_id: str,
    journal: Path,
) -> dict[str, Any]:
    returncode, payload = run_command(
        [
            str(PYTHON),
            str(PRACTICE_TOOL),
            "run-one",
            "--skill-id",
            skill_id,
        ],
        journal,
    )

    locked, problems = forge_locked(
        skill_id
    )

    if not locked:
        raise PermissionDrift(
            "Forge permission drift after counted practice: "
            + ", ".join(problems)
        )

    state = load(STATE)
    record = state["skills"][skill_id]

    attempts = int(
        record.get("attempts") or 0
    )

    successes = int(
        record.get("successful_attempts")
        or 0
    )

    knowledge = (
        record.get("knowledge_certified")
        is True
    )

    infrastructure = (
        isinstance(payload, dict)
        and (
            payload.get(
                "infrastructure_failure"
            )
            is True
            or payload.get("status")
            == "infrastructure_retry"
        )
    )

    passed = (
        returncode == 0
        and isinstance(payload, dict)
        and payload.get("ok") is True
    )

    if passed:
        if knowledge:
            next_action = (
                "run_safe_cadquery_evidence"
            )

            status = (
                "knowledge_certified_"
                "waiting_safe_cadquery"
            )
        else:
            next_round = min(
                successes + 1,
                3,
            )

            next_action = (
                "run_counted_practice_round_"
                f"{next_round}"
            )

            status = (
                "counted_practice_in_progress"
            )

        update_plan(
            skill_id,
            next_action,
            status,
        )

        return {
            "ok": True,
            "action":
                "counted_practice_pass",
            "skill_id": skill_id,
            "attempts": attempts,
            "successful_rounds": successes,
            "knowledge_certified": knowledge,
            "next_action": next_action,
            "journal": str(journal),
            "payload": payload,
        }

    if infrastructure:
        infrastructure_failures = int(
            record.get(
                "infrastructure_failures"
            )
            or 0
        )

        if infrastructure_failures >= 3:
            reason = (
                "Three consecutive counted-practice "
                "infrastructure failures require "
                "local review."
            )

            safety_stop(
                skill_id,
                reason,
            )

            return {
                "ok": False,
                "action":
                    "counted_infrastructure_safety_stop",
                "skill_id": skill_id,
                "journal": str(journal),
                "reason": reason,
                "payload": payload,
            }

        next_round = min(
            successes + 1,
            3,
        )

        update_plan(
            skill_id,
            (
                "retry_counted_practice_round_"
                f"{next_round}"
            ),
            "counted_infrastructure_retry_scheduled",
        )

        return {
            "ok": False,
            "action":
                "counted_infrastructure_retry_scheduled",
            "skill_id": skill_id,
            "journal": str(journal),
            "payload": payload,
        }

    reason = (
        "A counted semantic assessment failed. "
        "Automatic retries are paused so the "
        "failure can be studied honestly."
    )

    update_plan(
        skill_id,
        "study_failed_counted_assessment",
        "counted_semantic_failure_review",
        paused=True,
        review_required=True,
    )

    return {
        "ok": False,
        "action":
            "counted_semantic_failure_paused",
        "skill_id": skill_id,
        "returncode": returncode,
        "attempts": attempts,
        "successful_rounds": successes,
        "journal": str(journal),
        "reason": reason,
        "payload": payload,
    }


def cadquery_tick(
    skill_id: str,
    journal: Path,
) -> dict[str, Any]:
    paths = skill_paths(skill_id)

    receipt = (
        journal
        / "safe-cadquery-receipt.json"
    )

    command = [
        str(PYTHON),
        str(SAFE_CADQUERY),
        "run",
        "--skill-id",
        skill_id,
        "--receipt",
        str(receipt),
        "--expected-state-sha",
        sha256(STATE),
        "--expected-auto-sha",
        sha256(AUTOPILOT),
        "--expected-cert-sha",
        sha256(paths["certification"]),
        "--expected-manifest-sha",
        sha256(paths["manifest"]),
        "--expected-tests-sha",
        sha256(paths["tests"]),
    ]

    returncode, payload = run_command(
        command,
        journal,
    )

    locked, problems = forge_locked(
        skill_id
    )

    if not locked:
        raise PermissionDrift(
            "Forge permission drift after CadQuery validation: "
            + ", ".join(problems)
        )

    passed = (
        returncode == 0
        and isinstance(payload, dict)
        and payload.get(
            "classification"
        )
        == "SAFE_CADQUERY_EVIDENCE_PASS"
        and payload.get(
            "tool_evidence_validated"
        )
        is True
        and payload.get(
            "forge_allowed"
        )
        is False
    )

    if passed:
        complete_current(
            skill_id
        )

        return {
            "ok": True,
            "action":
                "safe_cadquery_pass_and_skill_complete",
            "skill_id": skill_id,
            "journal": str(journal),
            "receipt": str(receipt),
            "payload": payload,
            "next_action":
                "select_next_safe_candidate",
        }

    infrastructure = (
        isinstance(payload, dict)
        and payload.get(
            "classification"
        )
        == "SAFE_CADQUERY_INFRASTRUCTURE_RETRY"
    )

    if infrastructure:
        update_plan(
            skill_id,
            "retry_safe_cadquery_evidence",
            "safe_cadquery_infrastructure_retry",
        )

        return {
            "ok": False,
            "action":
                "safe_cadquery_infrastructure_retry",
            "skill_id": skill_id,
            "journal": str(journal),
            "payload": payload,
        }

    reason = (
        "CadQuery evidence failed a semantic "
        "validation check. Automatic learning "
        "is paused for local review."
    )

    update_plan(
        skill_id,
        "review_failed_cadquery_evidence",
        "safe_cadquery_semantic_failure_review",
        paused=True,
        review_required=True,
    )

    return {
        "ok": False,
        "action":
            "safe_cadquery_semantic_failure_paused",
        "skill_id": skill_id,
        "journal": str(journal),
        "reason": reason,
        "payload": payload,
    }


def scheduler_tick() -> dict[str, Any]:
    STATUS_ROOT.mkdir(
        parents=True,
        exist_ok=True,
    )

    with LOCK_PATH.open("a+") as lock:
        try:
            fcntl.flock(
                lock.fileno(),
                fcntl.LOCK_EX
                | fcntl.LOCK_NB,
            )
        except BlockingIOError:
            return write_status({
                "ok": True,
                "action":
                    "another_scheduler_tick_is_running",
            })

        ensure_environment()

        saved = load(
            STATUS_PATH,
            {},
        )

        autopilot = load(AUTOPILOT)

        if (
            saved.get("paused") is True
            or autopilot.get("paused") is True
            and autopilot.get(
                "review_required"
            )
            is True
        ):
            return write_status({
                "ok": True,
                "paused": True,
                "action":
                    "scheduler_paused_for_review",
                "reason":
                    saved.get("reason")
                    or autopilot.get(
                        "safety_stop_reason"
                    )
                    or "Local review is required.",
                "current_skill":
                    autopilot.get(
                        "current_skill"
                    ),
            })

        state = load(STATE)
        sprint = load(SPRINT)

        skill_id = autopilot.get(
            "current_skill"
        )

        snapshot: Path | None = None

        try:
            if isinstance(skill_id, str):
                record = (
                    state.get("skills") or {}
                ).get(skill_id)

                if not isinstance(record, dict):
                    raise RuntimeError(
                        "Current skill has no practice record."
                    )

                if (
                    record.get(
                        "cadquery_evidence_validated"
                    )
                    is True
                ):
                    snapshot = create_snapshot(
                        skill_id,
                        "complete",
                    )

                    complete_current(
                        skill_id
                    )

                    return write_status({
                        "ok": True,
                        "paused": False,
                        "action":
                            "completed_current_skill",
                        "completed_skill":
                            skill_id,
                        "snapshot":
                            str(snapshot),
                        "next_action":
                            "select_next_safe_candidate",
                    })

                safe, reasons = candidate_safety(
                    skill_id,
                    record,
                )

                if not safe:
                    raise RuntimeError(
                        "Current candidate is outside "
                        "the safe scheduler scope: "
                        + ", ".join(reasons)
                    )

            else:
                candidates = eligible_candidates()

                if not candidates:
                    autopilot["enabled"] = False
                    autopilot["paused"] = True
                    autopilot["review_required"] = False
                    autopilot["safety_stop"] = False
                    autopilot["updated_at"] = now()

                    set_safe_autopilot_fields(
                        autopilot
                    )

                    autopilot["current_plan"] = {
                        "skill_id": None,
                        "next_action":
                            "build_next_lane_safety_gate",
                        "status":
                            "safe_supported_curriculum_complete",
                        "forge_allowed": False,
                    }

                    sprint["current_skill"] = None
                    sprint["updated_at"] = now()

                    atomic(
                        AUTOPILOT,
                        autopilot,
                    )

                    atomic(
                        SPRINT,
                        sprint,
                    )

                    return write_status({
                        "ok": True,
                        "paused": True,
                        "action":
                            "safe_supported_curriculum_complete",
                        "remaining_safe_candidates": 0,
                        "next_action":
                            "build_next_lane_safety_gate",
                    })

                skill_id = candidates[0]

                snapshot = create_snapshot(
                    skill_id,
                    "enrol",
                )

                enroll_candidate(
                    skill_id
                )

                state = load(STATE)
                autopilot = load(AUTOPILOT)

            if snapshot is None:
                snapshot = create_snapshot(
                    skill_id,
                    "learning",
                )

            locked, problems = forge_locked(
                skill_id
            )

            if not locked:
                raise PermissionDrift(
                    "Forge gate is not locked before learning: "
                    + ", ".join(problems)
                )

            state = load(STATE)
            autopilot = load(AUTOPILOT)

            record = state["skills"][skill_id]

            qualification = (
                autopilot.setdefault(
                    "skills",
                    {},
                )
                .setdefault(
                    skill_id,
                    {
                        "qualification_runs": [],
                        "successful_qualifications": 0,
                        "consecutive_qualification_failures": 0,
                        "consecutive_infrastructure_failures": 0,
                    },
                )
            )

            rounds = successful_rounds(
                qualification
            )

            if (
                record.get(
                    "knowledge_certified"
                )
                is True
            ):
                stage = "safe-cadquery"

                journal = journal_directory(
                    skill_id,
                    stage,
                )

                result = cadquery_tick(
                    skill_id,
                    journal,
                )

            elif rounds != {1, 2, 3}:
                missing = sorted(
                    {1, 2, 3} - rounds
                )

                round_number = missing[0]
                stage = (
                    f"uncounted-round-{round_number}"
                )

                journal = journal_directory(
                    skill_id,
                    stage,
                )

                result = qualification_tick(
                    skill_id,
                    round_number,
                    journal,
                )

            else:
                successful_counted = int(
                    record.get(
                        "successful_attempts"
                    )
                    or 0
                )

                stage = (
                    "counted-round-"
                    f"{min(successful_counted + 1, 3)}"
                )

                journal = journal_directory(
                    skill_id,
                    stage,
                )

                result = counted_tick(
                    skill_id,
                    journal,
                )

            result.update({
                "paused":
                    result.get("action", "")
                    .endswith("_paused")
                    or result.get("action", "")
                    .endswith("_safety_stop"),
                "snapshot": str(snapshot),
                "scheduler_version": VERSION,
            })

            status = write_status(
                result
            )

            locked, problems = forge_locked(
                skill_id
            )

            if not locked:
                raise PermissionDrift(
                    "Forge gate changed after scheduler commit: "
                    + ", ".join(problems)
                )

            return status

        except PermissionDrift as exc:
            if (
                snapshot is not None
                and isinstance(skill_id, str)
            ):
                restore_snapshot(
                    snapshot,
                    skill_id,
                )

            safety_stop(
                skill_id
                if isinstance(skill_id, str)
                else None,
                str(exc),
            )

            return write_status({
                "ok": False,
                "paused": True,
                "action":
                    "permission_drift_safety_stop",
                "current_skill": skill_id,
                "reason": str(exc),
                "snapshot":
                    str(snapshot)
                    if snapshot
                    else None,
            })

        except Exception as exc:
            if (
                snapshot is not None
                and isinstance(skill_id, str)
            ):
                restore_snapshot(
                    snapshot,
                    skill_id,
                )

            safety_stop(
                skill_id
                if isinstance(skill_id, str)
                else None,
                (
                    "Unexpected safe scheduler failure: "
                    + str(exc)
                ),
            )

            status = write_status({
                "ok": False,
                "paused": True,
                "action":
                    "unexpected_scheduler_failure",
                "current_skill": skill_id,
                "reason": str(exc),
                "snapshot":
                    str(snapshot)
                    if snapshot
                    else None,
            })

            raise RuntimeError(
                json.dumps(status)
            ) from exc


def self_test() -> dict[str, Any]:
    ensure_environment()

    state_before = sha256(STATE)
    auto_before = sha256(AUTOPILOT)
    sprint_before = sha256(SPRINT)

    candidates = eligible_candidates()

    bounding_locked, bounding_problems = (
        forge_locked(
            "geometry_foundations.bounding-envelopes"
        )
    )

    if not bounding_locked:
        raise RuntimeError(
            "Bounding Envelopes Forge gate is unsafe: "
            + ", ".join(bounding_problems)
        )

    result = {
        "ok": True,
        "classification":
            "SAFE_AUTONOMOUS_SCHEDULER_SELF_TEST_PASS",
        "scheduler_version": VERSION,
        "safe_candidate_count":
            len(candidates),
        "first_safe_candidate":
            candidates[0]
            if candidates
            else None,
        "denylist": sorted(DENYLIST),
        "forge_unlock_allowed": False,
        "old_worker_state":
            service_state(OLD_WORKER),
    }

    if sha256(STATE) != state_before:
        raise RuntimeError(
            "Self-test changed practice state."
        )

    if sha256(AUTOPILOT) != auto_before:
        raise RuntimeError(
            "Self-test changed Autopilot state."
        )

    if sha256(SPRINT) != sprint_before:
        raise RuntimeError(
            "Self-test changed commercial sprint."
        )

    return result


def pause() -> dict[str, Any]:
    autopilot = load(AUTOPILOT)

    autopilot["enabled"] = False
    autopilot["paused"] = True
    autopilot["review_required"] = False
    autopilot["updated_at"] = now()

    set_safe_autopilot_fields(
        autopilot
    )

    atomic(
        AUTOPILOT,
        autopilot,
    )

    return write_status({
        "ok": True,
        "paused": True,
        "action":
            "scheduler_paused_manually",
        "current_skill":
            autopilot.get("current_skill"),
    })


def resume() -> dict[str, Any]:
    ensure_environment()

    autopilot = load(AUTOPILOT)

    skill_id = autopilot.get(
        "current_skill"
    )

    if isinstance(skill_id, str):
        locked, problems = forge_locked(
            skill_id
        )

        if not locked:
            raise RuntimeError(
                "Cannot resume with Forge permission drift: "
                + ", ".join(problems)
            )

    autopilot["enabled"] = True
    autopilot["paused"] = False
    autopilot["review_required"] = False
    autopilot["safety_stop"] = False
    autopilot["safety_stop_reason"] = None
    autopilot["updated_at"] = now()

    set_safe_autopilot_fields(
        autopilot
    )

    atomic(
        AUTOPILOT,
        autopilot,
    )

    return write_status({
        "ok": True,
        "paused": False,
        "action":
            "scheduler_resumed_locally",
        "current_skill": skill_id,
    })


def main() -> int:
    parser = argparse.ArgumentParser(
        description=(
            "SHiRE Academy safe autonomous "
            "supported-curriculum scheduler"
        )
    )

    parser.add_argument(
        "command",
        choices=(
            "self-test",
            "tick",
            "status",
            "pause",
            "resume",
        ),
    )

    args = parser.parse_args()

    if args.command == "self-test":
        payload = self_test()
    elif args.command == "tick":
        payload = scheduler_tick()
    elif args.command == "status":
        payload = live_status()
    elif args.command == "pause":
        payload = pause()
    else:
        payload = resume()

    print(json.dumps(
        payload,
        indent=2,
        sort_keys=True,
        ensure_ascii=False,
    ))

    return 0


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