From 19c66698008c17485ec848897fc93a545a3be811 Mon Sep 17 00:00:00 2001 From: Jerry Yan <792602257@qq.com> Date: Wed, 15 Jul 2026 18:08:48 +0800 Subject: [PATCH] feat(cloud): add durable cancellation support to task repository - Add nullable cancel_requested_at column (migration 0012) - Widen ScheduledTaskStatus/TerminalTaskStatus to include cancelled - Add CancellationRequestStatus + request_task_cancellation() to CloudRepository protocol and SQLAlchemy implementation - renew_lease() now returns LeaseRenewalResult, surfacing whether cancellation is pending, instead of a bare status string - reap_expired_leases() resolves pending-cancellation tasks to cancelled instead of requeuing/failing them - record_task_result() accepts cancelled and clears cancel_requested_at on any terminal write Note: internal_api/api.py's renew_assignment route still compares renew_lease()'s return value against a bare string; it will be updated in the next task (Internal Host<->Cloud protocol) to consume LeaseRenewalResult and populate the new cancel_requested wire field. --- openspec/changes/task-cancellation/tasks.md | 20 +- packages/cloud-platform/cloud/db_models.py | 1 + .../versions/0012_task_cancellation.py | 22 ++ packages/cloud-platform/cloud/repository.py | 27 +- packages/cloud-platform/cloud/scheduler.py | 5 +- packages/cloud-platform/cloud/schema.py | 2 +- .../cloud-platform/cloud/sql_repository.py | 83 ++++-- tests/test_cloud_repository_contract.py | 257 +++++++++++++++++- tests/test_cloud_task_progress_visibility.py | 3 +- 9 files changed, 385 insertions(+), 35 deletions(-) create mode 100644 packages/cloud-platform/cloud/migrations/versions/0012_task_cancellation.py diff --git a/openspec/changes/task-cancellation/tasks.md b/openspec/changes/task-cancellation/tasks.md index b44c1bb..e1dab15 100644 --- a/openspec/changes/task-cancellation/tasks.md +++ b/openspec/changes/task-cancellation/tasks.md @@ -8,16 +8,16 @@ ## 2. Cloud persistence: schema and repository -- [ ] 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`. -- [ ] 2.2 Add `cancel_requested_at` to `ScheduledTaskRow` (`db_models.py`) and to the `ScheduledTask` dataclass (`cloud/scheduler.py`). -- [ ] 2.3 Widen `ScheduledTaskStatus` (`cloud/scheduler.py`) to include `"cancelled"`. -- [ ] 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. -- [ ] 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. -- [ ] 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"`. -- [ ] 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. -- [ ] 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. -- [ ] 2.9 Ensure `record_task_result` and the immediate `queued`-cancellation path both clear `cancel_requested_at` on reaching any terminal status. -- [ ] 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`. +- [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 diff --git a/packages/cloud-platform/cloud/db_models.py b/packages/cloud-platform/cloud/db_models.py index d23783a..2dbd981 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/migrations/versions/0012_task_cancellation.py b/packages/cloud-platform/cloud/migrations/versions/0012_task_cancellation.py new file mode 100644 index 0000000..033c319 --- /dev/null +++ b/packages/cloud-platform/cloud/migrations/versions/0012_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 = "0012_task_cancellation" +down_revision = "0011_planner_decision_log_reflection" +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 2929cce..f57b454 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 @@ -485,7 +501,7 @@ class CloudRepository(Protocol): lease_expires_at: datetime, now: datetime, progress: AssignmentProgressSnapshot | None = None, - ) -> LeaseRenewalStatus: ... + ) -> LeaseRenewalResult: ... def record_task_result( self, @@ -500,6 +516,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 1764eae..337093d 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 = "0011_planner_decision_log_reflection" +HEAD_REVISION = "0012_task_cancellation" class SchemaVersionError(RuntimeError): diff --git a/packages/cloud-platform/cloud/sql_repository.py b/packages/cloud-platform/cloud/sql_repository.py index 238099d..153af3a 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( @@ -2223,6 +2273,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/tests/test_cloud_repository_contract.py b/tests/test_cloud_repository_contract.py index b595e56..db3c335 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: @@ -1810,3 +1811,251 @@ def test_record_planner_decision_stores_null_rationale_and_thinking( assert decisions[0].thinking 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_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