"""OpenClaw Live Ops Space scene state. This service maps the public-safe AI Agent autonomous runtime control readback into a small animation contract. It never reads raw sessions, SQLite, secrets, or runtime logs, and it never performs runtime actions. """ from __future__ import annotations from datetime import UTC, datetime from typing import Any from src.services.ai_agent_autonomous_runtime_control import ( build_ai_agent_autonomous_runtime_control_with_live_readback, ) _SCHEMA_VERSION = "openclaw_live_ops_scene_state_v1" _SOURCE_SCHEMA_VERSION = "ai_agent_autonomous_runtime_control_v1" _ZONES = [ { "zone_id": "control", "label": "OpenClaw Control", "kind": "control_plane", "position": {"x": 48, "y": 38}, }, { "zone_id": "mcp", "label": "MCP", "kind": "tool_context", "position": {"x": 18, "y": 30}, }, { "zone_id": "rag", "label": "RAG", "kind": "knowledge_retrieval", "position": {"x": 24, "y": 62}, }, { "zone_id": "playbook", "label": "PlayBook", "kind": "procedure_trust", "position": {"x": 54, "y": 70}, }, { "zone_id": "verifier", "label": "Verifier", "kind": "post_apply_verification", "position": {"x": 78, "y": 34}, }, { "zone_id": "telegram", "label": "Telegram Receipt", "kind": "receipt_delivery", "position": {"x": 82, "y": 66}, }, { "zone_id": "deploy", "label": "Deploy Readback", "kind": "release_truth", "position": {"x": 48, "y": 18}, }, ] _ZONE_BY_STAGE = { "mcp_context": "mcp", "service_log_evidence": "rag", "executor_log_projection": "rag", "playbook_trust": "playbook", "post_apply_verifier": "verifier", "telegram_receipt": "telegram", "trace_ledger": "control", "work_item_progress": "control", "deploy_readback": "deploy", "km_writeback": "playbook", } _ZONE_POSITIONS = {zone["zone_id"]: zone["position"] for zone in _ZONES} async def load_openclaw_live_ops_scene_state() -> dict[str, Any]: """Build scene state from the live autonomous runtime control readback.""" runtime_control = await build_ai_agent_autonomous_runtime_control_with_live_readback() return build_openclaw_live_ops_scene_state(runtime_control) def build_openclaw_live_ops_scene_state( runtime_control: dict[str, Any], ) -> dict[str, Any]: """Map autonomous runtime control into the OpenClaw scene contract.""" if runtime_control.get("schema_version") != _SOURCE_SCHEMA_VERSION: raise ValueError(f"runtime_control.schema_version must be {_SOURCE_SCHEMA_VERSION}") readback = _dict(runtime_control.get("runtime_receipt_readback")) trace = _dict(readback.get("trace_ledger")) progress = _dict(readback.get("work_item_progress")) work_items = _work_items(progress) agents = _agents(trace, work_items) source_connected = bool(trace) or bool(work_items) safety = _dict(trace.get("public_safety")) payload = { "schema_version": _SCHEMA_VERSION, "generated_at": datetime.now(UTC).isoformat(timespec="seconds"), "status": "live_readback_connected" if source_connected else "waiting_for_live_readback", "experience": { "name": "OpenClaw Live Ops Space", "mode": "continuous_animated_operations_room", "route": "/zh-TW/openclaw/live-ops-space", }, "source": { "source_endpoint": "/api/v1/agents/agent-autonomous-runtime-control", "source_schema_version": runtime_control.get("schema_version"), "deploy_readback_marker": _dict(runtime_control.get("program_status")).get( "deploy_readback_marker" ), "trace_ledger_schema_version": trace.get("schema_version"), "work_item_progress_schema_version": progress.get("schema_version"), "live_source_connected": source_connected, }, "room": { "room_id": "openclaw-live-ops", "layout": "isometric_ops_room", "zones": _ZONES, "animation_loop": { "enabled": True, "tick_ms": 4200, "motion_model": "idle_walk_work_wait_verify", }, }, "agents": agents, "work_items": work_items, "rollups": { "zone_count": len(_ZONES), "agent_count": len(agents), "work_item_count": len(work_items), "animated_entity_count": len(agents) + len(work_items), "completed_work_item_count": sum( 1 for item in work_items if item.get("status") == "completed" ), "blocked_work_item_count": sum( 1 for item in work_items if item.get("status") == "blocked" ), "source_stage_count": _int(trace.get("stage_count")), "recorded_stage_count": _int(trace.get("recorded_stage_count")), }, "boundaries": { "raw_session_read_allowed": False, "sqlite_read_allowed": False, "secret_value_display_allowed": False, "internal_reasoning_display_allowed": False, "telegram_unredacted_payload_display_allowed": False, "runtime_action_performed": False, "host_or_k8s_write_performed": False, "uses_public_safe_trace_ledger": safety.get("reads_raw_sessions") is not True, }, } _validate_scene_state(payload) return payload def _agents(trace: dict[str, Any], work_items: list[dict[str, Any]]) -> list[dict[str, Any]]: stages = [ item for item in _list(trace.get("stages")) if isinstance(item, dict) and str(item.get("stage_id") or "") ] if not stages: stages = [ { "stage_id": "work_item_progress", "display_name": "Work Item Progress", "recorded": bool(work_items), "total": len(work_items), "recent": 0, } ] agents: list[dict[str, Any]] = [] for index, stage in enumerate(stages[:8]): stage_id = str(stage.get("stage_id") or f"stage_{index}") zone_id = _ZONE_BY_STAGE.get(stage_id, "control") position = _position_for(zone_id, index) state = _agent_state(stage) agents.append( { "agent_id": f"openclaw-{stage_id.replace('_', '-')}", "label": str(stage.get("display_name") or stage_id).strip(), "zone_id": zone_id, "state": state, "animation": _animation_for_state(state, index), "position": position, "current_task": { "stage_id": stage_id, "recorded": stage.get("recorded") is True, "total": _int(stage.get("total")), "recent": _int(stage.get("recent")), "feeds_learning": stage.get("feeds_learning") is True, "required_for_closed_loop": ( stage.get("required_for_closed_loop") is True ), }, } ) return agents def _work_items(progress: dict[str, Any]) -> list[dict[str, Any]]: ordered = [ item for item in _list(progress.get("ordered_items")) if isinstance(item, dict) and str(item.get("work_item_id") or "") ] items: list[dict[str, Any]] = [] for index, item in enumerate(ordered[:12]): status = str(item.get("status") or "pending") zone_id = _zone_for_work_item(item) items.append( { "work_item_id": str(item.get("work_item_id") or ""), "title": str(item.get("title") or item.get("work_item_id") or ""), "priority": str(item.get("priority") or ""), "status": status, "zone_id": zone_id, "animation": _animation_for_state(_state_from_status(status), index), "position": _position_for(zone_id, index + 2), "next_controlled_action": str( item.get("next_controlled_action") or "" ), "exit_criteria": str(item.get("exit_criteria") or ""), } ) return items def _zone_for_work_item(item: dict[str, Any]) -> str: text = " ".join( str(item.get(key) or "") for key in ("work_item_id", "title", "next_controlled_action", "exit_criteria") ).lower() if "telegram" in text: return "telegram" if "verifier" in text or "verify" in text: return "verifier" if "km" in text or "knowledge" in text: return "playbook" if "deploy" in text: return "deploy" if "mcp" in text: return "mcp" if "rag" in text or "log" in text: return "rag" return "control" def _position_for(zone_id: str, index: int) -> dict[str, int]: base = _dict(_ZONE_POSITIONS.get(zone_id)) x = _int(base.get("x")) + ((index % 3) - 1) * 4 y = _int(base.get("y")) + ((index % 2) * 4) return {"x": max(6, min(92, x)), "y": max(8, min(86, y))} def _agent_state(stage: dict[str, Any]) -> str: if stage.get("recorded") is not True: return "waiting" if _int(stage.get("recent")) > 0: return "working" if stage.get("feeds_learning") is True: return "verified" return "idle" def _state_from_status(status: str) -> str: if status == "completed": return "verified" if status == "blocked": return "blocked" if status == "in_progress": return "working" return "waiting" def _animation_for_state(state: str, index: int) -> str: if state == "verified": return "pulse_verified" if state == "blocked": return "waiting_blink" if state == "working": return "typing_loop" if index % 2 else "walk_and_work_loop" if state == "waiting": return "idle_scan_loop" return "idle_breathing_loop" def _validate_scene_state(payload: dict[str, Any]) -> None: if payload.get("schema_version") != _SCHEMA_VERSION: raise ValueError(f"schema_version must be {_SCHEMA_VERSION}") if not _list(_dict(payload.get("room")).get("zones")): raise ValueError("room.zones must be present") if not _list(payload.get("agents")): raise ValueError("agents must be present") boundaries = _dict(payload.get("boundaries")) for key in ( "raw_session_read_allowed", "sqlite_read_allowed", "secret_value_display_allowed", "internal_reasoning_display_allowed", "telegram_unredacted_payload_display_allowed", "runtime_action_performed", "host_or_k8s_write_performed", ): if boundaries.get(key) is not False: raise ValueError(f"boundaries.{key} must remain false") def _dict(value: Any) -> dict[str, Any]: return value if isinstance(value, dict) else {} def _list(value: Any) -> list[Any]: return value if isinstance(value, list) else [] def _int(value: Any) -> int: try: return int(value) except (TypeError, ValueError): return 0