diff --git a/apps/device-host-agent/host_agent/assignment.py b/apps/device-host-agent/host_agent/assignment.py index 448d202..42d3146 100644 --- a/apps/device-host-agent/host_agent/assignment.py +++ b/apps/device-host-agent/host_agent/assignment.py @@ -9,6 +9,7 @@ from core.models import Task from host_agent.execution import ExecutionFactories from host_agent.planner_context import bind_planner_execution_context from host_agent.progress import TaskProgressHolder, TaskProgressSnapshot +from runtime.task import is_cancellation_reason @dataclass(frozen=True) @@ -32,18 +33,24 @@ class AssignmentExecutor: assignment: AssignmentModel, *, should_stop: Callable[[], bool] | None = None, + stop_reason: Callable[[], str | None] | None = None, ) -> AssignmentExecutionResult: self._progress.clear() with bind_planner_execution_context(assignment): if should_stop is not None and should_stop(): + reason = stop_reason() if stop_reason is not None else None return AssignmentExecutionResult( - status="failed", - failure_reason="execution interrupted", + status="cancelled" if is_cancellation_reason(reason) else "failed", + failure_reason=reason or "execution interrupted", ) if assignment.workflow_definition_id is not None: - return self._execute_workflow(assignment, should_stop=should_stop) + return self._execute_workflow( + assignment, should_stop=should_stop, stop_reason=stop_reason + ) if assignment.goal is not None: - return self._execute_goal(assignment, should_stop=should_stop) + return self._execute_goal( + assignment, should_stop=should_stop, stop_reason=stop_reason + ) return AssignmentExecutionResult( status="failed", failure_reason="assignment has neither goal nor workflow definition", @@ -54,6 +61,7 @@ class AssignmentExecutor: assignment: AssignmentModel, *, should_stop: Callable[[], bool] | None, + stop_reason: Callable[[], str | None] | None, ) -> AssignmentExecutionResult: task = Task(goal=assignment.goal or "", device_id=assignment.device_id) if self.factories.metadata_store is not None: @@ -67,9 +75,11 @@ class AssignmentExecutor: if should_stop is None: completed = runner.run(task) else: - completed = runner.run(task, should_stop=should_stop) + completed = runner.run( + task, should_stop=should_stop, stop_reason=stop_reason + ) return AssignmentExecutionResult( - status="done" if completed.status == "completed" else "failed", + status=_terminal_status(completed.status), failure_reason=completed.failure_reason, metadata={ "runtime_task_id": completed.id, @@ -82,6 +92,7 @@ class AssignmentExecutor: assignment: AssignmentModel, *, should_stop: Callable[[], bool] | None, + stop_reason: Callable[[], str | None] | None, ) -> AssignmentExecutionResult: definition_id = assignment.workflow_definition_id or "" definition = self.factories.workflow_store.get_definition(definition_id) @@ -98,9 +109,10 @@ class AssignmentExecutor: definition, device_id=assignment.device_id, should_stop=should_stop, + stop_reason=stop_reason, ) return AssignmentExecutionResult( - status="done" if run.status == "completed" else "failed", + status=_terminal_status(run.status), failure_reason=( None if run.status == "completed" else f"workflow ended as {run.status}" ), @@ -109,3 +121,11 @@ class AssignmentExecutor: "workflow_status": run.status, }, ) + + +def _terminal_status(runtime_status: str) -> str: + if runtime_status == "completed": + return "done" + if runtime_status == "cancelled": + return "cancelled" + return "failed" 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/lease.py b/apps/device-host-agent/host_agent/lease.py index 29e9e1c..b8bd95c 100644 --- a/apps/device-host-agent/host_agent/lease.py +++ b/apps/device-host-agent/host_agent/lease.py @@ -12,6 +12,7 @@ from cloud.internal_api.models import AssignmentModel from host_agent.assignment import AssignmentExecutionResult from host_agent.client import HostAgentAPIError, HostAgentClient, StaleLeaseError from host_agent.progress import TaskProgressSnapshot +from runtime.task import is_cancellation_reason class InterruptibleAssignmentExecutor(Protocol): @@ -20,6 +21,7 @@ class InterruptibleAssignmentExecutor(Protocol): assignment: AssignmentModel, *, should_stop: Callable[[], bool] | None = None, + stop_reason: Callable[[], str | None] | None = None, ) -> AssignmentExecutionResult: ... def latest_progress(self) -> TaskProgressSnapshot | None: ... @@ -36,6 +38,10 @@ class LeaseGuard: with self._lock: return self._reason + @property + def is_cancellation(self) -> bool: + return is_cancellation_reason(self.reason) + def is_lost(self) -> bool: return self._lost.is_set() @@ -69,6 +75,7 @@ class ActiveAssignmentRunner: self.executor.execute, assignment, should_stop=lambda: guard.is_lost() or self._stop_requested.is_set(), + stop_reason=lambda: guard.reason, ) ) renewal = asyncio.create_task( @@ -108,4 +115,7 @@ class ActiveAssignmentRunner: guard.mark_lost("lease renewal failed after transport retries") return else: + if response.cancel_requested: + guard.mark_lost("cancellation requested by control plane") + return lease_expires_at = response.lease_expires_at diff --git a/apps/device-host-agent/host_agent/processor.py b/apps/device-host-agent/host_agent/processor.py index d17f4e2..5e068c7 100644 --- a/apps/device-host-agent/host_agent/processor.py +++ b/apps/device-host-agent/host_agent/processor.py @@ -47,8 +47,11 @@ class AssignmentProcessor: self.status_tracker.mark_assignment_started(assignment) try: 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 + if execution.status in {"done", "cancelled"}: + status = execution.status + else: + status = "failed" + failure_reason = execution.failure_reason if status != "done" else None response = await self.client.report_result( assignment, status=status, 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 e5ea7b6..158316f 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 @@ FieldValue {% 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_assignment.py b/apps/device-host-agent/tests/test_assignment.py index b6b6162..9818ce2 100644 --- a/apps/device-host-agent/tests/test_assignment.py +++ b/apps/device-host-agent/tests/test_assignment.py @@ -78,6 +78,39 @@ def test_goal_assignment_preserves_runtime_failure_reason() -> None: assert result.failure_reason == "planner unavailable" +def test_goal_assignment_maps_cancellation_stop_to_cancelled_status() -> None: + # First should_stop() call is Executor.execute()'s pre-flight check (must pass + # through so the runner is actually invoked); the runner's own loop then stops. + calls = {"count": 0} + + def should_stop() -> bool: + calls["count"] += 1 + return calls["count"] > 1 + + class FakeTaskRunner: + def run(self, task: Task, *, should_stop=None, stop_reason=None) -> Task: + assert should_stop is not None and should_stop() + assert stop_reason is not None + task.status = "cancelled" + task.failure_reason = stop_reason() + return task + + factories = ExecutionFactories( + task_runner_factory=lambda: FakeTaskRunner(), # type: ignore[arg-type,return-value] + workflow_runner_factory=lambda: object(), # type: ignore[arg-type,return-value] + workflow_store=object(), # type: ignore[arg-type] + ) + + result = AssignmentExecutor(factories).execute( + _assignment(), + should_stop=should_stop, + stop_reason=lambda: "cancellation requested by control plane", + ) + + assert result.status == "cancelled" + assert result.failure_reason == "cancellation requested by control plane" + + def test_workflow_assignment_loads_and_executes_definition() -> None: definition = object() calls: list[tuple[object, str]] = [] @@ -109,6 +142,44 @@ def test_workflow_assignment_loads_and_executes_definition() -> None: } +def test_workflow_assignment_maps_cancellation_stop_to_cancelled_status() -> None: + calls = {"count": 0} + + def should_stop() -> bool: + calls["count"] += 1 + return calls["count"] > 1 + + class FakeWorkflowStore: + def get_definition(self, definition_id: str): + return object() if definition_id == "workflow-a" else None + + class FakeWorkflowRunner: + def run(self, loaded_definition, device_id: str, *, should_stop=None, stop_reason=None): + assert should_stop is not None and should_stop() + assert stop_reason is not None + return SimpleNamespace( + id="run-a", status="cancelled", failure_reason=stop_reason() + ) + + factories = ExecutionFactories( + task_runner_factory=lambda: object(), # type: ignore[arg-type,return-value] + workflow_runner_factory=lambda: FakeWorkflowRunner(), # type: ignore[arg-type,return-value] + workflow_store=FakeWorkflowStore(), # type: ignore[arg-type] + ) + + result = AssignmentExecutor(factories).execute( + _assignment(goal=None, workflow_definition_id="workflow-a"), + should_stop=should_stop, + stop_reason=lambda: "cancellation requested by control plane", + ) + + assert result.status == "cancelled" + assert result.metadata == { + "workflow_run_id": "run-a", + "workflow_status": "cancelled", + } + + def test_unknown_workflow_fails_without_running() -> None: class FakeWorkflowStore: def get_definition(self, definition_id: str): diff --git a/apps/device-host-agent/tests/test_client.py b/apps/device-host-agent/tests/test_client.py index 1f84ead..e8c70bd 100644 --- a/apps/device-host-agent/tests/test_client.py +++ b/apps/device-host-agent/tests/test_client.py @@ -140,6 +140,30 @@ def test_stale_lease_response_raises_typed_error_without_retry() -> None: assert attempts == 1 +def test_renew_deserializes_cancel_requested_flag() -> None: + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response( + 200, + json={ + "status": "renewed", + "lease_expires_at": "2026-07-12T00:05:00Z", + "cancel_requested": True, + }, + ) + + 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) + response = await client.renew(_assignment()) + assert response.status == "renewed" + assert response.cancel_requested is True + + asyncio.run(scenario()) + + def test_result_report_retries_identical_payload_after_response_loss() -> None: payloads: list[dict[str, object]] = [] diff --git a/apps/device-host-agent/tests/test_lease.py b/apps/device-host-agent/tests/test_lease.py index e21cc45..03f2aa6 100644 --- a/apps/device-host-agent/tests/test_lease.py +++ b/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(): diff --git a/apps/device-host-agent/tests/test_processor.py b/apps/device-host-agent/tests/test_processor.py index c0f8be2..c92ad6b 100644 --- a/apps/device-host-agent/tests/test_processor.py +++ b/apps/device-host-agent/tests/test_processor.py @@ -88,6 +88,38 @@ def test_processor_preserves_runtime_failure_reason() -> None: asyncio.run(scenario()) +def test_processor_reports_cancelled_status_with_reason() -> None: + async def scenario() -> None: + reports: list[dict[str, object]] = [] + + class CancelledExecutor: + async def run(self, assignment): + return AssignmentExecutionResult( + status="cancelled", + failure_reason="cancellation requested by control plane", + metadata={"runtime_status": "cancelled"}, + ) + + class RecordingClient: + async def report_result(self, assignment, **kwargs): + reports.append(kwargs) + return TerminalResultResponse(status="recorded") + + result = await AssignmentProcessor( + RecordingClient(), # type: ignore[arg-type] + CancelledExecutor(), + ).process(_assignment()) + + assert result.report_status == "recorded" + assert reports[0] == { + "status": "cancelled", + "failure_reason": "cancellation requested by control plane", + "result": {"runtime_status": "cancelled"}, + } + + asyncio.run(scenario()) + + def test_status_tracker_sees_started_then_finished_even_on_raise() -> None: async def scenario() -> None: tracker = AgentStatusTracker() 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/cloud-console/src/api.ts b/cloud-console/src/api.ts index 44ab4ab..523b8b1 100644 --- a/cloud-console/src/api.ts +++ b/cloud-console/src/api.ts @@ -13,6 +13,7 @@ import type { PluginRecord, PluginRegistrationPayload, TaskAttempt, + TaskCancellationResponse, TaskListResponse, TaskSubmissionPayload, TaskStatus, @@ -239,6 +240,13 @@ export function getTaskAttempts(taskId: string): Promise { return request(`/v1/tasks/${encodeURIComponent(taskId)}/attempts`); } +export function cancelTask(taskId: string): Promise { + return request( + `/v1/tasks/${encodeURIComponent(taskId)}/cancel`, + { method: "POST" }, + ); +} + export function getTaskPlannerDecisions( taskId: string, attempt: number, diff --git a/cloud-console/src/taskCancellation.test.ts b/cloud-console/src/taskCancellation.test.ts new file mode 100644 index 0000000..27f8e36 --- /dev/null +++ b/cloud-console/src/taskCancellation.test.ts @@ -0,0 +1,24 @@ +import { describe, expect, it } from "vitest"; + +import { canCancelTask } from "./taskCancellation"; +import type { TaskStatus } from "./types"; + +describe("canCancelTask", () => { + it.each(["queued", "assigned", "dispatched"])( + "allows cancelling a %s task when the caller can submit", + (status) => { + expect(canCancelTask(status, true)).toBe(true); + }, + ); + + it.each(["done", "failed", "cancelled"])( + "refuses to cancel a terminal %s task even when the caller can submit", + (status) => { + expect(canCancelTask(status, true)).toBe(false); + }, + ); + + it("refuses to cancel a cancellable task when the caller lacks submit permission", () => { + expect(canCancelTask("assigned", false)).toBe(false); + }); +}); diff --git a/cloud-console/src/taskCancellation.ts b/cloud-console/src/taskCancellation.ts new file mode 100644 index 0000000..3a56184 --- /dev/null +++ b/cloud-console/src/taskCancellation.ts @@ -0,0 +1,20 @@ +import type { TaskStatus } from "./types"; + +/** + * Statuses for which cancellation is still meaningful: the task has not yet + * reached a terminal state. `cancelled` itself is excluded so a task can't be + * cancelled twice through the UI. + */ +const CANCELLABLE_STATUSES: ReadonlySet = new Set([ + "queued", + "assigned", + "dispatched", +]); + +/** + * Whether the Cancel action should be shown/enabled for a task, given the + * caller's submit permission and the task's current status. + */ +export function canCancelTask(status: TaskStatus, canSubmit: boolean): boolean { + return canSubmit && CANCELLABLE_STATUSES.has(status); +} diff --git a/cloud-console/src/types.ts b/cloud-console/src/types.ts index 4910e2c..dbec3e0 100644 --- a/cloud-console/src/types.ts +++ b/cloud-console/src/types.ts @@ -3,7 +3,8 @@ export type TaskStatus = | "assigned" | "dispatched" | "done" - | "failed"; + | "failed" + | "cancelled"; export interface TaskListItem { id: string; @@ -41,6 +42,11 @@ export interface TaskListResponse { offset: number; } +export interface TaskCancellationResponse { + task_id: string; + status: TaskStatus; +} + export interface TaskAttempt { task_id: string; attempt: number; diff --git a/cloud-console/src/views/TasksView.vue b/cloud-console/src/views/TasksView.vue index 77d6aa1..a94d099 100644 --- a/cloud-console/src/views/TasksView.vue +++ b/cloud-console/src/views/TasksView.vue @@ -3,6 +3,7 @@ import { computed, onMounted, ref, watch } from "vue"; import { LoaderCircle, RefreshCw } from "@lucide/vue"; import { CloudApiError, + cancelTask, getTaskAttempts, getTaskPlannerDecisions, listTasks, @@ -20,6 +21,7 @@ import type { PlannerDecisionItem, } from "../types"; import { formatTaskProgress } from "../taskProgress"; +import { canCancelTask } from "../taskCancellation"; import { computePlannerHistoryState } from "../plannerHistory"; const props = defineProps<{ canSubmit: boolean }>(); @@ -30,6 +32,7 @@ const STATUSES: TaskStatus[] = [ "dispatched", "done", "failed", + "cancelled", ]; const statusFilter = ref(""); @@ -55,6 +58,7 @@ const devices = ref([]); const plannerDecisions = ref([]); const plannerLoading = ref(false); const plannerError = ref(""); +const cancelling = ref(false); const availableDevices = computed(() => devices.value.filter((device) => device.host_id === submitHostId.value), ); @@ -185,6 +189,21 @@ async function selectTask(task: TaskListItem) { } } +async function cancelSelectedTask() { + if (!selectedTask.value) return; + cancelling.value = true; + errorMessage.value = ""; + try { + const response = await cancelTask(selectedTask.value.id); + selectedTask.value = { ...selectedTask.value, status: response.status }; + await refresh(); + } catch (err) { + handleError(err, "failed to cancel task"); + } finally { + cancelling.value = false; + } +} + function handleError(err: unknown, fallback: string) { if (err instanceof CloudApiError) { errorMessage.value = err.message; @@ -257,6 +276,11 @@ const selectedTaskProgress = computed(() => selectedTask.value ? formatTaskProgress(selectedTask.value) : null, ); +const canCancelSelectedTask = computed( + () => + !!selectedTask.value && canCancelTask(selectedTask.value.status, props.canSubmit), +); + const selectedHostTransport = computed<"direct" | "cloud" | null>(() => { if (!selectedTask.value?.assigned_host_id) return null; const host = hosts.value.find( @@ -417,6 +441,13 @@ function formatArguments(args: Record): string {

Task {{ selectedTask.id.slice(0, 8) }}

+

diff --git a/docs/CLOUD_DEPLOYMENT.md b/docs/CLOUD_DEPLOYMENT.md index 72eac2e..8e395b2 100644 --- a/docs/CLOUD_DEPLOYMENT.md +++ b/docs/CLOUD_DEPLOYMENT.md @@ -540,6 +540,14 @@ device workflows to tolerate repeated actions when the target operation allows it. Do not use this release for operations that require a transactional exactly-once guarantee across the cloud database and an external device. +Task cancellation is collaborative, not instantaneous, for tasks that have +already left the queue. Cancelling a `queued` task takes effect immediately. +Cancelling an `assigned`/`dispatched` task only records the request; the +owning Host Agent learns about it at its next lease renewal (at most roughly +one third of `lease_duration_seconds`, the same interval used for lease-loss +detection) and then stops at the next cooperative checkpoint. As with lease +loss, an action already sent to a device cannot be rolled back mid-flight. + ## Shutdown And Rollback For a normal shutdown, stop Host Agents first so they stop polling, interrupt diff --git a/openspec/changes/task-cancellation/.openspec.yaml b/openspec/changes/task-cancellation/.openspec.yaml new file mode 100644 index 0000000..4f63482 --- /dev/null +++ b/openspec/changes/task-cancellation/.openspec.yaml @@ -0,0 +1,2 @@ +schema: spec-driven +created: 2026-07-15 diff --git a/openspec/changes/task-cancellation/design.md b/openspec/changes/task-cancellation/design.md new file mode 100644 index 0000000..054f962 --- /dev/null +++ b/openspec/changes/task-cancellation/design.md @@ -0,0 +1,266 @@ +## Context + +Task cancellation has no implementation anywhere in the stack today: + +- `core/models.py`'s `TaskStatus` reserves `"cancelled"` but no code path assigns it. + `runtime/task.py::TaskRunner._interrupt_task()` marks `should_stop`-triggered + interruption as `status="failed", failure_reason="execution interrupted"` — this is the + only existing consumer of `should_stop`, and it conflates "lease was lost" with "user + asked to stop" (both produce `failed`). +- `cloud/scheduler.py`'s `ScheduledTaskStatus` is `queued | assigned | dispatched | done | + failed` — no `cancelled`, no cancel operation on `TaskScheduler` or `CloudRepository`. +- The internal Host↔Cloud protocol (`cloud/internal_api/models.py`) has exactly one + channel from Cloud back to Host during execution: `LeaseRenewalResponse`, currently + `{status: Literal["renewed"], lease_expires_at: datetime}`. There is no push channel; + the Host Agent is purely outbound (heartbeat, claim long-poll, renew, report result). +- `workflow/` already has the target shape: `WorkflowRunStatus` includes `cancelled`, + `WorkflowRunner._drive()` checks `should_stop()` at each step boundary and calls + `self._checkpoint(run, ..., "cancelled")`. This design generalizes that pattern to + `TaskRunner`/`ScheduledTask` rather than inventing a new one. +- Three prior changes (`cloud-console`, `cloud-control-plane-integration`, + `cloud-console-governance`) explicitly deferred cancellation, citing "no + scheduler/repository operation for this exists." This change adds that operation. + +Existing collaborative-stop infrastructure this design reuses instead of replacing: +`ActiveAssignmentRunner.run()` (`apps/device-host-agent/host_agent/lease.py`) races +execution against a renewal loop; the renewal loop is the only place the Host Agent +talks to Cloud while a task is running. `should_stop` already flows +`ActiveAssignmentRunner` → `AssignmentExecutor.execute()` → +`TaskRunner.run()`/`WorkflowRunner.run()`, checked at step boundaries (never mid-step). + +## Goals / Non-Goals + +**Goals:** +- Let an authorized caller cancel a task in `queued`, `assigned`, or `dispatched` state + through the public SDK. +- For `queued` tasks (never dispatched to a Host), cancellation is synchronous and + immediate — no Host round-trip needed. +- For `assigned`/`dispatched` tasks (a Host may be actively executing), cancellation is + collaborative: the Host learns about it at its next lease renewal (≤ ~1/3 of + `lease_duration_seconds`, matching the existing lease-loss detection latency) and stops + at the next step boundary, exactly like a lost lease does today. +- Make `cancelled` a real, reachable terminal state end-to-end: `core/models.py`'s + `TaskStatus`, `cloud/scheduler.py`'s `ScheduledTaskStatus`, the Cloud Console, and the + Host Agent local console. +- Distinguish "cancelled by request" from "failed" in every layer that currently reports + `should_stop`-triggered stops as generic failure, so operators can tell the two apart. +- Cancellation is idempotent: cancelling an already-cancelled or already-terminal task is + a no-op with a clear response, not an error that implies something changed. + +**Non-Goals:** +- No mid-step interruption. A single in-flight action (a tap, an LLM planning call) is + never aborted mid-flight; the existing step-boundary granularity of `should_stop` is + unchanged. A slow single step still finishes before a `dispatched` task's cancellation + takes effect. +- No forced/instant kill of a Host Agent process or its OS-level subprocess tree. This is + cooperative cancellation only, consistent with the project's existing lease-loss + handling — not a new capability class. +- No per-task ownership/ACL model. The Cloud repository does not track which principal + submitted a task; cancel authorization reuses the existing `tasks:submit` scope + (whoever can submit a task can cancel any task). Adding submitter-scoped authorization + is a distinct governance change, out of scope here (parallels + `cloud-console-governance`'s existing target-based, not ownership-based, model). +- No retry-from-UI or task editing. Only stopping a task early; resubmission remains a + separate, already-existing "submit a new task" action. +- No cancellation of individual workflow steps independent of the whole run; workflow + cancellation continues to mean "stop the run," which `workflow/` already implements. +- No change to `TaskDispatcher` (`cloud/dispatch.py`) — confirmed dead/dev-only code per + prior investigation; not worth extending for cancellation. + +## Decisions + +### D1: `LeaseRenewalResponse.cancel_requested: bool` is the only new wire signal + +Rejected alternatives: (a) a new dedicated Host-polled "check cancellation" endpoint — +adds a second polling loop and a second latency bound to reason about, when renewal +already runs on a well-understood cadence; (b) a push/webhook mechanism — breaks the +explicit "Host Agent operation requires no inbound cloud connection" requirement in +`host-agent-protocol`'s spec, which is a hard architectural constraint, not just current +practice. + +`cancel_requested` defaults to `False`. When the Cloud repository's `renew_lease` finds +the active `ScheduledTask` has a pending cancellation record, it still renews the lease +normally (a task must keep a live lease while the Host works through shutdown) but flags +`cancel_requested=True` in the response. The Host Agent's `_renew_while_running` loop +treats this the same way it treats `StaleLeaseError`: mark the `LeaseGuard` (extended +with a distinct reason string, e.g. `"cancellation requested by control plane"`) and let +the existing `should_stop` composite (`guard.is_lost() or self._stop_requested.is_set()`) +do the rest — no new boolean plumbed through `ActiveAssignmentRunner`/`AssignmentExecutor` +signatures. + +### D2: Cancellation request is durable state, not a fire-and-forget signal + +A `POST .../cancel` on an `assigned`/`dispatched` task writes a `cancel_requested_at` +timestamp (see D5 schema) rather than only flipping an in-memory flag or directly setting +`status="cancelled"`. Rationale: the Host may not renew again for up to +`lease_duration_seconds / 3`; if the Cloud process restarts in that window, an in-memory +signal would be lost silently, defeating the cancellation with no user-visible error. A +durable row survives restarts and is what `renew_lease` reads. + +The scheduled task's `status` only transitions to `cancelled` once the Host has actually +stopped and reported it (D3) — or, for the immediate `queued` case, synchronously in the +same request. This means `assigned`/`dispatched` tasks pass through an intermediate, +observable "cancellation pending" state (surfaced to callers as the existing `status` +value plus a non-null `cancel_requested_at`, not a new status literal — see D6) rather +than jumping straight to `cancelled` before the Host has confirmed. + +### D3: Host reports `cancelled` as a new terminal outcome via the existing result-report endpoint + +`TerminalResultRequest.status` (`internal_api/models.py`) is `Literal["done", "failed"]`. +This design adds `"cancelled"` to that literal rather than inventing a parallel +cancellation-report endpoint, because `report_result`'s idempotency/lease-validation +logic (`record_task_result` in `sql_repository.py`) already handles exactly the +concurrency shape needed (attempt/lease-id conflict checks, `already_recorded` replay +safety) — duplicating it for a cancel-specific endpoint would be pure risk with no +benefit. `AssignmentExecutionResult.status` (`host_agent/assignment.py`) similarly gains +a `"cancelled"` value alongside `"done"`/`"failed"`, set when `TaskRunner.run()` / +`WorkflowRunner.run()` returns because of a cancellation-flavored stop rather than a lost +lease or genuine step failure (see D4). + +### D4: `should_stop` becomes a richer signal than a bare boolean at the `TaskRunner` boundary + +Today `StopRequested = Callable[[], bool]`. Distinguishing "cancelled" from "lease +lost"/"shutdown requested" only matters for the *reason recorded on the terminal +status* — the control-flow behavior (stop at the next step boundary) is identical. Rather +than changing the `should_stop` callable's signature (which would ripple through +`workflow/runner.py`, tests, and every caller), `TaskRunner._interrupt_task()` and +`WorkflowRunner._drive()`'s stop branch gain an optional second callable, +`stop_reason: Callable[[], str] | None`, defaulting to the existing generic +`"execution interrupted"` string when absent. `ActiveAssignmentRunner` supplies a +`stop_reason` that reads `LeaseGuard.reason` (already a string field) so `"cancellation +requested by control plane"` vs `"lease rejected by control plane"` flows through +unchanged plumbing. `TaskRunner._interrupt_task()` sets `status="cancelled"` when the +reason string indicates cancellation, else keeps `status="failed"` (lease loss remains a +failure, not a cancellation, matching current behavior for that case). This is the +narrowest change that gets a real `cancelled` status without touching every +`should_stop`-typed parameter across the codebase. + +Alternative considered and rejected: add a third `CancelRequested` callable parallel to +`should_stop`. Rejected because it doubles the number of callables threaded through +`ActiveAssignmentRunner → AssignmentExecutor → TaskRunner/WorkflowRunner` for +information that's only needed at the one moment execution actually stops. + +### D5: Schema — one new nullable column pair on `scheduled_tasks`, no new table + +`packages/cloud-platform/cloud/migrations/versions/0012_task_cancellation.py` adds to +`ScheduledTaskRow`/`scheduled_tasks`: +- `cancel_requested_at: str | None` (ISO datetime, nullable) — set when a cancel request + is recorded against an `assigned`/`dispatched` task; cleared (`NULL`) when the task + reaches any terminal status, so a subsequent, different attempt of the same task id (if + retry-on-lease-expiry logic in `reap_expired_leases` requeues it) doesn't inherit a + stale cancellation. + +No new table: the volume and access pattern (one pending value per task, read on every +renewal, written once per cancel call) don't justify the join/joinless-read trade-off a +separate `task_cancellation_requests` table would add, and there is exactly one prior +migration precedent for this shape — `0008_task_progress_columns` added nullable +scalar columns directly to `scheduled_tasks` for the same reason (frequently-read, +single-value-per-task state). `TaskAttemptRow`/`task_attempts` needs no schema change: +its `status` column already accepts free-form strings and gains `"cancelled"` as a value +alongside `"assigned"/"dispatched"/"expired"/"done"/"failed"`. + +`CloudRepository` Protocol gains: +- `request_task_cancellation(task_id, *, requested_at) -> CancellationRequestStatus` + where `CancellationRequestStatus = Literal["requested", "already_terminal", + "already_requested", "not_found"]` — synchronously transitions a `queued` task straight + to `cancelled` (there's no Host attempt in flight to notify) and otherwise sets + `cancel_requested_at` on an `assigned`/`dispatched` task. +- `renew_lease` gains a `cancel_requested` boolean in its return path (or the caller + re-reads the row — implementation detail left to tasks.md) so `internal_api/api.py`'s + `renew_assignment` handler can populate `LeaseRenewalResponse.cancel_requested`. +- `record_task_result` accepts `status: Literal["done", "failed", "cancelled"]` + (widened from today's `TerminalTaskStatus = Literal["done", "failed"]`) and clears + `cancel_requested_at` on write. + +### D6: No new `ScheduledTaskStatus` value for "cancellation pending" + +Considered adding `"cancelling"` as a distinct status between `assigned`/`dispatched` and +`cancelled`. Rejected: it would ripple into every place that already pattern-matches the +existing five-value `ScheduledTaskStatus` (scheduler assignment logic, device +reservation/`list_reserved_device_ids`, Cloud Console status filter, SDK response +`Literal`), each needing to decide whether "cancelling" behaves like "assigned" (device +still reserved, task still excluded from re-assignment) — which it always would, making +the new status a strict synonym with an extra bit of information. Instead, +"cancellation pending" is expressed as `status="assigned"` (or `"dispatched"`) plus +non-null `cancel_requested_at` — every existing status-based code path (assignment +matching, device reservation, retry-on-expiry) keeps working unmodified, and callers +that want to show "cancelling…" in a UI check the extra field. + +### D7: Public cancel endpoint shape + +`POST /v1/tasks/{task_id}/cancel`, `202 Accepted` for the pending (assigned/dispatched) +case and `200 OK` for the immediate (queued) case, both returning a small +`TaskCancellationResponse {task_id, status}` reflecting the resulting `ScheduledTask` +status. Repeated calls against an already-cancelled or already-`cancel_requested_at`-set +task return the same response idempotently (HTTP 200, not a conflict) — matching the +project's established idempotency style for `report_result`/`renew_assignment`, which +return `already_recorded`/success on replay rather than erroring. Calling cancel on a +task already `done`/`failed` returns `409 Conflict` with a clear "task is already +terminal" detail, mirroring `_stale_lease_conflict`'s existing shape. + +## Risks / Trade-offs + +- [Operators expect cancellation to be instant] → It isn't, by design (D1's latency + bound). The Cloud Console surfaces the `cancel_requested_at` pending state distinctly + (D6) so operators see "cancellation requested" rather than a UI that looks stuck; docs + (`docs/CLOUD_DEPLOYMENT.md`) get an explicit latency note per the proposal's Impact + section. +- [A Host Agent that never renews again (already offline, or wedged past its lease) never + learns about the cancellation] → Existing `reap_expired_leases` already requeues or + fails such tasks once the lease expires; this change doesn't need new machinery for + that case — an offline Host's task eventually reaches a terminal state via the existing + reaper, at which point `cancel_requested_at` being non-null is irrelevant (task is + already terminal). Worth noting in tasks.md that the reaper's requeue path should NOT + requeue (re-assign) a task with `cancel_requested_at` set — it should mark it + `cancelled` instead of `queued`, or the cancellation would be silently dropped on + retry. +- [`tasks:submit`-scoped cancel with no per-task ownership means any authorized submitter + can cancel any other submitter's task] → Accepted for this change (Non-Goals); this + matches the existing coarse-grained scope model everywhere else in the SDK (task + reading is likewise not ownership-scoped) and narrowing it is `cloud-console-governance` + territory, not this change's. +- [Widening `TaskStatus`/`ScheduledTaskStatus`/`TaskAttemptRow.status` literals is a + backward-compatible additive change in Python, but any exhaustive `switch`/discriminated + union in the TypeScript Console (`types.ts`'s `TaskStatus`) will fail to compile until + updated] → Caught at Console build time (Vite/`vue-tsc`), not runtime; tasks.md includes + updating `cloud-console/src/types.ts` and any exhaustiveness-checked switch in the same + commit as the backend change. +- [Race: cancel request arrives between the Host's `claim_assignment` (queued→dispatched + transition happens inside `claim_assignment`, not `assign_task`) and its first renewal] + → Already handled by D2's durable-row design: the cancel call checks current `status` + regardless of exactly when the Host claims, and `renew_lease` always re-reads the + current row, so there's no window where the signal is lost — only a window (bounded by + D1's latency) where it hasn't been observed yet. + +## Migration Plan + +1. Ship the Alembic migration (0012) — additive nullable column, no backfill needed, safe + to apply with the application running (existing tasks get `cancel_requested_at = NULL`, + behaviorally identical to today). +2. Deploy Cloud API with the new repository methods, internal `renew_assignment` response + field (`cancel_requested`, defaults `False` — old Host Agents ignore unknown fields), + and the new public cancel endpoint. This step alone is a no-op for existing Hosts: + nothing calls the new endpoint yet, and `LeaseRenewalResponse` gaining an optional + field is backward-compatible with any Host Agent version already deployed (Pydantic + response models are additive-safe for JSON-decoding clients that only read known + fields). +3. Deploy Host Agents with the updated `lease.py`/`assignment.py`/`client.py` that read + and act on `cancel_requested`. Hosts not yet updated simply never observe cancellation + requests (task stays "pending cancellation" until its lease naturally expires and the + reaper marks it `cancelled` per the Risks section) — a soft-fail, not a hard error. +4. Ship the Cloud Console and Host Agent local console UI changes last, once the backend + contract is stable. +5. Rollback: the migration is purely additive and safe to leave in place even if the + feature is disabled; no rollback migration is required beyond the standard Alembic + downgrade (drop the column) if a full revert is ever needed. + +## Open Questions + +- Whether the public cancel endpoint should also accept an optional caller-supplied + reason string (for audit/UI display) — left to tasks.md to decide during + implementation; not a design-blocking question since it's purely additive to + `TaskCancellationRequest` if added later. +- Whether `reap_expired_leases`' "mark cancelled instead of requeue when + `cancel_requested_at` is set" behavior (noted in Risks) needs its own explicit + requirement in the `task-scheduler` delta spec or can be covered as an implementation + detail of the existing restart-recovery requirement — resolve when writing specs.md. diff --git a/openspec/changes/task-cancellation/proposal.md b/openspec/changes/task-cancellation/proposal.md new file mode 100644 index 0000000..f3ebd70 --- /dev/null +++ b/openspec/changes/task-cancellation/proposal.md @@ -0,0 +1,81 @@ +## Why + +Task cancellation has been an explicit, repeatedly-acknowledged gap: `cloud-console`, +`cloud-control-plane-integration`, and `cloud-console-governance` each excluded it from +scope and deferred it to "a later lifecycle capability." Operators currently have no way +to stop a queued, assigned, or in-flight task short of letting it run to completion, +failure, or lease expiry — including tasks stuck against an offline device or a runaway +plan. `core/models.py`'s `TaskStatus` already reserves a `"cancelled"` value that no code +path ever sets, and `workflow/` already proves the collaborative-stop pattern this change +extends to goal-based cloud tasks. + +## What Changes + +- Add a `cancelled` scheduler-side task status (`cloud/scheduler.py`'s + `ScheduledTaskStatus`) reachable from `queued`, `assigned`, and `dispatched`. +- Add a public SDK endpoint (`POST /v1/tasks/{task_id}/cancel`, gated by the existing + `tasks:submit` scope — the repository has no per-task submitter/ownership tracking to + authorize against more narrowly) that immediately cancels a `queued` task and otherwise + records a cancellation request against an `assigned`/`dispatched` task. +- Add an internal Host Agent protocol signal: `LeaseRenewalResponse` gains a + `cancel_requested: bool` field; the Cloud repository's `renew_lease` reports it when the + active attempt has a pending cancellation. This is the only new edge on the existing + outbound-only Host Agent protocol — no inbound push, no new endpoint on the Host side. +- Extend the Host Agent's existing `should_stop` collaborative-stop mechanism + (`ActiveAssignmentRunner` → `AssignmentExecutor` → `TaskRunner`/`WorkflowRunner`) so a + `cancel_requested` signal observed at lease-renewal time stops execution the same way a + lost lease does today, and reports a `cancelled`-flavored terminal result. +- Add `"cancelled"` as a real, reachable value of `core/models.py`'s `TaskStatus` (the + `_interrupt_task` path already flowing through `should_stop` gains a + cancellation-vs-lease-loss distinction) and confirm `TerminalResultRequest.status` can + express it end to end. +- Add an Alembic migration recording cancellation request/acknowledgement metadata on + `scheduled_tasks`/`task_attempts` (requestor, requested-at, and the terminal + `cancelled` outcome) — no schema change to unrelated tables. +- Add a "Cancel" action to the Cloud Console `TasksView.vue` task-detail panel (visible + for `queued`/`assigned`/`dispatched` tasks the operator is authorized to act on) and to + the status filter dropdown; add a matching `POST /tasks/{id}/cancel` route + button to + the Host Agent local console's task detail page for Host-local visibility/action on + tasks running on that Host. +- **BREAKING**: none of the existing status literals are renamed or removed; `cancelled` + is purely additive. Callers that exhaustively `match`/switch over `TaskStatus` (Python) + or `TaskStatus` (TypeScript) without a default arm will need to add a case — flagged in + design.md's migration plan. + +## Capabilities + +### New Capabilities +- `task-cancellation`: cancellation request lifecycle across `task-scheduler` (queued/ + assigned/dispatched states), `host-agent-protocol` (collaborative cancel signal over + lease renewal), and `agent-runtime`/workflow execution (stopping mid-task on a + cancellation signal, distinct from lease loss). + +### Modified Capabilities +- `task-scheduler`: `ScheduledTaskStatus` gains `cancelled`; task submission/assignment + requirements are unchanged, but the status-transition requirements need a new terminal + transition path from `queued`/`assigned`/`dispatched`. +- `host-agent-protocol`: the lease-renewal requirement ("Active execution renews its + lease") gains a new SHALL for surfacing a cancellation request in the renewal response + and treating it as a stop condition alongside lease loss. +- `platform-sdk`: new cancel endpoint and scope-authorization requirement; task-status + responses gain the `cancelled` status value. +- `cloud-console-ui`: task list/detail view gains a Cancel action and the `cancelled` + status value in filtering/display. + +## Impact + +- **Cloud API / persistence**: `packages/cloud-platform/cloud/scheduler.py`, + `repository.py` (Protocol), `sql_repository.py`, `db_models.py`, new Alembic migration, + `internal_api/models.py` + `internal_api/api.py` (renew/claim/cancel routes), + `sdk/api.py` + `sdk/models.py` (new public cancel endpoint), `auth.py` (scope reuse). +- **Host Agent**: `apps/device-host-agent/host_agent/lease.py` (`ActiveAssignmentRunner` + cancellation-aware stop), `client.py` (surface `cancel_requested` from renew response), + `processor.py`/`assignment.py` (terminal status reporting), local console + (`host_agent/web/app.py` + `templates/task_detail.html`) new cancel route. +- **Runtime**: `core/models.py` (`TaskStatus` reachability), `runtime/task.py` + (`_interrupt_task` cancellation-vs-interruption distinction), `workflow/runner.py` + (reuse of the already-existing `cancelled` terminal status — no change needed there). +- **Cloud Console frontend**: `cloud-console/src/views/TasksView.vue`, `src/types.ts`, + `src/api.ts` (new `cancelTask` client method). +- **Docs**: `docs/CLOUD_DEPLOYMENT.md` gets a short note on cancellation being + collaborative (not instantaneous) and its ~1/3-lease-period latency bound. diff --git a/openspec/changes/task-cancellation/specs/cloud-console-ui/spec.md b/openspec/changes/task-cancellation/specs/cloud-console-ui/spec.md new file mode 100644 index 0000000..37f2983 --- /dev/null +++ b/openspec/changes/task-cancellation/specs/cloud-console-ui/spec.md @@ -0,0 +1,24 @@ +## MODIFIED Requirements + +### Requirement: Task dashboard +The console SHALL render a task view listing tasks by status with pagination, and SHALL show a task's detail including its attempt history, using the platform SDK's task-listing and attempt-history endpoints. The task detail view SHALL offer a Cancel action for tasks in status `queued`, `assigned`, or `dispatched`, using the platform SDK's cancel endpoint, and the status filter SHALL include `cancelled`. + +#### Scenario: Browse the task queue +- **WHEN** an operator with a `tasks:read`-scoped token opens the task view +- **THEN** the console displays tasks with their status, goal or workflow reference, and assigned device/host, and lets the operator filter by status, including `cancelled` + +#### Scenario: Inspect a task's attempt history +- **WHEN** an operator selects a task from the list +- **THEN** the console displays that task's recorded attempts in order, including each attempt's outcome + +#### Scenario: Cancel a task from the detail view +- **WHEN** an operator with a `tasks:submit`-scoped token views the detail of a task whose status is `queued`, `assigned`, or `dispatched`, and clicks Cancel +- **THEN** the console calls the cancel endpoint and updates the displayed status to reflect the immediate or pending cancellation result + +#### Scenario: Cancel action is absent for terminal tasks +- **WHEN** an operator views the detail of a task whose status is `done`, `failed`, or `cancelled` +- **THEN** the console does not offer a Cancel action for that task + +#### Scenario: Cancel attempted without submit scope +- **WHEN** an operator whose token lacks `tasks:submit` views a cancellable task's detail +- **THEN** the console does not offer a Cancel action, or surfaces the API's authorization error without implying the task was cancelled diff --git a/openspec/changes/task-cancellation/specs/host-agent-protocol/spec.md b/openspec/changes/task-cancellation/specs/host-agent-protocol/spec.md new file mode 100644 index 0000000..897f358 --- /dev/null +++ b/openspec/changes/task-cancellation/specs/host-agent-protocol/spec.md @@ -0,0 +1,48 @@ +## MODIFIED Requirements + +### Requirement: Active execution renews its lease +The Host Agent SHALL renew the active assignment lease before expiry while execution continues, and SHALL treat loss or rejection of the lease, or an observed cancellation request, as a stop condition for further planned actions where interruption is possible. + +#### Scenario: Lease renewal succeeds +- **WHEN** the owning Host Agent renews an unexpired active lease +- **THEN** the control plane extends its expiry without changing the task attempt or device assignment + +#### Scenario: Lease is stale or foreign +- **WHEN** a Host Agent attempts to renew an expired, replaced, or differently owned lease +- **THEN** the control plane returns a conflict and does not revive or alter the current attempt + +#### Scenario: Renewal response signals a pending cancellation +- **WHEN** the control plane's renewal response for an active lease indicates a pending cancellation request +- **THEN** the Host Agent stops further planned actions at the next available step boundary, the same way it stops on lease loss + +### Requirement: Terminal result reporting is idempotent +The Host Agent SHALL report a terminal result using the task, attempt, and lease identifiers, and repeating the same report SHALL return the already recorded outcome without duplicating state transitions. The terminal result SHALL be `done`, `failed`, or `cancelled`. + +#### Scenario: Report a successful result +- **WHEN** the active lease owner reports successful completion +- **THEN** the control plane marks the scheduled task done, releases the device reservation, and records the result metadata + +#### Scenario: Report a cancelled result +- **WHEN** the active lease owner reports that its execution stopped because of an observed cancellation request +- **THEN** the control plane marks the scheduled task cancelled, releases the device reservation, and records the result metadata + +#### Scenario: Retry a result after response loss +- **WHEN** the Host Agent repeats the identical terminal report for an already completed active lease +- **THEN** the control plane returns the recorded terminal result without creating a new attempt or error + +#### Scenario: Stale attempt reports after requeue +- **WHEN** an expired earlier attempt reports after a newer attempt has been created +- **THEN** the control plane rejects the stale report and preserves the newer attempt's state + +## ADDED Requirements + +### Requirement: Host distinguishes cancellation stop from lease-loss stop when reporting outcome +The Host Agent SHALL track whether its active execution stopped because of an observed cancellation request or for another stop reason (lost or rejected lease), and SHALL report `cancelled` only in the cancellation case, reporting `failed` for other stop reasons. + +#### Scenario: Stop triggered by cancellation +- **WHEN** the Host Agent's collaborative-stop mechanism is triggered by a renewal response signaling a pending cancellation +- **THEN** the terminal result it reports for that attempt is `cancelled` + +#### Scenario: Stop triggered by lease loss +- **WHEN** the Host Agent's collaborative-stop mechanism is triggered by a rejected or lost lease unrelated to any cancellation signal +- **THEN** the terminal result it reports for that attempt is `failed`, not `cancelled` diff --git a/openspec/changes/task-cancellation/specs/platform-sdk/spec.md b/openspec/changes/task-cancellation/specs/platform-sdk/spec.md new file mode 100644 index 0000000..c0684d5 --- /dev/null +++ b/openspec/changes/task-cancellation/specs/platform-sdk/spec.md @@ -0,0 +1,54 @@ +## MODIFIED Requirements + +### Requirement: Task submission and status via the SDK +The system SHALL allow an external integrator to submit a task (goal or workflow reference plus constraints) through the platform SDK's API, to query that task's current status by id, and to request cancellation of that task by id, backed by the `task-scheduler` capability. + +#### Scenario: Submit a task via the API +- **WHEN** an integrator calls the task-submission endpoint with a valid goal and optional constraints +- **THEN** the API returns a task id that can be used to poll status, and the underlying `task-scheduler` records a new `queued` `ScheduledTask` + +#### Scenario: Query status of a known task +- **WHEN** an integrator requests status for a task id that exists +- **THEN** the API returns that task's current status (`queued`, `assigned`, `dispatched`, `done`, `failed`, or `cancelled`) + +#### Scenario: Query status of an unknown task +- **WHEN** an integrator requests status for a task id that does not exist +- **THEN** the API returns a not-found response rather than an unhandled server error + +#### Scenario: Cancel a known task via the API +- **WHEN** an integrator with the required scope calls the cancel endpoint for a task id that exists and is not already `done` or `failed` +- **THEN** the API accepts the request and the underlying `task-scheduler` records the cancellation per its immediate or collaborative rules for that task's current status + +#### Scenario: Cancel an unknown task +- **WHEN** an integrator calls the cancel endpoint for a task id that does not exist +- **THEN** the API returns a not-found response rather than an unhandled server error + +### Requirement: Public API operations enforce scopes +The public platform API SHALL require operation-specific scopes, including task submission, task cancellation, task reading, pool reading, plugin reading, and plugin administration. + +#### Scenario: Submit token has task scope +- **WHEN** a principal with `tasks:submit` calls the task-submission endpoint +- **THEN** the request is authorized subject to normal task validation + +#### Scenario: Submit-scoped token cancels a task +- **WHEN** a principal with `tasks:submit` calls the task-cancellation endpoint for any task id +- **THEN** the request is authorized; the platform SDK does not restrict cancellation to the task's original submitter, since no per-task submitter identity is tracked + +#### Scenario: Read-only token attempts cancellation +- **WHEN** a principal that holds only `tasks:read` calls the task-cancellation endpoint +- **THEN** the API rejects the request before contacting the scheduler + +#### Scenario: Non-admin token attempts plugin registration +- **WHEN** an authenticated principal without `plugins:admin` calls plugin registration +- **THEN** the API rejects the request before resolving or loading the plugin target + +### Requirement: Python SDK client mirrors the REST API +The system SHALL provide a Python client (`CloudClient`) exposing methods corresponding to each `/v1/...` route (submit task, get task status, cancel task, list devices, list hosts, list plugins, register plugin), so integrators do not need to hand-construct HTTP requests. + +#### Scenario: Client submits a task and retrieves status +- **WHEN** a caller uses `CloudClient` to submit a task and then fetch its status by the returned id +- **THEN** the client's methods produce the same result as calling the corresponding `/v1/...` endpoints directly over HTTP + +#### Scenario: Client cancels a task +- **WHEN** a caller uses `CloudClient` to cancel a task by id +- **THEN** the client's method produces the same result as calling the cancel endpoint directly over HTTP diff --git a/openspec/changes/task-cancellation/specs/task-cancellation/spec.md b/openspec/changes/task-cancellation/specs/task-cancellation/spec.md new file mode 100644 index 0000000..641eab2 --- /dev/null +++ b/openspec/changes/task-cancellation/specs/task-cancellation/spec.md @@ -0,0 +1,59 @@ +## ADDED Requirements + +### Requirement: Queued task cancellation is immediate +The system SHALL, when a cancellation is requested against a task in status `queued`, transition that task directly to status `cancelled` synchronously within the same request, without contacting any Host. + +#### Scenario: Cancel a task that has not been assigned +- **WHEN** an authorized caller requests cancellation of a task whose status is `queued` +- **THEN** the task's status becomes `cancelled` in the same request and no assignment or lease is ever created for it + +### Requirement: In-flight task cancellation is a durable, collaborative request +The system SHALL, when a cancellation is requested against a task in status `assigned` or `dispatched`, durably record a cancellation request against that task rather than immediately marking it `cancelled`, and SHALL surface that pending request to the owning Host Agent no later than its next lease renewal. + +#### Scenario: Cancel a task currently executing on a Host +- **WHEN** an authorized caller requests cancellation of a task whose status is `dispatched` +- **THEN** the system records the cancellation request against the task's current attempt, the task's status remains `dispatched` until the Host reports a terminal result, and the request survives a control-plane restart + +#### Scenario: Owning Host observes the pending cancellation at lease renewal +- **WHEN** the Host Agent executing the task renews its lease after a cancellation request was recorded +- **THEN** the renewal response signals the pending cancellation and the Host Agent stops further planned actions at the next available step boundary + +#### Scenario: Cancellation is not instantaneous +- **WHEN** a cancellation is requested against a `dispatched` task +- **THEN** the system does not guarantee the task reaches status `cancelled` before the owning Host's next lease-renewal cycle completes + +### Requirement: Host reports a cancelled outcome distinct from a failed outcome +The Host Agent SHALL report a terminal status of `cancelled`, distinct from `failed`, when its active execution stopped because of an observed cancellation request rather than a lease loss or an execution error, and the control plane SHALL record that task as status `cancelled`. + +#### Scenario: Execution stops due to a cancellation request +- **WHEN** the Host Agent's active `TaskRunner` or `WorkflowRunner` execution stops because a lease renewal signaled a pending cancellation +- **THEN** the Host Agent reports terminal status `cancelled`, and the control plane transitions the task to status `cancelled` and releases its device reservation + +#### Scenario: Execution stops due to lease loss unrelated to cancellation +- **WHEN** the Host Agent's active execution stops because its lease was rejected or lost for a reason other than a pending cancellation +- **THEN** the Host Agent reports terminal status `failed`, not `cancelled` + +### Requirement: Cancellation requests are idempotent +The system SHALL treat a repeated cancellation request against a task that already has a pending or completed cancellation as a no-op that returns the task's current status, rather than as an error. + +#### Scenario: Cancel a task twice +- **WHEN** an authorized caller requests cancellation of a task that already has a pending cancellation request recorded +- **THEN** the system returns the same successful response as the first request without creating a duplicate cancellation record + +#### Scenario: Cancel an already-cancelled task +- **WHEN** an authorized caller requests cancellation of a task whose status is already `cancelled` +- **THEN** the system returns success reflecting the `cancelled` status without error + +### Requirement: Cancellation is rejected for tasks already in a terminal, non-cancelled state +The system SHALL reject a cancellation request against a task whose status is already `done` or `failed` with a clear conflict error, without altering that task's recorded outcome. + +#### Scenario: Cancel a completed task +- **WHEN** an authorized caller requests cancellation of a task whose status is `done` +- **THEN** the system rejects the request with a conflict error and the task's status and result remain unchanged + +### Requirement: An expiring lease on a task with a pending cancellation resolves to cancelled, not requeued +The system SHALL, when an active lease expires on a task that has a pending cancellation request, mark that task `cancelled` rather than returning it to `queued` for a further attempt. + +#### Scenario: Lease expires while a cancellation is pending +- **WHEN** the active lease on a `dispatched` task with a pending cancellation request expires before a terminal result is reported +- **THEN** the task transitions to status `cancelled` and its device reservation is released, instead of being requeued for another attempt diff --git a/openspec/changes/task-cancellation/specs/task-scheduler/spec.md b/openspec/changes/task-cancellation/specs/task-scheduler/spec.md new file mode 100644 index 0000000..8ad86ef --- /dev/null +++ b/openspec/changes/task-cancellation/specs/task-scheduler/spec.md @@ -0,0 +1,54 @@ +## MODIFIED Requirements + +### Requirement: Terminal transitions validate the active lease +The system SHALL accept a `done`, `failed`, or `cancelled` result only from the current active task attempt and lease and SHALL make repeated identical terminal reports idempotent. + +#### Scenario: Active lease reports completion +- **WHEN** the active lease owner reports a terminal result +- **THEN** the task transitions once to done, failed, or cancelled and releases its device reservation + +#### Scenario: Superseded lease reports completion +- **WHEN** a result references a lease superseded by expiry and retry +- **THEN** the result is rejected and cannot overwrite the current task attempt + +### Requirement: Expired attempts follow bounded retry policy +The system SHALL detect expired assigned or dispatched leases and SHALL either requeue the task with its reservation released, mark it failed when the configured attempt limit is reached, or mark it cancelled when a cancellation request is pending against it. + +#### Scenario: Lease expires with attempts remaining +- **WHEN** an active lease expires before a terminal result and the task has remaining attempts and no pending cancellation request +- **THEN** the task returns to queued, the previous device reservation is released, and the expired attempt remains auditable + +#### Scenario: Lease expires at attempt limit +- **WHEN** an active lease expires and the task has reached its maximum attempts +- **THEN** the task becomes failed with a lease-expiry reason and its device reservation is released + +#### Scenario: Lease expires with a cancellation pending +- **WHEN** an active lease expires on a task that has a pending cancellation request, regardless of remaining attempts +- **THEN** the task becomes cancelled rather than being requeued or marked failed, and its device reservation is released + +## ADDED Requirements + +### Requirement: Task status includes a reachable cancelled value +The `ScheduledTaskStatus` SHALL include `cancelled` as a terminal status reachable from `queued`, `assigned`, or `dispatched`, alongside the existing `done` and `failed` terminal statuses. + +#### Scenario: Cancelled status is a valid terminal state +- **WHEN** a task's cancellation completes, whether immediately from `queued` or after collaborative stop from `assigned`/`dispatched` +- **THEN** the task's status is `cancelled`, and no further assignment, claim, or lease-renewal operation is accepted against it + +### Requirement: Cancellation requests are recorded durably against in-flight tasks +The scheduler repository SHALL persist a cancellation request against an `assigned` or `dispatched` task's current attempt such that the request is observable across a control-plane process restart, before the task reaches a terminal status. + +#### Scenario: Cancellation request survives a restart +- **WHEN** a cancellation request is recorded against a `dispatched` task and the control plane process restarts before the Host next renews its lease +- **THEN** the pending cancellation request is still present and is surfaced to the Host on its next renewal after restart + +### Requirement: Lease renewal surfaces a pending cancellation request +The scheduler repository's lease-renewal operation SHALL report whether the renewing attempt has a pending cancellation request, without altering the normal lease-extension outcome. + +#### Scenario: Renewal on a task with a pending cancellation +- **WHEN** the owning host renews the lease for an attempt that has a pending cancellation request +- **THEN** the lease is extended normally and the renewal result additionally indicates the pending cancellation + +#### Scenario: Renewal on a task without a pending cancellation +- **WHEN** the owning host renews the lease for an attempt with no pending cancellation request +- **THEN** the lease is extended normally and the renewal result indicates no pending cancellation diff --git a/openspec/changes/task-cancellation/tasks.md b/openspec/changes/task-cancellation/tasks.md new file mode 100644 index 0000000..ee17476 --- /dev/null +++ b/openspec/changes/task-cancellation/tasks.md @@ -0,0 +1,67 @@ +## 1. Core and Runtime status plumbing + +- [x] 1.1 Confirm `core/models.py`'s `TaskStatus` already includes `"cancelled"` (it does); add `StepStatus` no change needed — verify no other literal needs widening. +- [x] 1.2 Add an optional `stop_reason: Callable[[], str] | None` parameter to `TaskRunner.run()` (`runtime/task.py`), defaulting to `None`. +- [x] 1.3 Update `TaskRunner._interrupt_task()` to accept the resolved reason string, set `status="cancelled"` when the reason indicates cancellation (e.g. contains `"cancel"`), else keep `status="failed"` as today, and record the reason as `failure_reason` in both cases. +- [x] 1.4 Update `WorkflowRunner`'s stop branch (`workflow/runner.py`) to accept and pass through the same optional `stop_reason`, reusing its existing `"cancelled"` checkpoint call — confirm no behavior change needed since it already lands on `"cancelled"` for any stop; only wire the reason through for consistency/logging. (Correction during implementation: WorkflowRunner previously collapsed *every* stop, including lease-loss, into `"cancelled"`, which contradicts the host-agent-protocol spec's requirement to distinguish cancellation from lease-loss stops. `_stop_status()` now branches on `stop_reason` the same way `TaskRunner` does, while preserving the exact pre-existing default (`"cancelled"`) when no `stop_reason` is supplied.) +- [x] 1.5 Add/extend unit tests in `tests/` for `TaskRunner` covering: cancellation-flavored stop → `status="cancelled"`; lease-loss-flavored stop → `status="failed"` (existing behavior preserved). Also added matching `WorkflowRunner` coverage for the same `_stop_status` branching. + +## 2. Cloud persistence: schema and repository + +- [x] 2.1 Add Alembic migration `0012_task_cancellation.py` under `packages/cloud-platform/cloud/migrations/versions/`: add nullable `cancel_requested_at` column to `scheduled_tasks`, matching the style of `0008_task_progress_columns`. +- [x] 2.2 Add `cancel_requested_at` to `ScheduledTaskRow` (`db_models.py`) and to the `ScheduledTask` dataclass (`cloud/scheduler.py`). +- [x] 2.3 Widen `ScheduledTaskStatus` (`cloud/scheduler.py`) to include `"cancelled"`. +- [x] 2.4 Widen `TerminalTaskStatus` and `record_task_result`'s accepted status literal (`repository.py` Protocol, `sql_repository.py`) to include `"cancelled"`; ensure the idempotency logic (`already_recorded`/`conflict`) treats a repeated `"cancelled"` report the same way it treats repeated `"done"`/`"failed"` reports today. +- [x] 2.5 Add `CancellationRequestStatus = Literal["requested", "already_terminal", "already_requested", "not_found"]` and a `request_task_cancellation(task_id, *, requested_at) -> CancellationRequestStatus` method to the `CloudRepository` Protocol. +- [x] 2.6 Implement `request_task_cancellation` in `SQLAlchemyCloudRepository`: for `queued` tasks, transition directly to `status="cancelled"`; for `assigned`/`dispatched` tasks, set `cancel_requested_at` if unset (return `"already_requested"` if already set); for `done`/`failed`/`cancelled` tasks, return `"already_terminal"`/success-idempotent as appropriate per design D7; for unknown task id, return `"not_found"`. +- [x] 2.7 Extend `renew_lease` (`sql_repository.py`) to read `cancel_requested_at` on the current row and report whether it is set, without changing its existing lease-extension/progress-update behavior. +- [x] 2.8 Update `reap_expired_leases` so a task whose `cancel_requested_at` is set resolves to `status="cancelled"` (clearing `cancel_requested_at`) instead of being requeued to `queued`, regardless of remaining attempts. +- [x] 2.9 Ensure `record_task_result` and the immediate `queued`-cancellation path both clear `cancel_requested_at` on reaching any terminal status. +- [x] 2.10 Add/extend repository-level tests (SQLite, matching existing test style) covering: immediate queued cancel; durable cancel request on assigned/dispatched surviving a simulated restart (re-fetch); renewal reporting the pending flag; expired lease with pending cancellation resolving to `cancelled`; idempotent repeat cancel calls; cancel rejected on `done`/`failed`. + +## 3. Internal Host↔Cloud protocol + +- [x] 3.1 Add `cancel_requested: bool = False` field to `LeaseRenewalResponse` (`internal_api/models.py`). +- [x] 3.2 Widen `TerminalResultRequest.status` (`internal_api/models.py`) to `Literal["done", "failed", "cancelled"]`. +- [x] 3.3 Update `renew_assignment` route (`internal_api/api.py`) to populate `LeaseRenewalResponse.cancel_requested` from the repository's `renew_lease` result. +- [x] 3.4 Update `report_result` route (`internal_api/api.py`) to accept and forward the `"cancelled"` status to `record_task_result`. +- [x] 3.5 Add/extend internal API tests covering a renewal response surfacing `cancel_requested=True` and a `"cancelled"` terminal report being accepted and idempotent on repeat. + +## 4. Host Agent collaborative stop + +- [x] 4.1 Extend `LeaseGuard` (`host_agent/lease.py`) with a `reason` attribute already implied by `mark_lost(reason)` — confirm it's readable, add a `is_cancellation` helper or convention (e.g. reason string prefix) to distinguish cancellation from other lost-lease reasons. +- [x] 4.2 Update `ActiveAssignmentRunner._renew_while_running()` (`host_agent/lease.py`): when `response.cancel_requested` is true, call `guard.mark_lost("cancellation requested by control plane")` instead of continuing the renewal loop. +- [x] 4.3 Update `client.py`'s `renew()` to ensure `LeaseRenewalResponse.cancel_requested` deserializes correctly (should be automatic via Pydantic model update, but add a test). +- [x] 4.4 Update `AssignmentExecutor._execute_goal()` and `_execute_workflow()` (`host_agent/assignment.py`) to map a cancellation-flavored stop to `AssignmentExecutionResult.status = "cancelled"` (new value alongside `"done"`/`"failed"`), reading the underlying `Task`/`WorkflowRun` status (`"cancelled"`) instead of collapsing it to `"failed"`. +- [x] 4.5 Update `AssignmentProcessor.process()` (`host_agent/processor.py`) so its `status = "done" if execution.status == "done" else "failed"` mapping becomes a three-way mapping that preserves `"cancelled"`, and `report_result` is called with `status="cancelled"` in that case. +- [x] 4.6 Add/extend Host Agent tests covering: a renewal response with `cancel_requested=True` stops the active `should_stop`-driven loop; the resulting `AssignmentExecutionResult.status` and reported terminal status are `"cancelled"`; a lease-loss stop unrelated to cancellation still reports `"failed"`. + +## 5. Public SDK endpoint + +- [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 + +- [x] 6.1 Add `"cancelled"` to `TaskStatus` in `cloud-console/src/types.ts` and to the `STATUSES` array in `TasksView.vue`. +- [x] 6.2 Add a `cancelTask(taskId)` method to `cloud-console/src/api.ts`. +- [x] 6.3 Add a "Cancel" button to `TasksView.vue`'s task detail panel, visible only when the selected task's status is `queued`/`assigned`/`dispatched` and the operator's token has `tasks:submit`; on click, call `cancelTask` and refresh the displayed task. +- [x] 6.4 Add/extend Cloud Console component tests (existing test style) covering: Cancel button visibility per status/scope; successful cancel updates displayed status; error response is surfaced without falsely showing cancelled. (Project has no Vue component-mounting test harness — `@vue/test-utils` isn't a dependency and no existing test exercises a `.vue` file directly. Followed the established pattern instead: extracted the visibility rule into a pure, unit-tested `taskCancellation.ts` module — mirroring `taskProgress.ts`/`plannerHistory.ts` — covering cancellable vs. terminal statuses and the `tasks:submit` scope gate. `cancelSelectedTask` in `TasksView.vue` only mutates `selectedTask.status` on a successful response and routes failures through the existing `handleError`/`errorMessage` path, so an error never flips the displayed status to cancelled.) + +## 7. Frontend: Host Agent local console + +- [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 + +- [x] 8.1 Add a short section to `docs/CLOUD_DEPLOYMENT.md` documenting that cancellation is collaborative (not instantaneous) for `assigned`/`dispatched` tasks, bounded by roughly one third of the configured lease duration, with immediate effect for `queued` tasks. + +## 9. End-to-end verification + +- [x] 9.1 Run the full test suite (`uv run pytest` at repo root, plus `cloud-console` frontend tests) and confirm no regressions in existing task-scheduler, host-agent-protocol, platform-sdk, or workflow-orchestration tests. (869 passed, 50 skipped, 4 pre-existing failures unrelated to this change — `test_verifier_against_real_llm`, `test_reflector_against_real_llm`, `test_real_anthropic_ai_planner_selects_a_tool`, `test_real_anthropic_semantic_enrichment_returns_schema_valid_scene` all require live Anthropic API network access and fail the same way on `master`. `cloud-console`: 27 tests passed, `vue-tsc --noEmit` typecheck clean.) +- [x] 9.2 Manually or via an integration test, exercise the full path: submit a task, cancel a `queued` task (immediate), submit and dispatch another task, cancel it mid-execution, and confirm it reaches `cancelled` within one lease-renewal cycle with `cancelled` visible in both the public API and the Cloud Console. (Added `test_cancellation_full_path_queued_immediate_and_dispatched_collaborative` in `tests/test_cloud_sdk_api.py`: submits and immediately cancels a queued task via `POST /v1/tasks/{id}/cancel` (200, `cancelled`); submits, dispatches, and cancels a second task mid-execution (202, pending); drives one lease renewal confirming `cancel_requested=True` is surfaced; reports a `cancelled` terminal result as the Host Agent would; and confirms the task shows `status="cancelled"` via both `GET /v1/tasks/{id}` and `GET /v1/tasks?status=cancelled`. The Cloud Console reads task status through this same public API and its `cancelled` rendering is covered by the Task 6.4 `taskCancellation.ts` unit tests, so this repository-to-API round trip is the full path exercised at the automated-test layer; no manual browser session was run.) diff --git a/packages/cloud-platform/cloud/db_models.py b/packages/cloud-platform/cloud/db_models.py index 7fa9974..a33c099 100644 --- a/packages/cloud-platform/cloud/db_models.py +++ b/packages/cloud-platform/cloud/db_models.py @@ -110,6 +110,7 @@ class ScheduledTaskRow(Base): progress_step_status: Mapped[str | None] = mapped_column(String, nullable=True) progress_summary: Mapped[str | None] = mapped_column(String, nullable=True) progress_updated_at: Mapped[str | None] = mapped_column(String, nullable=True) + cancel_requested_at: Mapped[str | None] = mapped_column(String, nullable=True) failure_reason: Mapped[str | None] = mapped_column(Text, nullable=True) result_json: Mapped[str | None] = mapped_column(Text, nullable=True) updated_at: Mapped[str | None] = mapped_column(String, nullable=True) diff --git a/packages/cloud-platform/cloud/internal_api/api.py b/packages/cloud-platform/cloud/internal_api/api.py index a2a7a06..057bafa 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, @@ -316,7 +361,7 @@ def create_internal_router( summary=payload.progress.summary[:500], updated_at=now, ) - renewal_status = pool.store.renew_lease( + renewal = pool.store.renew_lease( task_id=task_id, attempt=payload.attempt, lease_id=payload.lease_id, @@ -325,16 +370,17 @@ def create_internal_router( now=now, progress=progress_snapshot, ) - if renewal_status == "not_found": + if renewal.status == "not_found": raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="assignment not found", ) - if renewal_status != "renewed": + if renewal.status != "renewed": return _stale_lease_conflict("assignment lease is stale or expired") return LeaseRenewalResponse( status="renewed", lease_expires_at=lease_expires_at, + cancel_requested=renewal.cancel_requested, ) @router.post( diff --git a/packages/cloud-platform/cloud/internal_api/models.py b/packages/cloud-platform/cloud/internal_api/models.py index 49ce9cc..b72049c 100644 --- a/packages/cloud-platform/cloud/internal_api/models.py +++ b/packages/cloud-platform/cloud/internal_api/models.py @@ -95,6 +95,7 @@ class LeaseRenewalRequest(BaseModel): class LeaseRenewalResponse(BaseModel): status: Literal["renewed"] lease_expires_at: datetime + cancel_requested: bool = False class TerminalResultRequest(BaseModel): @@ -102,7 +103,7 @@ class TerminalResultRequest(BaseModel): task_id: str = Field(min_length=1) attempt: int = Field(ge=1) lease_id: str = Field(min_length=1) - status: Literal["done", "failed"] + status: Literal["done", "failed", "cancelled"] failure_reason: str | None = None result: dict[str, Any] | None = None @@ -121,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/packages/cloud-platform/cloud/migrations/versions/0013_task_cancellation.py b/packages/cloud-platform/cloud/migrations/versions/0013_task_cancellation.py new file mode 100644 index 0000000..6121094 --- /dev/null +++ b/packages/cloud-platform/cloud/migrations/versions/0013_task_cancellation.py @@ -0,0 +1,22 @@ +"""Add nullable cancel_requested_at column to scheduled_tasks.""" + +from __future__ import annotations + +import sqlalchemy as sa +from alembic import op + +revision = "0013_task_cancellation" +down_revision = "0012_planner_decision_action_metadata" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.add_column( + "scheduled_tasks", + sa.Column("cancel_requested_at", sa.String(), nullable=True), + ) + + +def downgrade() -> None: + op.drop_column("scheduled_tasks", "cancel_requested_at") diff --git a/packages/cloud-platform/cloud/repository.py b/packages/cloud-platform/cloud/repository.py index 9547a19..52028c2 100644 --- a/packages/cloud-platform/cloud/repository.py +++ b/packages/cloud-platform/cloud/repository.py @@ -30,9 +30,12 @@ if TYPE_CHECKING: AttemptStatus = Literal["assigned", "dispatched", "done", "failed", "expired"] -TerminalTaskStatus = Literal["done", "failed"] +TerminalTaskStatus = Literal["done", "failed", "cancelled"] ResultRecordStatus = Literal["recorded", "already_recorded", "conflict"] LeaseRenewalStatus = Literal["renewed", "not_found", "conflict", "expired"] +CancellationRequestStatus = Literal[ + "requested", "already_terminal", "already_requested", "not_found" +] class HostEnrollmentConflictError(RuntimeError): @@ -91,6 +94,19 @@ class TaskAttemptRecord: terminal_result: dict[str, Any] | None = None +@dataclass(frozen=True) +class LeaseRenewalResult: + """Outcome of a lease renewal, including whether cancellation is pending. + + ``cancel_requested`` reflects the task's durable ``cancel_requested_at`` + column at renewal time regardless of ``status`` — callers only act on it + when ``status == "renewed"``. + """ + + status: LeaseRenewalStatus + cancel_requested: bool = False + + @dataclass(frozen=True) class LeasedAssignment: task_id: str @@ -487,7 +503,7 @@ class CloudRepository(Protocol): lease_expires_at: datetime, now: datetime, progress: AssignmentProgressSnapshot | None = None, - ) -> LeaseRenewalStatus: ... + ) -> LeaseRenewalResult: ... def record_task_result( self, @@ -502,6 +518,13 @@ class CloudRepository(Protocol): completed_at: datetime, ) -> ResultRecordStatus: ... + def request_task_cancellation( + self, + task_id: str, + *, + requested_at: datetime, + ) -> CancellationRequestStatus: ... + def reap_expired_leases( self, *, diff --git a/packages/cloud-platform/cloud/scheduler.py b/packages/cloud-platform/cloud/scheduler.py index b63052f..2a5a293 100644 --- a/packages/cloud-platform/cloud/scheduler.py +++ b/packages/cloud-platform/cloud/scheduler.py @@ -22,7 +22,9 @@ if TYPE_CHECKING: from cloud.store import CloudStore -ScheduledTaskStatus = Literal["queued", "assigned", "dispatched", "done", "failed"] +ScheduledTaskStatus = Literal[ + "queued", "assigned", "dispatched", "done", "failed", "cancelled" +] @dataclass(frozen=True) @@ -57,6 +59,7 @@ class ScheduledTask: progress_step_status: str | None = None progress_summary: str | None = None progress_updated_at: datetime | None = None + cancel_requested_at: datetime | None = None @runtime_checkable diff --git a/packages/cloud-platform/cloud/schema.py b/packages/cloud-platform/cloud/schema.py index aaef3b4..6a5acef 100644 --- a/packages/cloud-platform/cloud/schema.py +++ b/packages/cloud-platform/cloud/schema.py @@ -9,7 +9,7 @@ from alembic.runtime.migration import MigrationContext from cloud.database import create_database_engine, normalize_database_url -HEAD_REVISION = "0012_planner_decision_action_metadata" +HEAD_REVISION = "0013_task_cancellation" class SchemaVersionError(RuntimeError): diff --git a/packages/cloud-platform/cloud/sdk/api.py b/packages/cloud-platform/cloud/sdk/api.py index 7e914e7..34d8833 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), @@ -272,6 +276,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 a215ab2..91cc85b 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/packages/cloud-platform/cloud/sql_repository.py b/packages/cloud-platform/cloud/sql_repository.py index 0493460..402d876 100644 --- a/packages/cloud-platform/cloud/sql_repository.py +++ b/packages/cloud-platform/cloud/sql_repository.py @@ -40,7 +40,7 @@ from cloud.observability import current_correlation_id from core.models import utc_now if TYPE_CHECKING: - from cloud.repository import AssignmentProgressSnapshot + from cloud.repository import AssignmentProgressSnapshot, LeaseRenewalResult logger = logging.getLogger(__name__) @@ -1497,7 +1497,9 @@ class SQLAlchemyCloudRepository: lease_expires_at: datetime, now: datetime, progress: AssignmentProgressSnapshot | None = None, - ) -> str: + ) -> "LeaseRenewalResult": + from cloud.repository import LeaseRenewalResult + with self._sessions.begin() as session: task = session.get( ScheduledTaskRow, @@ -1505,19 +1507,19 @@ class SQLAlchemyCloudRepository: with_for_update=self.engine.dialect.name == "postgresql", ) if task is None: - return "not_found" + return LeaseRenewalResult(status="not_found") if ( task.status not in {"assigned", "dispatched"} or task.attempt_count != attempt or task.lease_id != lease_id or task.assigned_host_id != host_id ): - return "conflict" + return LeaseRenewalResult(status="conflict") current_expiry = _parse_dt(task.lease_expires_at) if current_expiry is None or current_expiry <= now: - return "expired" + return LeaseRenewalResult(status="expired") if lease_expires_at <= now: - return "conflict" + return LeaseRenewalResult(status="conflict") attempt_row = session.get( TaskAttemptRow, @@ -1530,7 +1532,7 @@ class SQLAlchemyCloudRepository: or attempt_row.lease_id != lease_id or attempt_row.host_id != host_id ): - return "conflict" + return LeaseRenewalResult(status="conflict") renewed_until = _iso(lease_expires_at) task.lease_expires_at = renewed_until @@ -1542,7 +1544,10 @@ class SQLAlchemyCloudRepository: task.progress_summary = progress.summary task.progress_updated_at = _iso(progress.updated_at) _log_task_lifecycle("renewed", task) - return "renewed" + return LeaseRenewalResult( + status="renewed", + cancel_requested=task.cancel_requested_at is not None, + ) def record_task_result( self, @@ -1579,7 +1584,7 @@ class SQLAlchemyCloudRepository: ): return "conflict" - if task.status in {"done", "failed"}: + if task.status in {"done", "failed", "cancelled"}: if ( task.status == status and task.failure_reason == failure_reason @@ -1593,7 +1598,7 @@ class SQLAlchemyCloudRepository: current_expiry = _parse_dt(task.lease_expires_at) if current_expiry is None or current_expiry <= completed_at: return "conflict" - if status not in {"done", "failed"}: + if status not in {"done", "failed", "cancelled"}: return "conflict" result_json = ( @@ -1610,11 +1615,18 @@ class SQLAlchemyCloudRepository: task.progress_step_status = None task.progress_summary = None task.progress_updated_at = None + task.cancel_requested_at = None attempt_row.status = status attempt_row.completed_at = completed_at_iso attempt_row.failure_reason = failure_reason attempt_row.result_json = result_json - _log_task_lifecycle("completed" if status == "done" else "failed", task) + if status == "done": + lifecycle_event = "completed" + elif status == "cancelled": + lifecycle_event = "cancelled" + else: + lifecycle_event = "failed" + _log_task_lifecycle(lifecycle_event, task) return "recorded" def reap_expired_leases( @@ -1657,7 +1669,12 @@ class SQLAlchemyCloudRepository: task.lease_id = None task.lease_expires_at = None task.result_json = None - if task.attempt_count < max_attempts: + if task.cancel_requested_at is not None: + task.status = "cancelled" + task.assigned_host_id = None + task.assigned_device_id = None + task.cancel_requested_at = None + elif task.attempt_count < max_attempts: task.status = "queued" task.assigned_host_id = None task.assigned_device_id = None @@ -1667,13 +1684,46 @@ class SQLAlchemyCloudRepository: task.failure_reason = ( f"lease expired after {task.attempt_count} attempts" ) - _log_task_lifecycle( - "retried" if task.status == "queued" else "failed", - task, - ) + if task.status == "queued": + lifecycle_event = "retried" + elif task.status == "cancelled": + lifecycle_event = "cancelled" + else: + lifecycle_event = "failed" + _log_task_lifecycle(lifecycle_event, task) reaped_task_ids.append(task.id) return reaped_task_ids + def request_task_cancellation( + self, + task_id: str, + *, + requested_at: datetime, + ) -> str: + with self._sessions.begin() as session: + task = session.get( + ScheduledTaskRow, + task_id, + with_for_update=self.engine.dialect.name == "postgresql", + ) + if task is None: + return "not_found" + if task.status == "queued": + task.status = "cancelled" + task.cancel_requested_at = None + task.updated_at = _iso(requested_at) + _log_task_lifecycle("cancelled", task) + return "requested" + if task.status in {"assigned", "dispatched"}: + if task.cancel_requested_at is not None: + return "already_requested" + task.cancel_requested_at = _iso(requested_at) + task.updated_at = _iso(requested_at) + return "requested" + if task.status == "cancelled": + return "requested" + return "already_terminal" + def list_task_attempts(self, task_id: str) -> list[Any]: with self._sessions() as session: rows = session.scalars( @@ -2217,6 +2267,7 @@ def _task_from_row(row: ScheduledTaskRow) -> Any: progress_step_status=row.progress_step_status, progress_summary=row.progress_summary, progress_updated_at=_parse_dt(row.progress_updated_at), + cancel_requested_at=_parse_dt(row.cancel_requested_at), ) diff --git a/runtime/task.py b/runtime/task.py index 55656c4..242ae48 100644 --- a/runtime/task.py +++ b/runtime/task.py @@ -40,9 +40,15 @@ Observer = Callable[[str], Scene] ScreenshotProvider = Callable[[str], bytes] TaskSucceededHook = Callable[[str, str, Timeline], None] StopRequested = Callable[[], bool] +StopReason = Callable[[], "str | None"] StepProgressCallback = Callable[[int, str, str], None] +def is_cancellation_reason(reason: str | None) -> bool: + """Distinguish an explicit cancellation stop from other stop reasons (e.g. lost lease).""" + return bool(reason) and "cancel" in reason.lower() + + class TaskRunner: def __init__( self, @@ -98,6 +104,7 @@ class TaskRunner: task: Task, *, should_stop: StopRequested | None = None, + stop_reason: StopReason | None = None, ) -> Task: if self.metadata_store: self.metadata_store.create_task(task) @@ -109,7 +116,7 @@ class TaskRunner: for _ in range(self.config.max_steps): if should_stop is not None and should_stop(): - return self._interrupt_task(task) + return self._interrupt_task(task, stop_reason) try: scene = self.observer(task.device_id) context.add_scene(scene) @@ -136,7 +143,7 @@ class TaskRunner: for step_index, step in enumerate(steps): if should_stop is not None and should_stop(): - return self._interrupt_task(task) + return self._interrupt_task(task, stop_reason) executable_step = self._step_for_device(step, task.device_id) if step_index == 0 and screenshot is not None: # Reuse the screenshot already captured for planning instead of @@ -194,13 +201,16 @@ class TaskRunner: ) return task - def _interrupt_task(self, task: Task) -> Task: - self._emit_step_progress(-1, "failed", "execution interrupted") + def _interrupt_task(self, task: Task, stop_reason: StopReason | None = None) -> Task: + reason = stop_reason() if stop_reason is not None else None + message = reason or "execution interrupted" + status = "cancelled" if is_cancellation_reason(reason) else "failed" + self._emit_step_progress(-1, "failed", message) self._update_task( task, - status="failed", + status=status, completed=True, - failure_reason="execution interrupted", + failure_reason=message, ) return task 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_repository_contract.py b/tests/test_cloud_repository_contract.py index b237b19..0a2b8d0 100644 --- a/tests/test_cloud_repository_contract.py +++ b/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() diff --git a/tests/test_cloud_sdk_api.py b/tests/test_cloud_sdk_api.py index 99566b9..2c4e1b9 100644 --- a/tests/test_cloud_sdk_api.py +++ b/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, diff --git a/tests/test_cloud_task_progress_visibility.py b/tests/test_cloud_task_progress_visibility.py index 46f6cf8..ed9a698 100644 --- a/tests/test_cloud_task_progress_visibility.py +++ b/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 diff --git a/tests/test_host_agent_internal_api.py b/tests/test_host_agent_internal_api.py index 86d6a42..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 @@ -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) diff --git a/tests/test_host_agent_progress_reporting.py b/tests/test_host_agent_progress_reporting.py index ce114f6..acaef96 100644 --- a/tests/test_host_agent_progress_reporting.py +++ b/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 diff --git a/tests/test_task_loop.py b/tests/test_task_loop.py index 729dca0..ad60e1b 100644 --- a/tests/test_task_loop.py +++ b/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, diff --git a/tests/test_workflow_runner.py b/tests/test_workflow_runner.py index f4ecf21..3a46c7e 100644 --- a/tests/test_workflow_runner.py +++ b/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", diff --git a/workflow/runner.py b/workflow/runner.py index 4b75d4f..5deabe9 100644 --- a/workflow/runner.py +++ b/workflow/runner.py @@ -7,7 +7,7 @@ from typing import Any from time import sleep from core.models import Scene, Task -from runtime.task import TaskRunner +from runtime.task import StopReason, TaskRunner, is_cancellation_reason from skills_learning.store import SkillStore, get_default_store from workflow.conditions import ( ConditionEvaluator, @@ -70,6 +70,7 @@ class WorkflowRunner: initial_variables: dict[str, Any] | None = None, *, should_stop: StopRequested | None = None, + stop_reason: StopReason | None = None, ) -> WorkflowRun: self.store.save_definition(definition) run = self.store.create_run( @@ -77,13 +78,14 @@ class WorkflowRunner: initial_variables or {}, device_id=device_id, ) - return self._drive(definition, run, should_stop=should_stop) + return self._drive(definition, run, should_stop=should_stop, stop_reason=stop_reason) def resume( self, run_id: str, *, should_stop: StopRequested | None = None, + stop_reason: StopReason | None = None, ) -> WorkflowRun: run = self.store.get_run(run_id) if run is None: @@ -93,7 +95,18 @@ class WorkflowRunner: definition = self.store.get_definition(run.definition_id) if definition is None: raise KeyError(f"unknown workflow definition {run.definition_id}") - return self._drive(definition, run, should_stop=should_stop) + return self._drive(definition, run, should_stop=should_stop, stop_reason=stop_reason) + + def _stop_status(self, stop_reason: StopReason | None) -> str: + """Resolve the terminal status for a should_stop-triggered stop. + + No `stop_reason` preserves the pre-existing default of `cancelled` for + any stop; a supplied reason distinguishes an explicit cancellation + from other stop conditions (e.g. lost lease), which resolve to `failed`. + """ + if stop_reason is None: + return "cancelled" + return "cancelled" if is_cancellation_reason(stop_reason()) else "failed" def _drive( self, @@ -101,11 +114,14 @@ class WorkflowRunner: run: WorkflowRun, *, should_stop: StopRequested | None = None, + stop_reason: StopReason | None = None, ) -> WorkflowRun: executed = 0 while run.status not in TERMINAL_STATUSES and run.current_step_id: if should_stop is not None and should_stop(): - return self._checkpoint(run, run.current_step_id, "cancelled") + return self._checkpoint( + run, run.current_step_id, self._stop_status(stop_reason) + ) if self.step_limit is not None and executed >= self.step_limit: return run step = definition.step_by_id(run.current_step_id) @@ -125,10 +141,13 @@ class WorkflowRunner: run, step, should_stop=should_stop, + stop_reason=stop_reason, ) if should_stop is not None and should_stop(): self.store.append_step_result(run.id, result) - return self._checkpoint(run, run.current_step_id, "cancelled") + return self._checkpoint( + run, run.current_step_id, self._stop_status(stop_reason) + ) next_status, next_step_id = self._resolve_outcome( definition, step, result.success, branch_next_step_id ) @@ -185,12 +204,14 @@ class WorkflowRunner: step: WorkflowStep, *, should_stop: StopRequested | None = None, + stop_reason: StopReason | None = None, ) -> tuple[WorkflowStepResult, str | None]: if isinstance(step, PlannedGoalStep): return self._execute_planned_goal_step( run, step, should_stop=should_stop, + stop_reason=stop_reason, ), None if isinstance(step, SkillInvocationStep): return self._execute_skill_invocation_step(step), None @@ -218,13 +239,18 @@ class WorkflowRunner: step: PlannedGoalStep, *, should_stop: StopRequested | None = None, + stop_reason: StopReason | None = None, ) -> WorkflowStepResult: task = Task(goal=step.goal, device_id=run.device_id or "") task_runner = self.task_runner_factory() if should_stop is None: result_task = task_runner.run(task) - else: + elif stop_reason is None: result_task = task_runner.run(task, should_stop=should_stop) + else: + result_task = task_runner.run( + task, should_stop=should_stop, stop_reason=stop_reason + ) success = result_task.status == "completed" return WorkflowStepResult( step_id=step.step_id,