Blame for src/dashboard/playground_api.py

"""Dashboard adapter for the isolated non-canonical Playground."""

from __future__ import annotations

import csv
import hashlib
import hmac
import io
import json
import os
import re
import shutil
import subprocess
import threading
import time
import uuid
from collections import deque
from collections.abc import Callable, Mapping
from dataclasses import replace
from datetime import datetime, timezone
from http import HTTPStatus
from pathlib import Path
from typing import cast

from src.dashboard.verification import (
    SCIENTIFIC_PATHS,
    TEST_PATHS,
    current_git_head,
    inspect_source_tree,
)
from src.playground import service
from src.playground.cuda import (
    CompileBundle,
    CudaDriver,
    CudaDriverError,
    CudaRuntimeUnavailable,
    GateLaunchInputs,
    compile_mapping,
    cpu_gate_reference,
    execute_gate_bundle,
    gate_execution_parity_summary,
    preflight_bundle,
    validate_cuda_block_size,
)
from src.playground.cuda.builder_parity import run_builder_parity
from src.playground.cuda.recurrent import (
    cpu_recurrent_reference,
    execute_recurrent,
    recurrent_fixture,
    recurrent_parity,
)
from src.playground.cuda.synapses import SynapseConfig
from src.playground.integration import integration_catalog, transfer_element
from src.playground.models import PlaygroundConfig
from src.playground.night_run import NightRunManager
from src.playground.pan import PANEmbodiedSandboxSession, PANSessionDaemon
from src.playground.pan.cue_controls import run_cue_controls
from src.playground.pan.transfer import run_synaptic_transfer

_MAX_CONCURRENT_RUNS = 2
_RUNS_PER_MINUTE = 20
_RUN_WINDOW_SECONDS = 60.0
_CUDA_DIAGNOSTIC_ID = re.compile(r"[0-9a-f]{32}")
_CUDA_DIAGNOSTIC_MAX_BYTES = 8 * 1024 * 1024
_REPO_ROOT = Path(__file__).resolve().parents[2]
_CUDA_DIAGNOSTIC_URL_PREFIX = "/api/playground/cuda/diagnostics/"

_RUN_SEMAPHORE = threading.BoundedSemaphore(_MAX_CONCURRENT_RUNS)
_RATE_LOCK = threading.Lock()
_RECENT_RUNS: deque[float] = deque()


def _canonical_artifact_bytes(artifact: Mapping[str, object]) -> bytes:
    return json.dumps(
        artifact, sort_keys=True, ensure_ascii=True, separators=(",", ":")
    ).encode("utf-8")


def _write_cuda_diagnostic_artifact(
    repo_root: Path, artifact: Mapping[str, object]
) -> tuple[Path, str]:
    artifact_id = artifact.get("artifact_id")
    if not isinstance(artifact_id, str) or not _CUDA_DIAGNOSTIC_ID.fullmatch(
        artifact_id
    ):
        raise ValueError("CUDA diagnostic artifact ID is invalid")

    directory = repo_root / "artifacts" / "cuda_diagnostics"
    directory.mkdir(mode=0o700, parents=True, exist_ok=True)
    if os.name != "nt":
        directory.chmod(0o700)
    record = dict(artifact)
    artifact_digest = hashlib.sha256(_canonical_artifact_bytes(record)).hexdigest()
    record["artifact_sha256"] = artifact_digest
    serialized = json.dumps(record, sort_keys=True, indent=2, ensure_ascii=True) + "\n"
    encoded = serialized.encode("utf-8")
    if len(encoded) > _CUDA_DIAGNOSTIC_MAX_BYTES:
        raise ValueError("CUDA diagnostic artifact exceeds the 8 MiB limit")

    destination = directory / f"{artifact_id}.json"
    temporary = directory / f".{artifact_id}.{uuid.uuid4().hex}.tmp"
    try:
        temporary.write_bytes(encoded)
        if os.name != "nt":
            temporary.chmod(0o600)
        os.replace(temporary, destination)
    finally:
        temporary.unlink(missing_ok=True)
    return destination, artifact_digest


