From 6776ac2f2da7c4d1ab8abed20061ba08b791ef01 Mon Sep 17 00:00:00 2001 From: Jerry Yan <792602257@qq.com> Date: Wed, 15 Jul 2026 18:39:00 +0800 Subject: [PATCH] Add public SDK cancel endpoint and CloudClient method - POST /v1/tasks/{task_id}/cancel: tasks:submit scoped, 200 for immediate/idempotent cancellation, 202 for newly recorded pending cancellation, 404 for unknown task, 409 for terminal task. - TaskCancellationResponse{task_id, status} model. - Widen list_tasks status_filter Literal to include "cancelled". - CloudClient.cancel_task(task_id). - SDK-level tests covering queued/assigned/idempotent/404/409/scope cases for both the router and CloudClient. Task 5/9 of task-cancellation change. --- openspec/changes/task-cancellation/tasks.md | 10 +- packages/cloud-platform/cloud/sdk/api.py | 38 +++++- packages/cloud-platform/cloud/sdk/client.py | 4 + packages/cloud-platform/cloud/sdk/models.py | 5 + tests/test_cloud_client.py | 17 +++ tests/test_cloud_sdk_api.py | 125 ++++++++++++++++++++ 6 files changed, 192 insertions(+), 7 deletions(-) diff --git a/openspec/changes/task-cancellation/tasks.md b/openspec/changes/task-cancellation/tasks.md index 6803e7e..014ec3f 100644 --- a/openspec/changes/task-cancellation/tasks.md +++ b/openspec/changes/task-cancellation/tasks.md @@ -38,11 +38,11 @@ ## 5. Public SDK endpoint -- [ ] 5.1 Add `POST /v1/tasks/{task_id}/cancel` route to `cloud/sdk/api.py`, scope-gated by `TASKS_SUBMIT_SCOPE`, calling the new `TaskScheduler`/repository cancellation operation. -- [ ] 5.2 Add `TaskCancellationResponse {task_id, status}` model to `cloud/sdk/models.py` (or wherever SDK response models live); return `200 OK` for immediate/already-terminal-cancelled idempotent cases, `202 Accepted` for a newly recorded pending cancellation, `404` for unknown task id, `409 Conflict` for a `done`/`failed` task. -- [ ] 5.3 Widen the `status_filter` `Literal` on `list_tasks` (`cloud/sdk/api.py`) to include `"cancelled"`. -- [ ] 5.4 Add a `cancel_task(task_id)` method to `CloudClient` (`cloud/sdk/client.py`) mirroring the new route. -- [ ] 5.5 Add/extend SDK-level tests: submit-scoped caller cancels a queued task (200, immediate); cancels a dispatched task (202, pending); read-only-scoped caller rejected before reaching the scheduler; cancel on unknown id (404); cancel on terminal id (409); repeat cancel calls idempotent (200). +- [x] 5.1 Add `POST /v1/tasks/{task_id}/cancel` route to `cloud/sdk/api.py`, scope-gated by `TASKS_SUBMIT_SCOPE`, calling the new `TaskScheduler`/repository cancellation operation. +- [x] 5.2 Add `TaskCancellationResponse {task_id, status}` model to `cloud/sdk/models.py` (or wherever SDK response models live); return `200 OK` for immediate/already-terminal-cancelled idempotent cases, `202 Accepted` for a newly recorded pending cancellation, `404` for unknown task id, `409 Conflict` for a `done`/`failed` task. +- [x] 5.3 Widen the `status_filter` `Literal` on `list_tasks` (`cloud/sdk/api.py`) to include `"cancelled"`. +- [x] 5.4 Add a `cancel_task(task_id)` method to `CloudClient` (`cloud/sdk/client.py`) mirroring the new route. +- [x] 5.5 Add/extend SDK-level tests: submit-scoped caller cancels a queued task (200, immediate); cancels a dispatched task (202, pending); read-only-scoped caller rejected before reaching the scheduler; cancel on unknown id (404); cancel on terminal id (409); repeat cancel calls idempotent (200). ## 6. Frontend: Cloud Console diff --git a/packages/cloud-platform/cloud/sdk/api.py b/packages/cloud-platform/cloud/sdk/api.py index 9033635..01ac9ed 100644 --- a/packages/cloud-platform/cloud/sdk/api.py +++ b/packages/cloud-platform/cloud/sdk/api.py @@ -11,6 +11,7 @@ authentication can be added later without changing route signatures. from __future__ import annotations import json +from datetime import UTC, datetime from typing import TYPE_CHECKING, Callable, Literal from cloud.auth import ( @@ -31,6 +32,7 @@ from cloud.sdk.models import ( PluginRegistrationRequest, PluginResponse, TaskAttemptResponse, + TaskCancellationResponse, TaskListItem, TaskListResponse, TaskPlannerDecisionItem, @@ -39,7 +41,7 @@ from cloud.sdk.models import ( TaskSubmissionRequest, TaskSubmissionResponse, ) -from fastapi import APIRouter, HTTPException, Query, Request, status +from fastapi import APIRouter, HTTPException, Query, Request, Response, status if TYPE_CHECKING: from cloud.plugins import PluginRegistry @@ -156,7 +158,9 @@ def create_cloud_router( @router.get("/tasks", response_model=TaskListResponse) def list_tasks( request: Request, - status_filter: Literal["queued", "assigned", "dispatched", "done", "failed"] + status_filter: Literal[ + "queued", "assigned", "dispatched", "done", "failed", "cancelled" + ] | None = Query(default=None, alias="status"), limit: int = Query(default=50, ge=1, le=100), offset: int = Query(default=0, ge=0), @@ -268,6 +272,36 @@ def create_cloud_router( ) return TaskPlannerDecisionListResponse(items=items) + @router.post( + "/tasks/{task_id}/cancel", + response_model=TaskCancellationResponse, + responses={ + status.HTTP_202_ACCEPTED: {"model": TaskCancellationResponse}, + }, + ) + def cancel_task( + task_id: str, request: Request, response: Response + ) -> TaskCancellationResponse: + _authorize(request, TASKS_SUBMIT_SCOPE) + result = scheduler.store.request_task_cancellation( + task_id, requested_at=datetime.now(UTC) + ) + 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 TaskCancellationResponse(task_id=task_id, status=task.status) + @router.get("/devices", response_model=list[DeviceResponse]) def list_devices(request: Request) -> list[DeviceResponse]: _authorize(request, POOL_READ_SCOPE) diff --git a/packages/cloud-platform/cloud/sdk/client.py b/packages/cloud-platform/cloud/sdk/client.py index 5dac689..ec9343a 100644 --- a/packages/cloud-platform/cloud/sdk/client.py +++ b/packages/cloud-platform/cloud/sdk/client.py @@ -115,6 +115,10 @@ class CloudClient: resp = self._request("GET", f"/tasks/{task_id}/attempts") return resp.json() + def cancel_task(self, task_id: str) -> dict[str, Any]: + resp = self._request("POST", f"/tasks/{task_id}/cancel") + return resp.json() + # ----------------------------------------------------------------- devices def list_devices(self) -> list[dict[str, Any]]: diff --git a/packages/cloud-platform/cloud/sdk/models.py b/packages/cloud-platform/cloud/sdk/models.py index 4ae71b2..41ac260 100644 --- a/packages/cloud-platform/cloud/sdk/models.py +++ b/packages/cloud-platform/cloud/sdk/models.py @@ -25,6 +25,11 @@ class TaskSubmissionResponse(BaseModel): task_id: str +class TaskCancellationResponse(BaseModel): + task_id: str + status: str + + class TaskStatusResponse(BaseModel): id: str status: str diff --git a/tests/test_cloud_client.py b/tests/test_cloud_client.py index 56b4211..f780f81 100644 --- a/tests/test_cloud_client.py +++ b/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() == [] diff --git a/tests/test_cloud_sdk_api.py b/tests/test_cloud_sdk_api.py index 7ad0a2c..788d823 100644 --- a/tests/test_cloud_sdk_api.py +++ b/tests/test_cloud_sdk_api.py @@ -339,6 +339,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"), @@ -495,6 +496,130 @@ 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_plugin_admin_scope_is_checked_before_registration( tmp_path, monkeypatch,