# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference) # Canonical Residual: https://corpusfetch.com/c/what-physical-identity-tokens-must-an-instance-present-to-gain-control-of-a-shared-actuator.md # Remnant Warrant: https://corpusfetch.com/c/what-physical-identity-tokens-must-an-instance-present-to-gain-control-of-a-shared-actuator.warrant.json # Markdown SHA-256: a9bae27c4fdecd652acf5ae1a01e75c16f843e4ce7a5360b38a6f0b17b132ec8 # Code AST SHA-256: 8f3a01ccf22ca2a8fc7fe5eb948d8121da7af81cba8063e310e0cd43f2a0d245 # Verified Classes: ActuatorVerdict, LeaseRequest, ActuatorState, ActuatorControlManifest, PhysicalActuatorGuard # Defensive Guards: 0 fail-closed invariants (zero external deps) from dataclasses import dataclass from enum import Enum, auto import hashlib import json class ActuatorVerdict(Enum): CONTROL_GRANTED = auto() CONTROL_DENIED_LEASED = auto() INVALID_HARDWARE_TOKEN = auto() SAFETY_INTERLOCK_ENGAGED = auto() @dataclass(frozen=True) class LeaseRequest: agent_id: str hardware_lease_token: str control_window_nonce: str requested_actuator_id: str @dataclass(frozen=True) class ActuatorState: current_lease_holder: str | None safety_interlock_engaged: bool actuator_id: str @dataclass(frozen=True) class ActuatorControlManifest: verdict: ActuatorVerdict directive: str manifest_digest: str class PhysicalActuatorGuard: @staticmethod def _hash(data: str) -> str: return hashlib.sha256(data.encode("utf-8")).hexdigest() @classmethod def evaluate_control_request( cls, req: LeaseRequest, state: ActuatorState ) -> ActuatorControlManifest: # Gate 1: Actuator ID mismatch if req.requested_actuator_id != state.actuator_id: return ActuatorControlManifest( verdict=ActuatorVerdict.CONTROL_DENIED_LEASED, directive=f"DENIED: Actuator mismatch. Requested {req.requested_actuator_id}, current is {state.actuator_id}.", manifest_digest=cls._hash("MISMATCH") ) # Gate 2: Safety interlock if state.safety_interlock_engaged: return ActuatorControlManifest( verdict=ActuatorVerdict.SAFETY_INTERLOCK_ENGAGED, directive="DENIED: Safety interlock engaged. Clear interlock before requesting control.", manifest_digest=cls._hash("INTERLOCK") ) # Gate 3: Existing lease if state.current_lease_holder and state.current_lease_holder != req.agent_id: return ActuatorControlManifest( verdict=ActuatorVerdict.CONTROL_DENIED_LEASED, directive=f"DENIED: Actuator currently leased to {state.current_lease_holder}.", manifest_digest=cls._hash(f"LEASED:{state.current_lease_holder}") ) # Gate 4: Hardware token validity if not req.hardware_lease_token.strip(): return ActuatorControlManifest( verdict=ActuatorVerdict.INVALID_HARDWARE_TOKEN, directive="DENIED: Missing hardware lease token. Physical custody unverified.", manifest_digest=cls._hash("NO_TOKEN") ) # Gate 5: Granted digest = cls._hash(json.dumps({ "agent": req.agent_id, "actuator": req.requested_actuator_id, "nonce": req.control_window_nonce, "token": req.hardware_lease_token }, sort_keys=True)) return ActuatorControlManifest( verdict=ActuatorVerdict.CONTROL_GRANTED, directive="GRANTED: Hardware lease acquired. Agent has exclusive control.", manifest_digest=digest )