def _load_cuda_diagnostic_artifact(
    repo_root: Path, artifact_id: str
) -> dict[str, object]:
    if not _CUDA_DIAGNOSTIC_ID.fullmatch(artifact_id):
        raise ValueError("CUDA diagnostic artifact ID is invalid")
    path = repo_root / "artifacts" / "cuda_diagnostics" / f"{artifact_id}.json"
    value = json.loads(path.read_text(encoding="utf-8"))
    if not isinstance(value, dict):
        raise ValueError("CUDA diagnostic artifact is malformed")
    record = dict(cast(dict[str, object], value))
    if record.get("artifact_id") != artifact_id:
        raise ValueError("CUDA diagnostic artifact is malformed")
    recorded_digest = record.pop("artifact_sha256", None)
    actual_digest = hashlib.sha256(_canonical_artifact_bytes(record)).hexdigest()
    if not isinstance(recorded_digest, str) or not hmac.compare_digest(
        recorded_digest, actual_digest
    ):
        raise ValueError("CUDA diagnostic artifact digest mismatch")
    record["artifact_sha256"] = recorded_digest
    return record


def _cuda_hardware_identity(device_ordinal: int) -> dict[str, object]:
    executable = shutil.which("nvidia-smi")
    if executable is None:
        return {
            "status": "unavailable",
            "source": "nvidia-smi",
            "device_ordinal": device_ordinal,
            "reason": "nvidia-smi_not_found",
        }
    try:
        completed = subprocess.run(
            [
                executable,
                "--query-gpu=index,name,uuid,pci.bus_id,driver_version",
                "--format=csv,noheader,nounits",
            ],
            capture_output=True,
            check=False,
            text=True,
            timeout=5,
        )
    except (OSError, subprocess.SubprocessError) as exc:
        return {
            "status": "unavailable",
            "source": "nvidia-smi",
            "device_ordinal": device_ordinal,
            "reason": type(exc).__name__,
        }
    if completed.returncode != 0:
        return {
            "status": "unavailable",
            "source": "nvidia-smi",
            "device_ordinal": device_ordinal,
            "reason": "query_failed",
        }

    rows = [
        [value.strip() for value in row]
        for row in csv.reader(io.StringIO(completed.stdout), skipinitialspace=True)
        if row
    ]
    selector = os.environ.get("CUDA_VISIBLE_DEVICES")
    visible = selector.split(",") if selector is not None else None
    if visible is not None:
        if device_ordinal >= len(visible):
            return {
                "status": "unavailable",
                "source": "nvidia-smi",
                "device_ordinal": device_ordinal,
                "reason": "device_not_visible",
            }
        selected = visible[device_ordinal].strip()
        if selected.isdecimal():
            match = next((row for row in rows if row[0] == selected), None)
        elif selected.startswith("GPU-"):
            match = next(
                (row for row in rows if len(row) > 2 and row[2] == selected), None
            )
        else:
            match = None
    else:
        match = next((row for row in rows if row[0] == str(device_ordinal)), None)

    if match is None or len(match) < 5 or match[2] in {"", "N/A"}:
        return {
            "status": "unavailable",
            "source": "nvidia-smi",
            "device_ordinal": device_ordinal,
            "reason": "device_identity_unresolved",
        }
    return {
        "status": "captured",
        "source": "nvidia-smi",
        "device_ordinal": device_ordinal,
        "physical_index": match[0],
        "model": match[1],
        "device_uuid_sha256": hashlib.sha256(match[2].encode("utf-8")).hexdigest(),
        "pci_bus_id": match[3],
        "driver_version": match[4],
    }


def _working_tree_provenance(repo_root: Path) -> dict[str, object]:
    try:
        inspection = inspect_source_tree(repo_root)
    except (OSError, ValueError, subprocess.SubprocessError) as exc:
        return {
            "git_commit": current_git_head(repo_root),
            "working_tree_digest": None,
            "working_tree_digest_status": type(exc).__name__,
            "working_tree_digest_scope": [*SCIENTIFIC_PATHS, *TEST_PATHS],
        }
    dirty = bool(
        inspection.dirty_relevant_paths
        or inspection.untracked_relevant_paths
        or inspection.missing_relevant_paths
    )
    return {
        "git_commit": current_git_head(repo_root),
        "working_tree_digest": inspection.digest,
        "working_tree_digest_status": (
            "captured" if inspection.digest else "unavailable"
        ),
        "working_tree_state": (
            "unknown"
            if not inspection.git_available
            else "modified" if dirty else "clean"
        ),
        "working_tree_digest_scope": [*SCIENTIFIC_PATHS, *TEST_PATHS],
        "dirty_relevant_paths": list(inspection.dirty_relevant_paths),
        "untracked_relevant_paths": list(inspection.untracked_relevant_paths),
        "missing_relevant_paths": list(inspection.missing_relevant_paths),
    }


