diff --git a/src/k1link/simulation/__init__.py b/src/k1link/simulation/__init__.py index f6ebb62..19ec4d7 100644 --- a/src/k1link/simulation/__init__.py +++ b/src/k1link/simulation/__init__.py @@ -3,6 +3,7 @@ from k1link.simulation.contracts import ( AckermannControlSetpoint, AuthorityProfile, + CommandAuthorityScope, ControlProfile, ControlSetpoint, DifferentialControlSetpoint, @@ -38,6 +39,7 @@ __all__ = [ "AckermannControlSetpoint", "AuthorityProfile", "CheckStatus", + "CommandAuthorityScope", "ControlProfile", "ControlSetpoint", "DifferentialControlSetpoint", diff --git a/src/k1link/simulation/contracts.py b/src/k1link/simulation/contracts.py index 4b7a84a..f87e2d0 100644 --- a/src/k1link/simulation/contracts.py +++ b/src/k1link/simulation/contracts.py @@ -52,6 +52,10 @@ class ControlProfile(StrEnum): ROVER_SPEED_YAW_RATE_V1 = "rover-speed-yaw-rate/v1" +class CommandAuthorityScope(StrEnum): + VIRTUAL_ONLY = "virtual-only" + + @dataclass(frozen=True, slots=True) class ProviderPin: identifier: str @@ -418,6 +422,8 @@ class AckermannControlSetpoint: speed_mps: float steering_normalized: float profile: ControlProfile = ControlProfile.ROVER_SPEED_STEERING_V1 + authority_scope: CommandAuthorityScope = CommandAuthorityScope.VIRTUAL_ONLY + source: str = "simulation-orchestrator" def __post_init__(self) -> None: _command_header( @@ -434,6 +440,7 @@ class AckermannControlSetpoint: 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") + _command_provenance(self.authority_scope, self.source) def to_dict(self) -> dict[str, object]: return { @@ -442,6 +449,8 @@ class AckermannControlSetpoint: "command_id": self.command_id, "sequence": self.sequence, "profile": self.profile.value, + "authority_scope": self.authority_scope.value, + "source": self.source, "issued_at_sim_ns": self.issued_at_sim_ns, "valid_until_sim_ns": self.valid_until_sim_ns, "authority_generation": self.authority_generation, @@ -461,6 +470,8 @@ class DifferentialControlSetpoint: speed_mps: float yaw_rate_rps: float profile: ControlProfile = ControlProfile.ROVER_SPEED_YAW_RATE_V1 + authority_scope: CommandAuthorityScope = CommandAuthorityScope.VIRTUAL_ONLY + source: str = "simulation-orchestrator" def __post_init__(self) -> None: _command_header( @@ -475,6 +486,7 @@ class DifferentialControlSetpoint: _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") + _command_provenance(self.authority_scope, self.source) def to_dict(self) -> dict[str, object]: return { @@ -483,6 +495,8 @@ class DifferentialControlSetpoint: "command_id": self.command_id, "sequence": self.sequence, "profile": self.profile.value, + "authority_scope": self.authority_scope.value, + "source": self.source, "issued_at_sim_ns": self.issued_at_sim_ns, "valid_until_sim_ns": self.valid_until_sim_ns, "authority_generation": self.authority_generation, @@ -500,8 +514,9 @@ def control_setpoint_from_dict(value: object) -> ControlSetpoint: raise SimulationContractError("unsupported control-setpoint schema") try: profile = ControlProfile(_string(document, "profile")) + authority_scope = CommandAuthorityScope(_string(document, "authority_scope")) except ValueError as exc: - raise SimulationContractError("unknown control profile") from exc + raise SimulationContractError("unknown control profile or authority scope") from exc run_id = _string(document, "run_id") command_id = _string(document, "command_id") sequence = _integer(document, "sequence") @@ -519,6 +534,8 @@ def control_setpoint_from_dict(value: object) -> ControlSetpoint: authority_generation=authority_generation, speed_mps=speed_mps, steering_normalized=_number(document, "steering_normalized"), + authority_scope=authority_scope, + source=_string(document, "source"), ) return DifferentialControlSetpoint( run_id=run_id, @@ -529,6 +546,8 @@ def control_setpoint_from_dict(value: object) -> ControlSetpoint: authority_generation=authority_generation, speed_mps=speed_mps, yaw_rate_rps=_number(document, "yaw_rate_rps"), + authority_scope=authority_scope, + source=_string(document, "source"), ) @@ -552,6 +571,12 @@ def _command_header( raise SimulationContractError("command authority generation must be positive") +def _command_provenance(authority_scope: CommandAuthorityScope, source: str) -> None: + if authority_scope is not CommandAuthorityScope.VIRTUAL_ONLY: + raise SimulationContractError("S1 commands require virtual-only authority scope") + _identifier(source, "command source") + + def _identifier(value: str, label: str) -> str: if not IDENTIFIER_PATTERN.fullmatch(value): raise SimulationContractError(f"{label} is not a safe identifier") diff --git a/tests/test_simulation_s1_run_store.py b/tests/test_simulation_s1_run_store.py index 36a15b7..25a92a7 100644 --- a/tests/test_simulation_s1_run_store.py +++ b/tests/test_simulation_s1_run_store.py @@ -9,6 +9,7 @@ import pytest from k1link.simulation import ( AckermannControlSetpoint, AuthorityProfile, + CommandAuthorityScope, DifferentialControlSetpoint, ProviderPin, QualificationArtifact, @@ -223,6 +224,8 @@ def test_ackermann_and_differential_commands_are_typed_and_sequenced( store.submit_command(second) assert store.list_commands("run-s1-001") == (first, second) + assert first.to_dict()["authority_scope"] == "virtual-only" + assert first.to_dict()["source"] == "simulation-orchestrator" def test_commands_fail_closed_on_state_sequence_authority_and_ttl(tmp_path: Path) -> None: @@ -280,6 +283,21 @@ def test_command_contract_rejects_nonfinite_and_direct_actuator_authority() -> N heartbeat_timeout_monotonic_ns=100, direct_actuator_setpoints_allowed=True, ) + with pytest.raises(SimulationContractError, match="virtual-only"): + replace( + AckermannControlSetpoint( + run_id="run-s1-001", + command_id="command-scope", + sequence=1, + issued_at_sim_ns=0, + valid_until_sim_ns=100, + authority_generation=1, + speed_mps=0.0, + steering_normalized=0.0, + ), + authority_scope="physical", # type: ignore[arg-type] + ) + assert CommandAuthorityScope.VIRTUAL_ONLY.value == "virtual-only" with pytest.raises(SimulationContractError, match="valid explicit UTC"): replace(_run(), created_at_utc="2026-02-30T17:00:00Z")