Merge branch 'worktree-task-cancellation': task cancellation feature

# Conflicts:
#	packages/cloud-platform/cloud/schema.py

tests/test_cloud_client.py

@@ -123,6 +123,22 @@ def test_client_unknown_task_raises(tmp_path) -> None:
         client.get_task_status("does-not-exist")
 
 
+def test_client_cancel_task_round_trip(tmp_path) -> None:
+    client, _ = _client_and_pool(tmp_path)
+    task_id = client.submit_task(goal="cancel me")["task_id"]
+
+    cancelled = client.cancel_task(task_id)
+
+    assert cancelled == {"task_id": task_id, "status": "cancelled"}
+    assert client.get_task_status(task_id)["status"] == "cancelled"
+
+
+def test_client_cancel_unknown_task_raises(tmp_path) -> None:
+    client, _ = _client_and_pool(tmp_path)
+    with pytest.raises(httpx.HTTPStatusError):
+        client.cancel_task("does-not-exist")
+
+
 def test_client_applies_bearer_token_to_every_public_method(tmp_path) -> None:
     token = "sdk-secret"
     provider = ConfiguredBearerAuthProvider(
@@ -152,6 +168,7 @@ def test_client_applies_bearer_token_to_every_public_method(tmp_path) -> None:
     assert client.get_task_status(task_id)["status"] == "queued"
     assert client.list_tasks()["total"] == 1
     assert client.get_task_attempts(task_id) == []
+    assert client.cancel_task(task_id) == {"task_id": task_id, "status": "cancelled"}
     assert client.list_devices() == []
     assert client.list_hosts() == []
     assert client.list_plugins() == []

tests/test_cloud_repository_contract.py

@@ -967,7 +967,8 @@ def test_active_lease_renews_for_owning_host(database_url: str) -> None:
             now=now + timedelta(seconds=30),
         )
 
-        assert status == "renewed"
+        assert status.status == "renewed"
+        assert status.cancel_requested is False
         task = database.repository.get_task(task_id)
         assert task is not None
         assert task.lease_expires_at == renewed_expiry
@@ -1031,7 +1032,7 @@ def test_stale_or_foreign_lease_renewal_conflicts(
             now=now + timedelta(seconds=30),
         )
 
-        assert status == "conflict"
+        assert status.status == "conflict"
         task = database.repository.get_task(task_id)
         assert task is not None
         assert task.lease_expires_at == initial_expiry
@@ -1078,7 +1079,7 @@ def test_expired_or_missing_lease_cannot_be_renewed(database_url: str) -> None:
                 host_id=host_id,
                 lease_expires_at=now + timedelta(minutes=2),
                 now=now + timedelta(seconds=2),
-            )
+            ).status
             == "expired"
         )
         assert (
@@ -1089,7 +1090,7 @@ def test_expired_or_missing_lease_cannot_be_renewed(database_url: str) -> None:
                 host_id=host_id,
                 lease_expires_at=now + timedelta(minutes=2),
                 now=now,
-            )
+            ).status
             == "not_found"
         )
     finally:
