diff --git a/openspec/changes/cloud-control-plane-integration/tasks.md b/openspec/changes/cloud-control-plane-integration/tasks.md index 3b3e06e..ed8057f 100644 --- a/openspec/changes/cloud-control-plane-integration/tasks.md +++ b/openspec/changes/cloud-control-plane-integration/tasks.md @@ -35,7 +35,7 @@ ## 5. Host Agent Internal API -- [ ] 5.1 Add versioned `/internal/v1` request/response models for heartbeat snapshots, long-poll claim, lease renewal, and terminal result reporting. +- [x] 5.1 Add versioned `/internal/v1` request/response models for heartbeat snapshots, long-poll claim, lease renewal, and terminal result reporting. - [ ] 5.2 Add atomic heartbeat/snapshot validation and device ownership-conflict handling before delegating to `DevicePool`. - [ ] 5.3 Add long-poll assignment delivery that returns at most one claimed task and produces a normal empty timeout response. - [ ] 5.4 Add lease-renewal and idempotent terminal-result endpoints with typed stale-lease conflicts. diff --git a/packages/cloud-platform/cloud/internal_api/__init__.py b/packages/cloud-platform/cloud/internal_api/__init__.py new file mode 100644 index 0000000..7ac19c3 --- /dev/null +++ b/packages/cloud-platform/cloud/internal_api/__init__.py @@ -0,0 +1 @@ +"""Versioned protocol surface used by outbound-only Device Host Agents.""" diff --git a/packages/cloud-platform/cloud/internal_api/models.py b/packages/cloud-platform/cloud/internal_api/models.py new file mode 100644 index 0000000..768118c --- /dev/null +++ b/packages/cloud-platform/cloud/internal_api/models.py @@ -0,0 +1,72 @@ +from __future__ import annotations + +from datetime import datetime +from typing import Any, Literal + +from pydantic import BaseModel, Field + + +class DeviceSnapshotModel(BaseModel): + device_id: str = Field(min_length=1) + driver_type: str = Field(min_length=1) + status: Literal["idle", "busy", "offline", "error"] + capability_tags: list[str] = Field(default_factory=list) + + +class HeartbeatRequest(BaseModel): + host_id: str = Field(min_length=1) + address: str | None = None + devices: list[DeviceSnapshotModel] = Field(default_factory=list) + + +class HeartbeatResponse(BaseModel): + host_id: str + accepted_devices: int + received_at: datetime + + +class ClaimRequest(BaseModel): + host_id: str = Field(min_length=1) + timeout_seconds: float = Field(default=20.0, ge=0, le=60) + + +class AssignmentModel(BaseModel): + task_id: str + attempt: int = Field(ge=1) + lease_id: str + lease_expires_at: datetime + host_id: str + device_id: str + goal: str | None = None + workflow_definition_id: str | None = None + + +class ClaimResponse(BaseModel): + assignment: AssignmentModel | None = None + timed_out: bool = False + + +class LeaseRenewalRequest(BaseModel): + host_id: str = Field(min_length=1) + task_id: str = Field(min_length=1) + attempt: int = Field(ge=1) + lease_id: str = Field(min_length=1) + + +class LeaseRenewalResponse(BaseModel): + status: Literal["renewed"] + lease_expires_at: datetime + + +class TerminalResultRequest(BaseModel): + host_id: str = Field(min_length=1) + task_id: str = Field(min_length=1) + attempt: int = Field(ge=1) + lease_id: str = Field(min_length=1) + status: Literal["done", "failed"] + failure_reason: str | None = None + result: dict[str, Any] | None = None + + +class TerminalResultResponse(BaseModel): + status: Literal["recorded", "already_recorded"] diff --git a/tests/test_host_agent_protocol_models.py b/tests/test_host_agent_protocol_models.py new file mode 100644 index 0000000..210195c --- /dev/null +++ b/tests/test_host_agent_protocol_models.py @@ -0,0 +1,103 @@ +from __future__ import annotations + +from datetime import UTC, datetime + +import pytest +from pydantic import ValidationError + +from cloud.internal_api.models import ( + AssignmentModel, + ClaimRequest, + ClaimResponse, + DeviceSnapshotModel, + HeartbeatRequest, + LeaseRenewalRequest, + TerminalResultRequest, +) + + +def test_heartbeat_models_complete_device_snapshot() -> None: + heartbeat = HeartbeatRequest( + host_id="host-a", + address="10.0.0.1:9000", + devices=[ + DeviceSnapshotModel( + device_id="device-a", + driver_type="wda", + status="idle", + capability_tags=["ios", "physical"], + ) + ], + ) + + assert heartbeat.host_id == "host-a" + assert heartbeat.devices[0].capability_tags == ["ios", "physical"] + + +@pytest.mark.parametrize("status", ["unreachable", "unknown", ""]) +def test_device_snapshot_rejects_invalid_host_status(status: str) -> None: + with pytest.raises(ValidationError): + DeviceSnapshotModel( + device_id="device-a", + driver_type="wda", + status=status, + ) + + +def test_claim_response_supports_assignment_and_normal_timeout() -> None: + assignment = AssignmentModel( + task_id="task-a", + attempt=1, + lease_id="lease-a", + lease_expires_at=datetime(2026, 7, 12, tzinfo=UTC), + host_id="host-a", + device_id="device-a", + goal="open settings", + ) + + assert ClaimResponse(assignment=assignment).assignment == assignment + assert ClaimResponse(timed_out=True).model_dump() == { + "assignment": None, + "timed_out": True, + } + + +@pytest.mark.parametrize("timeout_seconds", [-1, 61]) +def test_claim_timeout_is_bounded(timeout_seconds: float) -> None: + with pytest.raises(ValidationError): + ClaimRequest(host_id="host-a", timeout_seconds=timeout_seconds) + + +def test_renewal_and_result_require_attempt_and_lease_identity() -> None: + renewal = LeaseRenewalRequest( + host_id="host-a", + task_id="task-a", + attempt=2, + lease_id="lease-a", + ) + result = TerminalResultRequest( + host_id="host-a", + task_id="task-a", + attempt=2, + lease_id="lease-a", + status="failed", + failure_reason="device offline", + result={"step": 3}, + ) + + assert renewal.attempt == result.attempt == 2 + assert result.failure_reason == "device offline" + + +@pytest.mark.parametrize( + "payload", + [ + {"host_id": "", "task_id": "task-a", "attempt": 1, "lease_id": "lease"}, + {"host_id": "host-a", "task_id": "", "attempt": 1, "lease_id": "lease"}, + {"host_id": "host-a", "task_id": "task-a", "attempt": 0, "lease_id": "lease"}, + {"host_id": "host-a", "task_id": "task-a", "attempt": 1, "lease_id": ""}, + ], +) +def test_renewal_rejects_incomplete_identity(payload: dict[str, object]) -> None: + with pytest.raises(ValidationError): + LeaseRenewalRequest.model_validate(payload)