QUELLCODE / PY
agent.py
src/hardware_box/controller/agent.py
1# OpenAss: nur privater, experimenteller Gebrauch; auf eigene Verantwortung.2# Niemals Unternehmensrechner, Produktivsysteme, Industrieanlagen oder3# sicherheitskritische Systeme anschließen oder bedienen.4# Haftungshinweis und gesetzliche Grenzen: ../../../DISCLAIMER.txt.5import asyncio6import dataclasses7import os8import secrets9import time1011from hardware_box.logging.audit import Audit12from hardware_box.openai.client import AstraClient, DemoAgent, calls_from, text_from13from hardware_box.safety.policy import Policy, SafetyError, MUTATIONS141516ACTIVE = {"RUNNING", "WAITING_CONFIRMATION"}171819class Controller:20 def __init__(self, settings, capture, hid, agent_factory=None):21 self.settings, self.capture, self.hid = settings, capture, hid22 self.agent_factory = agent_factory or (DemoAgent if settings.demo else AstraClient)23 self.state = "READY"24 self.error, self.message, self.task_text = "", "", ""25 self.current_action, self.pending, self.frame = None, None, None26 self.audit, self.policy, self.client = None, None, None27 self.worker, self.monitor, self._decision = None, None, None28 self._lifecycle = asyncio.Lock()2930 async def start(self, task):31 async with self._lifecycle:32 if self.worker and not self.worker.done():33 raise SafetyError("Bereits eine Sitzung aktiv")34 if not task.strip() or len(task) > 4000:35 raise SafetyError("Auftrag fehlt oder ist zu lang")36 if not self.settings.demo and (not self.settings.calibration_confirmed or not self.settings.allow_screenshot_upload):37 raise SafetyError("Kalibrierung und Screenshot-Übermittlung müssen lokal konfiguriert sein")38 self.policy = Policy(self.settings)39 self.audit = Audit(self.settings)40 self.client = self.agent_factory(self.settings, os.environ.get("OPENAI_API_KEY", ""))41 try:42 await self.hid.start()43 self.frame = await self.capture.snapshot()44 # Firmware requires a recent physical ARM button press, independently of this call.45 self.audit.event("start", task=task, demo=self.settings.demo, model=self.settings.model)46 await self.hid.arm()47 except BaseException:48 await self.hid.stop()49 await self.client.close()50 raise51 self.task_text, self.error, self.message = task, "", ""52 self.state = "RUNNING"53 self.worker = asyncio.create_task(self._run(), name="computer-agent")54 self.monitor = asyncio.create_task(self._watch(), name="session-guard")5556 async def _watch(self):57 try:58 while self.worker and not self.worker.done():59 await asyncio.sleep(0.2)60 self.policy.check_alive()61 health = self.hid.health_snapshot()62 if health.get("fault"):63 raise SafetyError("HID-Verbindung fehlerhaft")64 capture_health = self.capture.health()65 if capture_health["error"] or capture_health["age"] > 2:66 raise SafetyError("Capture unterbrochen oder Bild veraltet")67 except asyncio.CancelledError:68 return69 except Exception as exc:70 self.error, self.state = str(exc), "FAULT"71 if self.worker:72 self.worker.cancel()73 await self.hid.stop()7475 async def _run(self):76 try:77 self.audit.frame(self.frame)78 response = await self.client.start(self.task_text, self.frame)79 while True:80 self.policy.check_alive()81 self.audit.response(response)82 calls = calls_from(response)83 if not calls:84 self.message = text_from(response)85 if not self.message:86 raise SafetyError("API endet ohne Ergebnis oder Computer-Aktion")87 self.frame = await self.capture.snapshot()88 self.audit.frame(self.frame)89 self.state = "RESULT_REVIEW"90 self.audit.event("result_review", message=self.message)91 break92 outputs = []93 for call in calls:94 # Full preflight of each batch BEFORE its first HID input.95 for action in call["actions"]:96 self.policy.validate(action, self.frame.geometry)97 if action["type"] in MUTATIONS:98 self.hid.preflight(action, self.frame.geometry)99 checks = call.get("pending_safety_checks", [])100 if checks:101 await self._approve({"type": "api_safety_check", "checks": checks}, call["call_id"])102 for action in call["actions"]:103 await self._execute(action, call["call_id"])104 self.frame = await self.capture.snapshot()105 self.audit.frame(self.frame)106 output = {"type": "computer_call_output", "call_id": call["call_id"],107 "output": {"type": "computer_screenshot", "image_url": self.frame.data_url,108 "detail": "original"}}109 if checks:110 output["acknowledged_safety_checks"] = checks111 outputs.append(output)112 response = await self.client.observe(outputs)113 except asyncio.CancelledError:114 if self.state not in {"PAUSED", "STOPPED", "FAULT"}:115 self.state = "STOPPED"116 except Exception as exc:117 self.error, self.state = str(exc), "FAULT"118 if self.audit:119 try:120 self.audit.event("fault", error=self.error)121 except OSError:122 pass123 finally:124 self.pending, self.current_action = None, None125 if self._decision and not self._decision.done():126 self._decision.cancel()127 # Stop irrespective of logger/API availability.128 await self.hid.stop()129 if self.client:130 await self.client.close()131 if self.monitor and self.monitor is not asyncio.current_task():132 self.monitor.cancel()133134 async def _execute(self, action, call_id):135 self.policy.validate(action, self.frame.geometry)136 self.policy.check_alive()137 self.current_action = action138 if action["type"] in MUTATIONS:139 self.hid.preflight(action, self.frame.geometry)140 await self._approve(action, call_id)141 approved_hash = self.frame.sha256142 await self.policy.reserve()143 latest = await self.capture.snapshot()144 if latest.sha256 != approved_hash:145 raise SafetyError("Bild nach Freigabe verändert; keine Eingabe ausgeführt")146 self.frame = latest147 self.audit.event("action_intent", call_id=call_id, action=action, screenshot=self.frame.sha256)148 await self.hid.execute(action, self.frame.geometry)149 self.audit.event("action_acknowledged", call_id=call_id, action=action)150 elif action["type"] == "wait":151 await self.policy.reserve()152 await asyncio.sleep(0.75)153 self.audit.event("wait", call_id=call_id)154 else:155 await self.policy.reserve()156 # Always get a NEW frame after input completion and a short settle period.157 after = time.monotonic()158 await asyncio.sleep(self.settings.settle_seconds)159 self.frame = await self.capture.snapshot(after=after)160 self.current_action = None161162 async def _approve(self, action, call_id):163 self.policy.check_alive()164 # Refresh what the human sees, rather than displaying the old API input image.165 self.frame = await self.capture.snapshot()166 self.audit.frame(self.frame)167 nonce = secrets.token_urlsafe(24)168 self.pending = {"nonce": nonce, "action": action, "call_id": call_id,169 "screenshot_sha256": self.frame.sha256,170 "expires_at": time.time() + self.settings.approval_timeout_seconds}171 self._decision = asyncio.get_running_loop().create_future()172 self.state = "WAITING_CONFIRMATION"173 self.audit.event("approval_requested", action=action, call_id=call_id, screenshot=self.frame.sha256)174 try:175 approved = await asyncio.wait_for(self._decision, self.settings.approval_timeout_seconds)176 except TimeoutError as exc:177 raise SafetyError("Freigabe abgelaufen") from exc178 finally:179 self.pending = None180 if not approved:181 raise SafetyError("Aktion vom Benutzer abgelehnt")182 self.audit.event("approved", action=action, call_id=call_id)183 self.state = "RUNNING"184185 async def confirm(self, nonce, screenshot_sha256, approved):186 if self.state != "WAITING_CONFIRMATION" or not self.pending or not self._decision or self._decision.done():187 raise SafetyError("Keine offene Freigabe")188 if not secrets.compare_digest(nonce, self.pending["nonce"]) or screenshot_sha256 != self.pending["screenshot_sha256"]:189 raise SafetyError("Freigabe passt nicht zur Aktion und zum angezeigten Screenshot")190 if time.time() > self.pending["expires_at"]:191 raise SafetyError("Freigabe abgelaufen")192 expected_pending, expected_future = self.pending, self._decision193 if approved:194 # Strict default; blinking cursors/animations may require a new planning session.195 latest = await self.capture.snapshot()196 if self.pending is not expected_pending or self._decision is not expected_future or expected_future.done() or self.state != "WAITING_CONFIRMATION":197 raise SafetyError("Freigabe wurde während der Bildprüfung ungültig")198 if time.time() > expected_pending["expires_at"]:199 raise SafetyError("Freigabe während der Bildprüfung abgelaufen")200 if latest.sha256 != screenshot_sha256:201 expected_future.set_result(False)202 raise SafetyError("Bild seit Freigabeanzeige verändert; Sitzung neu planen")203 expected_future.set_result(approved)204205 async def stop(self, pause=False):206 async with self._lifecycle:207 self.state = "PAUSED" if pause else "STOPPED"208 if self.monitor:209 self.monitor.cancel()210 if self.worker and not self.worker.done():211 self.worker.cancel()212 # This call ceases heartbeat and attempts release/disarm; physical interlock is independent.213 await self.hid.stop()214 if self.worker:215 await asyncio.gather(self.worker, return_exceptions=True)216 self.pending, self.current_action = None, None217 if self.audit:218 self.audit.event("paused" if pause else "stopped")219220 async def manual(self, action):221 async with self._lifecycle:222 if self.state in ACTIVE or (self.worker and not self.worker.done()):223 raise SafetyError("Für manuelle Tests die Sitzung zuerst stoppen")224 self.policy, self.audit = Policy(self.settings), Audit(self.settings)225 self.frame = await self.capture.snapshot()226 self.policy.validate(action, self.frame.geometry)227 self.hid.preflight(action, self.frame.geometry)228 try:229 await self.hid.start()230 self.audit.event("manual_start", action=action, demo=self.settings.demo)231 await self.hid.arm()232 except BaseException:233 await self.hid.stop()234 raise235 self.state, self.error, self.message = "RUNNING", "", ""236 self.client = None237 self.worker = asyncio.create_task(self._manual_run(action))238 self.monitor = asyncio.create_task(self._watch())239240 async def _manual_run(self, action):241 try:242 await self._execute(action, "manual")243 self.state = "STOPPED"244 except asyncio.CancelledError:245 pass246 except Exception as exc:247 self.state, self.error = "FAULT", str(exc)248 finally:249 self.pending, self.current_action = None, None250 await self.hid.stop()251 if self.monitor:252 self.monitor.cancel()253254 def review_result(self, verified):255 if self.state != "RESULT_REVIEW":256 raise SafetyError("Kein Ergebnis zur Prüfung vorhanden")257 self.state = "COMPLETED" if verified else "STOPPED"258 self.audit.event("human_result_review", verified=verified)259260 def status(self):261 return {"state": self.state, "demo": self.settings.demo, "error": self.error,262 "message": self.message, "task": self.task_text, "action": self.current_action,263 "pending": self.pending, "session_id": self.audit.session_id if self.audit else None,264 "actions": self.policy.actions if self.policy else 0,265 "api_calls": self.client.api_calls if self.client else 0,266 "total_tokens": self.client.total_tokens if self.client else 0,267 "geometry": dataclasses.asdict(self.frame.geometry) if self.frame else None,268 "hid": self.hid.health_snapshot(), "capture": self.capture.health()}