diff --git a/apps/device-host-agent/host_agent/processor.py b/apps/device-host-agent/host_agent/processor.py new file mode 100644 index 0000000..dd94718 --- /dev/null +++ b/apps/device-host-agent/host_agent/processor.py @@ -0,0 +1,43 @@ +from __future__ import annotations + +from dataclasses import dataclass +from typing import Literal, Protocol + +from cloud.internal_api.models import AssignmentModel +from host_agent.assignment import AssignmentExecutionResult +from host_agent.client import HostAgentClient + + +class ActiveAssignmentExecutor(Protocol): + async def run(self, assignment: AssignmentModel) -> AssignmentExecutionResult: ... + + +@dataclass(frozen=True) +class AssignmentProcessingResult: + execution: AssignmentExecutionResult + report_status: Literal["recorded", "already_recorded"] + + +class AssignmentProcessor: + def __init__( + self, + client: HostAgentClient, + active_executor: ActiveAssignmentExecutor, + ) -> None: + self.client = client + self.active_executor = active_executor + + async def process(self, assignment: AssignmentModel) -> AssignmentProcessingResult: + execution = await self.active_executor.run(assignment) + status = "done" if execution.status == "done" else "failed" + failure_reason = execution.failure_reason if status == "failed" else None + response = await self.client.report_result( + assignment, + status=status, + failure_reason=failure_reason, + result=dict(execution.metadata), + ) + return AssignmentProcessingResult( + execution=execution, + report_status=response.status, + ) diff --git a/apps/device-host-agent/tests/test_client.py b/apps/device-host-agent/tests/test_client.py index c7e4f99..78e167d 100644 --- a/apps/device-host-agent/tests/test_client.py +++ b/apps/device-host-agent/tests/test_client.py @@ -1,6 +1,7 @@ from __future__ import annotations import asyncio +import json from datetime import UTC, datetime import httpx @@ -131,3 +132,37 @@ def test_stale_lease_response_raises_typed_error_without_retry() -> None: asyncio.run(scenario()) assert attempts == 1 + + +def test_result_report_retries_identical_payload_after_response_loss() -> None: + payloads: list[dict[str, object]] = [] + + def handler(request: httpx.Request) -> httpx.Response: + payloads.append(json.loads(request.content)) + if len(payloads) == 1: + raise httpx.ReadError("response lost", request=request) + return httpx.Response(200, json={"status": "already_recorded"}) + + async def scenario() -> None: + async with httpx.AsyncClient( + transport=httpx.MockTransport(handler), + base_url="https://control.example", + ) as http_client: + client = HostAgentClient( + _config(), + http_client=http_client, + sleep=lambda delay: asyncio.sleep(0), + ) + response = await client.report_result( + _assignment(), + status="failed", + failure_reason="planner unavailable", + result={"runtime_status": "failed"}, + ) + + assert response.status == "already_recorded" + + asyncio.run(scenario()) + assert len(payloads) == 2 + assert payloads[0] == payloads[1] + assert payloads[0]["failure_reason"] == "planner unavailable" diff --git a/apps/device-host-agent/tests/test_processor.py b/apps/device-host-agent/tests/test_processor.py new file mode 100644 index 0000000..7484f74 --- /dev/null +++ b/apps/device-host-agent/tests/test_processor.py @@ -0,0 +1,87 @@ +from __future__ import annotations + +import asyncio +from datetime import UTC, datetime + +from cloud.internal_api.models import AssignmentModel, TerminalResultResponse +from host_agent.assignment import AssignmentExecutionResult +from host_agent.processor import AssignmentProcessor + + +def _assignment() -> AssignmentModel: + return 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", + ) + + +def test_processor_reports_success_with_execution_metadata() -> None: + async def scenario() -> None: + reports: list[dict[str, object]] = [] + + class SuccessfulExecutor: + async def run(self, assignment): + return AssignmentExecutionResult( + status="done", + failure_reason="ignored success detail", + metadata={"runtime_task_id": "runtime-a"}, + ) + + class RecordingClient: + async def report_result(self, assignment, **kwargs): + reports.append(kwargs) + return TerminalResultResponse(status="recorded") + + result = await AssignmentProcessor( + RecordingClient(), # type: ignore[arg-type] + SuccessfulExecutor(), + ).process(_assignment()) + + assert result.report_status == "recorded" + assert reports == [ + { + "status": "done", + "failure_reason": None, + "result": {"runtime_task_id": "runtime-a"}, + } + ] + + asyncio.run(scenario()) + + +def test_processor_preserves_runtime_failure_reason() -> None: + async def scenario() -> None: + reports: list[dict[str, object]] = [] + + class FailedExecutor: + async def run(self, assignment): + return AssignmentExecutionResult( + status="failed", + failure_reason="planner unavailable", + metadata={"runtime_status": "failed"}, + ) + + class RecordingClient: + async def report_result(self, assignment, **kwargs): + reports.append(kwargs) + return TerminalResultResponse(status="already_recorded") + + result = await AssignmentProcessor( + RecordingClient(), # type: ignore[arg-type] + FailedExecutor(), + ).process(_assignment()) + + assert result.report_status == "already_recorded" + assert result.execution.failure_reason == "planner unavailable" + assert reports[0] == { + "status": "failed", + "failure_reason": "planner unavailable", + "result": {"runtime_status": "failed"}, + } + + asyncio.run(scenario()) diff --git a/openspec/changes/cloud-control-plane-integration/tasks.md b/openspec/changes/cloud-control-plane-integration/tasks.md index e3993f2..7e66ec6 100644 --- a/openspec/changes/cloud-control-plane-integration/tasks.md +++ b/openspec/changes/cloud-control-plane-integration/tasks.md @@ -57,7 +57,7 @@ - [x] 7.3 Compose local `TaskRunner` and `WorkflowRunner` factories without importing cloud concerns into Runtime-owned packages. - [x] 7.4 Execute goal assignments through the configured Runtime Planner/Executor and workflow assignments through the existing workflow runner. - [x] 7.5 Run lease renewal alongside active execution and stop further interruptible actions after confirmed lease loss. -- [ ] 7.6 Normalize and report successful/failed terminal outcomes, including Runtime failure reasons, with idempotent retries after response loss. +- [x] 7.6 Normalize and report successful/failed terminal outcomes, including Runtime failure reasons, with idempotent retries after response loss. - [ ] 7.7 Implement graceful shutdown that stops polling, finishes or interrupts current work according to lease policy, and performs a final heartbeat when possible. - [ ] 7.8 Add fake-driver end-to-end tests for one host, multiple hosts, NAT-style outbound-only operation, control-plane restart, Host Agent restart, and lease loss.