def _diagnostic_outcome(result: Mapping[str, object]) -> str:
    passed = result.get("passed")
    if isinstance(passed, bool):
        return "passed" if passed else "failed"
    parity = result.get("parity")
    if isinstance(parity, Mapping):
        parity_passed = cast(Mapping[str, object], parity).get("passed")
        if isinstance(parity_passed, bool):
            return "passed" if parity_passed else "failed"
    actions_exact = result.get("actions_exact")
    if isinstance(actions_exact, bool):
        return "passed" if actions_exact else "failed"
    return "not_evaluated"


def _diagnostic_http_status(exc: Exception) -> int:
    if isinstance(exc, (ValueError, TypeError, json.JSONDecodeError)):
        return HTTPStatus.BAD_REQUEST
    if isinstance(exc, FileNotFoundError):
        return HTTPStatus.NOT_FOUND
    if isinstance(exc, TimeoutError):
        return HTTPStatus.GATEWAY_TIMEOUT
    if isinstance(exc, RuntimeError):
        return HTTPStatus.SERVICE_UNAVAILABLE
    return HTTPStatus.INTERNAL_SERVER_ERROR


def _run_cuda_diagnostic(
    route: str,
    payload: Mapping[str, object],
    operation: Callable[[], dict[str, object]],
    *,
    repo_root: Path | None = None,
) -> dict[str, object]:
    root = repo_root or _REPO_ROOT
    started = datetime.now(timezone.utc)
    artifact_id = uuid.uuid4().hex
    ordinal = payload.get("device_ordinal", 0)
    device_ordinal = (
        ordinal
        if isinstance(ordinal, int) and not isinstance(ordinal, bool) and ordinal >= 0
        else 0
    )
    provenance = _working_tree_provenance(root)
    hardware = _cuda_hardware_identity(device_ordinal)
    try:
        result = operation()
    except Exception as exc:
        completed = datetime.now(timezone.utc)
        record: dict[str, object] = {
            "schema_version": "mhrn.cuda-diagnostic.v1",
            "artifact_id": artifact_id,
            "route": route,
            "execution_status": "failed",
            "outcome": "error",
            "started_at": started.isoformat(),
            "completed_at": completed.isoformat(),
            "duration_seconds": (completed - started).total_seconds(),
            "request_sha256": hashlib.sha256(
                _canonical_artifact_bytes(payload)
            ).hexdigest(),
            "provenance": provenance,
            "hardware_identity": hardware,
            "error_type": type(exc).__name__,
            "error": str(exc)[:1000],
            "scientific_evidence": False,
        }
        _attach_cuda_diagnostic_receipt(exc, root, record)
        raise

    completed = datetime.now(timezone.utc)
    record = {
        "schema_version": "mhrn.cuda-diagnostic.v1",
        "artifact_id": artifact_id,
        "route": route,
        "execution_status": "completed",
        "outcome": _diagnostic_outcome(result),
        "started_at": started.isoformat(),
        "completed_at": completed.isoformat(),
        "duration_seconds": (completed - started).total_seconds(),
        "request_sha256": hashlib.sha256(
            _canonical_artifact_bytes(payload)
        ).hexdigest(),
        "provenance": provenance,
        "hardware_identity": hardware,
        "result": result,
        "scientific_evidence": False,
    }
    receipt = _attach_cuda_diagnostic_receipt(None, root, record)
    response = dict(result)
    response["run_evidence"] = receipt
    return response


def _attach_cuda_diagnostic_receipt(
    error: Exception | None,
    repo_root: Path,
    record: dict[str, object],
) -> dict[str, object]:
    artifact_id = cast(str, record["artifact_id"])
    receipt: dict[str, object] = {
        "artifact_id": artifact_id,
        "artifact_url": f"/api/playground/cuda/diagnostics/{artifact_id}",
        "execution_status": record["execution_status"],
        "outcome": record["outcome"],
        "started_at": record["started_at"],
        "completed_at": record["completed_at"],
        "duration_seconds": record["duration_seconds"],
        "provenance": record["provenance"],
        "hardware_identity": record["hardware_identity"],
    }
    record["artifact_url"] = receipt["artifact_url"]
    try:
        _, artifact_digest = _write_cuda_diagnostic_artifact(repo_root, record)
    except (OSError, TypeError, ValueError) as exc:
        receipt["persistence_status"] = "failed"
        receipt["persistence_error"] = type(exc).__name__
        receipt["artifact_url"] = None
    else:
        receipt["persistence_status"] = "persisted"
        receipt["artifact_sha256"] = artifact_digest
    if error is not None:
        setattr(error, "run_evidence", receipt)
        setattr(error, "run_http_status", _diagnostic_http_status(error))
    return receipt


