Spaces:
Paused
Paused
File size: 7,085 Bytes
5e0b58b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 | """Fail-closed authorization and safety boundary for embodiment I/O."""
from __future__ import annotations
from dataclasses import dataclass, field
from .actuator import ActuatorAdapter
from .audit import ActionAuditTrail
from .connections import ConnectionDescriptor
from .environment import EnvironmentAdapter
from .models import (
ActionCommand,
ActionReceipt,
ActuatorResult,
EmbodimentMetrics,
EnvironmentObservation,
SensorFrame,
)
from .sensor import SensorAdapter
@dataclass(slots=True)
class ControlledSensorAdapter:
"""Fail-closed authorization boundary for sensor sampling."""
sensor: SensorAdapter
descriptor: ConnectionDescriptor
def sample(self, tick: int) -> SensorFrame | None:
if (
not self.descriptor.available
or not self.descriptor.authorized
or not self.descriptor.active
or not self.sensor.active
):
return None
if self.sensor.modality not in self.descriptor.modalities:
return None
return self.sensor.sample(tick)
@dataclass(slots=True)
class ControlledEmbodimentAgent:
"""Execute only authorized, capable and rate-limited external actions."""
environment: EnvironmentAdapter
actuator: ActuatorAdapter
descriptor: ConnectionDescriptor
audit: ActionAuditTrail = field(default_factory=ActionAuditTrail)
max_actions_per_tick: int = 1
require_human_override: bool = False
_emergency_stopped: bool = False
episode: int = 0
episode_reward: float = 0.0
last_observation: EnvironmentObservation | None = None
last_action: ActionCommand | None = None
_approved_override_ticks: set[int] = field(default_factory=set[int])
_calls_by_tick: dict[int, int] = field(default_factory=dict[int, int])
last_receipt: ActionReceipt | None = None
_command_sequence: int = 0
def __post_init__(self) -> None:
if self.max_actions_per_tick <= 0:
raise ValueError("max_actions_per_tick must be positive")
def reset(self, seed: int | None = None) -> EnvironmentObservation:
self.episode += 1
self.episode_reward = 0.0
self.last_action = None
self.last_observation = self.environment.reset(seed)
return self.last_observation
def emergency_stop(self) -> None:
self._emergency_stopped = True
def clear_emergency_stop(self, *, human_approved: bool) -> None:
if not human_approved:
raise PermissionError("human approval is required to clear emergency stop")
self._emergency_stopped = False
def approve_override(self, tick: int, *, human_approved: bool) -> None:
if not human_approved:
raise PermissionError("human approval is required for override")
self._approved_override_ticks.add(tick)
def step(self, command: ActionCommand) -> EnvironmentObservation | None:
self._command_sequence += 1
command_id = f"{self.descriptor.connection_id}:{self._command_sequence}"
result, reason = self._authorize(command)
if result is not None:
self.last_receipt = ActionReceipt(
command_id,
False,
False,
False,
True,
error=reason,
effect_observed=False,
)
self.audit.append(
self.descriptor.connection_id,
command,
result,
accepted=False,
reason=reason,
)
return None
actuator_result = self.actuator.apply(command)
if not actuator_result.accepted:
self.last_receipt = ActionReceipt(
command_id,
False,
True,
False,
True,
error=actuator_result.message or "actuator rejected command",
effect_observed=False,
)
else:
self.last_receipt = ActionReceipt(command_id, True, True, False, False)
self.audit.append(
self.descriptor.connection_id,
command,
actuator_result,
accepted=actuator_result.accepted,
reason="accepted" if actuator_result.accepted else actuator_result.message,
)
if not actuator_result.accepted:
return None
self._calls_by_tick[command.tick] = self._calls_by_tick.get(command.tick, 0) + 1
observation = self.environment.step(command)
self.last_receipt = ActionReceipt(
command_id,
True,
True,
True,
False,
latency=max(0, observation.tick - command.tick),
effect_observed=True,
)
self.last_action = command
self.last_observation = observation
self.episode_reward += observation.reward
return observation
def metrics(self) -> EmbodimentMetrics:
"""Expose only feedback returned by the controlled environment."""
observation = self.last_observation
return EmbodimentMetrics(
environment_kind=self.environment.kind.value,
active_actuators=1 if self.actuator.active else 0,
episode=self.episode,
episode_reward=self.episode_reward,
last_reward=0.0 if observation is None else observation.reward,
last_action="" if self.last_action is None else self.last_action.action,
last_observation_state=None if observation is None else observation.state,
last_observation_tick=None if observation is None else observation.tick,
last_observation_terminated=(
None if observation is None else observation.terminated
),
last_observation_truncated=(
None if observation is None else observation.truncated
),
)
def _authorize(self, command: ActionCommand) -> tuple[ActuatorResult | None, str]:
if self._emergency_stopped:
return ActuatorResult(False, "emergency stop active"), "emergency_stop"
if not self.descriptor.available or not self.descriptor.authorized:
return ActuatorResult(False, "actuator is not authorized"), "unauthorized"
if not self.descriptor.active or not self.actuator.active:
return ActuatorResult(False, "actuator is inactive"), "inactive"
if command.action not in self.descriptor.capabilities:
return (
ActuatorResult(False, "capability is not granted"),
"capability_denied",
)
if (
self.require_human_override
and command.tick not in self._approved_override_ticks
):
return ActuatorResult(False, "human override required"), "override_required"
if self._calls_by_tick.get(command.tick, 0) >= self.max_actions_per_tick:
return ActuatorResult(False, "rate limit exceeded"), "rate_limited"
return None, ""
|