165afd24bc
Some checks failed
CD Pipeline / workflow-shape (push) Successful in 0s
CD Pipeline / cancel-stale-cd (push) Has been skipped
CD Pipeline / tests (push) Failing after 1m7s
CD Pipeline / build-and-deploy (push) Has been skipped
CD Pipeline / post-deploy-checks (push) Has been skipped
324 lines
11 KiB
Python
324 lines
11 KiB
Python
"""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
|