feat(simulation): add S1 qualification run repository
This commit is contained in:
DCCONSTRUCTIONS
@@ -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",
|
||||
]
|
||||
|
||||
@@ -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
|
||||
@@ -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)
|
||||
Reference in New Issue
Block a user