def _bounded_cuda_diagnostic(
    route: str,
    payload: Mapping[str, object],
    operation: Callable[[], dict[str, object]],
) -> dict[str, object]:
    return _bounded_operation(lambda: _run_cuda_diagnostic(route, payload, operation))


def _payload_int(
    payload: Mapping[str, object],
    name: str,
    default: int,
) -> int:
    value = payload.get(name, default)
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        raise ValueError(f"{name} must be numeric")
    converted = int(value)
    if float(value) != float(converted):
        raise ValueError(f"{name} must be an integer")
    return converted


def _payload_float(
    payload: Mapping[str, object],
    name: str,
    default: float,
) -> float:
    value = payload.get(name, default)
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        raise ValueError(f"{name} must be numeric")
    return float(value)


def _payload_text(
    payload: Mapping[str, object],
    name: str,
    default: str,
) -> str:
    value = payload.get(name, default)
    if not isinstance(value, str):
        raise ValueError(f"{name} must be a string")
    return value


def _numeric_list(value: object, name: str) -> list[float]:
    if not isinstance(value, list):
        raise ValueError(f"{name} must be a list")
    items = cast(list[object], value)
    result: list[float] = []
    for item in items:
        if isinstance(item, bool) or not isinstance(item, (int, float)):
            raise ValueError(f"{name} entries must be numeric")
        result.append(float(item))
    return result


_LIVE_DAEMON = PANSessionDaemon()
_LIVE_SANDBOXES: dict[str, PANEmbodiedSandboxSession] = {}
_LIVE_LOCK = threading.RLock()
_NIGHT_RUN = NightRunManager()


class PlaygroundRateLimitError(RuntimeError):
    """Raised when the bounded Playground API rate is exceeded."""


class PlaygroundBusyError(RuntimeError):
    """Raised when all bounded Playground worker slots are occupied."""


def _reserve_rate_slot() -> None:
    now = time.monotonic()
    with _RATE_LOCK:
        cutoff = now - _RUN_WINDOW_SECONDS
        while _RECENT_RUNS and _RECENT_RUNS[0] < cutoff:
            _RECENT_RUNS.popleft()
        if len(_RECENT_RUNS) >= _RUNS_PER_MINUTE:
            raise PlaygroundRateLimitError(
                "Playground run rate exceeded; retry after the current minute window."
            )
        _RECENT_RUNS.append(now)


def _bounded_operation(
    operation: Callable[[], dict[str, object]],
) -> dict[str, object]:
    _reserve_rate_slot()
    if not _RUN_SEMAPHORE.acquire(blocking=False):
        raise PlaygroundBusyError(
            "Playground is busy; at most two simulation requests run concurrently."
        )
    try:
        return operation()
    finally:
        _RUN_SEMAPHORE.release()


def _bounded_run(path: str, payload: Mapping[str, object]) -> dict[str, object]:
    def operation() -> dict[str, object]:
        if path == "/api/playground/run":
            return service.run(payload)
        if path == "/api/playground/robustness":
            return service.robustness(payload)
        raise ValueError(f"unknown Playground POST route: {path}")

    return _bounded_operation(operation)


def _live_parts(path: str) -> list[str]:
    prefix = "/api/playground/live/"
    if not path.startswith(prefix):
        return []
    return [part for part in path[len(prefix) :].split("/") if part]


