diff --git a/autowsgr/server/device_lease.py b/autowsgr/server/device_lease.py
new file mode 100644
index 00000000..cca12d09
--- /dev/null
+++ b/autowsgr/server/device_lease.py
@@ -0,0 +1,81 @@
+"""Exclusive ownership for operations that drive the shared emulator."""
+
+from __future__ import annotations
+
+import threading
+from dataclasses import dataclass
+from functools import wraps
+from typing import TYPE_CHECKING, Any
+
+
+if TYPE_CHECKING:
+    from collections.abc import Awaitable, Callable
+
+
+class DeviceOperationBusyError(RuntimeError):
+    """Raised when another operation already owns the shared device."""
+
+
+@dataclass(frozen=True)
+class DeviceOperationToken:
+    """Opaque ownership token used for compare-and-release semantics."""
+
+    owner: str
+    identity: object
+
+
+class DeviceOperationLease:
+    """A non-blocking, token-owned lease for the shared emulator."""
+
+    def __init__(self) -> None:
+        self._lock = threading.Lock()
+        self._token: DeviceOperationToken | None = None
+
+    def acquire(self, owner: str) -> DeviceOperationToken:
+        """Acquire ownership immediately or report that the device is busy."""
+        with self._lock:
+            if self._token is not None:
+                raise DeviceOperationBusyError(f'设备正由 {self._token.owner} 使用')
+            token = DeviceOperationToken(owner=owner, identity=object())
+            self._token = token
+            return token
+
+    def release(self, token: DeviceOperationToken) -> None:
+        """Release ownership only when the exact active token is supplied."""
+        with self._lock:
+            if self._token is token:
+                self._token = None
+
+    @property
+    def owner(self) -> str | None:
+        """Return the current owner for status and diagnostics."""
+        with self._lock:
+            return self._token.owner if self._token is not None else None
+
+
+device_operation_lease = DeviceOperationLease()
+
+
+def exclusive_device_operation(
+    owner: str,
+) -> Callable[[Callable[..., Awaitable[Any]]], Callable[..., Awaitable[Any]]]:
+    """Reject concurrent HTTP device operations with a consistent 409 response."""
+    from fastapi import HTTPException
+
+    def decorator(
+        handler: Callable[..., Awaitable[Any]],
+    ) -> Callable[..., Awaitable[Any]]:
+        @wraps(handler)
+        async def wrapped(*args: Any, **kwargs: Any) -> Any:
+            try:
+                token = device_operation_lease.acquire(owner)
+            except DeviceOperationBusyError as error:
+                raise HTTPException(status_code=409, detail=str(error)) from error
+            try:
+                return await handler(*args, **kwargs)
+            finally:
+                device_operation_lease.release(token)
+
+        return wrapped
+
+    return decorator
diff --git a/autowsgr/server/routes/game.py b/autowsgr/server/routes/game.py
index 355d713a..00f719d5 100644
--- a/autowsgr/server/routes/game.py
+++ b/autowsgr/server/routes/game.py
@@ -7,6 +7,7 @@
 from fastapi import APIRouter, HTTPException
 
 from autowsgr.infra.logger import get_logger
+from autowsgr.server.device_lease import exclusive_device_operation
 from autowsgr.server.schemas import ApiResponse
 from autowsgr.server.serializers import (
     serialize_build_queue,
@@ -14,7 +15,6 @@
     serialize_fleet,
     serialize_resources,
 )
-from autowsgr.server.task_manager import task_manager
 
 from ..main import get_context
 
@@ -25,6 +25,7 @@
 
 
 @router.get('/api/game/acquisition', response_model=ApiResponse)
