feat: add task deadlines and cancellation

CHANGELOG.md

@@ -28,6 +28,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   distinguish `READY_TO_ATTEMPT`, `NOT_READY`, and `INDETERMINATE` without
   running an agent task. Registry, `AgentKit`, and provider-diagnostics helpers
   expose the same async checks.
+- Absolute `AgentTask.deadline` values, `AgentKit.run(timeout=...)`, typed
+  `AgentTaskTimeoutError`, and `FinishReason.TIMED_OUT` provide bounded task
+  execution across provider startup and reused-process queueing.
+- Built-in runtimes and `AgentKit.cancel()` now return immutable
+  `CancellationReceipt` values with explicit request dispositions; legacy
+  third-party `cancel()` methods returning `None` remain supported.
 
 ### Changed
 
@@ -69,6 +75,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   during static preflight instead of surfacing later or being silently ignored.
 - A configured model allow-list now fails closed when model selection is
   provider-native and therefore cannot be verified locally.
+- Cancellation now binds to one active task generation, unregisters without a
+  cancellable lock window, emits at most one timeout/cancellation terminal, and
+  gives reusable provider-process teardown a bounded grace period under repeated
+  cancellation. Runtime instances stay quarantined from new work while detached
+  teardown remains pending.
 
 ## 0.4.0 - 2026-07-02
 

README.md