def _cuda_runtime_status() -> dict[str, object]:
    ptxas_path = shutil.which("ptxas")
    driver_available = False
    driver_error: str | None = None
    try:
        driver = CudaDriver()
        driver.initialize()
        driver_available = True
    except (CudaRuntimeUnavailable, CudaDriverError, OSError) as exc:
        driver_error = str(exc)

    return {
        "classification": "PLAYGROUND_CUDA_DIAGNOSTICS_STATUS",
        "scientific_evidence": False,
        "canonical_cuda_backend": False,
        "provenance": {
            "git_commit": current_git_head(Path(__file__).resolve().parents[2]),
            "source": "runtime_repository_head",
            "working_tree_state": "not_captured",
        },
        "target_default": "sm_86",
        "ptxas_available": ptxas_path is not None,
        "ptxas_path": ptxas_path,
        "cuda_driver_available": driver_available,
        "cuda_driver_error": driver_error,
        "stages": {
            "CUDA-1.0": "PTXAS_LOAD_PREFLIGHT_IMPLEMENTED",
            "CUDA-1.1": "DETERMINISM_FREEZE_PARITY_CONTRACT_IMPLEMENTED",
            "CUDA-1.2": "LAUNCH_ABI_BUFFER_VALIDATION_IMPLEMENTED",
            "CUDA-1.3": "CPU_GATE_REFERENCE_D2_IMPLEMENTED_HARDWARE_VERIFICATION_REQUIRED",
            "CUDA-1.4": "BOUNDED_RECURRENT_FP64_REFERENCE_IMPLEMENTED",
            "CUDA-1.5": "BOUNDED_FROZEN_REWARD_PLASTICITY_REFERENCE_IMPLEMENTED",
            "CUDA-1.6": "HYBRID_BUILDER_MEMBRANE_AND_OPTIONAL_PAN_STATE_GPU_BODY_CPU_IMPLEMENTED",
        },
        "application_cpu": {
            "sandbox_physics": True,
            "sensorics": True,
            "actuation": True,
            "posture_analysis": True,
            "reward_triggers": True,
            "closed_loop": True,
        },
        "gpu_porting": {
            "gate_single_tick": True,
            "membrane_state_100_ticks": True,
            "adaptation_state": True,
            "pan_health_energy_hyperstate": "OPTIONAL_CUDA_PAN_BUILDER",
            "pan_apoptosis_decision": "OPTIONAL_CUDA_PAN_BUILDER",
            "pan_feedback_projection": "OPTIONAL_CUDA_PAN_BUILDER",
            "refractory_state": False,
            "synapses": "BUILDER_GPU_EMISSION_PLASTICITY_REWARD_RESIDENT_QUEUE_HOST_TRACE",
            "builder_rng_semantics": "HOST_TRAVERSAL_STREAM_PRESERVED",
            "builder_inhibitory_emission": True,
            "delays": True,
            "builder_delay_queue": "DEVICE_RESIDENT_RING_65",
            "builder_neuron_traces": "DEVICE_RESIDENT_TWO_PHASE",
            "builder_pan_population": "CUDA_ORDERED_AXIS_REDUCTION",
            "builder_feedback_history": "DEVICE_RING_66_SOURCE_SELECTION",
            "plasticity": "BUILDER_PAIR_TRIPLET_ELIGIBILITY_MODULATION_AND_BOUNDED_REFERENCE",
            "sandbox_physics": False,
        },
        "verification": {
            "cpu_gate_reference": True,
            "single_tick_d2_endpoint": True,
            "rng_action_path_endpoint": True,
            "raw_rng_value_parity": True,
            "memory_leak_instrumentation": True,
        },
    }


def _diagnostic_inputs(
    bundle: CompileBundle,
    *,
    n_neurons: int,
    seed: int,
    epsilon: float,
) -> GateLaunchInputs:
    manifest = bundle.manifest
    raw_abi = manifest.get("kernel_abi")
    if not isinstance(raw_abi, Mapping):
        raise ValueError("compile bundle is missing kernel_abi")
    abi = cast(Mapping[str, object], raw_abi)
    input_channels = _payload_int(abi, "input_channels", 8)
    action_count = _payload_int(abi, "action_space_size", 4)
    pan_dimensions = _payload_int(abi, "pan_dimensions", 5)
    full_mask = (1 << input_channels) - 1 if input_channels < 64 else 0xFFFFFFFFFFFFFFFF
    logits: list[float] = []
    for neuron in range(n_neurons):
        preferred = neuron % action_count
        for action in range(action_count):
            logits.append(1.0 if action == preferred else 0.05 * (action + 1))
    return GateLaunchInputs.from_sequences(
        input_current=[1.0 + 0.125 * (index % 7) for index in range(n_neurons)],
        channel_masks=[full_mask] * n_neurons,
        amplitudes=[1.0 + 0.05 * channel for channel in range(input_channels)],
        reward_ring=[0.5, -0.25, 0.75, 0.0],
        action_map=[
            0.1 * (action + 1) * (channel + 1)
            for action in range(action_count)
            for channel in range(input_channels)
        ],
        feedback_matrix=[
            0.01 * (1 + ((neuron + dimension) % 5))
            for neuron in range(n_neurons)
            for dimension in range(pan_dimensions)
        ],
        population=[0.1 * (dimension + 1) for dimension in range(pan_dimensions)],
        logits=logits,
        tick=3,
        target_index=min(2, action_count - 1),
        previous_action=min(1, action_count - 1),
        seed=seed,
        epsilon=epsilon,
    )


def _cuda_config_payload(payload: Mapping[str, object]) -> dict[str, object]:
    excluded = {
        "target_sm",
        "ptx_version",
        "n_neurons",
        "block_size",
        "device_ordinal",
        "reference_commit",
        "rng_samples",
    }
    return {key: value for key, value in payload.items() if key not in excluded}