+@exclusive_device_operation('api:game-acquisition')
 async def game_acquisition() -> ApiResponse:
     """从出征面板截图 OCR 识别今日舰船 (X/500) 与战利品 (X/50) 获取数量。
 
@@ -35,9 +36,6 @@ async def game_acquisition() -> ApiResponse:
     except RuntimeError as e:
         raise HTTPException(status_code=503, detail=str(e)) from e
 
-    if task_manager.is_running:
-        raise HTTPException(status_code=409, detail='任务执行中，无法查询获取数量')
-
     from autowsgr.ops.navigate import goto_page
     from autowsgr.ui.map.page import MapPage
 
diff --git a/autowsgr/server/routes/ops.py b/autowsgr/server/routes/ops.py
index 9411b252..531f5708 100644
--- a/autowsgr/server/routes/ops.py
+++ b/autowsgr/server/routes/ops.py
@@ -9,8 +9,8 @@
 from pydantic import BaseModel
 
 from autowsgr.infra.logger import get_logger
+from autowsgr.server.device_lease import exclusive_device_operation
 from autowsgr.server.schemas import ApiResponse
-from autowsgr.server.task_manager import task_manager
 
 from ..main import get_context
 
@@ -20,16 +20,11 @@
 router = APIRouter(tags=['ops'])
 
 
-def _require_idle() -> None:
-    """检查是否有任务正在运行。"""
-    if task_manager.is_running:
-        raise HTTPException(status_code=409, detail='任务执行中，无法操作')
-
-
 # ── 远征收取 ──
 
 
 @router.post('/api/expedition/check', response_model=ApiResponse)
+@exclusive_device_operation('api:expedition-check')
 async def expedition_check() -> ApiResponse:
     """检查并收取已完成的远征。"""
     try:
@@ -37,8 +32,6 @@ async def expedition_check() -> ApiResponse:
     except RuntimeError as e:
         raise HTTPException(status_code=503, detail=str(e)) from e
 
-    _require_idle()
-
     from autowsgr.ops.expedition import collect_expedition
 
     try:
@@ -61,10 +54,11 @@ class ExpeditionAutoCheckRequest(BaseModel):
 
 
 @router.post('/api/expedition/auto_check', response_model=ApiResponse)
+@exclusive_device_operation('api:expedition-auto-check')
 async def expedition_auto_check(request: ExpeditionAutoCheckRequest) -> ApiResponse:
     """自动远征检查（挂机专用）。
 
-    不受 _require_idle 限制，顺带领取任务奖励并根据战斗任务状态智能执行浴室维修。
+    顺带领取任务奖励并根据调用方配置决定是否执行浴室维修。
     """
     try:
         ctx = get_context()
@@ -92,11 +86,7 @@ async def expedition_auto_check(request: ExpeditionAutoCheckRequest) -> ApiRespo
         results['rewards_error'] = str(e)
 
     # 3. 浴室维修
-    if task_manager.is_running:
-        _log.info('[API] 自动远征检查: 战斗任务进行中，跳过浴室维修')
-        results['repair_skipped'] = True
-        results['repair_reason'] = '战斗任务进行中'
-    elif not request.allow_repair:
+    if not request.allow_repair:
         _log.info('[API] 自动远征检查: 前端禁止维修（队列中还有后续任务），跳过浴室维修')
         results['repair_skipped'] = True
         results['repair_reason'] = '队列中有后续战斗任务'
@@ -130,6 +120,7 @@ class BuildStartRequest(BaseModel):
 
 
 @router.post('/api/build/collect', response_model=ApiResponse)
+@exclusive_device_operation('api:build-collect')
 async def build_collect() -> ApiResponse:
     """收取已完成的建造。"""
     try:
@@ -137,8 +128,6 @@ async def build_collect() -> ApiResponse:
     except RuntimeError as e:
         raise HTTPException(status_code=503, detail=str(e)) from e
 
-    _require_idle()
-
     from autowsgr.ops import collect_built_ships
 
     try:
@@ -150,6 +139,7 @@ async def build_collect() -> ApiResponse:
 
 
 @router.post('/api/build/start', response_model=ApiResponse)
+@exclusive_device_operation('api:build-start')
 async def build_start(request: BuildStartRequest) -> ApiResponse:
     """开始建造。"""
     try:
@@ -157,8 +147,6 @@ async def build_start(request: BuildStartRequest) -> ApiResponse:
     except RuntimeError as e:
         raise HTTPException(status_code=503, detail=str(e)) from e
 
-    _require_idle()
-
     from autowsgr.ops import BuildRecipe, build_ship
 
     recipe = BuildRecipe(
@@ -186,6 +174,7 @@ async def build_start(request: BuildStartRequest) -> ApiResponse:
 
 
 @router.post('/api/reward/collect', response_model=ApiResponse)
+@exclusive_device_operation('api:reward-collect')
 async def reward_collect() -> ApiResponse:
     """收取任务奖励。"""
     try:
@@ -193,8 +182,6 @@ async def reward_collect() -> ApiResponse:
     except RuntimeError as e:
         raise HTTPException(status_code=503, detail=str(e)) from e
 
-    _require_idle()
-
     from autowsgr.ops import collect_rewards
 
     try:
@@ -216,6 +203,7 @@ class CookRequest(BaseModel):
 
 
 @router.post('/api/cook', response_model=ApiResponse)
+@exclusive_device_operation('api:cook')
 async def cook_action(request: CookRequest) -> ApiResponse:
     """食堂烹饪。"""
     try:
@@ -223,8 +211,6 @@ async def cook_action(request: CookRequest) -> ApiResponse:
     except RuntimeError as e:
         raise HTTPException(status_code=503, detail=str(e)) from e
 
-    _require_idle()
-
     from autowsgr.ops import cook
 
     try:
@@ -241,6 +227,7 @@ async def cook_action(request: CookRequest) -> ApiResponse:
 
 
 @router.post('/api/repair/bath', response_model=ApiResponse)
+@exclusive_device_operation('api:repair-bath')
 async def repair_bath() -> ApiResponse:
     """浴室修理。"""
     try:
@@ -248,8 +235,6 @@ async def repair_bath() -> ApiResponse:
     except RuntimeError as e:
         raise HTTPException(status_code=503, detail=str(e)) from e
 
-    _require_idle()
-
     from autowsgr.ops import repair_in_bath
 
     try:
@@ -267,6 +252,7 @@ class RepairShipRequest(BaseModel):
 
 
 @router.post('/api/repair/ship', response_model=ApiResponse)
+@exclusive_device_operation('api:repair-ship')
 async def repair_ship(request: RepairShipRequest) -> ApiResponse:
     """使用浴室修理指定名称的舰船。
 
