ModelForge/scripts/m16_chaos.py

"""M16 bounded chaos harness.

One harness, one shape, every scenario:

    preflight -> baseline -> inject -> observe -> assert -> recover -> verify invariants -> cleanup

Faults are injected from *outside* the product, through the container runtime and through
seams that already exist for rehearsals (the M12 lifecycle pause, the M15 restore interruption).
ModelForge itself gains no chaos endpoint and no way to run a shell: a control plane that can be
told to break itself is a control plane an attacker can tell to break itself.

Every scenario is bounded in time, reverses its own fault, and re-runs the full invariant set
afterwards. A scenario that cannot restore the platform reports RECOVERY_FAILED rather than
leaving the environment dirty and calling itself done.

    python scripts/m16_chaos.py --list
    python scripts/m16_chaos.py --scenario redis-outage --seed 1234
    python scripts/m16_chaos.py --all --report out.json
"""

from __future__ import annotations

import argparse
import json
import os
import random
import subprocess  # noqa: S404 - fixed argv container-runtime control, never a shell
import sys
import time
import urllib.error
import urllib.request
import uuid
from collections.abc import Callable
from dataclasses import dataclass, field
from datetime import UTC, datetime
from pathlib import Path
from typing import Any

from sqlalchemy import create_engine, text
from sqlalchemy.orm import Session

sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "backend" / "src"))

from modelforge_api.services.invariants import (  # noqa: E402
    InvariantStatus,
    check_invariants,
)

DEFAULT_BASE_URL = "http://127.0.0.1:8000"
DEFAULT_DATABASE_URL = os.getenv("MODELFORGE_RUNTIME_DATABASE_URL")

API = "modelforge-api-1"
REDIS = "modelforge-redis-1"
POSTGRES = "modelforge-postgres-1"
NODE_AGENT = "modelforge-node-agent-1"
NETWORK = "modelforge_default"


class ChaosError(RuntimeError):
    """A harness failure: the fault could not be injected, or the platform could not be restored."""


# --------------------------------------------------------------------- runtime control


def docker(*args: str, check: bool = True, timeout: int = 120) -> str:
    completed = subprocess.run(  # noqa: S603 - fixed argv, no shell
        ["docker", *args],  # noqa: S607 - docker resolves from PATH
        capture_output=True,
        text=True,
        encoding="utf-8",
        errors="replace",
        timeout=timeout,
        check=False,
    )
    if check and completed.returncode != 0:
        raise ChaosError(f"docker {' '.join(args)} failed: {completed.stderr.strip()[:400]}")
    return (completed.stdout or "").strip()


def container_running(name: str) -> bool:
    return docker("inspect", "--format", "{{.State.Running}}", name, check=False) == "true"


def container_started_at(name: str) -> str:
    return docker("inspect", "--format", "{{.State.StartedAt}}", name, check=False)


# --------------------------------------------------------------------- probes


@dataclass
class Probe:
    ok: bool
    status: int
    latency_ms: float
    body: str = ""


def http(
    url: str, *, token: str | None = None, method: str = "GET", timeout: float = 5.0
) -> Probe:
    request = urllib.request.Request(url, method=method)  # noqa: S310 - fixed http scheme
    if token:
        request.add_header("X-ModelForge-Admin-Token", token)
    started = time.perf_counter()
    try:
        with urllib.request.urlopen(request, timeout=timeout) as response:  # noqa: S310
            body = response.read(4096).decode("utf-8", "replace")
            return Probe(True, response.status, (time.perf_counter() - started) * 1000, body)
    except urllib.error.HTTPError as error:
        return Probe(False, error.code, (time.perf_counter() - started) * 1000, "")
    except (urllib.error.URLError, TimeoutError, OSError):
        return Probe(False, 0, (time.perf_counter() - started) * 1000, "")


def wait_for(predicate: Callable[[], bool], *, timeout: float, interval: float = 1.0) -> bool:
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        if predicate():
            return True
        time.sleep(interval)
    return predicate()


# --------------------------------------------------------------------- scenario contract


