From 18f053e64b1d74b958701c7871d55465f0babe1e Mon Sep 17 00:00:00 2001
From: Jerry Yan <792602257@qq.com>
Date: Wed, 15 Jul 2026 19:13:40 +0800
Subject: [PATCH] Add Host Agent local console task cancellation
(task-cancellation 7.1-7.3)
- New internal API route POST /internal/v1/hosts/{host_id}/tasks/{task_id}/cancel,
authenticated via the host's own bearer credential (authorize_host) with an
ownership check, since host tokens carry no scopes and cannot reach the
public SDK's tasks:submit-scoped cancel endpoint.
- HostAgentClient.cancel_task() calls the new internal route directly.
- create_console_app() gains a cancel_task callable with automatic default
wiring from host_client, so production app.py needs no changes.
- Local console: POST /tasks/{task_id}/cancel route resolves the local
execution id to its Cloud source_task_id before cancelling, and the task
detail page/template show a Cancel button plus notice/error banners.
- Tests across all three layers: internal API route, Jinja2 template
rendering, and FastAPI console route behavior.
---
apps/device-host-agent/host_agent/client.py | 12 ++
apps/device-host-agent/host_agent/web/app.py | 50 +++++
.../host_agent/web/templates/task_detail.html | 8 +
.../tests/host_agent/web/conftest.py | 7 +
.../tests/host_agent/web/test_templates.py | 37 ++++
apps/device-host-agent/tests/test_web_app.py | 165 +++++++++++++++-
openspec/changes/task-cancellation/tasks.md | 6 +-
.../cloud-platform/cloud/internal_api/api.py | 47 ++++-
.../cloud/internal_api/models.py | 5 +
tests/test_host_agent_internal_api.py | 186 ++++++++++++++++++
10 files changed, 518 insertions(+), 5 deletions(-)
diff --git a/apps/device-host-agent/host_agent/client.py b/apps/device-host-agent/host_agent/client.py
index 20758f3..76bf447 100644
--- a/apps/device-host-agent/host_agent/client.py
+++ b/apps/device-host-agent/host_agent/client.py
@@ -14,6 +14,7 @@ from cloud.internal_api.models import (
DeviceSnapshotModel,
HeartbeatResponse,
HostEnrollmentResponse,
+ HostTaskCancellationResponse,
HostTaskSubmissionResponse,
LeaseRenewalResponse,
TaskProgressModel,
@@ -219,6 +220,17 @@ class HostAgentClient:
"control plane returned malformed success payload"
) from exc
+ async def cancel_task(self, task_id: str) -> HostTaskCancellationResponse:
+ response = await self._client.request(
+ "POST",
+ f"/internal/v1/hosts/{self.config.host_id}/tasks/{task_id}/cancel",
+ json={"host_id": self.config.host_id},
+ headers={"Authorization": f"Bearer {self.config.token}"},
+ )
+ if not response.is_success:
+ _raise_api_error(response)
+ return HostTaskCancellationResponse.model_validate(response.json())
+
async def claim(self) -> AssignmentModel | None:
response = await self._request(
"POST",
diff --git a/apps/device-host-agent/host_agent/web/app.py b/apps/device-host-agent/host_agent/web/app.py
index af47f6b..6815795 100644
--- a/apps/device-host-agent/host_agent/web/app.py
+++ b/apps/device-host-agent/host_agent/web/app.py
@@ -41,7 +41,9 @@ CSRF_FORM_FIELD = "csrf_token"
_LOOPBACK_BIND_HOSTS = frozenset({"127.0.0.1", "localhost", "::1"})
TaskSubmissionCallable = Callable[..., Awaitable[str]]
+TaskCancellationCallable = Callable[..., Awaitable[Any]]
AUTOMATIC_DEVICE_VALUE = "__automatic__"
+_TERMINAL_LOCAL_TASK_STATUSES = frozenset({"completed", "failed", "cancelled"})
_ENV = jinja2.Environment(
loader=jinja2.FileSystemLoader(Path(__file__).parent / "templates"),
@@ -217,6 +219,7 @@ def create_console_app(
enrollment_client: HostAgentEnrollmentClient | None,
host_client: HostAgentClient | None = None,
submit_self_task: TaskSubmissionCallable | None = None,
+ cancel_task: TaskCancellationCallable | None = None,
metadata_store: TaskMetadataStore | None = None,
timeline: Timeline | None = None,
executor: AssignmentExecutor | None = None,
@@ -225,6 +228,8 @@ def create_console_app(
cookie_secure = config.console_bind_host not in _LOOPBACK_BIND_HOSTS
if submit_self_task is None and host_client is not None:
submit_self_task = host_client.submit_self_task
+ if cancel_task is None and host_client is not None:
+ cancel_task = host_client.cancel_task
submission_available = submit_self_task is not None
def _running_devices() -> list[dict[str, str]]:
@@ -709,6 +714,7 @@ def create_console_app(
@app.get("/tasks/{task_id}", response_class=HTMLResponse)
async def task_detail_page(
task_id: str,
+ request: Request,
session: SessionState = Depends(require_session),
) -> HTMLResponse:
if metadata_store is None:
@@ -738,13 +744,57 @@ def create_console_app(
if task.get(key) is not None
]
timeline_steps = [_timeline_step_context(record) for record in timeline_records]
+ can_cancel = (
+ cancel_task is not None
+ and task.get("source_task_id") is not None
+ and task.get("status") not in _TERMINAL_LOCAL_TASK_STATUSES
+ )
+ cancel_notice = (
+ "Cancellation requested. It may take a moment to take effect."
+ if request.query_params.get("cancelled") == "1"
+ else None
+ )
+ cancel_error = (
+ "Failed to request cancellation. Try again."
+ if request.query_params.get("cancel_error") == "1"
+ else None
+ )
return _render(
"task_detail.html",
title=f"Task {task_id}",
session=session,
+ csrf_token=session.csrf_token,
task=task,
task_rows=task_rows,
timeline_steps=timeline_steps,
+ can_cancel=can_cancel,
+ cancel_notice=cancel_notice,
+ cancel_error=cancel_error,
)
+ @app.post("/tasks/{task_id}/cancel")
+ async def tasks_cancel(
+ task_id: str,
+ session: SessionState = Depends(require_csrf),
+ ) -> Response:
+ if metadata_store is None:
+ raise HTTPException(
+ status_code=503, detail="task metadata store not configured"
+ )
+ task = await asyncio.to_thread(metadata_store.get_task, task_id)
+ if task is None:
+ raise HTTPException(status_code=404, detail="task not found")
+ source_task_id = task.get("source_task_id")
+ if cancel_task is None or source_task_id is None:
+ return RedirectResponse(
+ url=f"/tasks/{task_id}?cancel_error=1", status_code=303
+ )
+ try:
+ await cancel_task(source_task_id)
+ except HostAgentAPIError:
+ return RedirectResponse(
+ url=f"/tasks/{task_id}?cancel_error=1", status_code=303
+ )
+ return RedirectResponse(url=f"/tasks/{task_id}?cancelled=1", status_code=303)
+
return app
diff --git a/apps/device-host-agent/host_agent/web/templates/task_detail.html b/apps/device-host-agent/host_agent/web/templates/task_detail.html
index c474126..03c8f0f 100644
--- a/apps/device-host-agent/host_agent/web/templates/task_detail.html
+++ b/apps/device-host-agent/host_agent/web/templates/task_detail.html
@@ -42,6 +42,14 @@
| Field | Value |
|---|
{% for row in task_rows %}| {{ row[0] }} | {{ row[1] }} |
{% endfor %}
+ {% if cancel_notice %}{{ cancel_notice }}
{% endif %}
+ {% if cancel_error %}{{ cancel_error }}
{% endif %}
+ {% if can_cancel %}
+
+ {% endif %}
Timeline
{% if not timeline_steps %}
No timeline records.
diff --git a/apps/device-host-agent/tests/host_agent/web/conftest.py b/apps/device-host-agent/tests/host_agent/web/conftest.py
index 1eaee7d..f0aacb2 100644
--- a/apps/device-host-agent/tests/host_agent/web/conftest.py
+++ b/apps/device-host-agent/tests/host_agent/web/conftest.py
@@ -193,6 +193,9 @@ def make_task_detail_context(
task: dict[str, Any] | None = None,
task_rows: list[tuple[str, Any]] | None = None,
timeline_steps: list[dict[str, Any]] | None = None,
+ can_cancel: bool = False,
+ cancel_notice: str | None = None,
+ cancel_error: str | None = None,
) -> dict[str, Any]:
if task is None:
task = {
@@ -227,7 +230,11 @@ def make_task_detail_context(
return {
"title": "Task task-001",
"session": session,
+ "csrf_token": session.csrf_token,
"task": task,
"task_rows": task_rows,
"timeline_steps": timeline_steps,
+ "can_cancel": can_cancel,
+ "cancel_notice": cancel_notice,
+ "cancel_error": cancel_error,
}
diff --git a/apps/device-host-agent/tests/host_agent/web/test_templates.py b/apps/device-host-agent/tests/host_agent/web/test_templates.py
index 5be0f47..a7f38d4 100644
--- a/apps/device-host-agent/tests/host_agent/web/test_templates.py
+++ b/apps/device-host-agent/tests/host_agent/web/test_templates.py
@@ -77,6 +77,43 @@ def test_task_detail_renders(env, sample_session) -> None:
assert "Timeline
" in html
+def test_task_detail_shows_cancel_button_for_non_terminal_task(
+ env, sample_session
+) -> None:
+ html = env.get_template("task_detail.html").render(
+ **make_task_detail_context(sample_session, can_cancel=True)
+ )
+ assert 'action="/tasks/task-001/cancel"' in html
+ assert "Cancel task" in html
+
+
+def test_task_detail_hides_cancel_button_for_terminal_task(
+ env, sample_session
+) -> None:
+ html = env.get_template("task_detail.html").render(
+ **make_task_detail_context(sample_session, can_cancel=False)
+ )
+ assert 'action="/tasks/task-001/cancel"' not in html
+
+
+def test_task_detail_renders_cancel_notice_and_error(env, sample_session) -> None:
+ html = env.get_template("task_detail.html").render(
+ **make_task_detail_context(
+ sample_session,
+ cancel_notice="Cancellation requested. It may take a moment to take effect.",
+ )
+ )
+ assert 'id="cancel-notice"' in html
+
+ html = env.get_template("task_detail.html").render(
+ **make_task_detail_context(
+ sample_session,
+ cancel_error="Failed to request cancellation. Try again.",
+ )
+ )
+ assert 'id="cancel-error"' in html
+
+
# ---------------------------------------------------------------------------
# 6.4 XSS-probe tests (parametrised over templates with operator-influenced
# string fields set to )
diff --git a/apps/device-host-agent/tests/test_web_app.py b/apps/device-host-agent/tests/test_web_app.py
index 3927225..e609ac8 100644
--- a/apps/device-host-agent/tests/test_web_app.py
+++ b/apps/device-host-agent/tests/test_web_app.py
@@ -3,10 +3,12 @@ from __future__ import annotations
import re
from collections.abc import Awaitable, Callable
from datetime import UTC, datetime
+from typing import Any
from fastapi.testclient import TestClient
from cloud.internal_api.models import AssignmentModel
+from core.models import Task
from device.manager import DeviceManager
from host_agent.client import HostAgentAPIError, HostTaskSubmissionUnknownError
from host_agent.config import HostAgentConfig
@@ -22,6 +24,7 @@ from storage.task_metadata import TaskMetadataStore
CSRF_PATTERN = re.compile(r'name="csrf_token" value="([^"]+)"')
TaskSubmissionCallable = Callable[[str, str | None], Awaitable[str]]
+TaskCancellationCallable = Callable[[str], Awaitable[Any]]
def _build_client(
@@ -29,6 +32,7 @@ def _build_client(
*,
create_account: bool = True,
submit_self_task: TaskSubmissionCallable | None = None,
+ cancel_task: TaskCancellationCallable | None = None,
include_metadata_store: bool = True,
) -> tuple[TestClient, dict]:
config = HostAgentConfig(
@@ -62,6 +66,7 @@ def _build_client(
session_manager=session_manager,
enrollment_client=None,
submit_self_task=submit_self_task,
+ cancel_task=cancel_task,
metadata_store=metadata_store,
)
client = TestClient(app)
@@ -795,4 +800,162 @@ def test_submitted_redirect_does_not_include_goal_text(tmp_path) -> None:
assert "bearer-token-deadbeef" not in response.headers["location"]
follow = client.get(response.headers["location"])
assert secret_goal not in follow.text
- assert "bearer-token-deadbeef" not in follow.text
+
+
+def _make_cancellation_recorder(
+ *,
+ raise_api_error: HostAgentAPIError | None = None,
+) -> tuple[TaskCancellationCallable, dict]:
+ captured: dict = {}
+
+ async def cancel(task_id: str) -> None:
+ captured["task_id"] = task_id
+ if raise_api_error is not None:
+ raise raise_api_error
+
+ return cancel, captured
+
+
+def _seed_local_task(
+ metadata_store: TaskMetadataStore,
+ *,
+ status: str = "running",
+ source_task_id: str | None = "cloud-task-1",
+) -> str:
+ task = Task(goal="open settings", device_id="dev-1", status=status)
+ metadata_store.create_task(
+ task, source_task_id=source_task_id, source_attempt=1
+ )
+ return task.id
+
+
+def test_task_detail_page_shows_cancel_button_for_non_terminal_task(
+ tmp_path,
+) -> None:
+ cancel, _ = _make_cancellation_recorder()
+ client, context = _build_client(tmp_path, cancel_task=cancel)
+ execution_id = _seed_local_task(context["metadata_store"], status="running")
+ _login(client)
+
+ response = client.get(f"/tasks/{execution_id}")
+
+ assert response.status_code == 200
+ assert f'action="/tasks/{execution_id}/cancel"' in response.text
+ assert "Cancel task" in response.text
+
+
+def test_task_detail_page_hides_cancel_button_for_terminal_task(tmp_path) -> None:
+ cancel, _ = _make_cancellation_recorder()
+ client, context = _build_client(tmp_path, cancel_task=cancel)
+ execution_id = _seed_local_task(context["metadata_store"], status="completed")
+ _login(client)
+
+ response = client.get(f"/tasks/{execution_id}")
+
+ assert response.status_code == 200
+ assert f'action="/tasks/{execution_id}/cancel"' not in response.text
+
+
+def test_task_detail_page_hides_cancel_button_when_client_unavailable(
+ tmp_path,
+) -> None:
+ client, context = _build_client(tmp_path, cancel_task=None)
+ execution_id = _seed_local_task(context["metadata_store"], status="running")
+ _login(client)
+
+ response = client.get(f"/tasks/{execution_id}")
+
+ assert response.status_code == 200
+ assert f'action="/tasks/{execution_id}/cancel"' not in response.text
+
+
+def test_cancel_task_success_calls_client_with_cloud_task_id_and_redirects(
+ tmp_path,
+) -> None:
+ cancel, captured = _make_cancellation_recorder()
+ client, context = _build_client(tmp_path, cancel_task=cancel)
+ execution_id = _seed_local_task(
+ context["metadata_store"], status="running", source_task_id="cloud-task-99"
+ )
+ csrf_token = _login(client)
+
+ response = client.post(
+ f"/tasks/{execution_id}/cancel",
+ data={"csrf_token": csrf_token},
+ follow_redirects=False,
+ )
+
+ assert response.status_code == 303
+ assert response.headers["location"] == f"/tasks/{execution_id}?cancelled=1"
+ assert captured == {"task_id": "cloud-task-99"}
+
+ follow = client.get(response.headers["location"])
+ assert "cancel-notice" in follow.text
+
+
+def test_cancel_task_client_error_redirects_with_cancel_error(tmp_path) -> None:
+ cancel, _ = _make_cancellation_recorder(
+ raise_api_error=HostAgentAPIError(502, "control plane unavailable")
+ )
+ client, context = _build_client(tmp_path, cancel_task=cancel)
+ execution_id = _seed_local_task(context["metadata_store"], status="running")
+ csrf_token = _login(client)
+
+ response = client.post(
+ f"/tasks/{execution_id}/cancel",
+ data={"csrf_token": csrf_token},
+ follow_redirects=False,
+ )
+
+ assert response.status_code == 303
+ assert response.headers["location"] == f"/tasks/{execution_id}?cancel_error=1"
+
+ follow = client.get(response.headers["location"])
+ assert "cancel-error" in follow.text
+
+
+def test_cancel_task_unknown_execution_id_returns_404(tmp_path) -> None:
+ cancel, captured = _make_cancellation_recorder()
+ client, _ = _build_client(tmp_path, cancel_task=cancel)
+ csrf_token = _login(client)
+
+ response = client.post(
+ "/tasks/does-not-exist/cancel",
+ data={"csrf_token": csrf_token},
+ )
+
+ assert response.status_code == 404
+ assert captured == {}
+
+
+def test_unauthenticated_cancel_redirects_to_login_without_calling_client(
+ tmp_path,
+) -> None:
+ cancel, captured = _make_cancellation_recorder()
+ client, context = _build_client(tmp_path, cancel_task=cancel)
+ execution_id = _seed_local_task(context["metadata_store"], status="running")
+
+ response = client.post(
+ f"/tasks/{execution_id}/cancel",
+ data={},
+ follow_redirects=False,
+ )
+
+ assert response.status_code == 303
+ assert response.headers["location"] == "/login"
+ assert captured == {}
+
+
+def test_cancel_task_without_csrf_token_is_rejected(tmp_path) -> None:
+ cancel, captured = _make_cancellation_recorder()
+ client, context = _build_client(tmp_path, cancel_task=cancel)
+ execution_id = _seed_local_task(context["metadata_store"], status="running")
+ _login(client)
+
+ response = client.post(
+ f"/tasks/{execution_id}/cancel",
+ data={"csrf_token": "wrong-token"},
+ )
+
+ assert response.status_code == 403
+ assert captured == {}
diff --git a/openspec/changes/task-cancellation/tasks.md b/openspec/changes/task-cancellation/tasks.md
index be8bc03..d9a31d4 100644
--- a/openspec/changes/task-cancellation/tasks.md
+++ b/openspec/changes/task-cancellation/tasks.md
@@ -53,9 +53,9 @@
## 7. Frontend: Host Agent local console
-- [ ] 7.1 Add a `POST /tasks/{id}/cancel` route to the Host Agent local console (`host_agent/web/app.py`) that calls through to the same cancellation path used by the collaborative-stop mechanism for a locally-tracked task, consistent with existing local console read routes.
-- [ ] 7.2 Add a Cancel button to `templates/task_detail.html` for tasks not yet in a terminal state.
-- [ ] 7.3 Add/extend local console tests covering the new route and template rendering.
+- [x] 7.1 Add a `POST /tasks/{id}/cancel` route to the Host Agent local console (`host_agent/web/app.py`) that calls through to the same cancellation path used by the collaborative-stop mechanism for a locally-tracked task, consistent with existing local console read routes. (Correction during implementation: the Host Agent's own internal-API bearer credential — issued by `RepositoryHostAuthProvider` — carries an empty `scopes` frozenset and is authorized only via `Principal.require_host()` identity checks, not scopes. It therefore cannot call the public SDK's `tasks:submit`-scoped `POST /v1/tasks/{task_id}/cancel` endpoint from design.md/section 5. Added a new internal API route, `POST /internal/v1/hosts/{host_id}/tasks/{task_id}/cancel` (`cloud/internal_api/api.py`), authenticated the same way as the existing host self-submission route (`authorize_host`), with an added ownership check rejecting tasks whose `constraints.target_host_id != host_id` with 404. This mirrors the pre-existing pattern where hosts already self-serve create/execute their own tasks over this credential, and does not conflict with design.md's Non-Goal — that constraint (no per-task ACL; anyone with `tasks:submit` can cancel any task) is scoped to the public SDK layer, not the internal Host↔Cloud API. `HostAgentClient.cancel_task()` calls this new route directly rather than proxying to the public SDK.)
+- [x] 7.2 Add a Cancel button to `templates/task_detail.html` for tasks not yet in a terminal state.
+- [x] 7.3 Add/extend local console tests covering the new route and template rendering.
## 8. Docs
diff --git a/packages/cloud-platform/cloud/internal_api/api.py b/packages/cloud-platform/cloud/internal_api/api.py
index 1bbaf74..4dc2795 100644
--- a/packages/cloud-platform/cloud/internal_api/api.py
+++ b/packages/cloud-platform/cloud/internal_api/api.py
@@ -10,7 +10,7 @@ from time import monotonic
from typing import TYPE_CHECKING
from uuid import uuid4
-from fastapi import APIRouter, HTTPException, Request, status
+from fastapi import APIRouter, HTTPException, Request, Response, status
from fastapi.responses import JSONResponse
from cloud.auth import (
@@ -29,6 +29,7 @@ from cloud.internal_api.models import (
HostGovernancePolicyModel,
HostEnrollmentRequest,
HostEnrollmentResponse,
+ HostTaskCancellationResponse,
HostTaskSubmissionRequest,
HostTaskSubmissionResponse,
LeaseRenewalRequest,
@@ -250,6 +251,50 @@ def create_internal_router(
)
return HostTaskSubmissionResponse(task_id=task_id)
+ @router.post(
+ "/hosts/{host_id}/tasks/{task_id}/cancel",
+ response_model=HostTaskCancellationResponse,
+ responses={
+ status.HTTP_202_ACCEPTED: {"model": HostTaskCancellationResponse},
+ },
+ )
+ def cancel_host_task(
+ host_id: str,
+ task_id: str,
+ request: Request,
+ response: Response,
+ ) -> HostTaskCancellationResponse:
+ authorize_host(request, host_id)
+ if scheduler is None:
+ raise HTTPException(
+ status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+ detail="task cancellation is unavailable",
+ )
+ task = scheduler.store.get_task(task_id)
+ if task is None or task.constraints.target_host_id != host_id:
+ raise HTTPException(
+ status_code=status.HTTP_404_NOT_FOUND,
+ detail=f"task {task_id!r} not found",
+ )
+ result = scheduler.store.request_task_cancellation(
+ task_id, requested_at=utc_now()
+ )
+ if result == "not_found":
+ raise HTTPException(
+ status_code=status.HTTP_404_NOT_FOUND,
+ detail=f"task {task_id!r} not found",
+ )
+ if result == "already_terminal":
+ raise HTTPException(
+ status_code=status.HTTP_409_CONFLICT,
+ detail=f"task {task_id!r} has already reached a terminal state",
+ )
+ task = scheduler.store.get_task(task_id)
+ assert task is not None
+ if result == "requested" and task.status != "cancelled":
+ response.status_code = status.HTTP_202_ACCEPTED
+ return HostTaskCancellationResponse(task_id=task_id, status=task.status)
+
@router.post(
"/hosts/{host_id}/assignments/claim",
response_model=ClaimResponse,
diff --git a/packages/cloud-platform/cloud/internal_api/models.py b/packages/cloud-platform/cloud/internal_api/models.py
index 8104ed2..19d85e7 100644
--- a/packages/cloud-platform/cloud/internal_api/models.py
+++ b/packages/cloud-platform/cloud/internal_api/models.py
@@ -122,6 +122,11 @@ class HostTaskSubmissionResponse(BaseModel):
task_id: str
+class HostTaskCancellationResponse(BaseModel):
+ task_id: str
+ status: str
+
+
class StaleLeaseConflict(BaseModel):
code: Literal["stale_lease"] = "stale_lease"
detail: str
diff --git a/tests/test_host_agent_internal_api.py b/tests/test_host_agent_internal_api.py
index ae71dd3..c90cdee 100644
--- a/tests/test_host_agent_internal_api.py
+++ b/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