Spaces:

GaryLeckey
/

aureon-os

Running

aureon-os / scripts /validation /pytest_no_skip_shards.py
Deploy Aureon OS (deploy/selfhost stack: nginx edge + fusion API x2 + console + HNC organ + redis + postgres; paper mode) (part 19)
da0ce7b verified 1 day ago
Raw History Blame Contribute Delete
44.5 kB
#!/usr/bin/env python3
"""Deterministic, state-safe pytest collection and shard execution.

The module has two roles:

* a command-line harness that creates/verifies a canonical collection manifest
  and runs exactly one deterministic file shard; and
* a small pytest plugin (loaded with ``-p`` by the harness) that records exact
  collected node IDs, collection problems, loaded plugins, and test outcomes.

The harness never repairs, deletes, or restores operational state.  A changed
fingerprint is terminal evidence: the current receipt records it and every
later shard for the same manifest is refused.
"""

from __future__ import annotations

import argparse
import hashlib
import importlib.metadata
import json
import os
import re
import subprocess
import sys
import tempfile
import time
from collections import Counter, defaultdict
from collections.abc import Callable, Mapping, Sequence
from dataclasses import asdict, dataclass
from pathlib import Path
from typing import Any

SCHEMA = "aureon.pytest-no-skip-shards.v1"
PLUGIN_MODULE = "scripts.validation.pytest_no_skip_shards"
REPORT_OPTION = "--aureon-shard-report"

# This list is intentionally explicit.  Missing files are fingerprinted as
# missing so creation during collection or a shard is also detected.
OPERATIONAL_PATHS: tuple[str, ...] = (
    "aureon/exchanges/.kraken_nonce",
    "real_portfolio_state.json",
    "thoughts.jsonl",
    "trade_logger.log",
    "state/lambda_history.json",
    "state/lighthouse_event.jsonl",
    "state/capital_cfd_last_exchange_trace.json",
    "state/capital_cfd_pending_execution.json",
    "state/capital_shadow_promotions.jsonl",
    "state/aureon_capital_tradable_asset_registry.json",
    "state/unified_order_lifecycle_latest.json",
    "state/unified_order_lifecycle_events.jsonl",
)

SAFETY_ENVIRONMENT: Mapping[str, str] = {
    "AUREON_AUDIT_MODE": "1",
    "AUREON_LIVE": "0",
    "AUREON_LIVE_TRADING": "0",
    "AUREON_OBSERVER_MODE": "dry_run",
    "AUREON_DRY_RUN": "1",
    "AUREON_LLM_OFFLINE": "1",
    "AUREON_DISABLE_LLM_HTTP": "1",
    "AUREON_SUPPRESS_IMPORT_SIDE_EFFECTS": "1",
    "AUREON_ALLOW_PAID_PROVIDERS": "false",
    "AUREON_PROVIDER_MODE": "offline",
    "LIVE": "0",
    "DRY_RUN": "1",
    "OFFLINE": "1",
    "NO_NETWORK": "1",
    "BINANCE_DRY_RUN": "true",
    "KRAKEN_DRY_RUN": "true",
    "ALPACA_DRY_RUN": "true",
    "CAPITAL_DEMO": "true",
    "OANDA_ENVIRONMENT": "practice",
    "AWS_EC2_METADATA_DISABLED": "true",
    "HF_HUB_OFFLINE": "1",
    "TRANSFORMERS_OFFLINE": "1",
    "PIP_NO_INDEX": "1",
    "PYTHONDONTWRITEBYTECODE": "1",
    "PYTHON_DOTENV_DISABLED": "1",
    "PYTEST_ADDOPTS": "",
    "HTTP_PROXY": "http://127.0.0.1:9",
    "HTTPS_PROXY": "http://127.0.0.1:9",
    "ALL_PROXY": "http://127.0.0.1:9",
    "NO_PROXY": "",
    "http_proxy": "http://127.0.0.1:9",
    "https_proxy": "http://127.0.0.1:9",
    "all_proxy": "http://127.0.0.1:9",
    "no_proxy": "",
}

_PROVIDER_PREFIXES = (
    "ALPACA_",
    "ANTHROPIC_",
    "AWS_",
    "BINANCE_",
    "BYBIT_",
    "CAPITAL_",
    "CLOUDFLARE_",
    "COINBASE_",
    "GEMINI_",
    "KRAKEN_",
    "KUCOIN_",
    "OANDA_",
    "OPENAI_",
    "STRIPE_",
    "SUPABASE_",
)
_CREDENTIAL_NAME = re.compile(
    r"(?:^|_)(?:API_KEY|API_SECRET|ACCESS_KEY|ACCESS_TOKEN|AUTH_TOKEN|BEARER_TOKEN|"
    r"CLIENT_SECRET|CONNECTION_STRING|COOKIE|CREDENTIAL|CREDENTIALS|DATABASE_URL|PASSWORD|"
    r"PASSPHRASE|PRIVATE_KEY|REFRESH_TOKEN|SECRET|SECRET_KEY|SERVICE_ACCOUNT|"
    r"SESSION_TOKEN|TOKEN)(?:$|_)",
)
_EXTERNAL_PYTEST_ENV = {"PYTEST_DISABLE_PLUGIN_AUTOLOAD", "PYTEST_PLUGINS"}
_ACTIVE_PYTEST_CONFIG: Any | None = None