@@ -278,8 +264,6 @@ async def repair_ship(request: RepairShipRequest) -> ApiResponse:
     except RuntimeError as e:
         raise HTTPException(status_code=503, detail=str(e)) from e
 
-    _require_idle()
-
     from autowsgr.ops.repair import repair_ship_by_name
 
     try:
@@ -310,6 +294,7 @@ class DestroyRequest(BaseModel):
 
 
 @router.post('/api/destroy', response_model=ApiResponse)
+@exclusive_device_operation('api:destroy')
 async def destroy_action(request: DestroyRequest) -> ApiResponse:
     """解装/解体舰船。"""
     try:
@@ -317,8 +302,6 @@ async def destroy_action(request: DestroyRequest) -> ApiResponse:
     except RuntimeError as e:
         raise HTTPException(status_code=503, detail=str(e)) from e
 
-    _require_idle()
-
     from autowsgr.ops import destroy_ships
     from autowsgr.types import ShipType
 
diff --git a/autowsgr/server/routes/system.py b/autowsgr/server/routes/system.py
index 437f726d..2918a222 100644
--- a/autowsgr/server/routes/system.py
+++ b/autowsgr/server/routes/system.py
@@ -4,10 +4,15 @@
 
 import asyncio
 
-from fastapi import APIRouter
+from fastapi import APIRouter, HTTPException
 from pydantic import BaseModel
 
 from autowsgr.infra.logger import get_logger
+from autowsgr.server.device_lease import (
+    DeviceOperationBusyError,
+    device_operation_lease,
+    exclusive_device_operation,
+)
 from autowsgr.server.schemas import ApiResponse
 from autowsgr.server.task_manager import task_manager
 
@@ -27,6 +32,7 @@ class SystemStartRequest(BaseModel):
 
 
 @router.post('/start', response_model=ApiResponse)
+@exclusive_device_operation('api:system-start')
 async def system_start(request: SystemStartRequest) -> ApiResponse:
     """启动系统 (连接模拟器、启动游戏)。"""
     async with _main.lifecycle_lock:
@@ -69,9 +75,16 @@ async def system_stop() -> ApiResponse:
                 error='任务未在超时前停止，系统上下文仍保持活动状态',
             )
 
-        _main._ctx = None
-        _log.info('[System] 系统已停止')
-        return ApiResponse(success=True, message='系统已停止')
+        try:
+            lease_token = device_operation_lease.acquire('api:system-stop')
+        except DeviceOperationBusyError as error:
+            raise HTTPException(status_code=409, detail=str(error)) from error
+        try:
+            _main._ctx = None
+            _log.info('[System] 系统已停止')
+            return ApiResponse(success=True, message='系统已停止')
+        finally:
+            device_operation_lease.release(lease_token)
 
 
 @router.get('/status', response_model=ApiResponse)
