Spaces:
Paused
Paused
File size: 9,778 Bytes
31226fd | 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 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 | """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 .loop_contract import DispatchBudget, valid_tick
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 valid_tick(tick)
or self.sensor.sensor_id != self.descriptor.connection_id
or 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
frame = self.sensor.sample(tick)
if (
frame.tick != tick
or not valid_tick(frame.tick)
or frame.sensor_id != self.sensor.sensor_id
or frame.modality != self.sensor.modality
):
return None
return frame
@dataclass(slots=True)
class ControlledEmbodimentAgent:
"""Execute only authorized, capable and rate-limited external actions.
Dispatch is serialized by the owning runtime. An attempted dispatch consumes
budget even on rejection or failure. Unknown effects latch the emergency stop;
neither software retry nor reset is permission to clear that stop.
"""
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])
last_receipt: ActionReceipt | None = None
_command_sequence: int = 0
max_pending_overrides: int = 128
_dispatch_budget: DispatchBudget = field(default_factory=DispatchBudget, init=False)
def __post_init__(self) -> None:
self._dispatch_budget.denial(0, self.max_actions_per_tick)
if (
type(self.max_pending_overrides) is not int
or self.max_pending_overrides <= 0
):
raise ValueError("max_pending_overrides must be a positive integer")
def reset(self, seed: int | None = None) -> EnvironmentObservation:
observation = self.environment.reset(seed)
self.episode += 1
self.episode_reward = 0.0
self.last_action = None
self.last_receipt = None
self.last_observation = observation
self._dispatch_budget.reset()
self._approved_override_ticks.clear()
return 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")
if not valid_tick(tick):
raise ValueError("override tick must be a non-negative integer")
if self._dispatch_budget.tick is not None and tick < self._dispatch_budget.tick:
raise ValueError("cannot approve an expired override tick")
if (
tick not in self._approved_override_ticks
and len(self._approved_override_ticks) >= self.max_pending_overrides
):
raise ValueError("pending override capacity exceeded")
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
self._dispatch_budget.reserve(command.tick, self.max_actions_per_tick)
self._approved_override_ticks = {
tick for tick in self._approved_override_ticks if tick >= command.tick
}
accepted = False
phase = "actuator_error"
try:
actuator_result = self.actuator.apply(command)
accepted = actuator_result.accepted
self.last_receipt = ActionReceipt(
command_id,
accepted,
True,
False,
not accepted,
error=None if accepted else actuator_result.message,
effect_observed=None if accepted else False,
)
phase = "audit_error"
self.audit.append(
self.descriptor.connection_id,
command,
actuator_result,
accepted=accepted,
reason="accepted" if accepted else actuator_result.message,
)
if not accepted:
return None
phase = "feedback_error"
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
except Exception as error:
self.emergency_stop()
reason = f"{phase}:{type(error).__name__}"
self.last_receipt = ActionReceipt(
command_id,
accepted,
True,
False,
True,
error=reason,
effect_observed=None,
)
if phase != "audit_error":
self.audit.append(
self.descriptor.connection_id,
command,
ActuatorResult(False, reason),
accepted=False,
reason=reason,
)
raise
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 (
command.actuator_id != self.descriptor.connection_id
or command.actuator_id != self.actuator.actuator_id
):
return ActuatorResult(False, "actuator target mismatch"), "target_mismatch"
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",
)
reason = self._dispatch_budget.denial(command.tick, self.max_actions_per_tick)
if reason is not None:
return ActuatorResult(False, reason), reason
if (
self.require_human_override
and command.tick not in self._approved_override_ticks
):
return ActuatorResult(False, "human override required"), "override_required"
return None, ""
|