class HarnessError(RuntimeError):
    """Fail-closed harness error with no state recovery side effect."""


@dataclass(frozen=True)
class ProcessCapture:
    argv: tuple[str, ...]
    returncode: int
    stdout: str
    stderr: str
    duration_seconds: float
    timed_out: bool = False


Runner = Callable[[Sequence[str], Path, Mapping[str, str], float | None], ProcessCapture]


def _canonical_bytes(value: Any) -> bytes:
    return json.dumps(value, ensure_ascii=False, separators=(",", ":"), sort_keys=True).encode("utf-8")


def _sha256_bytes(value: bytes) -> str:
    return hashlib.sha256(value).hexdigest()


def _sha256_file(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for block in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(block)
    return digest.hexdigest()


def _normalise_distribution(name: str) -> str:
    return re.sub(r"[-_.]+", "-", name).lower()


def _safe_relative_path(root: Path, value: str | Path) -> str:
    candidate = Path(value)
    resolved = candidate.resolve() if candidate.is_absolute() else (root / candidate).resolve()
    try:
        relative = resolved.relative_to(root.resolve())
    except ValueError as exc:
        raise HarnessError(f"path escapes repository root: {value}") from exc
    return relative.as_posix()


def _is_sensitive_environment_name(name: str) -> bool:
    upper = name.upper()
    return upper.startswith(_PROVIDER_PREFIXES) or bool(_CREDENTIAL_NAME.search(upper))


def safe_subprocess_environment(
    inherited: Mapping[str, str] | None = None,
) -> tuple[dict[str, str], tuple[str, ...]]:
    """Return a credential-scrubbed, fail-closed subprocess environment."""

    source = os.environ if inherited is None else inherited
    safe: dict[str, str] = {}
    scrubbed: list[str] = []
    for key, value in source.items():
        if key.upper() in _EXTERNAL_PYTEST_ENV or _is_sensitive_environment_name(key):
            scrubbed.append(key)
            continue
        safe[key] = value
    safe.update(SAFETY_ENVIRONMENT)
    return safe, tuple(sorted(scrubbed, key=str.upper))


def isolate_runtime_writers(environment: Mapping[str, str], temporary: Path) -> dict[str, str]:
    """Route supported append-only runtime writers into the shard temp root."""

    isolated = dict(environment)
    isolated.update(
        {
            "AUREON_THOUGHT_BUS_PATH": str(temporary / "thoughts.jsonl"),
            "AUREON_BUS_TRACE_DIR": str(temporary / "bus-traces"),
            "AUREON_HNC_TRACE_PATH": str(temporary / "hnc-live-trace.jsonl"),
        }
    )
    return isolated


def installed_pytest_plugin_inventory() -> list[dict[str, str]]:
    """Return a deterministic inventory of auto-loadable pytest entry points."""

    discovered = importlib.metadata.entry_points()
    selected = (
        discovered.select(group="pytest11")
        if hasattr(discovered, "select")
        else discovered.get("pytest11", [])
    )
    inventory: list[dict[str, str]] = []
    for entry_point in selected:
        distribution = getattr(entry_point, "dist", None)
        distribution_name = ""
        version = ""
        if distribution is not None:
            distribution_name = str(distribution.metadata.get("Name") or "")
            version = str(distribution.version or "")
        inventory.append(
            {
                "distribution": distribution_name,
                "entry_point": str(entry_point.name),
                "module": str(entry_point.value),
                "version": version,
            }
        )
    return sorted(
        inventory,
        key=lambda row: (
            _normalise_distribution(row["distribution"]),
            row["entry_point"],
            row["module"],
            row["version"],
        ),
    )


def require_pytest_socket(plugin_inventory: Sequence[Mapping[str, str]]) -> None:
    installed = {_normalise_distribution(str(entry.get("distribution") or "")) for entry in plugin_inventory}
    if "pytest-socket" not in installed:
        raise HarnessError(
            "pytest-socket is required for a no-network collection/shard; install it before running the harness"
        )


def fingerprint_operational_paths(
    root: Path,
    paths: Sequence[str] = OPERATIONAL_PATHS,
) -> dict[str, dict[str, Any]]:
    """Fingerprint exact protected paths without following symlinks."""

    root = root.resolve()
    fingerprints: dict[str, dict[str, Any]] = {}
    for raw_path in paths:
        relative = _safe_relative_path(root, raw_path)
        path = root / Path(relative)
        if not path.exists() and not path.is_symlink():
            fingerprints[relative] = {"exists": False}
            continue
        stat = path.lstat()
        common = {
            "exists": True,
            "mtime_ns": stat.st_mtime_ns,
            "size": stat.st_size,
        }
        if path.is_symlink():
            target = os.readlink(path)
            fingerprints[relative] = {
                **common,
                "kind": "symlink",
                "target": target,
                "sha256": _sha256_bytes(os.fsencode(target)),
            }
        elif path.is_file():
            fingerprints[relative] = {**common, "kind": "file", "sha256": _sha256_file(path)}
        elif path.is_dir():
            fingerprints[relative] = {**common, "kind": "directory"}
        else:
            fingerprints[relative] = {**common, "kind": "other"}
    return fingerprints


def operational_changes(
    before: Mapping[str, Mapping[str, Any]],
    after: Mapping[str, Mapping[str, Any]],
) -> list[dict[str, Any]]:
    changes: list[dict[str, Any]] = []
    for path in sorted(set(before) | set(after)):
        old = before.get(path)
        new = after.get(path)
        if old != new:
            changes.append({"path": path, "before": old, "after": new})
    return changes


def default_runner(
    argv: Sequence[str],
    cwd: Path,
    env: Mapping[str, str],
    timeout_seconds: float | None,
) -> ProcessCapture:
    started = time.monotonic()
    try:
        completed = subprocess.run(
            list(argv),
            cwd=cwd,
            env=dict(env),
            capture_output=True,
            text=True,
            timeout=timeout_seconds,
            check=False,
        )
        return ProcessCapture(
            argv=tuple(str(part) for part in argv),
            returncode=completed.returncode,
            stdout=completed.stdout,
            stderr=completed.stderr,
            duration_seconds=time.monotonic() - started,
        )
    except subprocess.TimeoutExpired as exc:
        stdout = exc.stdout.decode(errors="replace") if isinstance(exc.stdout, bytes) else (exc.stdout or "")
        stderr = exc.stderr.decode(errors="replace") if isinstance(exc.stderr, bytes) else (exc.stderr or "")
        return ProcessCapture(
            argv=tuple(str(part) for part in argv),
            returncode=124,
            stdout=stdout,
            stderr=stderr,
            duration_seconds=time.monotonic() - started,
            timed_out=True,
        )
    except OSError as exc:
        return ProcessCapture(
            argv=tuple(str(part) for part in argv),
            returncode=127,
            stdout="",
            stderr=f"pytest process could not start: {exc}",
            duration_seconds=time.monotonic() - started,
        )


def _pytest_command(
    *,
    root: Path,
    config_path: Path,
    report_path: Path,
    cache_path: Path,
    targets: Sequence[str],
    collect_only: bool,
) -> list[str]:
    command = [
        sys.executable,
        "-m",
        "pytest",
        "--rootdir",
        str(root),
        "-c",
        str(config_path),
        "-p",
        PLUGIN_MODULE,
        "--disable-socket",
        "--strict-config",
        "-o",
        f"cache_dir={cache_path}",
        f"{REPORT_OPTION}={report_path}",
    ]
    if collect_only:
        command.append("--collect-only")
    command.extend(targets)
    return command


def _load_json(path: Path, description: str) -> dict[str, Any]:
    try:
        value = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as exc:
        raise HarnessError(f"unable to read {description} at {path}: {exc}") from exc
    if not isinstance(value, dict):
        raise HarnessError(f"{description} must be a JSON object: {path}")
    return value


def _write_new_json(path: Path, value: Mapping[str, Any]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    try:
        with path.open("x", encoding="utf-8", newline="\n") as handle:
            json.dump(value, handle, ensure_ascii=False, indent=2, sort_keys=True)
            handle.write("\n")
    except FileExistsError as exc:
        raise HarnessError(f"refusing to overwrite existing evidence: {path}") from exc


def _validate_plugin_report(report: Mapping[str, Any]) -> None:
    if report.get("schema") != SCHEMA:
        raise HarnessError("pytest plugin report schema mismatch")
    if not report.get("socket_blocker_loaded"):
        raise HarnessError("pytest-socket was not loaded by pytest")
    if not report.get("socket_blocking_requested"):
        raise HarnessError("pytest did not honor --disable-socket")
    if not isinstance(report.get("items"), list):
        raise HarnessError("pytest plugin report has no item list")
    if not isinstance(report.get("collection_errors"), list):
        raise HarnessError("pytest plugin report has no collection error list")
    if not isinstance(report.get("collection_skips"), list):
        raise HarnessError("pytest plugin report has no collection skip list")


def _normalise_items(root: Path, raw_items: Sequence[Mapping[str, Any]]) -> list[dict[str, str]]:
    items: list[dict[str, str]] = []
    for raw_item in raw_items:
        node_id = str(raw_item.get("node_id") or "")
        source_file = str(raw_item.get("source_file") or "")
        if not node_id or not source_file:
            raise HarnessError("every collected item must retain a node ID and source file")
        items.append({"node_id": node_id, "source_file": _safe_relative_path(root, source_file)})
    items.sort(key=lambda item: item["node_id"])
    duplicates = sorted(
        node_id for node_id, count in Counter(item["node_id"] for item in items).items() if count > 1
    )
    if duplicates:
        raise HarnessError(f"duplicate collected node IDs: {duplicates[:5]}")
    return items


def _canonical_skip(skip: Mapping[str, Any]) -> dict[str, str]:
    return {
        "node_id": str(skip.get("node_id") or ""),
        "reason": str(skip.get("reason") or ""),
    }


def assign_lpt_shards(items: Sequence[Mapping[str, str]], shard_count: int) -> list[dict[str, Any]]:
    """Assign whole files by deterministic longest-processing-time approximation."""

    if shard_count < 1:
        raise HarnessError("shard count must be positive")
    by_file: dict[str, list[str]] = defaultdict(list)
    for item in items:
        by_file[str(item["source_file"])].append(str(item["node_id"]))
    if not by_file:
        raise HarnessError("collection contained no test items")
    if shard_count > len(by_file):
        raise HarnessError(f"shard count {shard_count} exceeds collected source-file count {len(by_file)}")

    assignments: list[list[str]] = [[] for _ in range(shard_count)]
    loads = [0] * shard_count
    ordered_files = sorted(by_file, key=lambda path: (-len(by_file[path]), path))
    for source_file in ordered_files:
        shard_index = min(range(shard_count), key=lambda index: (loads[index], index))
        assignments[shard_index].append(source_file)
        loads[shard_index] += len(by_file[source_file])

    shards: list[dict[str, Any]] = []
    for index, files in enumerate(assignments, start=1):
        ordered = sorted(files)
        node_ids = sorted(node_id for source in ordered for node_id in by_file[source])
        shards.append(
            {
                "files": ordered,
                "item_count": len(node_ids),
                "node_ids": node_ids,
                "shard": index,
            }
        )
    return shards


def _assignment_proofs(
    items: Sequence[Mapping[str, str]], shards: Sequence[Mapping[str, Any]]
) -> dict[str, Any]:
    collected_node_ids = sorted(str(item["node_id"]) for item in items)
    assigned_node_ids = [str(node_id) for shard in shards for node_id in shard["node_ids"]]
    counts = Counter(assigned_node_ids)
    duplicate_assignments = sorted(node_id for node_id, count in counts.items() if count != 1)
    missing = sorted(set(collected_node_ids) - set(assigned_node_ids))
    unexpected = sorted(set(assigned_node_ids) - set(collected_node_ids))
    if duplicate_assignments or missing or unexpected or len(assigned_node_ids) != len(collected_node_ids):
        raise HarnessError("shard assignment is not disjoint and exhaustive")
    collected_files = sorted({str(item["source_file"]) for item in items})
    assigned_files = [str(path) for shard in shards for path in shard["files"]]
    if len(assigned_files) != len(set(assigned_files)) or sorted(assigned_files) != collected_files:
        raise HarnessError("source-file shard assignment is not disjoint and exhaustive")
    return {
        "assigned_item_count": len(assigned_node_ids),
        "assigned_node_ids_sha256": _sha256_bytes(_canonical_bytes(sorted(assigned_node_ids))),
        "collected_item_count": len(collected_node_ids),
        "collected_node_ids_sha256": _sha256_bytes(_canonical_bytes(collected_node_ids)),
        "disjoint": True,
        "exhaustive": True,
        "source_file_count": len(collected_files),
        "source_files_sha256": _sha256_bytes(_canonical_bytes(collected_files)),
    }


def build_manifest(
    *,
    root: Path,
    config_path: Path,
    collection_targets: Sequence[str],
    shard_count: int,
    plugin_inventory: Sequence[Mapping[str, str]],
    plugin_report: Mapping[str, Any],
    baseline: Mapping[str, Mapping[str, Any]],
) -> dict[str, Any]:
    _validate_plugin_report(plugin_report)
    collection_errors = plugin_report["collection_errors"]
    if collection_errors:
        raise HarnessError(f"pytest collection reported {len(collection_errors)} error(s)")
    items = _normalise_items(root, plugin_report["items"])
    shards = assign_lpt_shards(items, shard_count)
    proofs = _assignment_proofs(items, shards)
    root = root.resolve()
    payload: dict[str, Any] = {
        "collection": {
            "collection_skip_count": len(plugin_report["collection_skips"]),
            "collection_skips": sorted(
                (_canonical_skip(skip) for skip in plugin_report["collection_skips"]),
                key=lambda skip: (skip["node_id"], skip["reason"]),
            ),
            "items": items,
            "loaded_plugins": plugin_report.get("loaded_plugins", []),
            "targets": [_safe_relative_path(root, target) for target in collection_targets],
        },
        "operational_paths": list(OPERATIONAL_PATHS),
        "operational_state_baseline": dict(baseline),
        "proofs": proofs,
        "pytest_config": _safe_relative_path(root, config_path),
        "pytest_plugin_inventory": list(plugin_inventory),
        "safety": {
            "credential_environment_scrubbed": True,
            "external_pytest_option_injection_removed": sorted(_EXTERNAL_PYTEST_ENV),
            "network_blocker": "pytest-socket --disable-socket",
            "provider_environment": dict(SAFETY_ENVIRONMENT),
        },
        "schema": SCHEMA,
        "shard_count": shard_count,
        "shards": shards,
    }
    payload["manifest_sha256"] = _sha256_bytes(_canonical_bytes(payload))
    return payload


def verify_manifest(manifest: Mapping[str, Any]) -> dict[str, Any]:
    if manifest.get("schema") != SCHEMA:
        raise HarnessError("manifest schema mismatch")
    claimed_digest = str(manifest.get("manifest_sha256") or "")
    payload = dict(manifest)
    payload.pop("manifest_sha256", None)
    actual_digest = _sha256_bytes(_canonical_bytes(payload))
    if not claimed_digest or claimed_digest != actual_digest:
        raise HarnessError("manifest digest mismatch")
    items = manifest.get("collection", {}).get("items")
    shards = manifest.get("shards")
    if not isinstance(items, list) or not isinstance(shards, list):
        raise HarnessError("manifest collection/shards are malformed")
    expected_shards = assign_lpt_shards(items, int(manifest.get("shard_count") or 0))
    if shards != expected_shards:
        raise HarnessError("manifest shard assignment is not canonical LPT output")
    proofs = _assignment_proofs(items, shards)
    if manifest.get("proofs") != proofs:
        raise HarnessError("manifest assignment proofs do not verify")
    if manifest.get("operational_paths") != list(OPERATIONAL_PATHS):
        raise HarnessError("manifest operational path contract differs from this harness")
    return {"manifest_sha256": claimed_digest, "proofs": proofs}


def _process_payload(capture: ProcessCapture) -> dict[str, Any]:
    payload = asdict(capture)
    payload["argv"] = list(capture.argv)
    payload["stdout_sha256"] = _sha256_bytes(capture.stdout.encode("utf-8"))
    payload["stderr_sha256"] = _sha256_bytes(capture.stderr.encode("utf-8"))
    return payload


def _collection_receipt_path(manifest_path: Path) -> Path:
    return manifest_path.with_name(f"{manifest_path.name}.collection-receipt.json")


def _receipt_directory(manifest_path: Path) -> Path:
    return manifest_path.with_name(f"{manifest_path.name}.receipts")


def _shard_receipt_path(manifest_path: Path, shard_index: int) -> Path:
    return _receipt_directory(manifest_path) / f"shard-{shard_index:04d}.json"


def _prior_mutation_receipts(manifest_path: Path, manifest_sha256: str) -> list[Path]:
    receipt_dir = _receipt_directory(manifest_path)
    if not receipt_dir.exists():
        return []
    blocked: list[Path] = []
    for path in sorted(receipt_dir.glob("shard-*.json")):
        receipt = _load_json(path, "shard receipt")
        if receipt.get("manifest_sha256") != manifest_sha256:
            raise HarnessError(f"receipt belongs to another manifest: {path}")
        if receipt.get("state_mutation_detected"):
            blocked.append(path)
    return blocked


def collect_and_write_manifest(
    *,
    root: Path,
    config_path: Path,
    manifest_path: Path,
    shard_count: int,
    collection_targets: Sequence[str] = ("tests",),
    timeout_seconds: float | None = 900.0,
    runner: Runner = default_runner,
    plugin_inventory: Sequence[Mapping[str, str]] | None = None,
) -> dict[str, Any]:
    root = root.resolve()
    config_path = config_path.resolve()
    manifest_path = manifest_path.resolve()
    if manifest_path.exists():
        raise HarnessError(f"refusing to overwrite existing manifest: {manifest_path}")
    receipt_path = _collection_receipt_path(manifest_path)
    if receipt_path.exists():
        raise HarnessError(f"refusing to overwrite existing collection receipt: {receipt_path}")
    if not config_path.is_file():
        raise HarnessError(f"pytest config does not exist: {config_path}")
    config_path = root / Path(_safe_relative_path(root, config_path))
    manifest_path = root / Path(_safe_relative_path(root, manifest_path))
    inventory = list(installed_pytest_plugin_inventory() if plugin_inventory is None else plugin_inventory)
    require_pytest_socket(inventory)
    environment, scrubbed = safe_subprocess_environment()
    before = fingerprint_operational_paths(root)

    with tempfile.TemporaryDirectory(prefix="aureon-pytest-collect-") as temp_dir:
        temporary = Path(temp_dir)
        environment = isolate_runtime_writers(environment, temporary)
        report_path = temporary / "pytest-report.json"
        command = _pytest_command(
            root=root,
            config_path=config_path,
            report_path=report_path,
            cache_path=temporary / "pytest-cache",
            targets=collection_targets,
            collect_only=True,
        )
        capture = runner(command, root, environment, timeout_seconds)
        plugin_report_error = ""
        try:
            plugin_report = (
                _load_json(report_path, "pytest collection report") if report_path.exists() else None
            )
        except HarnessError as exc:
            plugin_report = None
            plugin_report_error = str(exc)

    after = fingerprint_operational_paths(root)
    changes = operational_changes(before, after)
    receipt: dict[str, Any] = {
        "collection_errors": plugin_report.get("collection_errors", []) if plugin_report else [],
        "collection_skip_count": len(plugin_report.get("collection_skips", [])) if plugin_report else 0,
        "credential_keys_scrubbed": list(scrubbed),
        "operational_state_after": after,
        "operational_state_before": before,
        "operational_state_changes": changes,
        "process": _process_payload(capture),
        "pytest_report_error": plugin_report_error,
        "schema": SCHEMA,
        "state_mutation_detected": bool(changes),
    }
    _write_new_json(receipt_path, receipt)
    if changes:
        raise HarnessError("operational state changed during pytest collection; manifest creation aborted")
    if capture.timed_out:
        raise HarnessError("pytest collection timed out")
    if plugin_report is None:
        raise HarnessError("pytest did not produce its structured collection report")
    _validate_plugin_report(plugin_report)
    if capture.returncode != 0:
        error_count = len(plugin_report.get("collection_errors", []))
        raise HarnessError(
            f"pytest collection exited {capture.returncode} with {error_count} collection error(s)"
        )
    if plugin_report["collection_skips"]:
        raise HarnessError(
            f"pytest collection reported {len(plugin_report['collection_skips'])} skip(s)"
        )
    manifest = build_manifest(
        root=root,
        config_path=config_path,
        collection_targets=collection_targets,
        shard_count=shard_count,
        plugin_inventory=inventory,
        plugin_report=plugin_report,
        baseline=before,
    )
    _write_new_json(manifest_path, manifest)
    return manifest


def _acquire_lock(lock_path: Path, manifest_sha256: str, shard_index: int) -> None:
    lock_path.parent.mkdir(parents=True, exist_ok=True)
    try:
        with lock_path.open("x", encoding="utf-8", newline="\n") as handle:
            json.dump(
                {
                    "manifest_sha256": manifest_sha256,
                    "pid": os.getpid(),
                    "shard": shard_index,
                },
                handle,
                sort_keys=True,
            )
            handle.write("\n")
    except FileExistsError as exc:
        raise HarnessError(f"another shard run or an unreviewed interrupted run owns {lock_path}") from exc


def execute_one_shard(
    *,
    root: Path,
    manifest_path: Path,
    shard_index: int,
    timeout_seconds: float | None = 1800.0,
    runner: Runner = default_runner,
    plugin_inventory: Sequence[Mapping[str, str]] | None = None,
) -> dict[str, Any]:
    root = root.resolve()
    manifest_path = manifest_path.resolve()
    manifest_path = root / Path(_safe_relative_path(root, manifest_path))
    manifest = _load_json(manifest_path, "pytest shard manifest")
    verification = verify_manifest(manifest)
    manifest_sha256 = verification["manifest_sha256"]
    if shard_index < 1 or shard_index > len(manifest["shards"]):
        raise HarnessError(f"shard index must be between 1 and {len(manifest['shards'])}")
    inventory = list(installed_pytest_plugin_inventory() if plugin_inventory is None else plugin_inventory)
    require_pytest_socket(inventory)
    if inventory != manifest["pytest_plugin_inventory"]:
        raise HarnessError("installed pytest plugin inventory differs from the collection manifest")
    prior_mutations = _prior_mutation_receipts(manifest_path, manifest_sha256)
    if prior_mutations:
        raise HarnessError(
            f"prior shard mutated operational state; further shards are blocked: {prior_mutations[0]}"
        )
    receipt_path = _shard_receipt_path(manifest_path, shard_index)
    if receipt_path.exists():
        raise HarnessError(f"refusing to overwrite existing shard receipt: {receipt_path}")

    baseline = manifest["operational_state_baseline"]
    before = fingerprint_operational_paths(root)
    baseline_changes = operational_changes(baseline, before)
    if baseline_changes:
        receipt = {
            "manifest_sha256": manifest_sha256,
            "operational_state_before": before,
            "operational_state_changes": baseline_changes,
            "schema": SCHEMA,
            "shard": shard_index,
            "state_mutation_detected": True,
            "status": "baseline_mismatch_before_run",
        }
        _write_new_json(receipt_path, receipt)
        raise HarnessError("operational state no longer matches the collection baseline; shard not started")

    config_path = root / Path(str(manifest["pytest_config"]))
    shard = manifest["shards"][shard_index - 1]
    environment, scrubbed = safe_subprocess_environment()
    receipt_dir = _receipt_directory(manifest_path)
    lock_path = receipt_dir / ".active-shard.lock"
    _acquire_lock(lock_path, manifest_sha256, shard_index)
    capture: ProcessCapture | None = None
    plugin_report: dict[str, Any] | None = None
    plugin_report_error = ""
    after = before
    try:
        with tempfile.TemporaryDirectory(prefix=f"aureon-pytest-shard-{shard_index:04d}-") as temp_dir:
            temporary = Path(temp_dir)
            environment = isolate_runtime_writers(environment, temporary)
            report_path = temporary / "pytest-report.json"
            command = _pytest_command(
                root=root,
                config_path=config_path,
                report_path=report_path,
                cache_path=temporary / "pytest-cache",
                targets=shard["files"],
                collect_only=False,
            )
            capture = runner(command, root, environment, timeout_seconds)
            try:
                plugin_report = (
                    _load_json(report_path, "pytest shard report") if report_path.exists() else None
                )
            except HarnessError as exc:
                plugin_report = None
                plugin_report_error = str(exc)
        after = fingerprint_operational_paths(root)
    finally:
        # This is a harness coordination lock, never an operational state file.
        # Failure to remove it is fail-closed: later shards refuse to start.
        try:
            lock_path.unlink()
        except FileNotFoundError:
            pass

    if capture is None:
        raise HarnessError("pytest shard runner returned no process capture")
    changes = operational_changes(before, after)
    status = "passed"
    selection_changes: dict[str, list[str]] = {"missing": [], "unexpected": []}
    if changes:
        status = "state_mutation_detected"
    elif capture.timed_out:
        status = "timed_out"
    elif plugin_report is None:
        status = "missing_pytest_report"
    else:
        try:
            _validate_plugin_report(plugin_report)
        except HarnessError as exc:
            plugin_report_error = str(exc)
            status = "invalid_pytest_report"
        else:
            actual_node_ids = sorted(str(item["node_id"]) for item in plugin_report["items"])
            expected_node_ids = sorted(str(node_id) for node_id in shard["node_ids"])
            selection_changes = {
                "missing": sorted(set(expected_node_ids) - set(actual_node_ids)),
                "unexpected": sorted(set(actual_node_ids) - set(expected_node_ids)),
            }
            if selection_changes["missing"] or selection_changes["unexpected"]:
                status = "selection_mismatch"
            elif plugin_report["collection_errors"]:
                status = "collection_error"
            elif capture.returncode != 0:
                status = "tests_failed"
            elif plugin_report.get("runtime_skip_count", 0) or plugin_report["collection_skips"]:
                status = "runtime_skips_detected"

    receipt = {
        "collection_errors": plugin_report.get("collection_errors", []) if plugin_report else [],
        "collection_skip_count": len(plugin_report.get("collection_skips", [])) if plugin_report else 0,
        "credential_keys_scrubbed": list(scrubbed),
        "expected_item_count": int(shard["item_count"]),
        "manifest_sha256": manifest_sha256,
        "operational_state_after": after,
        "operational_state_before": before,
        "operational_state_changes": changes,
        "process": _process_payload(capture),
        "pytest_report_error": plugin_report_error,
        "pytest_results": plugin_report,
        "runtime_skip_count": int(plugin_report.get("runtime_skip_count", 0)) if plugin_report else 0,
        "schema": SCHEMA,
        "selection_changes": selection_changes,
        "shard": shard_index,
        "state_mutation_detected": bool(changes),
        "status": status,
    }
    _write_new_json(receipt_path, receipt)
    return receipt


def verify_existing_manifest(
    *,
    root: Path,
    manifest_path: Path,
    plugin_inventory: Sequence[Mapping[str, str]] | None = None,
) -> dict[str, Any]:
    root = root.resolve()
    manifest_path = root / Path(_safe_relative_path(root, manifest_path.resolve()))
    manifest = _load_json(manifest_path, "pytest shard manifest")
    verification = verify_manifest(manifest)
    inventory = list(installed_pytest_plugin_inventory() if plugin_inventory is None else plugin_inventory)
    require_pytest_socket(inventory)
    if inventory != manifest["pytest_plugin_inventory"]:
        raise HarnessError("installed pytest plugin inventory differs from the collection manifest")
    current = fingerprint_operational_paths(root)
    changes = operational_changes(manifest["operational_state_baseline"], current)
    if changes:
        raise HarnessError("operational state differs from the collection baseline")
    blocked = _prior_mutation_receipts(manifest_path, verification["manifest_sha256"])
    if blocked:
        raise HarnessError(f"manifest is blocked by a state-mutation receipt: {blocked[0]}")
    return {
        **verification,
        "collection_skip_count": int(manifest["collection"]["collection_skip_count"]),
        "operational_state_matches": True,
        "shard_count": int(manifest["shard_count"]),
    }


def _plugin_state(config: Any) -> dict[str, Any]:
    state = getattr(config, "_aureon_shard_report_state", None)
    if state is None:
        state = {
            "collection_errors": [],
            "collection_skips": [],
            "items": [],
            "runtime_phases": [],
        }
        config._aureon_shard_report_state = state
    return state


def pytest_addoption(parser: Any) -> None:
    group = parser.getgroup("aureon-state-safe-shards")
    group.addoption(
        REPORT_OPTION,
        action="store",
        default=None,
        dest="aureon_shard_report",
        help="write Aureon pytest JSON report",
    )


def pytest_configure(config: Any) -> None:
    global _ACTIVE_PYTEST_CONFIG
    _ACTIVE_PYTEST_CONFIG = config
    _plugin_state(config)


def pytest_collectreport(report: Any) -> None:
    config = _ACTIVE_PYTEST_CONFIG
    if config is None:
        return
    state = _plugin_state(config)
    if report.failed:
        state["collection_errors"].append({"node_id": str(report.nodeid), "reason": str(report.longrepr)})
    elif report.skipped:
        state["collection_skips"].append({"node_id": str(report.nodeid), "reason": str(report.longrepr)})


def pytest_collection_finish(session: Any) -> None:
    root = Path(str(session.config.rootpath)).resolve()
    state = _plugin_state(session.config)
    items: list[dict[str, str]] = []
    for item in session.items:
        item_path = Path(str(item.path)).resolve()
        try:
            source_file = item_path.relative_to(root).as_posix()
        except ValueError:
            source_file = str(item_path)
        items.append({"node_id": str(item.nodeid), "source_file": source_file})
    state["items"] = items


def pytest_runtest_logreport(report: Any) -> None:
    config = _ACTIVE_PYTEST_CONFIG
    if config is None:
        return
    state = _plugin_state(config)
    state["runtime_phases"].append(
        {
            "duration_seconds": float(report.duration),
            "node_id": str(report.nodeid),
            "outcome": str(report.outcome),
            "was_xfail": str(getattr(report, "wasxfail", "")),
            "when": str(report.when),
        }
    )


def _normalise_loaded_plugin_path(root: Path, raw_path: str) -> str:
    path = Path(raw_path)
    if not path.is_absolute():
        return raw_path.replace("\\", "/")
    try:
        return path.resolve().relative_to(root).as_posix()
    except ValueError:
        return path.name


def _loaded_plugin_inventory(config: Any) -> list[dict[str, Any]]:
    root = Path(str(config.rootpath)).resolve()
    plugin_manager = config.pluginmanager
    distribution_by_plugin = {
        id(plugin): {
            "distribution": str(distribution.project_name or ""),
            "version": str(distribution.version or ""),
        }
        for plugin, distribution in plugin_manager.list_plugin_distinfo()
    }
    loaded: list[dict[str, Any]] = []
    for name, plugin in plugin_manager.list_name_plugin():
        if plugin is None:
            continue
        plugin_file = str(getattr(plugin, "__file__", "") or "")
        row: dict[str, Any] = {
            "module": str(getattr(plugin, "__name__", plugin.__class__.__module__)),
            "registered_as": _normalise_loaded_plugin_path(root, str(name)),
        }
        if plugin_file:
            row["source"] = _normalise_loaded_plugin_path(root, plugin_file)
        row.update(distribution_by_plugin.get(id(plugin), {}))
        loaded.append(row)
    return sorted(
        loaded,
        key=lambda row: (
            str(row.get("registered_as", "")),
            str(row.get("module", "")),
            str(row.get("distribution", "")),
        ),
    )


def pytest_sessionfinish(session: Any, exitstatus: int) -> None:
    config = session.config
    report_path_raw = config.getoption("aureon_shard_report")
    if not report_path_raw:
        return
    state = _plugin_state(config)
    loaded_plugins = _loaded_plugin_inventory(config)
    socket_loaded = any(
        _normalise_distribution(str(plugin.get("distribution") or "")) == "pytest-socket"
        for plugin in loaded_plugins
    )
    try:
        socket_requested = bool(config.getoption("disable_socket"))
    except (AttributeError, ValueError):
        socket_requested = False
    runtime_phases = state["runtime_phases"]
    runtime_skip_count = len({phase["node_id"] for phase in runtime_phases if phase["outcome"] == "skipped"})
    payload = {
        "collection_errors": state["collection_errors"],
        "collection_skips": state["collection_skips"],
        "items": state["items"],
        "loaded_plugins": loaded_plugins,
        "pytest_exitstatus": int(exitstatus),
        "runtime_phases": runtime_phases,
        "runtime_skip_count": runtime_skip_count,
        "schema": SCHEMA,
        "socket_blocker_loaded": socket_loaded,
        "socket_blocking_requested": socket_requested,
    }
    path = Path(str(report_path_raw))
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(
        json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True) + "\n", encoding="utf-8"
    )


def pytest_unconfigure(config: Any) -> None:
    global _ACTIVE_PYTEST_CONFIG
    if _ACTIVE_PYTEST_CONFIG is config:
        _ACTIVE_PYTEST_CONFIG = None


def _parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--repo-root",
        type=Path,
        default=Path(__file__).resolve().parents[2],
        help="repository root (default: inferred from this script)",
    )
    parser.add_argument("--manifest", type=Path, required=True, help="canonical manifest JSON path")
    parser.add_argument("--pytest-config", type=Path, default=Path("pyproject.toml"))
    parser.add_argument("--shard-count", type=int, default=8)
    parser.add_argument("--timeout-seconds", type=float, default=None)
    parser.add_argument("--collection-target", action="append", default=[])
    mode = parser.add_mutually_exclusive_group(required=True)
    mode.add_argument("--manifest-only", action="store_true")
    mode.add_argument("--one-shard", type=int, metavar="INDEX", help="run one 1-based shard")
    mode.add_argument("--verify-existing-manifest", action="store_true")
    return parser