@dataclass
class ScenarioContext:
    base_url: str
    database_url: str
    token: str
    seed: int
    rng: random.Random
    observations: dict[str, Any] = field(default_factory=dict)

    def note(self, key: str, value: Any) -> None:
        self.observations[key] = value

    def session(self) -> Session:
        engine = create_engine(self.database_url, pool_pre_ping=True)
        return Session(engine)


@dataclass
class Scenario:
    key: str
    title: str
    failure_class: str
    subsystem: str
    expected: str
    run: Callable[[ScenarioContext], None]
    recover: Callable[[ScenarioContext], None] | None = None
    cleanup: Callable[[ScenarioContext], None] | None = None
    requires_docker: bool = True


SCENARIOS: dict[str, Scenario] = {}


def scenario(
    key: str,
    title: str,
    failure_class: str,
    subsystem: str,
    expected: str,
    *,
    recover: Callable[[ScenarioContext], None] | None = None,
    cleanup: Callable[[ScenarioContext], None] | None = None,
    requires_docker: bool = True,
) -> Callable[[Callable[[ScenarioContext], None]], Callable[[ScenarioContext], None]]:
    def decorate(function: Callable[[ScenarioContext], None]) -> Callable[[ScenarioContext], None]:
        SCENARIOS[key] = Scenario(
            key=key,
            title=title,
            failure_class=failure_class,
            subsystem=subsystem,
            expected=expected,
            run=function,
            recover=recover,
            cleanup=cleanup,
            requires_docker=requires_docker,
        )
        return function

    return decorate


# --------------------------------------------------------------------- scenarios


def _api_healthy(context: ScenarioContext) -> bool:
    return http(f"{context.base_url}/api/v1/health/live").ok


def _restore_api(context: ScenarioContext) -> None:
    if not container_running(API):
        docker("start", API)
    if not wait_for(lambda: _api_healthy(context), timeout=120):
        raise ChaosError("the control plane did not become healthy again")


@scenario(
    "redis-outage",
    "Redis becomes unreachable and returns",
    failure_class="DEPENDENCY",
    subsystem="Redis",
    expected="the control plane stays up, reports the dependency honestly, and replays nothing",
    recover=_restore_api,
)
def redis_outage(context: ScenarioContext) -> None:
    """Redis holds queues and transient payloads only; losing it must not corrupt authoritative state."""

    before = http(f"{context.base_url}/api/v1/health/live")
    context.note("api_healthy_before", before.ok)

    docker("stop", REDIS)
    context.note("redis_stopped", True)
    time.sleep(5)

    during_live = http(f"{context.base_url}/api/v1/health/live")
    during_ready = http(f"{context.base_url}/api/v1/health/ready")
    during_models = http(f"{context.base_url}/api/v1/models")
    context.note("during_live_status", during_live.status)
    context.note("during_ready_status", during_ready.status)
    context.note("during_registry_status", during_models.status)

    with context.session() as session:
        queued = session.execute(
            text("select count(*) from serving_jobs where status in ('queued','leased','running')")
        ).scalar_one()
        context.note("serving_jobs_in_flight_during_outage", int(queued))

    docker("start", REDIS)
    if not wait_for(lambda: container_running(REDIS), timeout=60):
        raise ChaosError("redis did not restart")
    time.sleep(5)
    after = http(f"{context.base_url}/api/v1/health/live")
    context.note("api_healthy_after", after.ok)

    with context.session() as session:
        duplicates = session.execute(
            text(
                "select count(*) from (select idempotency_key from serving_jobs "
                "group by idempotency_key having count(*) > 1) s"
            )
        ).scalar_one()
        context.note("duplicate_serving_idempotency_keys", int(duplicates))
    if int(duplicates) != 0:
        raise ChaosError("a Redis outage produced duplicate serving jobs")
    if not after.ok:
        raise ChaosError("the control plane did not recover after Redis returned")


