feat(host-agent): report assignment outcomes

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,
+        )

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"

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())

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.