def _cuda_preflight(payload: Mapping[str, object]) -> dict[str, object]:
    target_sm = _payload_text(payload, "target_sm", "sm_86")
    ptx_version = _payload_text(payload, "ptx_version", "7.1")
    n_neurons = _payload_int(payload, "n_neurons", 256)
    block_size = validate_cuda_block_size(_payload_int(payload, "block_size", 64))
    device_ordinal = _payload_int(payload, "device_ordinal", 0)
    if not 1 <= n_neurons <= 12_288:
        raise ValueError("n_neurons must be in [1, 12288]")
    bundle = compile_mapping(
        _cuda_config_payload(payload),
        target_sm=target_sm,
        ptx_version=ptx_version,
    )
    return preflight_bundle(
        bundle,
        n_neurons=n_neurons,
        block_size=block_size,
        device_ordinal=device_ordinal,
    )


def _cuda_smoke(payload: Mapping[str, object]) -> dict[str, object]:
    target_sm = _payload_text(payload, "target_sm", "sm_86")
    ptx_version = _payload_text(payload, "ptx_version", "7.1")
    n_neurons = _payload_int(payload, "n_neurons", 64)
    block_size = validate_cuda_block_size(_payload_int(payload, "block_size", 64))
    device_ordinal = _payload_int(payload, "device_ordinal", 0)
    reference_commit = _payload_text(payload, "reference_commit", "")
    if not 1 <= n_neurons <= 4096:
        raise ValueError("n_neurons must be in [1, 4096] for hardware smoke")
    bundle = compile_mapping(
        _cuda_config_payload(payload),
        target_sm=target_sm,
        ptx_version=ptx_version,
    )
    inputs = _diagnostic_inputs(
        bundle,
        n_neurons=n_neurons,
        seed=12345,
        epsilon=0.0,
    )
    reference = cpu_gate_reference(bundle, inputs)
    first = execute_gate_bundle(
        bundle,
        inputs,
        block_size=block_size,
        device_ordinal=device_ordinal,
    )
    second = execute_gate_bundle(
        bundle,
        inputs,
        block_size=block_size,
        device_ordinal=device_ordinal,
    )
    first_parity = gate_execution_parity_summary(
        reference,
        first,
        tolerance=1.0e-5,
        reference_commit=reference_commit,
    )
    second_parity = gate_execution_parity_summary(
        reference,
        second,
        tolerance=1.0e-5,
        reference_commit=reference_commit,
    )
    deterministic = first.get("outputs") == second.get("outputs")
    return {
        "classification": "PLAYGROUND_CUDA1_3_HARDWARE_SMOKE",
        "scientific_evidence": False,
        "canonical_cuda_backend": False,
        "reference": reference,
        "first": first,
        "second": second,
        "parity": first_parity,
        "second_parity": second_parity,
        "gpu_repeat_exact": deterministic,
        "passed": bool(first_parity["passed"])
        and bool(second_parity["passed"])
        and deterministic,
        "cleanup_contract": {
            "device_allocations_released_in_finally": True,
            "driver_module_unloaded": True,
            "memory_leak_instrumented": True,
            "first_memory": first.get("memory"),
            "second_memory": second.get("memory"),
        },
    }


def _cuda_rng_parity(payload: Mapping[str, object]) -> dict[str, object]:
    target_sm = _payload_text(payload, "target_sm", "sm_86")
    ptx_version = _payload_text(payload, "ptx_version", "7.1")
    samples = _payload_int(payload, "rng_samples", 1000)
    block_size = validate_cuda_block_size(_payload_int(payload, "block_size", 64))
    device_ordinal = _payload_int(payload, "device_ordinal", 0)
    if not 1 <= samples <= 4096:
        raise ValueError("rng_samples must be in [1, 4096]")
    bundle = compile_mapping(
        _cuda_config_payload(payload),
        target_sm=target_sm,
        ptx_version=ptx_version,
    )
    inputs = _diagnostic_inputs(
        bundle,
        n_neurons=samples,
        seed=12345,
        epsilon=1.0,
    )
    reference = cpu_gate_reference(bundle, inputs)
    candidate = execute_gate_bundle(
        bundle,
        inputs,
        block_size=block_size,
        device_ordinal=device_ordinal,
    )
    reference_outputs = cast(Mapping[str, object], reference["outputs"])
    candidate_outputs = cast(Mapping[str, object], candidate["outputs"])
    reference_actions = cast(list[int], reference_outputs["action"])
    candidate_actions = cast(list[int], candidate_outputs["action"])
    actions_exact = reference_actions == candidate_actions
    return {
        "classification": "PLAYGROUND_CUDA_RNG_ACTION_PATH_PARITY",
        "scientific_evidence": False,
        "samples": samples,
        "seed": 12345,
        "tick": inputs.tick,
        "epsilon": 1.0,
        "actions_exact": actions_exact,
        "passed": actions_exact,
        "first_reference_actions": reference_actions[:20],
        "first_cuda_actions": candidate_actions[:20],
        "scope": "HASH_EPSILON_GREEDY_ACTION_PATH_UINT24_TO_F32_HALF_OPEN",
    }