@@ -1823,3 +1824,251 @@ def test_record_planner_decision_stores_null_rationale_and_thinking(
         assert decisions[0].expected_outcome is None
     finally:
         database.close()
+
+
+def test_cancel_queued_task_is_immediate(database_url: str) -> None:
+    database = CloudDatabase(database_url)
+    task_id = _unique_id("cancel-queued-task")
+    now = datetime(2026, 7, 15, 8, 0, tzinfo=UTC)
+
+    try:
+        database.repository.enqueue_task(
+            ScheduledTask(
+                id=task_id,
+                goal="cancel before assignment",
+                workflow_definition_id=None,
+                constraints=TaskConstraints(),
+                created_at=now,
+            )
+        )
+
+        status = database.repository.request_task_cancellation(
+            task_id, requested_at=now
+        )
+
+        assert status == "requested"
+        task = database.repository.get_task(task_id)
+        assert task is not None
+        assert task.status == "cancelled"
+        assert task.cancel_requested_at is None
+    finally:
+        database.close()
+
+
+def test_cancel_request_on_assigned_task_is_durable(database_url: str) -> None:
+    database = CloudDatabase(database_url)
+    host_id = _unique_id("cancel-durable-host")
+    device_id = _unique_id("cancel-durable-device")
+    task_id = _unique_id("cancel-durable-task")
+    now = datetime(2026, 7, 15, 8, 0, tzinfo=UTC)
+
+    try:
+        database.repository.upsert_host(host_id, address=None, last_seen_at=now)
+        database.repository.replace_host_devices(
+            host_id,
+            [_device(device_id, host_id)],
+        )
+        database.repository.enqueue_task(
+            ScheduledTask(
+                id=task_id,
+                goal="cancel while running",
+                workflow_definition_id=None,
+                constraints=TaskConstraints(),
+                created_at=now,
+            )
+        )
+        database.repository.assign_task(
+            task_id=task_id,
+            host_id=host_id,
+            device_id=device_id,
+            lease_id="durable-cancel-lease",
+            lease_expires_at=now + timedelta(minutes=5),
+            now=now,
+        )
+
+        status = database.repository.request_task_cancellation(
+            task_id, requested_at=now
+        )
+        assert status == "requested"
+
+        # Simulate a process restart by re-fetching the task from a fresh read.
+        task = database.repository.get_task(task_id)
+        assert task is not None
+        assert task.status == "assigned"
+        assert task.cancel_requested_at == now
+
+        repeat_status = database.repository.request_task_cancellation(
+            task_id, requested_at=now + timedelta(seconds=5)
+        )
+        assert repeat_status == "already_requested"
+        task = database.repository.get_task(task_id)
+        assert task is not None
+        assert task.cancel_requested_at == now
+    finally:
+        database.close()
+
+
+def test_renew_lease_reports_pending_cancellation(database_url: str) -> None:
+    database = CloudDatabase(database_url)
+    host_id = _unique_id("cancel-renew-host")
+    device_id = _unique_id("cancel-renew-device")
+    task_id = _unique_id("cancel-renew-task")
+    now = datetime(2026, 7, 15, 8, 0, tzinfo=UTC)
+
+    try:
+        database.repository.upsert_host(host_id, address=None, last_seen_at=now)
+        database.repository.replace_host_devices(
+            host_id,
+            [_device(device_id, host_id)],
+        )
+        database.repository.enqueue_task(
+            ScheduledTask(
+                id=task_id,
+                goal="report pending cancellation on renewal",
+                workflow_definition_id=None,
+                constraints=TaskConstraints(),
+                created_at=now,
+            )
+        )
+        database.repository.assign_task(
+            task_id=task_id,
+            host_id=host_id,
+            device_id=device_id,
+            lease_id="renew-cancel-lease",
+            lease_expires_at=now + timedelta(minutes=5),
+            now=now,
+        )
+        database.repository.request_task_cancellation(task_id, requested_at=now)
+
+        result = database.repository.renew_lease(
+            task_id=task_id,
+            attempt=1,
+            lease_id="renew-cancel-lease",
+            host_id=host_id,
+            lease_expires_at=now + timedelta(minutes=10),
+            now=now + timedelta(seconds=30),
+        )
+
+        assert result.status == "renewed"
+        assert result.cancel_requested is True
+    finally:
+        database.close()
+
+
+def test_expired_lease_with_pending_cancellation_resolves_to_cancelled(
+    database_url: str,
+) -> None:
+    database = CloudDatabase(database_url)
+    host_id = _unique_id("cancel-expiry-host")
+    device_id = _unique_id("cancel-expiry-device")
+    task_id = _unique_id("cancel-expiry-task")
+    now = datetime(2026, 7, 15, 8, 0, tzinfo=UTC)
+    expired_at = now + timedelta(seconds=10)
+    reaped_at = expired_at + timedelta(seconds=1)
+
+    try:
+        database.repository.upsert_host(host_id, address=None, last_seen_at=now)
+        database.repository.replace_host_devices(
+            host_id,
+            [_device(device_id, host_id)],
+        )
+        database.repository.enqueue_task(
+            ScheduledTask(
+                id=task_id,
+                goal="cancelled task must not be requeued",
+                workflow_definition_id=None,
+                constraints=TaskConstraints(),
+                created_at=now,
+            )
+        )
+        database.repository.assign_task(
+            task_id=task_id,
+            host_id=host_id,
+            device_id=device_id,
+            lease_id="cancel-then-expire-lease",
+            lease_expires_at=expired_at,
+            now=now,
+        )
+        database.repository.request_task_cancellation(task_id, requested_at=now)
+
+        reaped_task_ids = database.repository.reap_expired_leases(
+            now=reaped_at,
+            # A high attempt limit proves cancellation takes priority over retry.
+            max_attempts=5,
+        )
+        assert task_id in reaped_task_ids
+
+        task = database.repository.get_task(task_id)
+        assert task is not None
+        assert task.status == "cancelled"
+        assert task.cancel_requested_at is None
+        assert task.assigned_host_id is None
+        assert task.assigned_device_id is None
+    finally:
+        database.close()
+
+
+def test_cancel_request_rejected_on_terminal_task(database_url: str) -> None:
+    database = CloudDatabase(database_url)
+    host_id = _unique_id("cancel-terminal-host")
+    device_id = _unique_id("cancel-terminal-device")
+    task_id = _unique_id("cancel-terminal-task")
+    now = datetime(2026, 7, 15, 8, 0, tzinfo=UTC)
+
+    try:
+        database.repository.upsert_host(host_id, address=None, last_seen_at=now)
+        database.repository.replace_host_devices(
+            host_id,
+            [_device(device_id, host_id)],
+        )
+        database.repository.enqueue_task(
+            ScheduledTask(
+                id=task_id,
+                goal="already finished",
+                workflow_definition_id=None,
+                constraints=TaskConstraints(),
+                created_at=now,
+            )
+        )
+        database.repository.assign_task(
+            task_id=task_id,
+            host_id=host_id,
+            device_id=device_id,
+            lease_id="terminal-lease",
+            lease_expires_at=now + timedelta(minutes=5),
+            now=now,
+        )
+        database.repository.record_task_result(
+            task_id=task_id,
+            attempt=1,
+            lease_id="terminal-lease",
+            host_id=host_id,
+            status="done",
+            failure_reason=None,
+            terminal_result={"ok": True},
+            completed_at=now + timedelta(seconds=10),
+        )
+
+        status = database.repository.request_task_cancellation(
+            task_id, requested_at=now + timedelta(seconds=20)
+        )
+
+        assert status == "already_terminal"
+        task = database.repository.get_task(task_id)
+        assert task is not None
+        assert task.status == "done"
+    finally:
+        database.close()
+
+
+def test_cancel_request_unknown_task_not_found(database_url: str) -> None:
+    database = CloudDatabase(database_url)
+    now = datetime(2026, 7, 15, 8, 0, tzinfo=UTC)
+
+    try:
+        status = database.repository.request_task_cancellation(
+            _unique_id("missing-cancel-task"), requested_at=now
+        )
+        assert status == "not_found"
+    finally:
+        database.close()

tests/test_cloud_sdk_api.py

@@ -2,7 +2,7 @@
 
 from __future__ import annotations
 
-from datetime import UTC, datetime
+from datetime import UTC, datetime, timedelta
 
 import pytest
 
@@ -380,6 +380,7 @@ def test_submit_rejects_incomplete_or_foreign_target(tmp_path) -> None:
         ("get", "/v1/tasks", None, "tasks:read"),
         ("get", "/v1/tasks/missing/attempts", None, "tasks:read"),
         ("get", "/v1/tasks/missing/planner-decisions?attempt=0", None, "tasks:read"),
+        ("post", "/v1/tasks/missing/cancel", None, "tasks:submit"),
         ("get", "/v1/devices", None, "pool:read"),
         ("get", "/v1/hosts", None, "pool:read"),
         ("get", "/v1/plugins", None, "plugins:read"),
@@ -534,6 +535,193 @@ def test_list_task_attempts_returns_404_for_unknown_task(tmp_path) -> None:
     assert "does-not-exist" in resp.json()["detail"]
 
 
+def test_cancel_queued_task_transitions_immediately(tmp_path) -> None:
+    app, _, scheduler, _ = _build_app(tmp_path)
+    task_id = scheduler.submit(goal="cancel me")
+
+    resp = _client_for(app).post(f"/v1/tasks/{task_id}/cancel")
+
+    assert resp.status_code == 200, resp.text
+    assert resp.json() == {"task_id": task_id, "status": "cancelled"}
+    assert scheduler.store.get_task(task_id).status == "cancelled"
+
+
+def test_cancel_assigned_task_records_pending_request(tmp_path) -> None:
+    app, pool, scheduler, _ = _build_app(tmp_path)
+    pool.sync_host_devices(
+        "host-a",
+        [Device(id="device-a", driver_type="wda", status="idle")],  # type: ignore[arg-type]
+    )
+    task_id = scheduler.submit(goal="cancel me mid-flight")
+    scheduler.assign()
+
+    resp = _client_for(app).post(f"/v1/tasks/{task_id}/cancel")
+
+    assert resp.status_code == 202, resp.text
+    assert resp.json() == {"task_id": task_id, "status": "assigned"}
+    task = scheduler.store.get_task(task_id)
+    assert task.status == "assigned"
+    assert task.cancel_requested_at is not None
+
+
+def test_cancel_repeat_call_on_pending_request_is_idempotent(tmp_path) -> None:
+    app, pool, scheduler, _ = _build_app(tmp_path)
+    pool.sync_host_devices(
+        "host-a",
+        [Device(id="device-a", driver_type="wda", status="idle")],  # type: ignore[arg-type]
+    )
+    task_id = scheduler.submit(goal="cancel me twice")
+    scheduler.assign()
+    client = _client_for(app)
+
+    first = client.post(f"/v1/tasks/{task_id}/cancel")
+    second = client.post(f"/v1/tasks/{task_id}/cancel")
+
+    assert first.status_code == 202, first.text
+    assert second.status_code == 200, second.text
+    assert second.json() == {"task_id": task_id, "status": "assigned"}
+
+
+def test_cancel_repeat_call_on_already_cancelled_task_is_idempotent(
+    tmp_path,
+) -> None:
+    app, _, scheduler, _ = _build_app(tmp_path)
+    task_id = scheduler.submit(goal="cancel me twice")
+    client = _client_for(app)
+
+    first = client.post(f"/v1/tasks/{task_id}/cancel")
+    second = client.post(f"/v1/tasks/{task_id}/cancel")
+
+    assert first.status_code == 200, first.text
+    assert second.status_code == 200, second.text
+    assert second.json() == {"task_id": task_id, "status": "cancelled"}
+
+
+def test_cancel_unknown_task_returns_404(tmp_path) -> None:
+    app, _, _, _ = _build_app(tmp_path)
+    resp = _client_for(app).post("/v1/tasks/does-not-exist/cancel")
+    assert resp.status_code == 404, resp.text
+    assert "does-not-exist" in resp.json()["detail"]
+
+
+def test_cancel_terminal_task_returns_409(tmp_path) -> None:
+    app, pool, scheduler, _ = _build_app(tmp_path)
+    pool.sync_host_devices(
+        "host-a",
+        [Device(id="device-a", driver_type="wda", status="idle")],  # type: ignore[arg-type]
+    )
+    task_id = scheduler.submit(goal="finish me")
+    scheduler.assign()
+    task = scheduler.store.get_task(task_id)
+    scheduler.store.record_task_result(
+        task_id=task_id,
+        attempt=task.attempt_count,
+        lease_id=task.lease_id or "",
+        host_id=task.assigned_host_id or "",
+        status="done",
+        failure_reason=None,
+        terminal_result={"runtime_status": "completed"},
+        completed_at=datetime.now(UTC),
+    )
+
+    resp = _client_for(app).post(f"/v1/tasks/{task_id}/cancel")
+
+    assert resp.status_code == 409, resp.text
+    assert "terminal" in resp.json()["detail"]
+
+
+def test_cancel_scope_rejected_before_reaching_scheduler(tmp_path, monkeypatch) -> None:
+    provider = ConfiguredBearerAuthProvider(
+        [
+            BearerCredential(
+                principal_id="reader",
+                token="reader-token",
+                scopes=frozenset({"tasks:read"}),
+            )
+        ]
+    )
+    app, _, scheduler, _ = _build_app(tmp_path, auth_provider=provider)
+    called = False
+
+    def fail_if_called(*args, **kwargs):
+        nonlocal called
+        called = True
+        raise AssertionError("cancellation must not run before authorization")
+
+    monkeypatch.setattr(scheduler.store, "request_task_cancellation", fail_if_called)
+
+    resp = _client_for(app).post(
+        "/v1/tasks/does-not-exist/cancel",
+        headers={"Authorization": "Bearer reader-token"},
+    )
+
+    assert resp.status_code == 403
+    assert called is False
+
+
+def test_cancellation_full_path_queued_immediate_and_dispatched_collaborative(
+    tmp_path,
+) -> None:
+    """End-to-end exercise of the cancellation path (task-cancellation 9.2):
+    a queued task is cancelled immediately; a dispatched task's cancellation
+    is only recorded until the Host Agent's next lease renewal surfaces it
+    and reports back a cancelled terminal result, after which the task is
+    visible as cancelled via the public API's get and list endpoints."""
+    app, pool, scheduler, _ = _build_app(tmp_path)
+    pool.sync_host_devices(
+        "host-a",
+        [Device(id="device-a", driver_type="wda", status="idle")],  # type: ignore[arg-type]
+    )
+    client = _client_for(app)
+
+    queued_task_id = scheduler.submit(goal="cancel me while queued")
+    queued_cancel = client.post(f"/v1/tasks/{queued_task_id}/cancel")
+    assert queued_cancel.status_code == 200, queued_cancel.text
+    assert queued_cancel.json() == {"task_id": queued_task_id, "status": "cancelled"}
+
+    dispatched_task_id = scheduler.submit(goal="cancel me mid-execution")
+    scheduler.assign()
+    dispatched_cancel = client.post(f"/v1/tasks/{dispatched_task_id}/cancel")
+    assert dispatched_cancel.status_code == 202, dispatched_cancel.text
+    assert dispatched_cancel.json() == {
+        "task_id": dispatched_task_id,
+        "status": "assigned",
+    }
+
+    task = scheduler.store.get_task(dispatched_task_id)
+    renewed_at = datetime.now(UTC)
+    renewal = scheduler.store.renew_lease(
+        task_id=dispatched_task_id,
+        attempt=task.attempt_count,
+        lease_id=task.lease_id or "",
+        host_id=task.assigned_host_id or "",
+        lease_expires_at=renewed_at + timedelta(seconds=30),
+        now=renewed_at,
+    )
+    assert renewal.status == "renewed", renewal
+    assert renewal.cancel_requested is True
+
+    scheduler.store.record_task_result(
+        task_id=dispatched_task_id,
+        attempt=task.attempt_count,
+        lease_id=task.lease_id or "",
+        host_id=task.assigned_host_id or "",
+        status="cancelled",
+        failure_reason="cancellation requested by control plane",
+        terminal_result=None,
+        completed_at=datetime.now(UTC),
+    )
+
+    status_resp = client.get(f"/v1/tasks/{dispatched_task_id}")
+    assert status_resp.status_code == 200, status_resp.text
+    assert status_resp.json()["status"] == "cancelled"
+
+    list_resp = client.get("/v1/tasks", params={"status": "cancelled"})
+    assert list_resp.status_code == 200, list_resp.text
+    cancelled_ids = {t["id"] for t in list_resp.json()["items"]}
+    assert {queued_task_id, dispatched_task_id} <= cancelled_ids
+
+
 def test_plugin_admin_scope_is_checked_before_registration(
     tmp_path,
     monkeypatch,

tests/test_cloud_task_progress_visibility.py

@@ -90,7 +90,8 @@ def test_renew_lease_writes_progress_on_success(tmp_path) -> None:
             now=now + timedelta(seconds=5),
             progress=progress,
         )
-        assert result == "renewed"
+        assert result.status == "renewed"
+        assert result.cancel_requested is False
 
         task = database.repository.get_task(task_id)
         assert task is not None

tests/test_host_agent_internal_api.py

@@ -296,6 +296,192 @@ def test_host_policy_converges_and_disables_self_submission(tmp_path) -> None:
     assert disabled.status_code == 403
 
 
+def _enqueue_task_for_host(
+    pool: DevicePool,
+    *,
+    task_id: str,
+    host_id: str,
+    goal: str = "cancel me",
+) -> None:
+    pool.store.enqueue_task(
+        ScheduledTask(
+            id=task_id,
+            goal=goal,
+            workflow_definition_id=None,
+            constraints=TaskConstraints(target_host_id=host_id),
+            created_at=datetime.now(UTC),
+        )
+    )
+
+
+def _register_idle_device(
+    pool: DevicePool, *, host_id: str, device_id: str, now: datetime
+) -> None:
+    pool.store.upsert_host(host_id, address=None, last_seen_at=now)
+    pool.store.replace_host_devices(
+        host_id,
+        [
+            PooledDevice(
+                device_id=device_id,
+                host_id=host_id,
+                driver_type="wda",
+                status="idle",
+                synced_at=now,
+            )
+        ],
+    )
+
+
+def test_cancel_host_task_transitions_queued_task_to_cancelled_immediately(
+    tmp_path,
+) -> None:
+    client, pool = _build_client(tmp_path)
+    _enqueue_task_for_host(pool, task_id="task-1", host_id="host-a")
+
+    response = client.post(
+        "/internal/v1/hosts/host-a/tasks/task-1/cancel",
+        headers={"Authorization": "Bearer token-a"},
+    )
+
+    assert response.status_code == 200, response.text
+    assert response.json() == {"task_id": "task-1", "status": "cancelled"}
+    assert pool.store.get_task("task-1").status == "cancelled"  # type: ignore[union-attr]
+
+
+def test_cancel_host_task_returns_202_for_pending_assigned_task(tmp_path) -> None:
+    client, pool = _build_client(tmp_path)
+    now = datetime.now(UTC)
+    _enqueue_task_for_host(pool, task_id="task-2", host_id="host-a")
+    _register_idle_device(pool, host_id="host-a", device_id="device-a", now=now)
+    pool.store.assign_task(
+        task_id="task-2",
+        host_id="host-a",
+        device_id="device-a",
+        lease_id="lease-2",
+        lease_expires_at=now + timedelta(minutes=1),
+        now=now,
+    )
+
+    response = client.post(
+        "/internal/v1/hosts/host-a/tasks/task-2/cancel",
+        headers={"Authorization": "Bearer token-a"},
+    )
+
+    assert response.status_code == 202, response.text
+    body = response.json()
+    assert body["task_id"] == "task-2"
+    assert body["status"] == "assigned"
+    task = pool.store.get_task("task-2")
+    assert task is not None
+    assert task.cancel_requested_at is not None
+
+
+def test_cancel_host_task_repeat_call_on_pending_request_is_idempotent(
+    tmp_path,
+) -> None:
+    client, pool = _build_client(tmp_path)
+    now = datetime.now(UTC)
+    _enqueue_task_for_host(pool, task_id="task-3", host_id="host-a")
+    _register_idle_device(pool, host_id="host-a", device_id="device-a", now=now)
+    pool.store.assign_task(
+        task_id="task-3",
+        host_id="host-a",
+        device_id="device-a",
+        lease_id="lease-3",
+        lease_expires_at=now + timedelta(minutes=1),
+        now=now,
+    )
+    headers = {"Authorization": "Bearer token-a"}
+
+    first = client.post(
+        "/internal/v1/hosts/host-a/tasks/task-3/cancel", headers=headers
+    )
+    second = client.post(
+        "/internal/v1/hosts/host-a/tasks/task-3/cancel", headers=headers
+    )
+
+    assert first.status_code == 202, first.text
+    assert second.status_code == 200, second.text
+    assert second.json() == {"task_id": "task-3", "status": "assigned"}
+
+
+def test_cancel_host_task_rejects_terminal_task(tmp_path) -> None:
+    client, pool = _build_client(tmp_path)
+    now = datetime.now(UTC)
+    _enqueue_task_for_host(pool, task_id="task-4", host_id="host-a")
+    _register_idle_device(pool, host_id="host-a", device_id="device-a", now=now)
+    pool.store.assign_task(
+        task_id="task-4",
+        host_id="host-a",
+        device_id="device-a",
+        lease_id="lease-4",
+        lease_expires_at=now + timedelta(minutes=1),
+        now=now,
+    )
+    pool.store.record_task_result(
+        task_id="task-4",
+        attempt=1,
+        lease_id="lease-4",
+        host_id="host-a",
+        status="done",
+        failure_reason=None,
+        terminal_result=None,
+        completed_at=now,
+    )
+
+    response = client.post(
+        "/internal/v1/hosts/host-a/tasks/task-4/cancel",
+        headers={"Authorization": "Bearer token-a"},
+    )
+
+    assert response.status_code == 409, response.text
+
+
+def test_cancel_host_task_rejects_unknown_task_id(tmp_path) -> None:
+    client, _ = _build_client(tmp_path)
+
+    response = client.post(
+        "/internal/v1/hosts/host-a/tasks/does-not-exist/cancel",
+        headers={"Authorization": "Bearer token-a"},
+    )
+
+    assert response.status_code == 404
+
+
+def test_cancel_host_task_rejects_task_owned_by_other_host(tmp_path) -> None:
+    client, pool = _build_client(tmp_path)
+    _enqueue_task_for_host(pool, task_id="task-5", host_id="host-a")
+
+    response = client.post(
+        "/internal/v1/hosts/host-b/tasks/task-5/cancel",
+        headers={"Authorization": "Bearer token-b"},
+    )
+
+    assert response.status_code == 404
+    assert pool.store.get_task("task-5").status == "queued"  # type: ignore[union-attr]
+
+
+def test_cancel_host_task_rejects_mismatched_host_identity(tmp_path) -> None:
+    client, pool = _build_client(tmp_path)
+    _enqueue_task_for_host(pool, task_id="task-6", host_id="host-a")
+
+    response = client.post(
+        "/internal/v1/hosts/host-a/tasks/task-6/cancel",
+        headers={"Authorization": "Bearer token-b"},
+    )
+
+    assert response.status_code == 403
+    assert pool.store.get_task("task-6").status == "queued"  # type: ignore[union-attr]
+
+
+def test_cancel_host_task_requires_authentication(tmp_path) -> None:
+    client, _ = _build_client(tmp_path)
+
+    response = client.post("/internal/v1/hosts/host-a/tasks/task-7/cancel")
+
+    assert response.status_code == 401
+
+
 def test_planner_proxy_reserves_and_enforces_host_daily_token_budget(tmp_path) -> None:
     class FakePlannerClient:
         calls = 0
@@ -473,10 +659,32 @@ def test_lease_renewal_extends_active_assignment(tmp_path) -> None:
 
     assert response.status_code == 200
     assert response.json()["status"] == "renewed"
+    assert response.json()["cancel_requested"] is False
     renewed_expiry = datetime.fromisoformat(response.json()["lease_expires_at"])
     assert renewed_expiry > original_time + timedelta(seconds=30)
 
 
+def test_lease_renewal_surfaces_pending_cancellation(tmp_path) -> None:
+    client, pool = _build_client(tmp_path)
+    _seed_active_assignment(pool)
+    pool.store.request_task_cancellation("active-task", requested_at=datetime.now(UTC))
+
+    response = client.post(
+        "/internal/v1/hosts/host-a/assignments/active-task/renew",
+        headers={"Authorization": "Bearer token-a"},
+        json={
+            "host_id": "host-a",
+            "task_id": "active-task",
+            "attempt": 1,
+            "lease_id": "active-lease",
+        },
+    )
+
+    assert response.status_code == 200
+    assert response.json()["status"] == "renewed"
+    assert response.json()["cancel_requested"] is True
+
+
 def test_stale_renewal_returns_typed_conflict(tmp_path) -> None:
     client, pool = _build_client(tmp_path)
     _seed_active_assignment(pool)
@@ -526,6 +734,37 @@ def test_terminal_result_is_idempotent_through_internal_api(tmp_path) -> None:
     assert pool.store.get_task("active-task").status == "done"  # type: ignore[union-attr]
 
 
+def test_cancelled_terminal_result_is_accepted_and_idempotent(tmp_path) -> None:
+    client, pool = _build_client(tmp_path)
+    _seed_active_assignment(pool)
+    pool.store.request_task_cancellation("active-task", requested_at=datetime.now(UTC))
+    payload = {
+        "host_id": "host-a",
+        "task_id": "active-task",
+        "attempt": 1,
+        "lease_id": "active-lease",
+        "status": "cancelled",
+        "failure_reason": "cancellation requested by control plane",
+    }
+
+    first = client.post(
+        "/internal/v1/hosts/host-a/assignments/active-task/result",
+        headers={"Authorization": "Bearer token-a"},
+        json=payload,
+    )
+    repeated = client.post(
+        "/internal/v1/hosts/host-a/assignments/active-task/result",
+        headers={"Authorization": "Bearer token-a"},
+        json=payload,
+    )
+
+    assert first.status_code == 200
+    assert first.json()["status"] == "recorded"
+    assert repeated.status_code == 200
+    assert repeated.json()["status"] == "already_recorded"
+    assert pool.store.get_task("active-task").status == "cancelled"  # type: ignore[union-attr]
+
+
 def test_conflicting_repeated_result_returns_stale_lease_conflict(tmp_path) -> None:
     client, pool = _build_client(tmp_path)
     _seed_active_assignment(pool)

tests/test_host_agent_progress_reporting.py

@@ -58,6 +58,7 @@ class _FakeExecutor:
         assignment: AssignmentModel,
         *,
         should_stop: Any | None = None,
+        stop_reason: Any | None = None,
     ) -> Any:
         from host_agent.assignment import AssignmentExecutionResult
 