def main(argv: Sequence[str] | None = None) -> int:
    args = _parser().parse_args(argv)
    root = args.repo_root.resolve()
    manifest_path = args.manifest if args.manifest.is_absolute() else root / args.manifest
    config_path = args.pytest_config if args.pytest_config.is_absolute() else root / args.pytest_config
    collection_targets = tuple(args.collection_target or ["tests"])
    try:
        if args.manifest_only:
            timeout = 900.0 if args.timeout_seconds is None else args.timeout_seconds
            manifest = collect_and_write_manifest(
                root=root,
                config_path=config_path,
                manifest_path=manifest_path,
                shard_count=args.shard_count,
                collection_targets=collection_targets,
                timeout_seconds=timeout,
            )
            result = {
                "collection_skip_count": manifest["collection"]["collection_skip_count"],
                "item_count": manifest["proofs"]["collected_item_count"],
                "manifest": str(manifest_path),
                "manifest_sha256": manifest["manifest_sha256"],
                "shard_count": manifest["shard_count"],
                "status": "manifest_created",
            }
            exit_code = 0
        elif args.one_shard is not None:
            timeout = 1800.0 if args.timeout_seconds is None else args.timeout_seconds
            receipt = execute_one_shard(
                root=root,
                manifest_path=manifest_path,
                shard_index=args.one_shard,
                timeout_seconds=timeout,
            )
            result = {
                "receipt": str(_shard_receipt_path(manifest_path, args.one_shard)),
                "runtime_skip_count": receipt["runtime_skip_count"],
                "shard": args.one_shard,
                "status": receipt["status"],
            }
            exit_code = 0 if receipt["status"] == "passed" else 1
            if receipt["state_mutation_detected"]:
                exit_code = 4
        else:
            verification = verify_existing_manifest(root=root, manifest_path=manifest_path)
            result = {**verification, "manifest": str(manifest_path), "status": "verified"}
            exit_code = 0
    except HarnessError as exc:
        print(json.dumps({"error": str(exc), "status": "blocked"}, sort_keys=True), file=sys.stderr)
        return 4 if "state" in str(exc).lower() or "mutation" in str(exc).lower() else 2
    print(json.dumps(result, sort_keys=True))
    return exit_code


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