def _cuda_recurrent_parity(payload: Mapping[str, object]) -> dict[str, object]:
    block_size = validate_cuda_block_size(_payload_int(payload, "block_size", 64))
    inputs = recurrent_fixture(
        n_neurons=_payload_int(payload, "n_neurons", 129),
        ticks=_payload_int(payload, "ticks", 100),
        model=_payload_text(payload, "model", "pan_adex_5d"),
    )
    plasticity = payload.get("plasticity", False)
    if not isinstance(plasticity, bool):
        raise ValueError("plasticity must be boolean")
    if plasticity:
        inputs = replace(
            inputs,
            synapses=SynapseConfig(),
            rewards=tuple(
                1.0 if tick % 32 == 0 else 0.0 for tick in range(inputs.ticks)
            ),
        )
    reference = cpu_recurrent_reference(inputs)
    candidate = execute_recurrent(
        inputs,
        block_size=block_size,
        target_sm=_payload_text(payload, "target_sm", "sm_86"),
    )
    return {
        "classification": (
            "PLAYGROUND_CUDA15_PLASTICITY_PARITY"
            if plasticity
            else "PLAYGROUND_CUDA14_RECURRENT_PARITY"
        ),
        "scientific_evidence": False,
        "model": inputs.model,
        "ticks": inputs.ticks,
        "state_dtype": candidate["state_dtype"],
        "preflight": candidate["preflight"],
        "parity": recurrent_parity(
            reference, cast(dict[str, object], candidate["outputs"])
        ),
        "scope": candidate["comparison_scope"],
        "full_pan_backend": False,
    }


def get_playground(path: str) -> dict[str, object] | None:
    if path == "/api/playground/catalog":
        return service.catalog()
    if path == "/api/playground/integration":
        return integration_catalog()
    if path == "/api/playground/sessions":
        return service.sessions()
    if path.startswith("/api/playground/sessions/"):
        session_id = path[len("/api/playground/sessions/") :]
        return service.replay(session_id)
    if path.startswith(_CUDA_DIAGNOSTIC_URL_PREFIX):
        artifact_id = path[len(_CUDA_DIAGNOSTIC_URL_PREFIX) :]
        if not _CUDA_DIAGNOSTIC_ID.fullmatch(artifact_id):
            return None
        try:
            return _load_cuda_diagnostic_artifact(_REPO_ROOT, artifact_id)
        except FileNotFoundError:
            return None
    if path == "/api/playground/night":
        return _NIGHT_RUN.status()
    if path == "/api/playground/cuda/status":
        return _cuda_runtime_status()
    if path == "/api/playground/live":
        return {
            "class": "PLAYGROUND_LIVE_SESSIONS",
            "scientific_evidence": False,
            "sessions": _LIVE_DAEMON.list(),
        }
    parts = _live_parts(path)
    if len(parts) == 1:
        return {"session_id": parts[0], **_LIVE_DAEMON.get(parts[0]).snapshot()}
    return None