diff --git a/autowsgr/server/routes/task.py b/autowsgr/server/routes/task.py
index 418addd9..3e0e7a98 100644
--- a/autowsgr/server/routes/task.py
+++ b/autowsgr/server/routes/task.py
@@ -8,6 +8,7 @@
 from pydantic import Discriminator
 
 from autowsgr.infra.logger import get_logger
+from autowsgr.server.device_lease import DeviceOperationBusyError
 from autowsgr.server.schemas import (
     ApiResponse,
     CampaignRequest,
@@ -47,18 +48,21 @@ async def task_start(request: TaskRequestUnion) -> ApiResponse:  # type: ignore[
 
         ctx.stop_event = task_manager.stop_event
 
-        if isinstance(request, NormalFightRequest):
-            return await _start_normal_fight(ctx, request)
-        elif isinstance(request, EventFightRequest):
-            return await _start_event_fight(ctx, request)
-        elif isinstance(request, CampaignRequest):
-            return await _start_campaign(ctx, request)
-        elif isinstance(request, ExerciseRequest):
-            return await _start_exercise(ctx, request)
-        elif isinstance(request, DecisiveRequest):
-            return await _start_decisive(ctx, request)
-        else:
-            raise HTTPException(status_code=400, detail='未知的任务类型')
+        try:
+            if isinstance(request, NormalFightRequest):
+                return await _start_normal_fight(ctx, request)
+            elif isinstance(request, EventFightRequest):
+                return await _start_event_fight(ctx, request)
+            elif isinstance(request, CampaignRequest):
+                return await _start_campaign(ctx, request)
+            elif isinstance(request, ExerciseRequest):
+                return await _start_exercise(ctx, request)
+            elif isinstance(request, DecisiveRequest):
+                return await _start_decisive(ctx, request)
+            else:
+                raise HTTPException(status_code=400, detail='未知的任务类型')
+        except DeviceOperationBusyError as error:
+            raise HTTPException(status_code=409, detail=str(error)) from error
 
 
 @router.post('/stop', response_model=ApiResponse)
diff --git a/autowsgr/server/task_manager.py b/autowsgr/server/task_manager.py
index c6a3a4fa..f4be1c0a 100644
--- a/autowsgr/server/task_manager.py
+++ b/autowsgr/server/task_manager.py
@@ -11,6 +11,11 @@
 from typing import TYPE_CHECKING, Any
 
 from autowsgr.infra.logger import get_logger
+from autowsgr.server.device_lease import (
+    DeviceOperationLease,
+    DeviceOperationToken,
+    device_operation_lease,
+)
 from autowsgr.server.ws_manager import ws_manager
 
 
@@ -107,12 +112,13 @@ class TaskManager:
     所有战斗操作在后台线程执行，避免阻塞事件循环。
     """
 
-    def __init__(self) -> None:
+    def __init__(self, device_lease: DeviceOperationLease | None = None) -> None:
         self._current_task: TaskInfo | None = None
         self._executor_thread: threading.Thread | None = None
         self._stop_event = threading.Event()
         self._lock = threading.Lock()
         self._loop: asyncio.AbstractEventLoop | None = None
+        self._device_lease = device_lease or device_operation_lease
 
     @property
     def current_task(self) -> TaskInfo | None:
@@ -160,6 +166,7 @@ def start_task(
                 raise RuntimeError('已有任务正在运行')
 
             task_id = f'task_{uuid.uuid4().hex[:8]}'
+            lease_token = self._device_lease.acquire(f'task:{task_id}')
             self._current_task = TaskInfo(
                 task_id=task_id,
                 task_type=task_type,
@@ -172,10 +179,16 @@ def start_task(
             # 启动后台线程执行
             self._executor_thread = threading.Thread(
                 target=self._run_in_thread,
-                args=(executor,),
+                args=(executor, lease_token),
                 daemon=True,
             )
-            self._executor_thread.start()
+            try:
+                self._executor_thread.start()
+            except Exception:
+                self._current_task = None
+                self._executor_thread = None
+                self._device_lease.release(lease_token)
+                raise
 
             _log.info('[Task] 启动任务: {} ({})', task_id, task_type)
             return task_id
@@ -183,6 +196,7 @@ def start_task(
     def _run_in_thread(
         self,
         executor: Callable[[TaskInfo], TaskOutcome],
+        lease_token: DeviceOperationToken,
     ) -> None:
         """在后台线程中执行任务。"""
         assert self._current_task is not None
@@ -211,6 +225,7 @@ def _run_in_thread(
             _log.error('[Task] 任务失败: {} - {}', task.task_id, e)
 
         finally:
+            self._device_lease.release(lease_token)
             # 通过事件循环发送 WebSocket 通知
             if self._loop:
                 asyncio.run_coroutine_threadsafe(
diff --git a/testing/server/test_device_lease.py b/testing/server/test_device_lease.py
new file mode 100644
index 00000000..4d3f4ce5
--- /dev/null
+++ b/testing/server/test_device_lease.py
@@ -0,0 +1,72 @@
+"""Shared emulator ownership tests."""
+
+from __future__ import annotations
+
+import asyncio
+
+import pytest
+from fastapi import HTTPException
+
+from autowsgr.server.device_lease import (
+    DeviceOperationLease,
+    exclusive_device_operation,
+)
+
+
+def test_stale_token_cannot_release_new_owner() -> None:
+    """Only the exact current owner can release the device."""
+    lease = DeviceOperationLease()
+    first = lease.acquire('first')
+    lease.release(first)
+    second = lease.acquire('second')
+
+    lease.release(first)
+
+    assert lease.owner == 'second'
+    lease.release(second)
+    assert lease.owner is None
+
+
+def test_http_device_operation_rejects_busy_owner(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Device-driving HTTP operations return 409 instead of waiting."""
+    lease = DeviceOperationLease()
+    monkeypatch.setattr(
+        'autowsgr.server.device_lease.device_operation_lease',
+        lease,
+    )
+    active = lease.acquire('task:active')
+    called = False
+
+    @exclusive_device_operation('api:test')
+    async def operation() -> None:
+        nonlocal called
+        called = True
+
+    with pytest.raises(HTTPException) as exc_info:
+        asyncio.run(operation())
+
+    assert exc_info.value.status_code == 409
+    assert called is False
+    lease.release(active)
+
+
+def test_http_device_operation_releases_after_failure(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """An operation exception cannot leak device ownership."""
+    lease = DeviceOperationLease()
+    monkeypatch.setattr(
+        'autowsgr.server.device_lease.device_operation_lease',
+        lease,
+    )
+
+    @exclusive_device_operation('api:test')
+    async def operation() -> None:
+        raise RuntimeError('failed')
+
+    with pytest.raises(RuntimeError, match='failed'):
+        asyncio.run(operation())
+
+    assert lease.owner is None
diff --git a/testing/server/test_device_routes.py b/testing/server/test_device_routes.py
new file mode 100644
index 00000000..b3e214ff
--- /dev/null
+++ b/testing/server/test_device_routes.py
@@ -0,0 +1,53 @@
+"""Device-driving route conflict tests."""
+
+from __future__ import annotations
+
+import asyncio
+
+import pytest
+from fastapi import HTTPException
+
+from autowsgr.server import main as server_main
+from autowsgr.server.device_lease import device_operation_lease
+from autowsgr.server.routes import game, ops
+
+
+@pytest.mark.parametrize(
+    'route_call',
+    [
+        ops.expedition_check,
+        lambda: ops.expedition_auto_check(ops.ExpeditionAutoCheckRequest()),
+        ops.build_collect,
+        lambda: ops.build_start(ops.BuildStartRequest()),
+        ops.reward_collect,
+        lambda: ops.cook_action(ops.CookRequest()),
+        ops.repair_bath,
+        lambda: ops.repair_ship(ops.RepairShipRequest(ship_name='测试舰')),
+        lambda: ops.destroy_action(ops.DestroyRequest()),
+        game.game_acquisition,
+    ],
+)
+def test_device_routes_reject_active_owner_before_context_access(
+    monkeypatch: pytest.MonkeyPatch,
+    route_call: object,
+) -> None:
+    """Every direct device route shares the same non-blocking lease."""
+    context_reads = 0
+
+    def get_context() -> object:
+        nonlocal context_reads
+        context_reads += 1
+        return object()
+
+    monkeypatch.setattr(ops, 'get_context', get_context)
+    monkeypatch.setattr(game, 'get_context', get_context)
+    monkeypatch.setattr(server_main, '_ctx', object())
+    token = device_operation_lease.acquire('task:active')
+    try:
+        with pytest.raises(HTTPException) as exc_info:
+            asyncio.run(route_call())  # type: ignore[operator]
+    finally:
+        device_operation_lease.release(token)
+
+    assert exc_info.value.status_code == 409
+    assert context_reads == 0
diff --git a/testing/server/test_system_routes.py b/testing/server/test_system_routes.py
index ff1cc80d..b0691b97 100644
--- a/testing/server/test_system_routes.py
+++ b/testing/server/test_system_routes.py
@@ -11,6 +11,7 @@
 from fastapi import HTTPException
 
 from autowsgr.server import main as server_main
+from autowsgr.server.device_lease import device_operation_lease
 from autowsgr.server.routes import system, task
 from autowsgr.server.schemas import ExerciseRequest
 
@@ -118,6 +119,22 @@ def launch(_config_path: str) -> object:
     assert server_main._ctx is None
 
 
+def test_system_start_rejects_active_device_owner(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """System launch cannot connect or navigate while another operation owns the device."""
+    monkeypatch.setattr(server_main, '_ctx', None)
+    token = device_operation_lease.acquire('api:repair')
+    try:
+        with pytest.raises(HTTPException) as exc_info:
+            asyncio.run(system.system_start(system.SystemStartRequest()))
+    finally:
+        device_operation_lease.release(token)
+
+    assert exc_info.value.status_code == 409
+    assert server_main._ctx is None
+
+
 def test_system_stop_keeps_context_when_worker_does_not_finish(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
@@ -207,3 +224,22 @@ async def run_race() -> None:
 
     assert server_main._ctx is None
     assert started_contexts == []
+
+
+def test_system_stop_rejects_non_task_device_owner(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Context teardown cannot race an active direct device operation."""
+    ctx = object()
+    manager = _TerminalTaskManager(completed=True)
+    monkeypatch.setattr(server_main, '_ctx', ctx)
+    monkeypatch.setattr(system, 'task_manager', manager)
+    token = device_operation_lease.acquire('api:repair')
+    try:
+        with pytest.raises(HTTPException) as exc_info:
+            asyncio.run(system.system_stop())
+    finally:
+        device_operation_lease.release(token)
+
+    assert exc_info.value.status_code == 409
+    assert server_main._ctx is ctx
diff --git a/testing/server/test_task_manager.py b/testing/server/test_task_manager.py
index f775f5a7..c06deb29 100644
--- a/testing/server/test_task_manager.py
+++ b/testing/server/test_task_manager.py
@@ -5,6 +5,9 @@
 import threading
 import time
 
+import pytest
+
+from autowsgr.server.device_lease import DeviceOperationBusyError, DeviceOperationLease
 from autowsgr.server.task_manager import TaskManager, TaskOutcome, TaskStatus
 
 
@@ -109,3 +112,118 @@ def executor(_task: object) -> TaskOutcome:
     release_worker.set()
     assert manager.wait_for_completion(timeout=1) is True
     assert manager.current_task.status is TaskStatus.STOPPED
+
+
+def test_task_owns_device_until_worker_exits() -> None:
+    """A second task cannot start until the active worker releases the device."""
+    lease = DeviceOperationLease()
+    first_manager = TaskManager(device_lease=lease)
+    second_manager = TaskManager(device_lease=lease)
+    worker_started = threading.Event()
+    release_worker = threading.Event()
+
+    def blocking_executor(_task: object) -> TaskOutcome:
+        worker_started.set()
+        release_worker.wait(timeout=1)
+        return TaskOutcome.from_results([{'round': 1, 'success': True}])
+
+    first_manager.start_task(
+        task_type='normal_fight',
+        total_rounds=1,
+        executor=blocking_executor,
+    )
+    assert worker_started.wait(timeout=1)
+
+    with pytest.raises(DeviceOperationBusyError):
+        second_manager.start_task(
+            task_type='exercise',
+            total_rounds=1,
+            executor=lambda _task: TaskOutcome.from_results(
+                [{'round': 1, 'success': True}],
+            ),
+        )
+
+    assert second_manager.current_task is None
+
+    release_worker.set()
+    assert first_manager.wait_for_completion(timeout=1) is True
+
+    second_manager.start_task(
+        task_type='exercise',
+        total_rounds=1,
+        executor=lambda _task: TaskOutcome.from_results(
+            [{'round': 1, 'success': True}],
+        ),
+    )
+    assert second_manager.wait_for_completion(timeout=1) is True
+
+
+@pytest.mark.parametrize(
+    'executor',
+    [
+        lambda _task: TaskOutcome(results=[], success=False, error='failed'),
+        lambda _task: (_ for _ in ()).throw(RuntimeError('crashed')),
+    ],
+)
+def test_task_releases_device_after_failure(executor: object) -> None:
+    """Failed outcomes and unexpected exceptions both release ownership."""
+    lease = DeviceOperationLease()
+    manager = TaskManager(device_lease=lease)
+
+    manager.start_task(
+        task_type='normal_fight',
+        total_rounds=1,
+        executor=executor,  # type: ignore[arg-type]
+    )
+    assert manager.wait_for_completion(timeout=1) is True
+
+    token = lease.acquire('next-operation')
+    lease.release(token)
+
+
+def test_stop_request_does_not_release_device_before_worker_exit() -> None:
+    """Cooperative cancellation retains ownership while code still runs."""
+    lease = DeviceOperationLease()
+    manager = TaskManager(device_lease=lease)
+    worker_started = threading.Event()
+    release_worker = threading.Event()
+
+    def executor(_task: object) -> TaskOutcome:
+        worker_started.set()
+        release_worker.wait(timeout=1)
+        return TaskOutcome.from_results([{'round': 1, 'success': True}])
+
+    manager.start_task(task_type='normal_fight', total_rounds=1, executor=executor)
+    assert worker_started.wait(timeout=1)
+    assert manager.stop_task() is True
+
+    with pytest.raises(DeviceOperationBusyError):
+        lease.acquire('next-operation')
+
+    release_worker.set()
+    assert manager.wait_for_completion(timeout=1) is True
+    token = lease.acquire('next-operation')
+    lease.release(token)
+
+
+def test_thread_start_failure_rolls_back_task_and_device(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Failure to launch the worker cannot publish a task or leak ownership."""
+    lease = DeviceOperationLease()
+    manager = TaskManager(device_lease=lease)
+
+    def fail_start(_thread: threading.Thread) -> None:
+        raise RuntimeError('thread failed')
+
+    monkeypatch.setattr(threading.Thread, 'start', fail_start)
+
+    with pytest.raises(RuntimeError, match='thread failed'):
+        manager.start_task(
+            task_type='normal_fight',
+            total_rounds=1,
+            executor=lambda _task: TaskOutcome.from_results([]),
+        )
+
+    assert manager.current_task is None
+    assert lease.owner is None
diff --git a/testing/server/test_task_routes.py b/testing/server/test_task_routes.py
index c72d3419..27364ee3 100644
--- a/testing/server/test_task_routes.py
+++ b/testing/server/test_task_routes.py
@@ -9,6 +9,7 @@
 from fastapi import HTTPException
 
 from autowsgr.server import main as server_main
+from autowsgr.server.device_lease import DeviceOperationBusyError
 from autowsgr.server.routes import task
 from autowsgr.server.schemas import (
     CampaignRequest,
@@ -46,6 +47,24 @@ def test_task_start_requires_system_context(monkeypatch: pytest.MonkeyPatch) ->
     assert exc_info.value.status_code == 503
 
 
+def test_task_start_reports_device_conflict(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Task lease conflicts are returned synchronously as HTTP 409."""
+    manager = _TaskManager()
+    ctx = type('Context', (), {'stop_event': None})()
+
+    async def busy_start(_ctx: object, _request: ExerciseRequest) -> object:
+        raise DeviceOperationBusyError('设备正由 api:repair 使用')
+
+    monkeypatch.setattr(task, 'task_manager', manager)
+    monkeypatch.setattr(server_main, '_ctx', ctx)
+    monkeypatch.setattr(task, '_start_exercise', busy_start)
+
+    with pytest.raises(HTTPException) as exc_info:
+        asyncio.run(task.task_start(ExerciseRequest()))
+
+    assert exc_info.value.status_code == 409
+
+
 @pytest.mark.parametrize(
     ('task_request', 'helper_name'),
     [