diff --git a/README.md b/README.md index 648b8fe..184cb67 100644 --- a/README.md +++ b/README.md @@ -142,8 +142,14 @@ uv run missioncore-sim s0 doctor \ --json ``` -This accepts simulation infrastructure, not navigation behavior, safety -behavior, S1 command authority or real actuator control. +S1A now adds the first server-owned domain boundary: typed immutable +`QualificationRun` identities, legal lifecycle transitions, canonical +Ackermann/Differential commands, authority generation and TTL validation, and +an append-only crash-safe run/event/command/artifact repository. Active runs +recover as failed/interrupted rather than silently resuming, and reset creates +a new linked run/episode. This is repository-level lifecycle acceptance only; +PX4 command delivery, watchdog/failsafe behavior, navigation behavior and real +actuator control remain unaccepted. See the [Polygon product/SRS](docs/12_SIMULATION_POLYGON_PRODUCT_AND_SRS.md), [ADR 0015](docs/adr/0015-simulation-polygon-qualification-boundary.md) and the diff --git a/docs/01_IMPLEMENTATION_PLAN.md b/docs/01_IMPLEMENTATION_PLAN.md index 522b283..8fc5019 100644 --- a/docs/01_IMPLEMENTATION_PLAN.md +++ b/docs/01_IMPLEMENTATION_PLAN.md @@ -24,7 +24,7 @@ Each gate produces evidence and an explicit GO, PAUSE or BLOCKED result. | Plugin isolation | GO (laboratory control plane) — vendor backend/frontend and optional scene controls are plugin-owned; manifest/runtime descriptor parity, versioned handshake, lifecycle health and transport correlation fail closed while execution remains in-process | | K1 application control | GO (physical staged cycle) — after fixing the PCAP-proven `sint64` time field, one explicit UI launch completed all 14 canonical operations on one control session, reached live `SCANNING + project + init_ready`, displayed real points, then one explicit STOP returned K1 to unbound `READY`. No retry or fallback command was sent. Native-project reuse through LixelGO/USB remains an independent verification | | Stage 8 product storage | PAUSE — retention, replication, encryption, capacity monitoring and long-run browser/WASM stress remain deployment gates | -| Simulation Polygon | SIM S0 GO — dedicated Docker-free D-only worker, exact accepted pins, stock Ackermann PX4/ROS 2/Gazebo telemetry, authoritative clock, pause/2 ms step, 1×/2× RTF/resource baselines, repeated clean lifecycle and all nine digest-bound evidence claims passed target doctor. S1 orchestrator/control, navigation/safety acceptance and real actuator authority remain absent | +| Simulation Polygon | SIM S0 GO; S1A repository boundary implemented — dedicated Docker-free D-only worker and exact stack/clock/resource evidence are accepted. Typed immutable run identity, lifecycle journal, canonical rover command envelopes, authority/TTL checks, artifact index, reset identity and fail-closed restart recovery are implemented locally. Provider process orchestration, PX4 command delivery, watchdog/failsafe evidence, navigation/safety acceptance and real actuator authority remain absent | USB project copying remains optional ground truth rather than a blocker for the now-verified network path. Owner-operated LixelGO traffic verifies the MQTT @@ -150,8 +150,11 @@ uv run missioncore-sim s0 doctor --json Without the exact admitted target-worker evidence its correct result remains `INCOMPLETE`. S1 starts from the accepted S0 version/storage/time/process -boundary and must add server-owned lifecycle, canonical rover commands, -TTL/watchdog/failsafe evidence and immutable qualification runs. +boundary. Its S1A increment implements server-owned immutable run identity, +legal lifecycle transitions, canonical rover command envelopes, authority/TTL +admission, append-only artifacts and restart reconciliation. Process ownership, +PX4 delivery, heartbeat/watchdog execution and captured failsafe outcomes remain +the next S1 increments. ## Stage 0 — repository and host baseline diff --git a/docs/12_SIMULATION_POLYGON_PRODUCT_AND_SRS.md b/docs/12_SIMULATION_POLYGON_PRODUCT_AND_SRS.md index d2253f0..a8a80b0 100644 --- a/docs/12_SIMULATION_POLYGON_PRODUCT_AND_SRS.md +++ b/docs/12_SIMULATION_POLYGON_PRODUCT_AND_SRS.md @@ -503,6 +503,37 @@ match the exact candidate generation. K1, Nav2 and top-level UI do not block S1 backend acceptance. +#### S1A implementation status + +The first S1 increment is implemented in `k1link.simulation.contracts` and +`k1link.simulation.run_store`: + +- the immutable manifest owns run/episode identity, exact scenario/profile + digests, Mission Core revision, provider pins, host profile, seed, + reproducibility tier, clock domain and fail-closed authority; +- lifecycle and domain events are stored as exclusive, individually atomic, + contiguous records; state is reconstructed by replay and the immutable + manifest is never rewritten; +- only legal lifecycle edges are admitted; terminal history and its artifact + index are sealed; +- Ackermann speed/steering and Differential speed/yaw-rate commands require a + positive monotonic sequence, simulation-time expiry and the current authority + generation; non-finite values, over-limit TTL, stale generations, direct + actuator authority and commands outside `running` fail closed; +- replay/shadow cannot causally accept commands, while HIL and + controller-in-loop admission remain blocked pending a separate safety gate; +- restart reconciliation marks an active run `failed` with + `orchestrator-recovery-interrupted`; it never hides a resume; +- reset is accepted only after the prior run is terminal and creates a new + linked run and episode identity; +- registered artifacts use confined relative POSIX paths, SHA-256 and + append-only sequence records. + +The repository root is caller-supplied so unit tests and development remain +portable. The target Simulation Orchestrator must bind that root under the +accepted `/mnt/d/NDC_MISSIONCORE` storage boundary; S1 provider orchestration +and target evidence are not implied by this increment. + ### S2 - Gazebo LiDAR/odometry feed the ROS 2/Nav2 baseline. diff --git a/docs/adr/0015-simulation-polygon-qualification-boundary.md b/docs/adr/0015-simulation-polygon-qualification-boundary.md index a978f9e..3911dca 100644 --- a/docs/adr/0015-simulation-polygon-qualification-boundary.md +++ b/docs/adr/0015-simulation-polygon-qualification-boundary.md @@ -100,3 +100,28 @@ The doctor reports `INCOMPLETE` until it is running on the reviewed target with resolved/accepted pins and all required evidence. On the accepted profile generation and nine-claim confined evidence set it returned `GO`; it cannot infer that result from repository state alone. + +## S1A implementation + +`k1link.simulation.contracts` now defines the first product-owned +`QualificationRun`, provider/authority/artifact provenance, canonical +Ackermann and Differential rover setpoints, and qualification events. +`k1link.simulation.run_store` persists them as an immutable manifest plus +one-file-per-sequence append-only journals. Exclusive creation and file/directory +fsync prevent partially acknowledged records; loading rejects gaps, identity +drift, non-regular records and illegal lifecycle history. + +The aggregate derives its current state by replay. Terminal runs cannot accept +new events, commands or artifacts. A restart never silently resumes an active +run: reconciliation terminates it as `failed` with +`orchestrator-recovery-interrupted`. Reset requires a terminal parent and +creates a new linked run/episode. Commands are admitted only for a running +closed-loop virtual run with the current authority generation and bounded +simulation-time TTL. Replay/shadow commands, direct actuator authority and +ungated HIL/controller-in-loop admission fail closed. + +This increment proves the local lifecycle/repository contract. It does not yet +own provider processes, transmit a PX4 setpoint, execute heartbeat/watchdog +failsafe behavior, or qualify navigation/safety behavior. The target +orchestrator must place the caller-supplied repository root under the accepted +D-only worker boundary. diff --git a/src/k1link/simulation/__init__.py b/src/k1link/simulation/__init__.py index 68722a9..f6ebb62 100644 --- a/src/k1link/simulation/__init__.py +++ b/src/k1link/simulation/__init__.py @@ -1,5 +1,28 @@ """Mission Core qualification and simulation boundaries.""" +from k1link.simulation.contracts import ( + AckermannControlSetpoint, + AuthorityProfile, + ControlProfile, + ControlSetpoint, + DifferentialControlSetpoint, + ProviderPin, + QualificationArtifact, + QualificationEvent, + QualificationRun, + ReproducibilityTier, + RunKind, + RunState, + SimulationContractError, +) +from k1link.simulation.run_store import ( + QualificationRunConflictError, + QualificationRunIntegrityError, + QualificationRunNotFoundError, + QualificationRunStore, + QualificationRunStoreError, + QualificationRunTransitionError, +) from k1link.simulation.s0 import ( CheckStatus, DoctorVerdict, @@ -12,12 +35,31 @@ from k1link.simulation.s0 import ( ) __all__ = [ + "AckermannControlSetpoint", + "AuthorityProfile", "CheckStatus", + "ControlProfile", + "ControlSetpoint", + "DifferentialControlSetpoint", "DoctorVerdict", + "ProviderPin", + "QualificationArtifact", + "QualificationEvent", + "QualificationRun", + "QualificationRunConflictError", + "QualificationRunIntegrityError", + "QualificationRunNotFoundError", + "QualificationRunStore", + "QualificationRunStoreError", + "QualificationRunTransitionError", + "ReproducibilityTier", "RuntimeAcceptance", + "RunKind", + "RunState", "S0DoctorReport", "S0Profile", "S0ProfileError", + "SimulationContractError", "load_s0_profile", "run_s0_doctor", ] diff --git a/src/k1link/simulation/contracts.py b/src/k1link/simulation/contracts.py new file mode 100644 index 0000000..4b7a84a --- /dev/null +++ b/src/k1link/simulation/contracts.py @@ -0,0 +1,660 @@ +from __future__ import annotations + +import math +import re +from dataclasses import dataclass, replace +from datetime import UTC, datetime +from enum import StrEnum +from pathlib import PurePosixPath +from typing import Any, Final + +IDENTIFIER_PATTERN: Final = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$") +SHA256_PATTERN: Final = re.compile(r"^[a-f0-9]{64}$") +GIT_REVISION_PATTERN: Final = re.compile(r"^[a-f0-9]{7,64}$") + + +class SimulationContractError(ValueError): + """A Polygon domain object violates the admitted S1 contract.""" + + +class RunKind(StrEnum): + SIMULATION_CLOSED_LOOP = "simulation_closed_loop" + REPLAY_SHADOW = "replay_shadow" + DIGITAL_TWIN_CLOSED_LOOP = "digital_twin_closed_loop" + CONTROLLER_IN_LOOP = "controller_in_loop" + HIL = "hil" + PHYSICAL_SHADOW = "physical_shadow" + + +class RunState(StrEnum): + ADMITTED = "admitted" + STARTING = "starting" + RUNNING = "running" + PAUSED = "paused" + STOPPING = "stopping" + COMPLETED = "completed" + FAILED = "failed" + ABORTED = "aborted" + + @property + def terminal(self) -> bool: + return self in {RunState.COMPLETED, RunState.FAILED, RunState.ABORTED} + + +class ReproducibilityTier(StrEnum): + R0 = "R0" + R1 = "R1" + R2 = "R2" + + +class ControlProfile(StrEnum): + ROVER_SPEED_STEERING_V1 = "rover-speed-steering/v1" + ROVER_SPEED_YAW_RATE_V1 = "rover-speed-yaw-rate/v1" + + +@dataclass(frozen=True, slots=True) +class ProviderPin: + identifier: str + version: str + revision: str + digest: str | None = None + + def __post_init__(self) -> None: + _identifier(self.identifier, "provider identifier") + if not self.version.strip(): + raise SimulationContractError("provider version must not be empty") + if not self.revision.strip(): + raise SimulationContractError("provider revision must not be empty") + if self.digest is not None: + _sha256(self.digest, "provider digest") + + def to_dict(self) -> dict[str, object]: + return { + "identifier": self.identifier, + "version": self.version, + "revision": self.revision, + "digest": self.digest, + } + + @classmethod + def from_dict(cls, value: object) -> ProviderPin: + document = _object(value, "provider pin") + return cls( + identifier=_string(document, "identifier"), + version=_string(document, "version"), + revision=_string(document, "revision"), + digest=_optional_string(document, "digest"), + ) + + +@dataclass(frozen=True, slots=True) +class AuthorityProfile: + """The only command authority admitted before a physical safety gate.""" + + generation: int + command_ttl_max_ns: int + heartbeat_timeout_monotonic_ns: int + simulation_or_shadow_only: bool = True + actuator_authority: bool = False + navigation_or_safety_accepted: bool = False + direct_actuator_setpoints_allowed: bool = False + + def __post_init__(self) -> None: + if self.generation < 1: + raise SimulationContractError("authority generation must be positive") + if self.command_ttl_max_ns < 1: + raise SimulationContractError("command TTL limit must be positive") + if self.heartbeat_timeout_monotonic_ns < 1: + raise SimulationContractError("heartbeat timeout must be positive") + if ( + not self.simulation_or_shadow_only + or self.actuator_authority + or self.navigation_or_safety_accepted + or self.direct_actuator_setpoints_allowed + ): + raise SimulationContractError( + "S1 authority must remain simulation/shadow-only without actuator authority" + ) + + def to_dict(self) -> dict[str, object]: + return { + "generation": self.generation, + "command_ttl_max_ns": self.command_ttl_max_ns, + "heartbeat_timeout_monotonic_ns": self.heartbeat_timeout_monotonic_ns, + "simulation_or_shadow_only": self.simulation_or_shadow_only, + "actuator_authority": self.actuator_authority, + "navigation_or_safety_accepted": self.navigation_or_safety_accepted, + "direct_actuator_setpoints_allowed": self.direct_actuator_setpoints_allowed, + } + + @classmethod + def from_dict(cls, value: object) -> AuthorityProfile: + document = _object(value, "authority profile") + return cls( + generation=_integer(document, "generation"), + command_ttl_max_ns=_integer(document, "command_ttl_max_ns"), + heartbeat_timeout_monotonic_ns=_integer( + document, + "heartbeat_timeout_monotonic_ns", + ), + simulation_or_shadow_only=_boolean(document, "simulation_or_shadow_only"), + actuator_authority=_boolean(document, "actuator_authority"), + navigation_or_safety_accepted=_boolean( + document, + "navigation_or_safety_accepted", + ), + direct_actuator_setpoints_allowed=_boolean( + document, + "direct_actuator_setpoints_allowed", + ), + ) + + +@dataclass(frozen=True, slots=True) +class QualificationArtifact: + artifact_id: str + kind: str + relative_path: str + sha256: str + byte_length: int + source_of_record: bool + sequence: int = 0 + + def __post_init__(self) -> None: + _identifier(self.artifact_id, "artifact id") + _identifier(self.kind, "artifact kind") + _relative_path(self.relative_path) + _sha256(self.sha256, "artifact digest") + if self.byte_length < 0: + raise SimulationContractError("artifact byte length must not be negative") + if self.sequence < 0: + raise SimulationContractError("artifact sequence must not be negative") + + def to_dict(self) -> dict[str, object]: + return { + "artifact_id": self.artifact_id, + "kind": self.kind, + "relative_path": self.relative_path, + "sha256": self.sha256, + "byte_length": self.byte_length, + "source_of_record": self.source_of_record, + "sequence": self.sequence, + } + + @classmethod + def from_dict(cls, value: object) -> QualificationArtifact: + document = _object(value, "qualification artifact") + return cls( + artifact_id=_string(document, "artifact_id"), + kind=_string(document, "kind"), + relative_path=_string(document, "relative_path"), + sha256=_string(document, "sha256"), + byte_length=_integer(document, "byte_length"), + source_of_record=_boolean(document, "source_of_record"), + sequence=_integer(document, "sequence"), + ) + + +@dataclass(frozen=True, slots=True) +class QualificationRun: + run_id: str + episode_id: str + kind: RunKind + state: RunState + scenario_generation: str + scenario_sha256: str + profile_generation: str + profile_sha256: str + mission_core_commit: str + providers: tuple[ProviderPin, ...] + host_profile_id: str + host_profile_sha256: str + seed: int + reproducibility_tier: ReproducibilityTier + authority: AuthorityProfile + clock_domain: str + created_at_utc: str + started_at_utc: str | None = None + ended_at_utc: str | None = None + terminal_reason: str | None = None + parent_run_id: str | None = None + revision: int = 0 + artifacts: tuple[QualificationArtifact, ...] = () + + def __post_init__(self) -> None: + _identifier(self.run_id, "run id") + _identifier(self.episode_id, "episode id") + _identifier(self.scenario_generation, "scenario generation") + _sha256(self.scenario_sha256, "scenario digest") + _identifier(self.profile_generation, "profile generation") + _sha256(self.profile_sha256, "profile digest") + if not GIT_REVISION_PATTERN.fullmatch(self.mission_core_commit): + raise SimulationContractError("Mission Core commit must be a hexadecimal revision") + if not self.providers: + raise SimulationContractError("at least one provider pin is required") + provider_ids = [provider.identifier for provider in self.providers] + if len(provider_ids) != len(set(provider_ids)): + raise SimulationContractError("provider identifiers must be unique") + _identifier(self.host_profile_id, "host profile id") + _sha256(self.host_profile_sha256, "host profile digest") + if self.seed < 0: + raise SimulationContractError("run seed must not be negative") + if self.clock_domain != "gazebo:/clock" and self.kind in { + RunKind.SIMULATION_CLOSED_LOOP, + RunKind.DIGITAL_TWIN_CLOSED_LOOP, + }: + raise SimulationContractError("closed-loop simulation requires Gazebo /clock") + _utc_timestamp(self.created_at_utc, "created timestamp") + if self.started_at_utc is not None: + _utc_timestamp(self.started_at_utc, "started timestamp") + if self.ended_at_utc is not None: + _utc_timestamp(self.ended_at_utc, "ended timestamp") + if self.parent_run_id is not None: + _identifier(self.parent_run_id, "parent run id") + if self.parent_run_id == self.run_id: + raise SimulationContractError("a run cannot be its own parent") + if self.revision < 0: + raise SimulationContractError("run revision must not be negative") + if self.state.terminal != (self.ended_at_utc is not None): + raise SimulationContractError("terminal state and ended timestamp must agree") + if self.state.terminal != (self.terminal_reason is not None): + raise SimulationContractError("terminal state and terminal reason must agree") + if self.kind in {RunKind.CONTROLLER_IN_LOOP, RunKind.HIL}: + raise SimulationContractError( + f"{self.kind.value} requires a separate physical authority safety gate" + ) + + def with_runtime( + self, + *, + state: RunState, + revision: int, + started_at_utc: str | None, + ended_at_utc: str | None, + terminal_reason: str | None, + artifacts: tuple[QualificationArtifact, ...], + ) -> QualificationRun: + return replace( + self, + state=state, + revision=revision, + started_at_utc=started_at_utc, + ended_at_utc=ended_at_utc, + terminal_reason=terminal_reason, + artifacts=artifacts, + ) + + def to_manifest_dict(self) -> dict[str, object]: + return { + "schema_version": "missioncore.qualification-run/v1", + "run_id": self.run_id, + "episode_id": self.episode_id, + "kind": self.kind.value, + "initial_state": RunState.ADMITTED.value, + "scenario": { + "generation": self.scenario_generation, + "sha256": self.scenario_sha256, + }, + "profile": { + "generation": self.profile_generation, + "sha256": self.profile_sha256, + }, + "mission_core_commit": self.mission_core_commit, + "providers": [provider.to_dict() for provider in self.providers], + "host_profile": { + "id": self.host_profile_id, + "sha256": self.host_profile_sha256, + }, + "seed": self.seed, + "reproducibility_tier": self.reproducibility_tier.value, + "authority": self.authority.to_dict(), + "clock_domain": self.clock_domain, + "created_at_utc": self.created_at_utc, + "parent_run_id": self.parent_run_id, + } + + @classmethod + def from_manifest_dict(cls, value: object) -> QualificationRun: + document = _object(value, "qualification run") + if document.get("schema_version") != "missioncore.qualification-run/v1": + raise SimulationContractError("unsupported qualification-run schema") + if document.get("initial_state") != RunState.ADMITTED.value: + raise SimulationContractError("qualification run must be admitted immutably") + scenario = _object(document.get("scenario"), "scenario") + profile = _object(document.get("profile"), "profile") + host_profile = _object(document.get("host_profile"), "host profile") + providers = document.get("providers") + if not isinstance(providers, list): + raise SimulationContractError("providers must be an array") + try: + kind = RunKind(_string(document, "kind")) + tier = ReproducibilityTier(_string(document, "reproducibility_tier")) + except ValueError as exc: + raise SimulationContractError("qualification run contains an unknown enum") from exc + return cls( + run_id=_string(document, "run_id"), + episode_id=_string(document, "episode_id"), + kind=kind, + state=RunState.ADMITTED, + scenario_generation=_string(scenario, "generation"), + scenario_sha256=_string(scenario, "sha256"), + profile_generation=_string(profile, "generation"), + profile_sha256=_string(profile, "sha256"), + mission_core_commit=_string(document, "mission_core_commit"), + providers=tuple(ProviderPin.from_dict(provider) for provider in providers), + host_profile_id=_string(host_profile, "id"), + host_profile_sha256=_string(host_profile, "sha256"), + seed=_integer(document, "seed"), + reproducibility_tier=tier, + authority=AuthorityProfile.from_dict(document.get("authority")), + clock_domain=_string(document, "clock_domain"), + created_at_utc=_string(document, "created_at_utc"), + parent_run_id=_optional_string(document, "parent_run_id"), + ) + + +@dataclass(frozen=True, slots=True) +class QualificationEvent: + run_id: str + sequence: int + event_type: str + observed_at_utc: str + host_monotonic_ns: int + sim_time_ns: int | None + payload: dict[str, Any] + + def __post_init__(self) -> None: + _identifier(self.run_id, "event run id") + _identifier(self.event_type, "event type") + if self.sequence < 1: + raise SimulationContractError("event sequence must be positive") + _utc_timestamp(self.observed_at_utc, "event timestamp") + if self.host_monotonic_ns < 0: + raise SimulationContractError("host monotonic timestamp must not be negative") + if self.sim_time_ns is not None and self.sim_time_ns < 0: + raise SimulationContractError("simulation timestamp must not be negative") + _json_object(self.payload, "event payload") + + def to_dict(self) -> dict[str, object]: + return { + "schema_version": "missioncore.qualification-event/v1", + "run_id": self.run_id, + "sequence": self.sequence, + "event_type": self.event_type, + "observed_at_utc": self.observed_at_utc, + "host_monotonic_ns": self.host_monotonic_ns, + "sim_time_ns": self.sim_time_ns, + "payload": self.payload, + } + + @classmethod + def from_dict(cls, value: object) -> QualificationEvent: + document = _object(value, "qualification event") + if document.get("schema_version") != "missioncore.qualification-event/v1": + raise SimulationContractError("unsupported qualification-event schema") + sim_time = document.get("sim_time_ns") + if sim_time is not None and (not isinstance(sim_time, int) or isinstance(sim_time, bool)): + raise SimulationContractError("sim_time_ns must be an integer or null") + payload = _object(document.get("payload"), "event payload") + return cls( + run_id=_string(document, "run_id"), + sequence=_integer(document, "sequence"), + event_type=_string(document, "event_type"), + observed_at_utc=_string(document, "observed_at_utc"), + host_monotonic_ns=_integer(document, "host_monotonic_ns"), + sim_time_ns=sim_time, + payload=payload, + ) + + +@dataclass(frozen=True, slots=True) +class AckermannControlSetpoint: + run_id: str + command_id: str + sequence: int + issued_at_sim_ns: int + valid_until_sim_ns: int + authority_generation: int + speed_mps: float + steering_normalized: float + profile: ControlProfile = ControlProfile.ROVER_SPEED_STEERING_V1 + + def __post_init__(self) -> None: + _command_header( + self.run_id, + self.command_id, + self.sequence, + self.issued_at_sim_ns, + self.valid_until_sim_ns, + self.authority_generation, + ) + _finite(self.speed_mps, "speed") + _finite(self.steering_normalized, "steering") + if not -1.0 <= self.steering_normalized <= 1.0: + raise SimulationContractError("normalized steering must be within -1..1") + if self.profile is not ControlProfile.ROVER_SPEED_STEERING_V1: + raise SimulationContractError("Ackermann command has an incompatible profile") + + def to_dict(self) -> dict[str, object]: + return { + "schema_version": "missioncore.control-setpoint/v1", + "run_id": self.run_id, + "command_id": self.command_id, + "sequence": self.sequence, + "profile": self.profile.value, + "issued_at_sim_ns": self.issued_at_sim_ns, + "valid_until_sim_ns": self.valid_until_sim_ns, + "authority_generation": self.authority_generation, + "speed_mps": self.speed_mps, + "steering_normalized": self.steering_normalized, + } + + +@dataclass(frozen=True, slots=True) +class DifferentialControlSetpoint: + run_id: str + command_id: str + sequence: int + issued_at_sim_ns: int + valid_until_sim_ns: int + authority_generation: int + speed_mps: float + yaw_rate_rps: float + profile: ControlProfile = ControlProfile.ROVER_SPEED_YAW_RATE_V1 + + def __post_init__(self) -> None: + _command_header( + self.run_id, + self.command_id, + self.sequence, + self.issued_at_sim_ns, + self.valid_until_sim_ns, + self.authority_generation, + ) + _finite(self.speed_mps, "speed") + _finite(self.yaw_rate_rps, "yaw rate") + if self.profile is not ControlProfile.ROVER_SPEED_YAW_RATE_V1: + raise SimulationContractError("differential command has an incompatible profile") + + def to_dict(self) -> dict[str, object]: + return { + "schema_version": "missioncore.control-setpoint/v1", + "run_id": self.run_id, + "command_id": self.command_id, + "sequence": self.sequence, + "profile": self.profile.value, + "issued_at_sim_ns": self.issued_at_sim_ns, + "valid_until_sim_ns": self.valid_until_sim_ns, + "authority_generation": self.authority_generation, + "speed_mps": self.speed_mps, + "yaw_rate_rps": self.yaw_rate_rps, + } + + +ControlSetpoint = AckermannControlSetpoint | DifferentialControlSetpoint + + +def control_setpoint_from_dict(value: object) -> ControlSetpoint: + document = _object(value, "control setpoint") + if document.get("schema_version") != "missioncore.control-setpoint/v1": + raise SimulationContractError("unsupported control-setpoint schema") + try: + profile = ControlProfile(_string(document, "profile")) + except ValueError as exc: + raise SimulationContractError("unknown control profile") from exc + run_id = _string(document, "run_id") + command_id = _string(document, "command_id") + sequence = _integer(document, "sequence") + issued_at_sim_ns = _integer(document, "issued_at_sim_ns") + valid_until_sim_ns = _integer(document, "valid_until_sim_ns") + authority_generation = _integer(document, "authority_generation") + speed_mps = _number(document, "speed_mps") + if profile is ControlProfile.ROVER_SPEED_STEERING_V1: + return AckermannControlSetpoint( + run_id=run_id, + command_id=command_id, + sequence=sequence, + issued_at_sim_ns=issued_at_sim_ns, + valid_until_sim_ns=valid_until_sim_ns, + authority_generation=authority_generation, + speed_mps=speed_mps, + steering_normalized=_number(document, "steering_normalized"), + ) + return DifferentialControlSetpoint( + run_id=run_id, + command_id=command_id, + sequence=sequence, + issued_at_sim_ns=issued_at_sim_ns, + valid_until_sim_ns=valid_until_sim_ns, + authority_generation=authority_generation, + speed_mps=speed_mps, + yaw_rate_rps=_number(document, "yaw_rate_rps"), + ) + + +def _command_header( + run_id: str, + command_id: str, + sequence: int, + issued_at_sim_ns: int, + valid_until_sim_ns: int, + authority_generation: int, +) -> None: + _identifier(run_id, "command run id") + _identifier(command_id, "command id") + if sequence < 1: + raise SimulationContractError("command sequence must be positive") + if issued_at_sim_ns < 0: + raise SimulationContractError("command issue time must not be negative") + if valid_until_sim_ns <= issued_at_sim_ns: + raise SimulationContractError("command validity deadline must follow issue time") + if authority_generation < 1: + raise SimulationContractError("command authority generation must be positive") + + +def _identifier(value: str, label: str) -> str: + if not IDENTIFIER_PATTERN.fullmatch(value): + raise SimulationContractError(f"{label} is not a safe identifier") + return value + + +def _sha256(value: str, label: str) -> str: + if not SHA256_PATTERN.fullmatch(value): + raise SimulationContractError(f"{label} must be a lowercase SHA-256 digest") + return value + + +def _relative_path(value: str) -> str: + path = PurePosixPath(value) + if ( + not value + or value.startswith("/") + or "\\" in value + or path.is_absolute() + or path.as_posix() != value + or any(part in {"", ".", ".."} for part in path.parts) + ): + raise SimulationContractError("artifact path must be a confined relative POSIX path") + return value + + +def _utc_timestamp(value: str, label: str) -> str: + if not value.endswith("Z") or "T" not in value: + raise SimulationContractError(f"{label} must be an explicit UTC timestamp") + try: + parsed = datetime.fromisoformat(value.removesuffix("Z") + "+00:00") + except ValueError as exc: + raise SimulationContractError(f"{label} must be a valid explicit UTC timestamp") from exc + if parsed.tzinfo != UTC: + raise SimulationContractError(f"{label} must be an explicit UTC timestamp") + return value + + +def _finite(value: float, label: str) -> float: + if not math.isfinite(value): + raise SimulationContractError(f"{label} must be finite") + return value + + +def _object(value: object, label: str) -> dict[str, Any]: + if not isinstance(value, dict) or any(not isinstance(key, str) for key in value): + raise SimulationContractError(f"{label} must be an object") + return value + + +def _json_object(value: object, label: str) -> dict[str, Any]: + document = _object(value, label) + _json_value(document, label) + return document + + +def _json_value(value: object, label: str) -> None: + if value is None or isinstance(value, str | bool | int): + return + if isinstance(value, float): + _finite(value, label) + return + if isinstance(value, list): + for item in value: + _json_value(item, label) + return + if isinstance(value, dict) and all(isinstance(key, str) for key in value): + for item in value.values(): + _json_value(item, label) + return + raise SimulationContractError(f"{label} must contain only finite JSON values") + + +def _string(document: dict[str, Any], key: str) -> str: + value = document.get(key) + if not isinstance(value, str): + raise SimulationContractError(f"{key} must be a string") + return value + + +def _optional_string(document: dict[str, Any], key: str) -> str | None: + value = document.get(key) + if value is not None and not isinstance(value, str): + raise SimulationContractError(f"{key} must be a string or null") + return value + + +def _integer(document: dict[str, Any], key: str) -> int: + value = document.get(key) + if not isinstance(value, int) or isinstance(value, bool): + raise SimulationContractError(f"{key} must be an integer") + return value + + +def _number(document: dict[str, Any], key: str) -> float: + value = document.get(key) + if not isinstance(value, int | float) or isinstance(value, bool): + raise SimulationContractError(f"{key} must be a number") + return float(value) + + +def _boolean(document: dict[str, Any], key: str) -> bool: + value = document.get(key) + if not isinstance(value, bool): + raise SimulationContractError(f"{key} must be a boolean") + return value diff --git a/src/k1link/simulation/run_store.py b/src/k1link/simulation/run_store.py new file mode 100644 index 0000000..c905f44 --- /dev/null +++ b/src/k1link/simulation/run_store.py @@ -0,0 +1,583 @@ +from __future__ import annotations + +import json +import os +import shutil +import stat +import threading +from contextlib import suppress +from pathlib import Path +from typing import Any, Final +from uuid import uuid4 + +from k1link.artifacts import write_json_atomic +from k1link.simulation.contracts import ( + ControlSetpoint, + QualificationArtifact, + QualificationEvent, + QualificationRun, + RunKind, + RunState, + SimulationContractError, + control_setpoint_from_dict, +) + +MAX_RECORD_BYTES: Final = 1024 * 1024 +SEQUENCE_WIDTH: Final = 20 +ACTIVE_RECOVERY_STATES: Final = { + RunState.STARTING, + RunState.RUNNING, + RunState.PAUSED, + RunState.STOPPING, +} +COMMANDABLE_RUN_KINDS: Final = { + RunKind.SIMULATION_CLOSED_LOOP, + RunKind.DIGITAL_TWIN_CLOSED_LOOP, +} +TRANSITIONS: Final[dict[RunState, frozenset[RunState]]] = { + RunState.ADMITTED: frozenset({RunState.STARTING, RunState.ABORTED}), + RunState.STARTING: frozenset({RunState.RUNNING, RunState.FAILED, RunState.ABORTED}), + RunState.RUNNING: frozenset( + {RunState.PAUSED, RunState.STOPPING, RunState.FAILED, RunState.ABORTED} + ), + RunState.PAUSED: frozenset( + {RunState.RUNNING, RunState.STOPPING, RunState.FAILED, RunState.ABORTED} + ), + RunState.STOPPING: frozenset({RunState.COMPLETED, RunState.FAILED, RunState.ABORTED}), + RunState.COMPLETED: frozenset(), + RunState.FAILED: frozenset(), + RunState.ABORTED: frozenset(), +} + + +class QualificationRunStoreError(RuntimeError): + """Base error for the append-only qualification-run repository.""" + + +class QualificationRunNotFoundError(QualificationRunStoreError): + """The requested qualification run is not present.""" + + +class QualificationRunConflictError(QualificationRunStoreError): + """The caller used stale revision/sequence state or an existing identity.""" + + +class QualificationRunIntegrityError(QualificationRunStoreError): + """Stored qualification evidence violates its immutable contract.""" + + +class QualificationRunTransitionError(QualificationRunStoreError): + """A requested lifecycle transition is not allowed.""" + + +class QualificationRunStore: + """Single-writer, crash-safe, append-only Polygon run repository. + + The run manifest is immutable. Lifecycle, domain events, accepted commands + and artifact-index entries are one-file-per-record journals created with + exclusive filesystem semantics and fsynced before acknowledgement. + """ + + def __init__(self, root: Path) -> None: + self.root = root.expanduser().resolve() + self.root.mkdir(mode=0o700, parents=True, exist_ok=True) + _reject_symlink(self.root, "qualification repository") + _chmod_private(self.root) + self._lock = threading.RLock() + + def create(self, run: QualificationRun) -> QualificationRun: + if ( + run.state is not RunState.ADMITTED + or run.revision != 0 + or run.artifacts + or run.started_at_utc is not None + or run.ended_at_utc is not None + or run.terminal_reason is not None + ): + raise QualificationRunTransitionError( + "a new qualification run must be admitted at revision zero" + ) + final = self._run_path(run.run_id) + with self._lock: + if final.exists(): + raise QualificationRunConflictError("qualification run identity already exists") + staging = self.root / f".{run.run_id}.{uuid4().hex}.tmp" + try: + staging.mkdir(mode=0o700) + for name in ("events", "commands", "artifacts"): + (staging / name).mkdir(mode=0o700) + write_json_atomic(staging / "manifest.json", run.to_manifest_dict()) + _fsync_directory(staging) + try: + staging.rename(final) + except FileExistsError as exc: + raise QualificationRunConflictError( + "qualification run identity already exists" + ) from exc + _fsync_directory(self.root) + finally: + if staging.exists(): + shutil.rmtree(staging) + return self.load(run.run_id) + + def load(self, run_id: str) -> QualificationRun: + path = self._existing_run_path(run_id) + with self._lock: + manifest = QualificationRun.from_manifest_dict( + _read_json(path / "manifest.json", "run manifest") + ) + if manifest.run_id != run_id: + raise QualificationRunIntegrityError("manifest identity does not match its path") + events = self._read_events(path, run_id) + artifacts = self._read_artifacts(path, run_id) + return _replay_runtime(manifest, events, artifacts) + + def list_runs(self) -> tuple[QualificationRun, ...]: + with self._lock: + run_ids = tuple( + child.name + for child in sorted(self.root.iterdir(), key=lambda item: item.name) + if child.is_dir() and not child.name.startswith(".") + ) + return tuple(self.load(run_id) for run_id in run_ids) + + def transition( + self, + run_id: str, + new_state: RunState, + *, + expected_revision: int, + observed_at_utc: str, + host_monotonic_ns: int, + sim_time_ns: int | None = None, + reason: str | None = None, + ) -> QualificationRun: + with self._lock: + current = self.load(run_id) + if current.revision != expected_revision: + raise QualificationRunConflictError( + f"run revision is {current.revision}, not {expected_revision}" + ) + if new_state not in TRANSITIONS[current.state]: + raise QualificationRunTransitionError( + f"transition {current.state.value} -> {new_state.value} is not allowed" + ) + if new_state.terminal and (reason is None or not reason.strip()): + raise QualificationRunTransitionError("terminal transition requires a reason") + if not new_state.terminal and reason is not None: + raise QualificationRunTransitionError( + "a nonterminal transition cannot set a terminal reason" + ) + self._append_event_locked( + current, + event_type="lifecycle.state-changed", + observed_at_utc=observed_at_utc, + host_monotonic_ns=host_monotonic_ns, + sim_time_ns=sim_time_ns, + payload={ + "from": current.state.value, + "to": new_state.value, + "reason": reason, + }, + ) + return self.load(run_id) + + def append_event( + self, + run_id: str, + *, + event_type: str, + observed_at_utc: str, + host_monotonic_ns: int, + payload: dict[str, Any], + sim_time_ns: int | None = None, + expected_revision: int | None = None, + ) -> QualificationEvent: + if event_type == "lifecycle.state-changed": + raise QualificationRunTransitionError("lifecycle events must use transition()") + with self._lock: + current = self.load(run_id) + if current.state.terminal: + raise QualificationRunTransitionError( + "terminal qualification-run history is immutable" + ) + if expected_revision is not None and current.revision != expected_revision: + raise QualificationRunConflictError( + f"run revision is {current.revision}, not {expected_revision}" + ) + return self._append_event_locked( + current, + event_type=event_type, + observed_at_utc=observed_at_utc, + host_monotonic_ns=host_monotonic_ns, + sim_time_ns=sim_time_ns, + payload=payload, + ) + + def submit_command(self, command: ControlSetpoint) -> ControlSetpoint: + with self._lock: + run = self.load(command.run_id) + if run.kind not in COMMANDABLE_RUN_KINDS: + raise QualificationRunTransitionError( + f"{run.kind.value} cannot causally accept control commands" + ) + if run.state is not RunState.RUNNING: + raise QualificationRunTransitionError( + "control commands are accepted only while the run is running" + ) + if command.authority_generation != run.authority.generation: + raise QualificationRunConflictError("command authority generation is stale") + ttl_ns = command.valid_until_sim_ns - command.issued_at_sim_ns + if ttl_ns > run.authority.command_ttl_max_ns: + raise QualificationRunTransitionError("command TTL exceeds the authority limit") + directory = self._existing_run_path(command.run_id) / "commands" + expected_sequence = _next_sequence(directory) + if command.sequence != expected_sequence: + raise QualificationRunConflictError(f"command sequence must be {expected_sequence}") + _append_json_exclusive( + directory / _sequence_name(command.sequence), + command.to_dict(), + ) + return command + + def list_commands(self, run_id: str) -> tuple[ControlSetpoint, ...]: + path = self._existing_run_path(run_id) + with self._lock: + records = _read_sequence_directory(path / "commands", "command") + commands: list[ControlSetpoint] = [] + for sequence, record in records: + try: + command = control_setpoint_from_dict(record) + except SimulationContractError as exc: + raise QualificationRunIntegrityError(str(exc)) from exc + if command.run_id != run_id or command.sequence != sequence: + raise QualificationRunIntegrityError( + "command identity/sequence does not match its journal path" + ) + commands.append(command) + return tuple(commands) + + def register_artifact( + self, + run_id: str, + artifact: QualificationArtifact, + ) -> QualificationArtifact: + with self._lock: + run = self.load(run_id) + if run.state.terminal: + raise QualificationRunTransitionError( + "terminal qualification-run artifacts are sealed" + ) + directory = self._existing_run_path(run_id) / "artifacts" + if any(existing.artifact_id == artifact.artifact_id for existing in run.artifacts): + raise QualificationRunConflictError("artifact id is already registered") + sequence = _next_sequence(directory) + if artifact.sequence not in {0, sequence}: + raise QualificationRunConflictError(f"artifact sequence must be zero or {sequence}") + admitted = QualificationArtifact( + artifact_id=artifact.artifact_id, + kind=artifact.kind, + relative_path=artifact.relative_path, + sha256=artifact.sha256, + byte_length=artifact.byte_length, + source_of_record=artifact.source_of_record, + sequence=sequence, + ) + record = { + "schema_version": "missioncore.qualification-artifact/v1", + "run_id": run_id, + **admitted.to_dict(), + } + _append_json_exclusive(directory / _sequence_name(sequence), record) + return admitted + + def create_reset_episode( + self, + previous_run_id: str, + *, + run_id: str, + episode_id: str, + created_at_utc: str, + ) -> QualificationRun: + previous = self.load(previous_run_id) + if not previous.state.terminal: + raise QualificationRunTransitionError( + "reset requires the previous run to be terminal and sealed" + ) + replacement = QualificationRun( + run_id=run_id, + episode_id=episode_id, + kind=previous.kind, + state=RunState.ADMITTED, + scenario_generation=previous.scenario_generation, + scenario_sha256=previous.scenario_sha256, + profile_generation=previous.profile_generation, + profile_sha256=previous.profile_sha256, + mission_core_commit=previous.mission_core_commit, + providers=previous.providers, + host_profile_id=previous.host_profile_id, + host_profile_sha256=previous.host_profile_sha256, + seed=previous.seed, + reproducibility_tier=previous.reproducibility_tier, + authority=previous.authority, + clock_domain=previous.clock_domain, + created_at_utc=created_at_utc, + parent_run_id=previous.run_id, + ) + return self.create(replacement) + + def reconcile_interrupted( + self, + run_id: str, + *, + observed_at_utc: str, + host_monotonic_ns: int, + ) -> QualificationRun: + run = self.load(run_id) + if run.state not in ACTIVE_RECOVERY_STATES: + return run + return self.transition( + run_id, + RunState.FAILED, + expected_revision=run.revision, + observed_at_utc=observed_at_utc, + host_monotonic_ns=host_monotonic_ns, + reason="orchestrator-recovery-interrupted", + ) + + def _append_event_locked( + self, + run: QualificationRun, + *, + event_type: str, + observed_at_utc: str, + host_monotonic_ns: int, + sim_time_ns: int | None, + payload: dict[str, Any], + ) -> QualificationEvent: + sequence = run.revision + 1 + event = QualificationEvent( + run_id=run.run_id, + sequence=sequence, + event_type=event_type, + observed_at_utc=observed_at_utc, + host_monotonic_ns=host_monotonic_ns, + sim_time_ns=sim_time_ns, + payload=payload, + ) + directory = self._existing_run_path(run.run_id) / "events" + if _next_sequence(directory) != sequence: + raise QualificationRunConflictError("event journal changed concurrently") + _append_json_exclusive(directory / _sequence_name(sequence), event.to_dict()) + return event + + def _read_events(self, run_path: Path, run_id: str) -> tuple[QualificationEvent, ...]: + events: list[QualificationEvent] = [] + for sequence, record in _read_sequence_directory(run_path / "events", "event"): + try: + event = QualificationEvent.from_dict(record) + except SimulationContractError as exc: + raise QualificationRunIntegrityError(str(exc)) from exc + if event.run_id != run_id or event.sequence != sequence: + raise QualificationRunIntegrityError( + "event identity/sequence does not match its journal path" + ) + events.append(event) + return tuple(events) + + def _read_artifacts( + self, + run_path: Path, + run_id: str, + ) -> tuple[QualificationArtifact, ...]: + artifacts: list[QualificationArtifact] = [] + identifiers: set[str] = set() + for sequence, record in _read_sequence_directory( + run_path / "artifacts", + "artifact", + ): + if record.get("schema_version") != "missioncore.qualification-artifact/v1": + raise QualificationRunIntegrityError("unsupported qualification-artifact schema") + if record.get("run_id") != run_id: + raise QualificationRunIntegrityError("artifact run identity does not match") + try: + artifact = QualificationArtifact.from_dict(record) + except SimulationContractError as exc: + raise QualificationRunIntegrityError(str(exc)) from exc + if artifact.sequence != sequence: + raise QualificationRunIntegrityError( + "artifact sequence does not match its journal path" + ) + if artifact.artifact_id in identifiers: + raise QualificationRunIntegrityError("artifact journal contains a duplicate id") + identifiers.add(artifact.artifact_id) + artifacts.append(artifact) + return tuple(artifacts) + + def _run_path(self, run_id: str) -> Path: + try: + QualificationEvent( + run_id=run_id, + sequence=1, + event_type="identity.check", + observed_at_utc="1970-01-01T00:00:00Z", + host_monotonic_ns=0, + sim_time_ns=None, + payload={}, + ) + except SimulationContractError as exc: + raise QualificationRunIntegrityError(str(exc)) from exc + return self.root / run_id + + def _existing_run_path(self, run_id: str) -> Path: + path = self._run_path(run_id) + if not path.is_dir(): + raise QualificationRunNotFoundError("qualification run was not found") + _reject_symlink(path, "qualification run") + for name in ("events", "commands", "artifacts"): + directory = path / name + if not directory.is_dir(): + raise QualificationRunIntegrityError(f"run {name} journal is missing") + _reject_symlink(directory, f"run {name} journal") + return path + + +def _replay_runtime( + manifest: QualificationRun, + events: tuple[QualificationEvent, ...], + artifacts: tuple[QualificationArtifact, ...], +) -> QualificationRun: + state = RunState.ADMITTED + started_at_utc: str | None = None + ended_at_utc: str | None = None + terminal_reason: str | None = None + for event in events: + if state.terminal: + raise QualificationRunIntegrityError("event exists after terminal lifecycle state") + if event.event_type != "lifecycle.state-changed": + continue + source = event.payload.get("from") + target = event.payload.get("to") + reason = event.payload.get("reason") + if source != state.value or not isinstance(target, str): + raise QualificationRunIntegrityError("lifecycle journal does not match current state") + try: + new_state = RunState(target) + except ValueError as exc: + raise QualificationRunIntegrityError("lifecycle journal has an unknown state") from exc + if new_state not in TRANSITIONS[state]: + raise QualificationRunIntegrityError("lifecycle journal contains an illegal transition") + if new_state is RunState.RUNNING and started_at_utc is None: + started_at_utc = event.observed_at_utc + if new_state.terminal: + if not isinstance(reason, str) or not reason.strip(): + raise QualificationRunIntegrityError("terminal lifecycle event has no reason") + ended_at_utc = event.observed_at_utc + terminal_reason = reason + elif reason is not None: + raise QualificationRunIntegrityError("nonterminal lifecycle event has a reason") + state = new_state + try: + return manifest.with_runtime( + state=state, + revision=len(events), + started_at_utc=started_at_utc, + ended_at_utc=ended_at_utc, + terminal_reason=terminal_reason, + artifacts=artifacts, + ) + except SimulationContractError as exc: + raise QualificationRunIntegrityError(str(exc)) from exc + + +def _read_sequence_directory( + directory: Path, + label: str, +) -> tuple[tuple[int, dict[str, Any]], ...]: + _reject_symlink(directory, f"{label} journal") + records: list[tuple[int, dict[str, Any]]] = [] + expected = 1 + for path in sorted(directory.iterdir(), key=lambda item: item.name): + _reject_regular_file(path, f"{label} record") + if path.name != _sequence_name(expected): + raise QualificationRunIntegrityError(f"{label} journal is not a contiguous sequence") + records.append((expected, _read_json(path, f"{label} record"))) + expected += 1 + return tuple(records) + + +def _next_sequence(directory: Path) -> int: + return len(_read_sequence_directory(directory, "journal")) + 1 + + +def _sequence_name(sequence: int) -> str: + return f"{sequence:0{SEQUENCE_WIDTH}d}.json" + + +def _append_json_exclusive(path: Path, payload: object) -> None: + serialized = (json.dumps(payload, ensure_ascii=False, indent=2) + "\n").encode() + if len(serialized) > MAX_RECORD_BYTES: + raise QualificationRunIntegrityError("journal record exceeds the size limit") + flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL + if hasattr(os, "O_NOFOLLOW"): + flags |= os.O_NOFOLLOW + try: + descriptor = os.open(path, flags, 0o600) + except FileExistsError as exc: + raise QualificationRunConflictError("journal sequence already exists") from exc + try: + with os.fdopen(descriptor, "wb") as stream: + stream.write(serialized) + stream.flush() + os.fsync(stream.fileno()) + except BaseException: + path.unlink(missing_ok=True) + raise + _fsync_directory(path.parent) + + +def _read_json(path: Path, label: str) -> dict[str, Any]: + _reject_regular_file(path, label) + if path.stat().st_size > MAX_RECORD_BYTES: + raise QualificationRunIntegrityError(f"{label} exceeds the size limit") + try: + value = json.loads(path.read_text(encoding="utf-8")) + except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc: + raise QualificationRunIntegrityError(f"{label} is not valid UTF-8 JSON") from exc + if not isinstance(value, dict) or any(not isinstance(key, str) for key in value): + raise QualificationRunIntegrityError(f"{label} must be a JSON object") + return value + + +def _reject_regular_file(path: Path, label: str) -> None: + try: + mode = path.lstat().st_mode + except FileNotFoundError as exc: + raise QualificationRunIntegrityError(f"{label} is missing") from exc + if not stat.S_ISREG(mode): + raise QualificationRunIntegrityError(f"{label} must be a regular file") + + +def _reject_symlink(path: Path, label: str) -> None: + try: + mode = path.lstat().st_mode + except FileNotFoundError as exc: + raise QualificationRunIntegrityError(f"{label} is missing") from exc + if stat.S_ISLNK(mode): + raise QualificationRunIntegrityError(f"{label} must not be a symlink") + + +def _chmod_private(path: Path) -> None: + with suppress(OSError): + path.chmod(0o700) + + +def _fsync_directory(path: Path) -> None: + flags = os.O_RDONLY + if hasattr(os, "O_DIRECTORY"): + flags |= os.O_DIRECTORY + descriptor = os.open(path, flags) + try: + os.fsync(descriptor) + finally: + os.close(descriptor) diff --git a/tests/test_simulation_s1_run_store.py b/tests/test_simulation_s1_run_store.py new file mode 100644 index 0000000..36a15b7 --- /dev/null +++ b/tests/test_simulation_s1_run_store.py @@ -0,0 +1,409 @@ +from __future__ import annotations + +import json +from dataclasses import replace +from pathlib import Path + +import pytest + +from k1link.simulation import ( + AckermannControlSetpoint, + AuthorityProfile, + DifferentialControlSetpoint, + ProviderPin, + QualificationArtifact, + QualificationRun, + QualificationRunConflictError, + QualificationRunIntegrityError, + QualificationRunStore, + QualificationRunTransitionError, + ReproducibilityTier, + RunKind, + RunState, + SimulationContractError, +) + +SHA_A = "a" * 64 +SHA_B = "b" * 64 +SHA_C = "c" * 64 + + +def _run( + *, + run_id: str = "run-s1-001", + episode_id: str = "episode-s1-001", + kind: RunKind = RunKind.SIMULATION_CLOSED_LOOP, +) -> QualificationRun: + return QualificationRun( + run_id=run_id, + episode_id=episode_id, + kind=kind, + state=RunState.ADMITTED, + scenario_generation="stock-rover-v1", + scenario_sha256=SHA_A, + profile_generation="s1-ackermann-v1", + profile_sha256=SHA_B, + mission_core_commit="a19053dfd47577fb66cc63f605ba29ea2314f6aa", + providers=( + ProviderPin( + identifier="px4-autopilot", + version="v1.17.0", + revision="v1.17.0", + ), + ProviderPin( + identifier="gazebo", + version="harmonic", + revision="8.9.0", + ), + ), + host_profile_id="mission-gpu-s0", + host_profile_sha256=SHA_C, + seed=42, + reproducibility_tier=ReproducibilityTier.R1, + authority=AuthorityProfile( + generation=1, + command_ttl_max_ns=250_000_000, + heartbeat_timeout_monotonic_ns=500_000_000, + ), + clock_domain="gazebo:/clock", + created_at_utc="2026-07-24T17:00:00Z", + ) + + +def _start(store: QualificationRunStore, run_id: str = "run-s1-001") -> QualificationRun: + starting = store.transition( + run_id, + RunState.STARTING, + expected_revision=0, + observed_at_utc="2026-07-24T17:00:01Z", + host_monotonic_ns=1, + ) + return store.transition( + run_id, + RunState.RUNNING, + expected_revision=starting.revision, + observed_at_utc="2026-07-24T17:00:02Z", + host_monotonic_ns=2, + sim_time_ns=0, + ) + + +def test_run_manifest_is_immutable_and_runtime_replays_from_events(tmp_path: Path) -> None: + store = QualificationRunStore(tmp_path / "runs") + created = store.create(_run()) + + assert created.state is RunState.ADMITTED + manifest_before = (tmp_path / "runs/run-s1-001/manifest.json").read_bytes() + + running = _start(store) + restored = QualificationRunStore(tmp_path / "runs").load(running.run_id) + + assert restored.state is RunState.RUNNING + assert restored.revision == 2 + assert restored.started_at_utc == "2026-07-24T17:00:02Z" + assert (tmp_path / "runs/run-s1-001/manifest.json").read_bytes() == manifest_before + assert len(tuple((tmp_path / "runs/run-s1-001/events").iterdir())) == 2 + + +def test_lifecycle_rejects_stale_illegal_and_post_terminal_changes(tmp_path: Path) -> None: + store = QualificationRunStore(tmp_path) + store.create(_run()) + + with pytest.raises(QualificationRunTransitionError, match="not allowed"): + store.transition( + "run-s1-001", + RunState.RUNNING, + expected_revision=0, + observed_at_utc="2026-07-24T17:00:01Z", + host_monotonic_ns=1, + ) + + running = _start(store) + with pytest.raises(QualificationRunConflictError, match="revision"): + store.transition( + running.run_id, + RunState.PAUSED, + expected_revision=0, + observed_at_utc="2026-07-24T17:00:03Z", + host_monotonic_ns=3, + ) + + stopping = store.transition( + running.run_id, + RunState.STOPPING, + expected_revision=running.revision, + observed_at_utc="2026-07-24T17:00:03Z", + host_monotonic_ns=3, + ) + completed = store.transition( + running.run_id, + RunState.COMPLETED, + expected_revision=stopping.revision, + observed_at_utc="2026-07-24T17:00:04Z", + host_monotonic_ns=4, + reason="goal-reached", + ) + + assert completed.terminal_reason == "goal-reached" + assert completed.ended_at_utc == "2026-07-24T17:00:04Z" + with pytest.raises(QualificationRunTransitionError, match="immutable"): + store.append_event( + completed.run_id, + event_type="metric.observed", + observed_at_utc="2026-07-24T17:00:05Z", + host_monotonic_ns=5, + payload={"name": "distance", "value": 1.0}, + ) + + +def test_pause_resume_and_domain_event_share_monotonic_journal(tmp_path: Path) -> None: + store = QualificationRunStore(tmp_path) + store.create(_run()) + running = _start(store) + event = store.append_event( + running.run_id, + event_type="watchdog.heartbeat", + observed_at_utc="2026-07-24T17:00:03Z", + host_monotonic_ns=3, + sim_time_ns=2_000_000, + payload={"authority_generation": 1}, + expected_revision=2, + ) + paused = store.transition( + running.run_id, + RunState.PAUSED, + expected_revision=event.sequence, + observed_at_utc="2026-07-24T17:00:04Z", + host_monotonic_ns=4, + sim_time_ns=2_000_000, + ) + resumed = store.transition( + running.run_id, + RunState.RUNNING, + expected_revision=paused.revision, + observed_at_utc="2026-07-24T17:00:05Z", + host_monotonic_ns=5, + sim_time_ns=2_000_000, + ) + + assert event.sequence == 3 + assert resumed.revision == 5 + assert resumed.started_at_utc == "2026-07-24T17:00:02Z" + + +def test_ackermann_and_differential_commands_are_typed_and_sequenced( + tmp_path: Path, +) -> None: + store = QualificationRunStore(tmp_path) + store.create(_run()) + _start(store) + + first = AckermannControlSetpoint( + run_id="run-s1-001", + command_id="command-1", + sequence=1, + issued_at_sim_ns=10, + valid_until_sim_ns=200_000_010, + authority_generation=1, + speed_mps=1.5, + steering_normalized=-0.25, + ) + second = DifferentialControlSetpoint( + run_id="run-s1-001", + command_id="command-2", + sequence=2, + issued_at_sim_ns=20, + valid_until_sim_ns=200_000_020, + authority_generation=1, + speed_mps=0.5, + yaw_rate_rps=0.2, + ) + + store.submit_command(first) + store.submit_command(second) + + assert store.list_commands("run-s1-001") == (first, second) + + +def test_commands_fail_closed_on_state_sequence_authority_and_ttl(tmp_path: Path) -> None: + store = QualificationRunStore(tmp_path) + store.create(_run()) + command = AckermannControlSetpoint( + run_id="run-s1-001", + command_id="command-1", + sequence=1, + issued_at_sim_ns=0, + valid_until_sim_ns=100, + authority_generation=1, + speed_mps=1.0, + steering_normalized=0.0, + ) + + with pytest.raises(QualificationRunTransitionError, match="only while"): + store.submit_command(command) + _start(store) + with pytest.raises(QualificationRunConflictError, match="authority"): + store.submit_command(replace(command, authority_generation=2)) + with pytest.raises(QualificationRunConflictError, match="sequence"): + store.submit_command(replace(command, sequence=2)) + with pytest.raises(QualificationRunTransitionError, match="TTL"): + store.submit_command(replace(command, valid_until_sim_ns=250_000_001)) + + +def test_command_contract_rejects_nonfinite_and_direct_actuator_authority() -> None: + with pytest.raises(SimulationContractError, match="finite"): + AckermannControlSetpoint( + run_id="run-s1-001", + command_id="command-1", + sequence=1, + issued_at_sim_ns=0, + valid_until_sim_ns=100, + authority_generation=1, + speed_mps=float("nan"), + steering_normalized=0.0, + ) + with pytest.raises(SimulationContractError, match="-1..1"): + AckermannControlSetpoint( + run_id="run-s1-001", + command_id="command-1", + sequence=1, + issued_at_sim_ns=0, + valid_until_sim_ns=100, + authority_generation=1, + speed_mps=1.0, + steering_normalized=1.1, + ) + with pytest.raises(SimulationContractError, match="without actuator"): + AuthorityProfile( + generation=1, + command_ttl_max_ns=100, + heartbeat_timeout_monotonic_ns=100, + direct_actuator_setpoints_allowed=True, + ) + with pytest.raises(SimulationContractError, match="valid explicit UTC"): + replace(_run(), created_at_utc="2026-02-30T17:00:00Z") + + +def test_replay_and_ungated_hil_cannot_acquire_command_authority(tmp_path: Path) -> None: + store = QualificationRunStore(tmp_path) + store.create(_run(kind=RunKind.REPLAY_SHADOW)) + _start(store) + command = AckermannControlSetpoint( + run_id="run-s1-001", + command_id="command-1", + sequence=1, + issued_at_sim_ns=0, + valid_until_sim_ns=100, + authority_generation=1, + speed_mps=1.0, + steering_normalized=0.0, + ) + + with pytest.raises(QualificationRunTransitionError, match="cannot causally"): + store.submit_command(command) + with pytest.raises(SimulationContractError, match="separate physical"): + _run(kind=RunKind.HIL) + + +def test_artifact_index_is_append_only_confined_and_digest_bound(tmp_path: Path) -> None: + store = QualificationRunStore(tmp_path) + store.create(_run()) + admitted = store.register_artifact( + "run-s1-001", + QualificationArtifact( + artifact_id="run-manifest-copy", + kind="source.manifest", + relative_path="source/run-manifest.json", + sha256=SHA_A, + byte_length=123, + source_of_record=True, + ), + ) + + assert admitted.sequence == 1 + assert store.load("run-s1-001").artifacts == (admitted,) + with pytest.raises(QualificationRunConflictError, match="already registered"): + store.register_artifact("run-s1-001", admitted) + with pytest.raises(SimulationContractError, match="confined"): + QualificationArtifact( + artifact_id="escape", + kind="source.log", + relative_path="../outside.log", + sha256=SHA_A, + byte_length=1, + source_of_record=True, + ) + with pytest.raises(SimulationContractError, match="confined"): + replace(admitted, artifact_id="noncanonical", relative_path="source//artifact.json") + + +def test_reset_creates_new_episode_identity_without_overwriting_parent(tmp_path: Path) -> None: + store = QualificationRunStore(tmp_path) + store.create(_run()) + running = _start(store) + stopping = store.transition( + running.run_id, + RunState.STOPPING, + expected_revision=running.revision, + observed_at_utc="2026-07-24T17:00:03Z", + host_monotonic_ns=3, + ) + store.transition( + running.run_id, + RunState.ABORTED, + expected_revision=stopping.revision, + observed_at_utc="2026-07-24T17:00:04Z", + host_monotonic_ns=4, + reason="operator-reset", + ) + + reset = store.create_reset_episode( + running.run_id, + run_id="run-s1-002", + episode_id="episode-s1-002", + created_at_utc="2026-07-24T17:00:05Z", + ) + + assert reset.parent_run_id == "run-s1-001" + assert reset.state is RunState.ADMITTED + assert store.load("run-s1-001").state is RunState.ABORTED + assert {item.run_id for item in store.list_runs()} == {"run-s1-001", "run-s1-002"} + + +def test_restart_reconciliation_fails_active_run_without_hidden_resume(tmp_path: Path) -> None: + store = QualificationRunStore(tmp_path) + store.create(_run()) + _start(store) + + recovered = QualificationRunStore(tmp_path).reconcile_interrupted( + "run-s1-001", + observed_at_utc="2026-07-24T17:01:00Z", + host_monotonic_ns=1_000, + ) + + assert recovered.state is RunState.FAILED + assert recovered.terminal_reason == "orchestrator-recovery-interrupted" + assert len(QualificationRunStore(tmp_path).list_commands(recovered.run_id)) == 0 + + +def test_corrupt_or_noncontiguous_journal_fails_closed(tmp_path: Path) -> None: + store = QualificationRunStore(tmp_path) + store.create(_run()) + _start(store) + events = tmp_path / "run-s1-001/events" + (events / "00000000000000000002.json").rename(events / "00000000000000000003.json") + + with pytest.raises(QualificationRunIntegrityError, match="contiguous"): + store.load("run-s1-001") + + +def test_manifest_identity_tampering_is_detected(tmp_path: Path) -> None: + store = QualificationRunStore(tmp_path) + store.create(_run()) + manifest = tmp_path / "run-s1-001/manifest.json" + document = json.loads(manifest.read_text(encoding="utf-8")) + document["run_id"] = "other-run" + manifest.write_text(json.dumps(document), encoding="utf-8") + + with pytest.raises(QualificationRunIntegrityError, match="identity"): + store.load("run-s1-001")