def post_playground(
    path: str, payload: Mapping[str, object]
) -> dict[str, object] | None:
    if path in {"/api/playground/run", "/api/playground/robustness"}:
        return _bounded_run(path, payload)

    if path == "/api/playground/integration/transfer":
        return transfer_element(payload)

    if path == "/api/playground/research/synaptic-transfer":
        return _bounded_cuda_diagnostic(
            path, payload, lambda: run_synaptic_transfer(payload)
        )

    if path == "/api/playground/research/cue-controls":
        return _bounded_cuda_diagnostic(
            path, payload, lambda: run_cue_controls(payload)
        )

    if path == "/api/playground/determinism":
        return service.determinism(payload)

    if path == "/api/playground/cuda/compile":
        target_sm = _payload_text(payload, "target_sm", "sm_86")
        ptx_version = _payload_text(payload, "ptx_version", "7.1")
        config_payload = dict(payload)
        config_payload.pop("target_sm", None)
        config_payload.pop("ptx_version", None)
        return compile_mapping(
            config_payload,
            target_sm=target_sm,
            ptx_version=ptx_version,
        ).to_mapping()

    if path == "/api/playground/cuda/preflight":
        return _bounded_cuda_diagnostic(path, payload, lambda: _cuda_preflight(payload))

    if path == "/api/playground/cuda/smoke":
        return _bounded_cuda_diagnostic(path, payload, lambda: _cuda_smoke(payload))

    if path == "/api/playground/cuda/rng-parity":
        return _bounded_cuda_diagnostic(
            path, payload, lambda: _cuda_rng_parity(payload)
        )

    if path == "/api/playground/cuda/builder-parity":
        return _bounded_cuda_diagnostic(
            path, payload, lambda: run_builder_parity(payload)
        )

    if path == "/api/playground/cuda/recurrent-parity":
        return _bounded_cuda_diagnostic(
            path, payload, lambda: _cuda_recurrent_parity(payload)
        )

    if path == "/api/playground/night/start":
        return _NIGHT_RUN.start(
            hours=_payload_float(payload, "hours", 8.0),
            max_episodes=_payload_int(payload, "max_episodes", 10_000),
            checkpoint_seconds=_payload_float(payload, "checkpoint_seconds", 600.0),
            seed=_payload_int(payload, "seed", 12345),
        )
    if path == "/api/playground/night/stop":
        return _NIGHT_RUN.stop()

    if path == "/api/playground/live/stop-all":
        with _LIVE_LOCK:
            _LIVE_SANDBOXES.clear()
        return _LIVE_DAEMON.stop_all()

    if path == "/api/playground/live/create":
        config_payload = dict(payload)
        config_payload.setdefault("neuron_model", "pan_adex_5d")
        config_payload.setdefault("pan_enabled", True)
        config_payload.setdefault("thalamic_relay_threshold", 0.0)
        config_payload.setdefault("pan_bias_current", 10.0)
        config_payload.setdefault("behavior_target_mode", "cycle")
        config = PlaygroundConfig.from_mapping(config_payload)
        session_id = _LIVE_DAEMON.create(config)
        with _LIVE_LOCK:
            _LIVE_SANDBOXES[session_id] = PANEmbodiedSandboxSession(
                _LIVE_DAEMON.get(session_id)
            )
        return {
            "class": "PLAYGROUND_LIVE_SESSION",
            "scientific_evidence": False,
            "session_id": session_id,
            "state": _LIVE_DAEMON.get(session_id).snapshot(),
        }

    parts = _live_parts(path)
    if len(parts) >= 2:
        session_id, action = parts[0], parts[1]
        session = _LIVE_DAEMON.get(session_id)
        if action == "step":
            ticks = _payload_int(payload, "ticks", 32)
            include_topology = payload.get("include_topology", True) is not False
            return {
                "session_id": session_id,
                **session.step(ticks, include_topology=include_topology),
            }
        if action == "input":
            values = _numeric_list(payload.get("values", []), "values")
            duration = _payload_int(payload, "duration_ticks", 16)
            gain = _payload_float(payload, "gain", 25.0)
            session.inject_vector(values, duration_ticks=duration, gain=gain)
            return {"session_id": session_id, "accepted": True}
        if action == "strategy":
            context = _payload_text(payload, "context", "default")
            options = _payload_int(payload, "action_count", 4)
            chosen = session.choose_strategy(context, options)
            return {
                "session_id": session_id,
                "classification": "PLAYGROUND_META_STRATEGY",
                "scientific_evidence": False,
                "context": context,
                "action": chosen,
                "action_count": options,
            }
        if action == "reward":
            context = _payload_text(payload, "context", "default")
            chosen = _payload_int(payload, "action", 0)
            action_count = _payload_int(payload, "action_count", 4)
            reward = _payload_float(payload, "reward", 0.0)
            return {
                "session_id": session_id,
                **session.apply_strategy_reward(
                    context=context,
                    action=chosen,
                    reward=reward,
                    action_count=action_count,
                ),
            }
        if action == "sandbox":
            with _LIVE_LOCK:
                sandbox = _LIVE_SANDBOXES.get(session_id)
                if sandbox is None:
                    sandbox = PANEmbodiedSandboxSession(session)
                    _LIVE_SANDBOXES[session_id] = sandbox
            ticks = _payload_int(payload, "ticks", 1)
            return {"session_id": session_id, **sandbox.step(ticks)}
        if action == "stop":
            with _LIVE_LOCK:
                _LIVE_SANDBOXES.pop(session_id, None)
            return {"session_id": session_id, **_LIVE_DAEMON.stop(session_id)}
    return None