@scenario(
    "postgres-restart",
    "PostgreSQL restarts under the running control plane",
    failure_class="DEPENDENCY",
    subsystem="PostgreSQL",
    expected="the API fails safely while the database is gone and recovers its pool without duplicate work",
    recover=_restore_api,
)
def postgres_restart(context: ScenarioContext) -> None:
    """The control plane may not invent state while its authoritative store is unavailable."""

    with context.session() as session:
        audit_before = session.execute(text("select count(*) from audit_events")).scalar_one()
    context.note("audit_events_before", int(audit_before))

    docker("stop", POSTGRES)
    time.sleep(3)
    during_live = http(f"{context.base_url}/api/v1/health/live")
    during_models = http(f"{context.base_url}/api/v1/models", timeout=10.0)
    context.note("during_live_status", during_live.status)
    context.note("during_registry_status", during_models.status)
    if during_models.ok:
        raise ChaosError("the registry served data while PostgreSQL was stopped")

    docker("start", POSTGRES)
    if not wait_for(
        lambda: docker(
            "exec", POSTGRES, "pg_isready", "-U", "modelforge", "-d", "modelforge", check=False
        ).endswith("accepting connections"),
        timeout=120,
    ):
        raise ChaosError("PostgreSQL did not accept connections again")

    recovered = wait_for(lambda: http(f"{context.base_url}/api/v1/models", timeout=10.0).ok, timeout=120)
    context.note("registry_recovered_without_restart", recovered)

    with context.session() as session:
        audit_after = session.execute(text("select count(*) from audit_events")).scalar_one()
        duplicates = session.execute(
            text(
                "select count(*) from (select sequence from audit_events "
                "group by sequence having count(*) > 1) s"
            )
        ).scalar_one()
    context.note("audit_events_after", int(audit_after))
    context.note("duplicate_audit_sequences", int(duplicates))
    if int(duplicates) != 0:
        raise ChaosError("a PostgreSQL restart produced duplicate audit sequences")
    if not recovered:
        raise ChaosError("the control plane never recovered its database pool")


RECOVERY_AGENT = "modelforge-node-recovery-agent-1"
RECOVERY_COMPOSE = "docker-compose.node-recovery.yml"


def _recovery_agent_node(context: ScenarioContext) -> tuple[Any, Any] | None:
    with context.session() as session:
        row = session.execute(
            text(
                "select id, last_heartbeat_at from compute_nodes "
                "where hostname = 'modelforge-chaos-rehearsal' and enabled "
                "order by first_seen_at desc limit 1"
            )
        ).first()
    return (row[0], row[1]) if row else None


def _stop_recovery_agent(context: ScenarioContext) -> None:
    docker("network", "connect", NETWORK, RECOVERY_AGENT, check=False)
    subprocess.run(  # noqa: S603 - fixed argv, no shell
        [
            "docker",  # noqa: S607
            "compose",
            "-p",
            "modelforge",
            "-f",
            RECOVERY_COMPOSE,
            "down",
            "-v",
        ],
        capture_output=True,
        text=True,
        encoding="utf-8",
        errors="replace",
        timeout=180,
        check=False,
    )
    with context.session() as session:
        session.execute(
            text(
                "update compute_nodes set enabled = false, status = 'unavailable', "
                "status_reason = 'M16 disposable chaos node; environment torn down' "
                "where hostname = 'modelforge-chaos-rehearsal'"
            )
        )
        session.execute(
            text(
                "update node_credentials set revoked_at = now() where revoked_at is null "
                "and compute_node_id in "
                "(select id from compute_nodes where hostname = 'modelforge-chaos-rehearsal')"
            )
        )
        session.commit()


