257 lines
8.5 KiB
Python
257 lines
8.5 KiB
Python
from __future__ import annotations
|
|
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[2]
|
|
APP_ROOT = REPO_ROOT / "active-runtime" / "app"
|
|
if str(APP_ROOT) not in sys.path:
|
|
sys.path.insert(0, str(APP_ROOT))
|
|
|
|
from aios_hermes import HermesAdapter, HermesJobState, HermesWorker
|
|
from aios_openclaw import (
|
|
CommandState,
|
|
InProcessRuntimeTransport,
|
|
OpenClawAdapter,
|
|
)
|
|
from aios_v2 import (
|
|
ControlPlaneService,
|
|
ExecutionEvidence,
|
|
ExecutionOrchestrator,
|
|
ExecutionTracker,
|
|
RequestContext,
|
|
TaskState,
|
|
TenantIsolationError,
|
|
)
|
|
from aios_v2.persistence import DurableControlPlaneService, SQLiteSnapshotRepository
|
|
|
|
|
|
SIGNING_MATERIAL = b"candidate-test-signing-material-0001"
|
|
|
|
|
|
def context(
|
|
tenant_id: str,
|
|
principal: str,
|
|
*roles: str,
|
|
) -> RequestContext:
|
|
return RequestContext.create(
|
|
tenant_id=tenant_id,
|
|
principal_id=principal,
|
|
roles=roles,
|
|
trace_id=f"trace-{tenant_id}-{principal}",
|
|
)
|
|
|
|
|
|
def bootstrap(service: ControlPlaneService, tenant_id: str = "tenant-a"):
|
|
platform = context(tenant_id, "platform", "platform_admin")
|
|
service.register_tenant(
|
|
platform,
|
|
tenant_id=tenant_id,
|
|
display_name=tenant_id,
|
|
)
|
|
operator = context(
|
|
tenant_id,
|
|
"operator",
|
|
"tenant_admin",
|
|
"operator",
|
|
"approver",
|
|
"auditor",
|
|
"runtime",
|
|
)
|
|
return platform, operator
|
|
|
|
|
|
class CandidateFlowTests(unittest.TestCase):
|
|
def test_openclaw_full_flow_reconciles_completed_task(self) -> None:
|
|
service = ControlPlaneService()
|
|
_, operator = bootstrap(service)
|
|
agent = service.register_agent(
|
|
operator,
|
|
agent_id="openclaw-agent",
|
|
display_name="OpenClaw",
|
|
runtime="openclaw",
|
|
capabilities=("echo",),
|
|
)
|
|
task = service.create_task(
|
|
operator,
|
|
agent_id=agent.agent_id,
|
|
action="echo",
|
|
payload={"message": "hello"},
|
|
idempotency_key="openclaw-flow-1",
|
|
)
|
|
approved = service.transition_task(
|
|
operator,
|
|
task_id=task.task_id,
|
|
target_state=TaskState.APPROVED,
|
|
expected_revision=task.revision,
|
|
)
|
|
|
|
transport = InProcessRuntimeTransport(
|
|
lambda command: {
|
|
"state": CommandState.COMPLETED.value,
|
|
"output": {"echo": command.payload["message"]},
|
|
}
|
|
)
|
|
resolver = lambda tenant_id, key_id: SIGNING_MATERIAL
|
|
adapter = OpenClawAdapter(transport, resolver)
|
|
orchestrator = ExecutionOrchestrator(
|
|
service,
|
|
openclaw=adapter,
|
|
openclaw_signing_material_resolver=resolver,
|
|
)
|
|
|
|
dispatch = orchestrator.dispatch(operator, task_id=approved.task_id)
|
|
completed = service.get_task(operator, task.task_id)
|
|
|
|
self.assertEqual(dispatch.runtime, "openclaw")
|
|
self.assertEqual(dispatch.state, CommandState.COMPLETED.value)
|
|
self.assertEqual(completed.state, TaskState.COMPLETED)
|
|
self.assertEqual(completed.result, {"echo": "hello"})
|
|
|
|
def test_hermes_full_flow_reconciles_worker_result(self) -> None:
|
|
service = ControlPlaneService()
|
|
_, operator = bootstrap(service)
|
|
agent = service.register_agent(
|
|
operator,
|
|
agent_id="hermes-agent",
|
|
display_name="Hermes",
|
|
runtime="hermes",
|
|
capabilities=("transform",),
|
|
metadata={"cell_id": "cell-a", "profile_id": "profile-a"},
|
|
)
|
|
task = service.create_task(
|
|
operator,
|
|
agent_id=agent.agent_id,
|
|
action="transform",
|
|
payload={"value": 7},
|
|
idempotency_key="hermes-flow-1",
|
|
)
|
|
approved = service.transition_task(
|
|
operator,
|
|
task_id=task.task_id,
|
|
target_state=TaskState.APPROVED,
|
|
expected_revision=task.revision,
|
|
)
|
|
|
|
hermes = HermesAdapter(default_lease_seconds=10)
|
|
orchestrator = ExecutionOrchestrator(service, hermes=hermes)
|
|
dispatch = orchestrator.dispatch(operator, task_id=approved.task_id)
|
|
self.assertEqual(dispatch.runtime, "hermes")
|
|
self.assertEqual(dispatch.state, HermesJobState.QUEUED.value)
|
|
|
|
worker = HermesWorker(
|
|
hermes,
|
|
lambda job: {"transformed": int(job.payload["value"]) * 2},
|
|
worker_id="worker-a",
|
|
tenant_id="tenant-a",
|
|
cell_id="cell-a",
|
|
profile_id="profile-a",
|
|
heartbeat_seconds=0.05,
|
|
)
|
|
run = worker.run_once()
|
|
self.assertTrue(run.claimed)
|
|
self.assertEqual(run.state, HermesJobState.COMPLETED.value)
|
|
|
|
result = hermes.result(dispatch.runtime_execution_id, tenant_id="tenant-a")
|
|
completed = orchestrator.reconcile_hermes(operator, result)
|
|
self.assertEqual(completed.state, TaskState.COMPLETED)
|
|
self.assertEqual(completed.result, {"transformed": 14})
|
|
|
|
def test_cross_tenant_task_read_is_denied(self) -> None:
|
|
service = ControlPlaneService()
|
|
_, tenant_a = bootstrap(service, "tenant-a")
|
|
platform_b = context("tenant-b", "platform", "platform_admin")
|
|
service.register_tenant(
|
|
platform_b,
|
|
tenant_id="tenant-b",
|
|
display_name="Tenant B",
|
|
)
|
|
tenant_b = context("tenant-b", "auditor", "auditor")
|
|
|
|
agent = service.register_agent(
|
|
tenant_a,
|
|
agent_id="agent-a",
|
|
display_name="Agent A",
|
|
runtime="hermes",
|
|
capabilities=("work",),
|
|
)
|
|
task = service.create_task(
|
|
tenant_a,
|
|
agent_id=agent.agent_id,
|
|
action="work",
|
|
payload={},
|
|
idempotency_key="tenant-isolation-1",
|
|
)
|
|
|
|
with self.assertRaises(TenantIsolationError):
|
|
service.get_task(tenant_b, task.task_id)
|
|
|
|
def test_durable_service_recovers_state_after_restart(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
path = Path(temporary) / "state.sqlite3"
|
|
service = DurableControlPlaneService(SQLiteSnapshotRepository(path))
|
|
_, operator = bootstrap(service)
|
|
agent = service.register_agent(
|
|
operator,
|
|
agent_id="persistent-agent",
|
|
display_name="Persistent",
|
|
runtime="hermes",
|
|
capabilities=("persist",),
|
|
)
|
|
task = service.create_task(
|
|
operator,
|
|
agent_id=agent.agent_id,
|
|
action="persist",
|
|
payload={"ok": True},
|
|
idempotency_key="persist-1",
|
|
)
|
|
|
|
recovered = DurableControlPlaneService(SQLiteSnapshotRepository(path))
|
|
restored_task = recovered.get_task(operator, task.task_id)
|
|
self.assertEqual(restored_task.payload, {"ok": True})
|
|
self.assertEqual(restored_task.state, TaskState.CREATED)
|
|
|
|
def test_execution_tracking_persists_claim_heartbeat_and_result(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
tracker = ExecutionTracker(temporary)
|
|
claimed = tracker.claim(
|
|
ExecutionEvidence(
|
|
execution_id="exec-1",
|
|
tenant_id="tenant-a",
|
|
task_id="task-1",
|
|
agent_id="agent-1",
|
|
executor="codex",
|
|
branch="task/codex/example",
|
|
progress_total=3,
|
|
)
|
|
)
|
|
self.assertEqual(claimed.status, "claimed")
|
|
tracker.heartbeat(
|
|
tenant_id="tenant-a",
|
|
execution_id="exec-1",
|
|
status="running",
|
|
current_step="tests",
|
|
progress_completed=2,
|
|
last_commit="abc123",
|
|
)
|
|
tracker.complete(
|
|
tenant_id="tenant-a",
|
|
execution_id="exec-1",
|
|
status="complete",
|
|
test_status="passed",
|
|
test_summary={"passed": 5, "failed": 0},
|
|
)
|
|
restored = ExecutionTracker(temporary).get(
|
|
tenant_id="tenant-a",
|
|
execution_id="exec-1",
|
|
)
|
|
self.assertEqual(restored.status, "complete")
|
|
self.assertEqual(restored.progress_completed, 3)
|
|
self.assertEqual(restored.test_summary["passed"], 5)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|