@@ -178,5 +178,6 @@ built-in runtime populates it yet, so it is always an empty tuple today.
 - [Capability matrix](https://github.com/ebarti/agent-runtime-kit/blob/main/docs/capability-matrix.md)
 - [API stability](https://github.com/ebarti/agent-runtime-kit/blob/main/docs/api-stability.md)
 - [Live smoke tests](https://github.com/ebarti/agent-runtime-kit/blob/main/docs/live-smoke.md)
+- [Deadlines and cancellation](https://github.com/ebarti/agent-runtime-kit/blob/main/docs/task-control.md)
 - [Mestre migration notes](https://github.com/ebarti/agent-runtime-kit/blob/main/docs/mestre-migration.md)
 - [SDK evolution agent](https://github.com/ebarti/agent-runtime-kit/blob/main/docs/sdk-evolution-agent.md)

docs/api-stability.md

@@ -23,6 +23,11 @@ import the names from the top-level package instead.
   party can ship an adapter for a new runtime without forking the enum.
 - **Every runtime exposes the async lifecycle** (`aclose`, `async with`) declared
   by the `AgentRuntime` protocol. Stateless runtimes implement it as a no-op.
+- **Cancellation receipts describe requests, not rollback.** Built-in runtimes
+  return an immutable `CancellationReceipt`; the protocol still permits `None`
+  for existing third-party runtimes, which `AgentKit` reports as
+  `LEGACY_UNCONFIRMED`. `AgentTask.deadline` is keyword-only and an expired task
+  raises `AgentTaskTimeoutError` with `FinishReason.TIMED_OUT` event semantics.
 - **`finish_reason` values** come from `FinishReason`. The field is typed `str`
   for forward-compatibility, so new reasons can be added without a type break;
   compare against `FinishReason` members rather than bare literals.

docs/capability-matrix.md

@@ -12,6 +12,7 @@
 | Permission mapping | `permission_mode` | approval mode + sandbox | capabilities + policies |
 | Streaming output events | Yes — incremental `output.delta` while the SDK runs | No — a single `output.delta` at the end (non-streaming SDK) | Yes — from response chunks |
 | Tool audit events | Yes — streamed from message blocks | Yes — emitted from parsed `TurnResult` items after the turn | Yes — from tool chunks |
+| Deadlines and cancellation | Absolute deadline + coroutine cancellation | Absolute deadline + coroutine cancellation | Absolute deadline + coroutine cancellation |
 | `vendor.turn` events | No | No | Yes — from thought/unknown chunks |
 | Package availability | Sync, side-effect-free, package-only | Sync, side-effect-free, package-only | Sync, side-effect-free, package-only; never calls ADC |
 | Async readiness | Direct API key/OAuth/Bedrock bearer signal → `READY_TO_ATTEMPT`; provider/local chains → `INDETERMINATE` | Supported `AsyncCodex.account(refresh_token=False)` probe; absent account → `NOT_READY` | API key or off-loop ADC/project probe; missing → `NOT_READY` |
@@ -33,6 +34,11 @@ For every provider, malformed schemas are rejected locally before dispatch and
 returned structured values are validated locally after the SDK completes.
 `parsed_output_available` distinguishes valid JSON `null` from no parsed value.
 
+Deadlines cover provider startup and waits for a reused process. Cancellation
+receipts acknowledge a request at the coroutine boundary; they do not imply
+rollback of tool side effects that already completed. See
+[Deadlines and cancellation](task-control.md).
+
 ## Permission mapping
 
 A single portable `PermissionMode` maps to each vendor's native controls. The

docs/task-control.md

@@ -0,0 +1,61 @@
+# Deadlines and cancellation
+
+`AgentTask.deadline` is an absolute, timezone-aware `datetime`. It applies to
+the whole provider operation, including SDK startup and waiting for a reused
+provider process. `AgentKit.run(timeout=...)` is a convenience that converts a
+finite, non-negative number of seconds (or `timedelta`) into one absolute
+deadline at call time.
+
+```python
+from agent_runtime_kit import AgentKit, AgentTaskTimeoutError
+
+try:
+    result = await kit.run("codex", goal="Refactor the parser", timeout=30)
+except AgentTaskTimeoutError as exc:
+    print(exc.task_id, exc.deadline)
+```
+
+An expired deadline never starts the vendor SDK (and therefore emits no started
+event). A deadline expiry emits an `agent.task.failed` event with
+`finish_reason="timed_out"`, cancels the in-flight provider coroutine, gives its
+cleanup a bounded five-second grace period, and raises `AgentTaskTimeoutError`
+(which is both an `AgentRuntimeError` and `TimeoutError`). Direct adapter calls
+honor `AgentTask(deadline=...)` too. If provider teardown remains wedged after
+that grace, the runtime instance is quarantined and rejects new runs rather than
+overlapping them; it becomes reusable when the detached cleanup finally settles.
+
+To cancel a task started through `AgentKit`, keep its task id and use the same
+runtime instance or cached kind:
+
+```python
+import asyncio
+
+task_id = "index-repository"
+running = asyncio.create_task(
+    kit.run("claude", goal="Index the repository", task_id=task_id)
+)
+receipt = await kit.cancel("claude", task_id)
+
+try:
+    await running
+except asyncio.CancelledError:
+    pass
+
+print(receipt.disposition)
+```
+
+`cancel()` does not construct a runtime that has not already been cached, and
+it does not wait for the cancelled run to settle. Built-in adapters also expose
+the same method directly. `CancellationReceipt.disposition` distinguishes a
+new request, a repeated request, an inactive id, an unsupported hook, a failed
+hook, and a legacy runtime that returned no receipt.
+
+An active `(runtime, task_id)` identifies one run generation. If a legacy
+task-id-only cancellation hook is still settling, `AgentKit` keeps that id
+reserved rather than allowing the delayed hook to target a replacement run.
+
+A `REQUESTED` receipt confirms only that cancellation was requested at the
+runtime coroutine boundary. It does not promise rollback: commands, network
+requests, or other tool side effects that completed before cancellation may be
+permanent. Await the original run task to observe completion of provider
+cleanup before reusing related resources.

pyproject.toml

@@ -112,6 +112,7 @@ include = [
     "docs/quickstart.md",
     "docs/sdk-evolution-agent-design.md",
     "docs/sdk-evolution-agent.md",
+    "docs/task-control.md",
 ]
 
 [tool.ruff]

src/agent_runtime_kit/__init__.py

@@ -3,6 +3,7 @@
 from agent_runtime_kit._errors import (
     AgentRuntimeError,
     AgentRuntimeUnavailableError,
+    AgentTaskTimeoutError,
     OutputSchemaError,
     OutputTypeError,
     RuntimeNotRegisteredError,
@@ -18,6 +19,8 @@
     AgentTask,
     ArtifactRef,
     AvailabilityReason,
+    CancellationDisposition,
+    CancellationReceipt,
     EventSink,
     FilesystemAccess,
     FinishReason,
@@ -66,8 +69,11 @@
     "AgentRuntimeKind",
     "AgentRuntimeUnavailableError",
     "AgentTask",
+    "AgentTaskTimeoutError",
     "ArtifactRef",
     "AvailabilityReason",
+    "CancellationDisposition",
+    "CancellationReceipt",
     "COMPATIBILITY_MANIFEST",
     "DEFAULT_READINESS_TIMEOUT",
     "EventSink",

src/agent_runtime_kit/_control.py

@@ -0,0 +1,234 @@
+"""Portable task deadline and cancellation control.
+
+Cancellation is cooperative at the Python coroutine boundary.  A successful
+receipt means that cancellation was requested; it cannot promise that vendor
+tools have rolled back side effects that already happened.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from collections.abc import Awaitable, Callable
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from typing import Any, TypeVar
+
+from agent_runtime_kit._errors import AgentRuntimeError, AgentTaskTimeoutError
+from agent_runtime_kit._types import (
+    AgentRuntimeKind,
+    AgentTask,
+    CancellationDisposition,
+    CancellationReceipt,
+    FinishReason,
+)
+from agent_runtime_kit.events import safe_emit, task_failed_event
+
+_T = TypeVar("_T")
+_OPERATION_CLEANUP_TIMEOUT = 5.0
+
+
+@dataclass
+class _ActiveRun:
+    task: asyncio.Task[Any]
+    cancellation_requested: bool = False
+
+
+class RuntimeTaskController:
+    """Track active runs for one runtime and enforce their deadlines."""
+
+    def __init__(self, kind: AgentRuntimeKind | str) -> None:
+        self.kind = AgentRuntimeKind.coerce(kind)
+        self._active: dict[str, _ActiveRun] = {}
+        self._quarantined: set[asyncio.Future[Any]] = set()
+
+    async def run(
+        self,
+        task: AgentTask,
+        operation: Callable[[], Awaitable[_T]],
+    ) -> _T:
+        """Run ``operation`` under task-id registration and deadline control."""
+
+        current = asyncio.current_task()
+        if current is None:  # pragma: no cover - asyncio always supplies one here
+            raise AgentRuntimeError("task control requires an active asyncio task")
+        self._quarantined = {item for item in self._quarantined if not item.done()}
+        if self._quarantined:
+            raise AgentRuntimeError(
+                f"{self.kind} has a cancelled operation still cleaning up; "
+                "wait for cleanup before starting another run"
+            )
+        # These registry operations deliberately contain no await.  asyncio tasks
+        # share one event-loop thread, so the check-and-set is atomic without a
+        # cancellable lock acquisition.  The same is true of cleanup below.
+        existing = self._active.get(task.task_id)
+        if existing is not None:
+            raise AgentRuntimeError(
+                f"task_id {task.task_id!r} is already active for {self.kind}"
+            )
+        self._active[task.task_id] = _ActiveRun(current)
+
+        try:
+            return await self._run_with_deadline(task, operation)
+        finally:
+            active = self._active.get(task.task_id)
+            if active is not None and active.task is current:
+                del self._active[task.task_id]
+
+    async def cancel(
+        self,
+        task_id: str,
+        *,
+        _expected_task: asyncio.Task[Any] | None = None,
+    ) -> CancellationReceipt:
+        """Request cancellation of an active coroutine without blocking on it."""
+
+        active = self._active.get(task_id)
+        if active is None or (
+            _expected_task is not None and active.task is not _expected_task
+        ):
+            return self._receipt(task_id, CancellationDisposition.NOT_ACTIVE)
+        if active.cancellation_requested:
+            return self._receipt(task_id, CancellationDisposition.ALREADY_REQUESTED)
+        active.cancellation_requested = True
+        accepted = active.task.cancel()
+        if not accepted:
+            return self._receipt(
+                task_id,
+                CancellationDisposition.FAILED,
+                "the active asyncio task had already completed",
+            )
+        return self._receipt(task_id, CancellationDisposition.REQUESTED)
+
+    async def _run_with_deadline(
+        self,
+        task: AgentTask,
+        operation: Callable[[], Awaitable[_T]],
+    ) -> _T:
+        deadline = task.deadline
+        if deadline is None:
+            try:
+                return await operation()
+            except asyncio.CancelledError:
+                await self._emit_interrupted(task)
+                raise
+
+        remaining = (deadline - datetime.now(tz=timezone.utc)).total_seconds()
+        if remaining <= 0:
+            # An already-expired task never starts, including its started event.
+            # Emit the terminal first so cancellation of a blocking observability
+            # sink cannot strand the lifecycle after a misleading start.
+            await self._emit_timed_out(task, deadline)
+            raise AgentTaskTimeoutError(self.kind, task.task_id, deadline)
+
+        operation_task: asyncio.Future[_T] = asyncio.ensure_future(operation())
+        timed_out = False
+        try:
+            done, _ = await asyncio.wait(
+                {operation_task},
+                timeout=remaining,
+                return_when=asyncio.FIRST_COMPLETED,
+            )
+            if operation_task in done:
+                return await operation_task
+
+            # Transition before emitting: cancellation of a slow sink must never
+            # turn one timeout into a second, contradictory interrupted terminal.
+            timed_out = True
+            if not await _cancel_and_settle(operation_task):
+                self._quarantine(operation_task)
+            await self._emit_timed_out(task, deadline)
+            raise AgentTaskTimeoutError(self.kind, task.task_id, deadline)
+        except asyncio.CancelledError:
+            if not await _cancel_and_settle(operation_task):
+                self._quarantine(operation_task)
+            if not timed_out:
+                await self._emit_interrupted(task)
+            raise
+
+    async def _emit_interrupted(self, task: AgentTask) -> None:
+        await safe_emit(
+            task,
+            task_failed_event(
+                task,
+                self.kind,
+                error="task cancellation requested",
+                finish_reason=FinishReason.INTERRUPTED.value,
+            ),
+        )
+
+    async def _emit_timed_out(self, task: AgentTask, deadline: datetime) -> None:
+        await safe_emit(
+            task,
+            task_failed_event(
+                task,
+                self.kind,
+                error=f"task exceeded deadline {deadline.isoformat()}",
+                finish_reason=FinishReason.TIMED_OUT.value,
+            ),
+        )
+
+    def _receipt(
+        self,
+        task_id: str,
+        disposition: CancellationDisposition,
+        message: str | None = None,
+    ) -> CancellationReceipt:
+        return CancellationReceipt(
+            task_id=task_id,
+            kind=self.kind,
+            disposition=disposition,
+            message=message,
+        )
+
+    def _quarantine(self, operation: asyncio.Future[Any]) -> None:
+        self._quarantined.add(operation)
+
+        def release(completed: asyncio.Future[Any]) -> None:
+            self._quarantined.discard(completed)
+            _consume_operation_result(completed)
+
+        operation.add_done_callback(release)
+
+
+async def run_legacy_with_deadline(
+    task: AgentTask,
+    kind: AgentRuntimeKind | str,
+    operation: Callable[[], Awaitable[_T]],
+) -> _T:
+    """Enforce deadlines for third-party runtimes without native task control."""
+
+    # A fresh controller is intentional: AgentKit owns cancellation tracking for
+    # legacy runtimes, while this helper owns only this invocation's deadline and
+    # terminal event semantics.
+    controller = RuntimeTaskController(kind)
+    return await controller.run(task, operation)
+
+
+async def _cancel_and_settle(operation: asyncio.Future[Any]) -> bool:
+    """Give a cancelled operation bounded, repeat-cancel-safe cleanup time."""
+
+    operation.cancel()
+    deadline = asyncio.get_running_loop().time() + _OPERATION_CLEANUP_TIMEOUT
+    while not operation.done():
+        remaining = deadline - asyncio.get_running_loop().time()
+        if remaining <= 0:
+            operation.cancel()
+            return False
+        try:
+            await asyncio.wait_for(asyncio.shield(operation), timeout=remaining)
+        except asyncio.CancelledError:
+            continue
+        except asyncio.TimeoutError:
+            operation.cancel()
+            return False
+        except Exception:
+            break
+    _consume_operation_result(operation)
+    return True
+
+
+def _consume_operation_result(operation: asyncio.Future[Any]) -> None:
+    try:
+        operation.exception()
+    except BaseException:
+        return