tests/test_task_loop.py

@@ -115,6 +115,76 @@ def test_task_runner_stops_before_the_next_planned_action() -> None:
     assert actions == ["tap"]
 
 
+def test_task_runner_stop_reason_cancellation_yields_cancelled_status() -> None:
+    scene = Scene(width=10, height=20, elements=[])
+    stop_requested = False
+
+    def record_action(**kwargs):
+        nonlocal stop_requested
+        stop_requested = True
+        return {"ok": True}
+
+    runner = TaskRunner(
+        planner=ScriptedPlanner(
+            [
+                PlannedStep(action="tap", description="first", args={}),
+                PlannedStep(action="tap", description="second", args={}),
+            ]
+        ),
+        executor=Executor(
+            tools={"tap": record_action},
+            config=ExecutorConfig(max_retries=1, backoff_seconds=0),
+        ),
+        config=TaskRunnerConfig(max_steps=5),
+        observer=lambda device_id: scene,
+        screenshot_provider=lambda device_id: PNG_10X20,
+    )
+
+    result = runner.run(
+        Task(goal="perform two actions", device_id="phone"),
+        should_stop=lambda: stop_requested,
+        stop_reason=lambda: "cancellation requested by control plane",
+    )
+
+    assert result.status == "cancelled"
+    assert result.failure_reason == "cancellation requested by control plane"
+
+
+def test_task_runner_stop_reason_lease_loss_yields_failed_status() -> None:
+    scene = Scene(width=10, height=20, elements=[])
+    stop_requested = False
+
+    def record_action(**kwargs):
+        nonlocal stop_requested
+        stop_requested = True
+        return {"ok": True}
+
+    runner = TaskRunner(
+        planner=ScriptedPlanner(
+            [
+                PlannedStep(action="tap", description="first", args={}),
+                PlannedStep(action="tap", description="second", args={}),
+            ]
+        ),
+        executor=Executor(
+            tools={"tap": record_action},
+            config=ExecutorConfig(max_retries=1, backoff_seconds=0),
+        ),
+        config=TaskRunnerConfig(max_steps=5),
+        observer=lambda device_id: scene,
+        screenshot_provider=lambda device_id: PNG_10X20,
+    )
+
+    result = runner.run(
+        Task(goal="perform two actions", device_id="phone"),
+        should_stop=lambda: stop_requested,
+        stop_reason=lambda: "lease rejected by control plane",
+    )
+
+    assert result.status == "failed"
+    assert result.failure_reason == "lease rejected by control plane"
+
+
 def test_task_runner_persists_action_evidence_and_raw_ocr(tmp_path) -> None:
     scene = Scene(
         width=10,

tests/test_workflow_runner.py

@@ -166,6 +166,82 @@ def test_workflow_runner_stops_before_the_next_step(tmp_path) -> None:
     assert calls == ["first"]
 
 
+def test_workflow_runner_stop_reason_cancellation_yields_cancelled_status(
+    tmp_path,
+) -> None:
+    stop_requested = False
+    calls: list[str] = []
+
+    class StoppingTaskRunner:
+        def run(self, task: Task, *, should_stop=None, stop_reason=None) -> Task:
+            nonlocal stop_requested
+            calls.append(task.goal)
+            stop_requested = True
+            task.status = "completed"
+            return task
+
+    definition = WorkflowDefinition(
+        name="interruptible",
+        entry_step_id="first",
+        steps=[
+            PlannedGoalStep("first", "first", next_step_id="second"),
+            PlannedGoalStep("second", "second"),
+        ],
+    )
+    runner = WorkflowRunner(
+        _store(tmp_path),
+        task_runner_factory=lambda: StoppingTaskRunner(),  # type: ignore[arg-type]
+    )
+
+    run = runner.run(
+        definition,
+        "phone",
+        should_stop=lambda: stop_requested,
+        stop_reason=lambda: "cancellation requested by control plane",
+    )
+
+    assert run.status == "cancelled"
+    assert calls == ["first"]
+
+
+def test_workflow_runner_stop_reason_lease_loss_yields_failed_status(
+    tmp_path,
+) -> None:
+    stop_requested = False
+    calls: list[str] = []
+
+    class StoppingTaskRunner:
+        def run(self, task: Task, *, should_stop=None, stop_reason=None) -> Task:
+            nonlocal stop_requested
+            calls.append(task.goal)
+            stop_requested = True
+            task.status = "completed"
+            return task
+
+    definition = WorkflowDefinition(
+        name="interruptible",
+        entry_step_id="first",
+        steps=[
+            PlannedGoalStep("first", "first", next_step_id="second"),
+            PlannedGoalStep("second", "second"),
+        ],
+    )
+    runner = WorkflowRunner(
+        _store(tmp_path),
+        task_runner_factory=lambda: StoppingTaskRunner(),  # type: ignore[arg-type]
+    )
+
+    run = runner.run(
+        definition,
+        "phone",
+        should_stop=lambda: stop_requested,
+        stop_reason=lambda: "lease rejected by control plane",
+    )
+
+    assert run.status == "failed"
+    assert calls == ["first"]
+
+
 def test_workflow_runner_failing_planned_goal_marks_run_failed(tmp_path) -> None:
     definition = WorkflowDefinition(
         name="fail",