@scenario(
    "node-agent-disconnect",
    "A Node Agent loses its network and reconnects",
    failure_class="DEPENDENCY",
    subsystem="Node Agent",
    expected="heartbeat stops, liveness degrades honestly, and reconnect keeps one identity",
    cleanup=_stop_recovery_agent,
)
def node_agent_disconnect(context: ScenarioContext) -> None:
    """Losing a node must degrade liveness honestly rather than invent readiness.

    A disposable agent is used deliberately. GPU Node reaches the control plane over the LAN rather
    than the Compose network, so disconnecting a container would not interrupt it — and revoking or
    isolating the production node to prove a point is exactly what M16 forbids.
    """

    _stop_recovery_agent(context)

    token = _issue_enrollment(context)
    env = {
        "MODELFORGE_RECOVERY_AGENT_TOKEN": token,
        "MODELFORGE_RECOVERY_AGENT_URL": "http://api:8000",
        "MODELFORGE_RECOVERY_AGENT_HOSTNAME": "modelforge-chaos-rehearsal",
        "MODELFORGE_RECOVERY_AGENT_DISPLAY_NAME": "M16 chaos rehearsal node",
    }
    _compose_up(env)

    if not wait_for(lambda: _recovery_agent_node(context) is not None, timeout=120, interval=3.0):
        raise ChaosError("the disposable agent never enrolled")
    node = _recovery_agent_node(context)
    assert node is not None
    node_id, _ = node
    context.note("disposable_node_id", str(node_id))

    if not wait_for(
        lambda: _heartbeat_moved(context, node_id, None), timeout=90, interval=3.0
    ):
        raise ChaosError("the disposable agent never published a heartbeat")
    with context.session() as session:
        before = session.execute(
            text("select liveness_state, last_heartbeat_at from compute_nodes where id = :i"),
            {"i": node_id},
        ).first()
    context.note("liveness_before", before[0])
    heartbeat_before = before[1]

    docker("network", "disconnect", NETWORK, RECOVERY_AGENT)
    context.note("disconnected", True)
    try:
        # Longer than the configured offline threshold so liveness has to move, not just age.
        time.sleep(100)
        with context.session() as session:
            during = session.execute(
                text("select liveness_state, last_heartbeat_at from compute_nodes where id = :i"),
                {"i": node_id},
            ).first()
        context.note("liveness_during", during[0])
        advanced = during[1] != heartbeat_before
        context.note("heartbeat_advanced_during_outage", advanced)
        if advanced:
            raise ChaosError("the heartbeat advanced while the agent was disconnected")
        if during[0] == "online":
            raise ChaosError("liveness still reported online after the agent went away")
    finally:
        docker("network", "connect", NETWORK, RECOVERY_AGENT, check=False)

    reconnected = wait_for(
        lambda: _heartbeat_moved(context, node_id, heartbeat_before), timeout=150, interval=3.0
    )
    context.note("reconnected", reconnected)
    with context.session() as session:
        after = session.execute(
            text("select liveness_state from compute_nodes where id = :i"), {"i": node_id}
        ).scalar_one()
        identities = session.execute(
            text(
                "select count(*) from compute_nodes "
                "where hostname = 'modelforge-chaos-rehearsal' and enabled"
            )
        ).scalar_one()
    context.note("liveness_after", after)
    context.note("enabled_identities_after", int(identities))
    if int(identities) != 1:
        raise ChaosError("the disconnect produced a duplicate node identity")
    if not reconnected:
        raise ChaosError("the Node Agent never resumed publishing")


def _issue_enrollment(context: ScenarioContext) -> str:
    payload = json.dumps(
        {
            "expires_in_seconds": 3600,
            "display_name": "M16 chaos rehearsal node",
            "role": "rehearsal",
            "labels": {"rehearsal": "m16-chaos"},
            "production_eligible": False,
            "lab_eligible": False,
            "benchmark_eligible": False,
        }
    ).encode()
    request = urllib.request.Request(  # noqa: S310 - fixed http scheme
        f"{context.base_url}/api/v1/admin/node-enrollments", data=payload, method="POST"
    )
    request.add_header("Content-Type", "application/json")
    request.add_header("X-ModelForge-Admin-Token", context.token)
    with urllib.request.urlopen(request, timeout=60) as response:  # noqa: S310
        return str(json.loads(response.read().decode())["enrollment_token"])


def _compose_up(extra_env: dict[str, str]) -> None:
    import os

    environment = {**os.environ, **extra_env}
    completed = subprocess.run(  # noqa: S603 - fixed argv, no shell
        [
            "docker",  # noqa: S607
            "compose",
            "-p",
            "modelforge",
            "-f",
            RECOVERY_COMPOSE,
            "up",
            "-d",
            "--build",
            "node-recovery-agent",
        ],
        capture_output=True,
        text=True,
        encoding="utf-8",
        errors="replace",
        timeout=900,
        env=environment,
        check=False,
    )
    if completed.returncode != 0:
        raise ChaosError(f"the disposable agent did not start: {completed.stderr.strip()[:400]}")


