feat(runtime): add cancellation-aware stop_reason to TaskRunner and WorkflowRunner

- TaskRunner.run() and WorkflowRunner.run()/resume() accept an optional
  stop_reason callable alongside should_stop, distinguishing a genuine
  cancellation from other stop conditions (e.g. lost lease).
- is_cancellation_reason() shared helper added to runtime/task.py.
- WorkflowRunner._stop_status() now branches cancelled/failed based on
  stop_reason, correcting a prior blanket cancelled-on-any-stop behavior
  that conflicted with the host-agent-protocol spec's requirement to
  distinguish cancellation from lease-loss stops.
- Default behavior (stop_reason=None) is preserved exactly for both
  runners so existing callers/tests are unaffected.
- Task 1 of openspec change task-cancellation.

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,

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",