feat(host-protocol): define internal API models

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.

packages/cloud-platform/cloud/internal_api/__init__.py

@@ -0,0 +1 @@
+"""Versioned protocol surface used by outbound-only Device Host Agents."""

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"]

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)