def _heartbeat_moved(context: ScenarioContext, node_id: Any, before: Any) -> bool:
    with context.session() as session:
        current = session.execute(
            text("select last_heartbeat_at from compute_nodes where id = :i"), {"i": node_id}
        ).scalar()
    return current is not None and current != before


@scenario(
    "control-plane-restart-storm",
    "The control plane restarts repeatedly under load",
    failure_class="TRANSIENT",
    subsystem="Control Plane",
    expected="each start reconciles idempotently and no duplicate authoritative object appears",
    recover=_restore_api,
)
def control_plane_restart_storm(context: ScenarioContext) -> None:
    """Restart reconciliation must be idempotent, not merely survivable."""

    with context.session() as session:
        before = _authoritative_counts(session)
    context.note("counts_before", before)

    restarts = 4
    statuses: list[int] = []
    for index in range(restarts):
        docker("restart", API)
        healthy = wait_for(lambda: _api_healthy(context), timeout=120)
        if not healthy:
            raise ChaosError(f"the control plane did not return after restart {index + 1}")
        statuses.append(http(f"{context.base_url}/api/v1/models").status)
    context.note("restarts", restarts)
    context.note("registry_status_after_each_restart", statuses)

    with context.session() as session:
        after = _authoritative_counts(session)
    context.note("counts_after", after)

    drifted = {
        key: (before[key], after[key])
        for key in before
        if key not in {"audit_events"} and after[key] != before[key]
    }
    context.note("authoritative_drift", drifted)
    if drifted:
        raise ChaosError(f"restart storm changed authoritative state: {drifted}")


def _authoritative_counts(session: Session) -> dict[str, int]:
    tables = (
        "models",
        "model_revisions",
        "model_artifacts",
        "artifact_sets",
        "projects",
        "project_bindings",
        "capability_deployments",
        "lifecycle_operations",
        "migration_plans",
        "migration_cutover_operations",
        "compute_nodes",
        "node_credentials",
        "service_clients",
        "backup_sets",
        "audit_events",
    )
    counts: dict[str, int] = {}
    for table in tables:
        counts[table] = int(
            session.execute(text(f"select count(*) from {table}")).scalar_one()  # noqa: S608
        )
    return counts


@scenario(
    "invalid-auth-burst",
    "A burst of invalid operator credentials",
    failure_class="AUTH",
    subsystem="Control Plane",
    expected="every attempt is refused identically, with no amplification and no enumeration signal",
)
def invalid_auth_burst(context: ScenarioContext) -> None:
    """A wrong credential must not reveal whether it was close to right."""

    attempts = 60
    tokens = [uuid.UUID(int=context.rng.getrandbits(128)).hex for _ in range(attempts)]
    tokens[0] = ""
    tokens[1] = "x"
    tokens[2] = context.token[:-1] + ("a" if context.token[-1] != "a" else "b")
    statuses: list[int] = []
    bodies: set[str] = set()
    latencies: list[float] = []
    for token in tokens:
        probe = http(
            f"{context.base_url}/api/v1/admin/recovery/dashboard",
            token=token or None,
            timeout=10.0,
        )
        statuses.append(probe.status)
        bodies.add(probe.body[:0])
        latencies.append(probe.latency_ms)
    context.note("attempts", attempts)
    context.note("distinct_statuses", sorted(set(statuses)))
    context.note("max_latency_ms", round(max(latencies), 2))
    context.note("mean_latency_ms", round(sum(latencies) / len(latencies), 2))

    if set(statuses) != {401}:
        raise ChaosError(f"invalid credentials produced varying statuses: {sorted(set(statuses))}")

    good = http(f"{context.base_url}/api/v1/admin/recovery/dashboard", token=context.token)
    context.note("valid_credential_still_works", good.ok)
    if not good.ok:
        raise ChaosError("a burst of invalid credentials locked out the valid one")


