feat(host-agent): stop assignment execution collaboratively on cancellation

- LeaseGuard gains an is_cancellation convenience property
- ActiveAssignmentRunner marks the lease lost with a cancellation
  reason when a renewal response reports cancel_requested
- AssignmentExecutor threads stop_reason through to TaskRunner/
  WorkflowRunner and maps a cancellation-flavored stop to
  AssignmentExecutionResult.status = "cancelled" instead of "failed"
- AssignmentProcessor forwards a three-way done/cancelled/failed
  status when reporting the terminal result
- Add/extend tests across lease, assignment, processor, and client

apps/device-host-agent/tests/test_lease.py

@@ -29,7 +29,7 @@ def test_lease_renews_while_execution_is_active() -> None:
         renewed = asyncio.Event()
 
         class BlockingExecutor:
-            def execute(self, assignment, *, should_stop=None):
+            def execute(self, assignment, *, should_stop=None, stop_reason=None):
                 execution_started.set()
                 release_execution.wait(timeout=2)
                 return AssignmentExecutionResult(status="done")
@@ -61,13 +61,13 @@ def test_lease_renews_while_execution_is_active() -> None:
     asyncio.run(scenario())
 
 
-def test_stale_lease_stops_later_interruptible_actions() -> None:
+def test_cancel_requested_renewal_stops_execution_with_cancelled_status() -> None:
     async def scenario() -> None:
         first_action_started = Event()
         actions: list[str] = []
 
         class CooperativeExecutor:
-            def execute(self, assignment, *, should_stop=None):
+            def execute(self, assignment, *, should_stop=None, stop_reason=None):
                 assert should_stop is not None
                 actions.append("first")
                 first_action_started.set()
@@ -76,9 +76,59 @@ def test_stale_lease_stops_later_interruptible_actions() -> None:
                     Event().wait(0.001)
                 if not should_stop():
                     actions.append("second")
+                reason = stop_reason() if stop_reason is not None else None
                 return AssignmentExecutionResult(
-                    status="failed",
-                    failure_reason="execution interrupted",
+                    status="cancelled" if reason and "cancel" in reason else "failed",
+                    failure_reason=reason,
+                )
+
+            def latest_progress(self):
+                return None
+
+        class CancellingClient:
+            async def renew(self, assignment, *, progress=None):
+                assert await asyncio.to_thread(first_action_started.wait, 1)
+                return LeaseRenewalResponse(
+                    status="renewed",
+                    lease_expires_at=datetime.now(UTC) + timedelta(seconds=30),
+                    cancel_requested=True,
+                )
+
+        result = await asyncio.wait_for(
+            ActiveAssignmentRunner(
+                CancellingClient(),  # type: ignore[arg-type]
+                CooperativeExecutor(),
+            ).run(_assignment()),
+            timeout=1,
+        )
+
+        assert result.status == "cancelled"
+        assert result.failure_reason == "cancellation requested by control plane"
+        assert actions == ["first"]
+
+    asyncio.run(scenario())
+
+
+def test_stale_lease_stops_later_interruptible_actions() -> None:
+    async def scenario() -> None:
+        first_action_started = Event()
+        actions: list[str] = []
+
+        class CooperativeExecutor:
+            def execute(self, assignment, *, should_stop=None, stop_reason=None):
+                assert should_stop is not None
+                actions.append("first")
+                first_action_started.set()
+                assert first_action_started.wait(timeout=1)
+                while not should_stop():
+                    Event().wait(0.001)
+                if not should_stop():
+                    actions.append("second")
+                reason = stop_reason() if stop_reason is not None else None
+                assert reason == "lease rejected by control plane"
+                return AssignmentExecutionResult(
+                    status="cancelled" if reason and "cancel" in reason else "failed",
+                    failure_reason=reason,
                 )
 
             def latest_progress(self):
@@ -108,7 +158,7 @@ def test_renewal_loop_exits_when_execution_finishes() -> None:
         renew_calls = 0
 
         class ImmediateExecutor:
-            def execute(self, assignment, *, should_stop=None):
+            def execute(self, assignment, *, should_stop=None, stop_reason=None):
                 return AssignmentExecutionResult(status="done")
 
             def latest_progress(self):
@@ -142,7 +192,7 @@ def test_shutdown_request_stops_active_execution_cooperatively() -> None:
         execution_started = Event()
 
         class CooperativeExecutor:
-            def execute(self, assignment, *, should_stop=None):
+            def execute(self, assignment, *, should_stop=None, stop_reason=None):
                 assert should_stop is not None
                 execution_started.set()
                 while not should_stop():