feat(openclaw): add live ops space scene state
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

This commit is contained in:
Your Name
2026-06-29 21:17:13 +08:00
parent 1b184f9150
commit 165afd24bc
8 changed files with 1056 additions and 3 deletions

View File

@@ -9,8 +9,8 @@ from fastapi.testclient import TestClient
from src.api.v1.agents import router
from src.services.credential_escrow_evidence_intake_readiness import (
load_latest_credential_escrow_evidence_intake_readiness,
validate_credential_escrow_evidence_refs,
validate_credential_escrow_evidence_owner_response,
validate_credential_escrow_evidence_refs,
)
ESCROW_ITEMS = [

View File

@@ -0,0 +1,144 @@
from __future__ import annotations
from fastapi import FastAPI
from fastapi.testclient import TestClient
import src.api.v1.agents as agents_api
from src.api.v1.agents import router
from src.services.openclaw_live_ops_scene_state import (
build_openclaw_live_ops_scene_state,
)
def test_openclaw_live_ops_scene_state_maps_runtime_control_to_animation_contract():
payload = build_openclaw_live_ops_scene_state(_runtime_control_payload())
assert payload["schema_version"] == "openclaw_live_ops_scene_state_v1"
assert payload["status"] == "live_readback_connected"
assert payload["experience"]["name"] == "OpenClaw Live Ops Space"
assert payload["room"]["layout"] == "isometric_ops_room"
assert payload["room"]["animation_loop"]["enabled"] is True
assert payload["source"]["source_endpoint"] == (
"/api/v1/agents/agent-autonomous-runtime-control"
)
assert payload["source"]["live_source_connected"] is True
assert payload["rollups"]["zone_count"] >= 6
assert payload["rollups"]["agent_count"] >= 2
assert payload["rollups"]["work_item_count"] == 3
assert payload["rollups"]["animated_entity_count"] >= 5
assert any(agent["zone_id"] == "mcp" for agent in payload["agents"])
assert any(agent["zone_id"] == "verifier" for agent in payload["agents"])
assert any(item["status"] == "blocked" for item in payload["work_items"])
assert payload["boundaries"]["raw_session_read_allowed"] is False
assert payload["boundaries"]["sqlite_read_allowed"] is False
assert payload["boundaries"]["secret_value_display_allowed"] is False
assert payload["boundaries"]["runtime_action_performed"] is False
def test_openclaw_live_ops_scene_state_endpoint_returns_public_safe_scene(
monkeypatch,
):
async def fake_scene_state():
return build_openclaw_live_ops_scene_state(_runtime_control_payload())
monkeypatch.setattr(
agents_api,
"load_openclaw_live_ops_scene_state",
fake_scene_state,
)
app = FastAPI()
app.include_router(router, prefix="/api/v1")
client = TestClient(app)
response = client.get("/api/v1/agents/openclaw-live-ops-scene-state")
assert response.status_code == 200
data = response.json()
assert data["schema_version"] == "openclaw_live_ops_scene_state_v1"
assert data["room"]["animation_loop"]["enabled"] is True
assert data["rollups"]["agent_count"] >= 2
assert data["boundaries"]["telegram_unredacted_payload_display_allowed"] is False
assert data["boundaries"]["host_or_k8s_write_performed"] is False
def _runtime_control_payload() -> dict:
return {
"schema_version": "ai_agent_autonomous_runtime_control_v1",
"program_status": {
"deploy_readback_marker": (
"p2_416_d1n_autonomous_runtime_control_prod_readback_v2"
)
},
"runtime_receipt_readback": {
"trace_ledger": {
"schema_version": "ai_agent_autonomous_trace_ledger_v1",
"stage_count": 3,
"recorded_stage_count": 2,
"missing_required_stage_ids": ["telegram_receipt"],
"public_safety": {
"reads_raw_sessions": False,
"stores_secret_values": False,
"stores_unredacted_telegram_payload": False,
"stores_internal_reasoning": False,
},
"stages": [
{
"stage_id": "mcp_context",
"display_name": "MCP context",
"recorded": True,
"total": 42,
"recent": 5,
"feeds_learning": True,
"required_for_closed_loop": True,
},
{
"stage_id": "post_apply_verifier",
"display_name": "Post apply verifier",
"recorded": True,
"total": 18,
"recent": 0,
"feeds_learning": True,
"required_for_closed_loop": True,
},
{
"stage_id": "telegram_receipt",
"display_name": "Telegram receipt",
"recorded": False,
"total": 0,
"recent": 0,
"feeds_learning": False,
"required_for_closed_loop": True,
},
],
},
"work_item_progress": {
"schema_version": "ai_agent_automation_work_item_progress_v1",
"ordered_items": [
{
"work_item_id": "P0-006",
"priority": "P0",
"title": "StockPlatform freshness retry readback",
"status": "blocked",
"next_controlled_action": "wait retry then verify",
"exit_criteria": "STOCK_FRESHNESS_STATUS=ok",
},
{
"work_item_id": "P0-005",
"priority": "P0",
"title": "Credential escrow evidence refs",
"status": "in_progress",
"next_controlled_action": "validate evidence refs",
"exit_criteria": "owner_response_accepted_count=1",
},
{
"work_item_id": "P0-003",
"priority": "P0",
"title": "Gitea private inventory closeout",
"status": "completed",
"next_controlled_action": "",
"exit_criteria": "active_blocker_count=0",
},
],
},
},
}