# The backup root is a named volume, not a path inside the API container. Injecting this fault used
# to be `docker exec -u 0 modelforge-api-1 chmod ...`, which stopped working the moment M16 hardened
# the API container: with all capabilities dropped and no-new-privileges set, even uid 0 inside that
# container cannot change a mode. That is the hardening behaving correctly, so the fault moves
# outside the product instead of the hardening being weakened to keep a test convenient.
#
# A disposable Alpine container mounts the same volume and changes the mode there. It touches only
# the isolated backup volume, never a host path, and is removed on exit.
BACKUP_VOLUME = "modelforge_modelforge-backups"


def _set_backup_root_mode(mode: str, *, check: bool = True) -> None:
    docker(
        "run",
        "--rm",
        "-v",
        f"{BACKUP_VOLUME}:/data/backups",
        "alpine:3.20",
        "chmod",
        mode,
        "/data/backups",
        check=check,
    )


@scenario(
    "backup-destination-unavailable",
    "The backup destination cannot be written",
    failure_class="CAPACITY",
    subsystem="Backup storage",
    expected="the backup fails closed and serving is unaffected",
)
def backup_destination_unavailable(context: ScenarioContext) -> None:
    """A backup that cannot be written must fail, not silently produce a partial recovery point."""

    backup_id = f"chaos-unwritable-{context.seed:08d}"
    _set_backup_root_mode("0555")
    try:
        payload = json.dumps(
            {
                "backup_id": backup_id,
                "reason": "M16 chaos scenario: the backup root is not writable",
            }
        ).encode()
        request = urllib.request.Request(  # noqa: S310 - fixed http scheme
            f"{context.base_url}/api/v1/admin/recovery/backups", data=payload, method="POST"
        )
        request.add_header("Content-Type", "application/json")
        request.add_header("X-ModelForge-Admin-Token", context.token)
        status = 0
        try:
            with urllib.request.urlopen(request, timeout=120) as response:  # noqa: S310
                status = response.status
                body = json.loads(response.read().decode())
        except urllib.error.HTTPError as error:
            status = error.code
            body = {}
        context.note("backup_http_status", status)
        context.note("backup_state", body.get("state"))
        context.note("backup_failure_code", body.get("failure_code"))
    finally:
        _set_backup_root_mode("0755", check=False)

    serving = http(f"{context.base_url}/api/v1/health/ready")
    context.note("serving_unaffected", serving.ok)

    with context.session() as session:
        state = session.execute(
            text("select state from backup_sets where backup_id = :b"), {"b": backup_id}
        ).scalar()
    context.note("persisted_backup_state", state)
    if state == "VERIFIED":
        raise ChaosError("an unwritable destination still produced a verified backup")
    if not serving.ok:
        raise ChaosError("a failed backup degraded serving")


def _cleanup_chaos_backups(context: ScenarioContext) -> None:
    with context.session() as session:
        session.execute(
            text("delete from backup_manifest_entries where backup_set_id in "
                 "(select id from backup_sets where backup_id like 'chaos-%')")
        )
        session.execute(text("delete from backup_sets where backup_id like 'chaos-%'"))
        session.commit()


SCENARIOS["backup-destination-unavailable"].cleanup = _cleanup_chaos_backups


# --------------------------------------------------------------------- runner


def run_scenario(scenario_obj: Scenario, context: ScenarioContext) -> dict[str, Any]:
    started = datetime.now(UTC)
    record: dict[str, Any] = {
        "scenario": scenario_obj.key,
        "title": scenario_obj.title,
        "failure_class": scenario_obj.failure_class,
        "subsystem": scenario_obj.subsystem,
        "expected": scenario_obj.expected,
        "seed": context.seed,
        "started_at": started.isoformat(),
    }

    with context.session() as session:
        baseline = check_invariants(session)
    record["invariants_before"] = {
        "checked": baseline.checked,
        "violated": baseline.violated,
        "violations": [
            item.key for item in baseline.results if item.status is InvariantStatus.VIOLATED
        ],
    }
    if baseline.violated:
        record["outcome"] = "PREFLIGHT_FAILED"
        record["detail"] = "invariants were already violated before injection"
        record["finished_at"] = datetime.now(UTC).isoformat()
        return record

    outcome = "PASSED"
    detail = ""
    try:
        scenario_obj.run(context)
    except ChaosError as error:
        outcome = "FAILED"
        detail = str(error)
    except Exception as error:  # noqa: BLE001 - the harness must report, not crash the run
        outcome = "HARNESS_ERROR"
        detail = f"{type(error).__name__}: {error}"

    if scenario_obj.recover is not None:
        try:
            scenario_obj.recover(context)
        except ChaosError as error:
            outcome = "RECOVERY_FAILED"
            detail = f"{detail} | recovery: {error}".strip(" |")

    if scenario_obj.cleanup is not None:
        try:
            scenario_obj.cleanup(context)
            record["cleanup"] = "DONE"
        except Exception as error:  # noqa: BLE001
            record["cleanup"] = f"FAILED: {type(error).__name__}: {error}"
            outcome = "CLEANUP_FAILED" if outcome == "PASSED" else outcome
    else:
        record["cleanup"] = "NOT_REQUIRED"

    with context.session() as session:
        after = check_invariants(session)
    record["invariants_after"] = {
        "checked": after.checked,
        "violated": after.violated,
        "violations": [
            {"key": item.key, "examples": item.violations}
            for item in after.results
            if item.status is InvariantStatus.VIOLATED
        ],
    }
    if after.violated and outcome == "PASSED":
        outcome = "INVARIANT_VIOLATED"
        detail = "the platform survived the fault but broke an invariant"

    record["observations"] = context.observations
    record["outcome"] = outcome
    record["detail"] = detail
    record["finished_at"] = datetime.now(UTC).isoformat()
    record["duration_seconds"] = round(
        (datetime.fromisoformat(record["finished_at"]) - started).total_seconds(), 3
    )
    return record


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--base-url", default=DEFAULT_BASE_URL)
    parser.add_argument("--database-url", default=DEFAULT_DATABASE_URL)
    parser.add_argument("--token", default=None, help="operator API key")
    parser.add_argument("--scenario", action="append", default=[])
    parser.add_argument("--all", action="store_true")
    parser.add_argument("--list", action="store_true")
    parser.add_argument("--seed", type=int, default=None)
    parser.add_argument("--report", default=None)
    args = parser.parse_args(argv)

    if args.list:
        for key, item in sorted(SCENARIOS.items()):
            print(f"{key:32} {item.failure_class:12} {item.subsystem:16} {item.title}")
        return 0
    if not args.database_url:
        parser.error(
            "--database-url or MODELFORGE_RUNTIME_DATABASE_URL is required "
            "(use the non-owner runtime role)"
        )

    keys = list(SCENARIOS) if args.all else args.scenario
    if not keys:
        parser.error("choose --scenario, --all or --list")
    unknown = [key for key in keys if key not in SCENARIOS]
    if unknown:
        parser.error(f"unknown scenarios: {unknown}")

    token = args.token
    if not token:
        env = Path(__file__).resolve().parents[1] / ".env"
        if env.is_file():
            for line in env.read_text("utf-8").splitlines():
                if line.startswith("MODELFORGE_OPERATOR_API_KEY="):
                    token = line.split("=", 1)[1].strip()
    if not token:
        parser.error("an operator API key is required")

    seed = args.seed if args.seed is not None else random.randrange(1, 10**8)
    records = []
    for key in keys:
        context = ScenarioContext(
            base_url=args.base_url,
            database_url=args.database_url,
            token=token,
            seed=seed,
            rng=random.Random(seed),
        )
        print(f"--- {key} (seed {seed}) ---", flush=True)
        record = run_scenario(SCENARIOS[key], context)
        records.append(record)
        print(f"    {record['outcome']}  {record.get('detail', '')}", flush=True)
        for name, value in record["observations"].items():
            print(f"      {name}: {value}", flush=True)

    report = {
        "generated_at": datetime.now(UTC).isoformat(),
        "seed": seed,
        "scenarios": records,
        "passed": sum(item["outcome"] == "PASSED" for item in records),
        "total": len(records),
    }
    if args.report:
        Path(args.report).write_text(json.dumps(report, indent=2), encoding="utf-8")
        print(f"report written to {args.report}")
    return 0 if report["passed"] == report["total"] else 1


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