diff --git a/apps/device-host-agent/host_agent/assignment.py b/apps/device-host-agent/host_agent/assignment.py
index 42d3146..28ce190 100644
--- a/apps/device-host-agent/host_agent/assignment.py
+++ b/apps/device-host-agent/host_agent/assignment.py
@@ -2,7 +2,7 @@ from __future__ import annotations
 
 from collections.abc import Callable
 from dataclasses import dataclass, field
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 from cloud.internal_api.models import AssignmentModel
 from core.models import Task
@@ -11,6 +11,9 @@ from host_agent.planner_context import bind_planner_execution_context
 from host_agent.progress import TaskProgressHolder, TaskProgressSnapshot
 from runtime.task import is_cancellation_reason
 
+if TYPE_CHECKING:
+    from host_agent.mcp_lock import McpBusyTracker
+
 
 @dataclass(frozen=True)
 class AssignmentExecutionResult:
@@ -20,9 +23,15 @@ class AssignmentExecutionResult:
 
 
 class AssignmentExecutor:
-    def __init__(self, factories: ExecutionFactories) -> None:
+    def __init__(
+        self,
+        factories: ExecutionFactories,
+        *,
+        mcp_busy_tracker: McpBusyTracker | None = None,
+    ) -> None:
         self.factories = factories
         self._progress = TaskProgressHolder()
+        self._mcp_busy_tracker = mcp_busy_tracker
 
     def latest_progress(self) -> TaskProgressSnapshot | None:
         """Latest step progress reported by the currently-running assignment."""
@@ -36,6 +45,16 @@ class AssignmentExecutor:
         stop_reason: Callable[[], str | None] | None = None,
     ) -> AssignmentExecutionResult:
         self._progress.clear()
+        if self._mcp_busy_tracker is not None and (
+            assignment.device_id in self._mcp_busy_tracker.busy_device_ids()
+        ):
+            return AssignmentExecutionResult(
+                status="failed",
+                failure_reason=(
+                    f"device {assignment.device_id} is held by an active "
+                    "MCP session"
+                ),
+            )
         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
diff --git a/apps/device-host-agent/tests/test_assignment.py b/apps/device-host-agent/tests/test_assignment.py
index 9818ce2..b27180c 100644
--- a/apps/device-host-agent/tests/test_assignment.py
+++ b/apps/device-host-agent/tests/test_assignment.py
@@ -180,6 +180,63 @@ def test_workflow_assignment_maps_cancellation_stop_to_cancelled_status() -> Non
     }
 
 
+def test_execute_fails_fast_when_mcp_session_holds_device() -> None:
+    """Cloud assignment arriving for a device currently held by an MCP
+    session must fail immediately rather than fight for the device."""
+    from host_agent.mcp_lock import McpBusyTracker
+
+    tracker = McpBusyTracker()
+    tracker.acquire("phone-1", "sess-mcp")
+    executor = AssignmentExecutor(
+        _build_factories(),
+        mcp_busy_tracker=tracker,
+    )
+    assignment = _assignment(device_id="phone-1")
+    result = executor.execute(assignment)
+    assert result.status == "failed"
+    assert "MCP" in (result.failure_reason or "")
+
+
+def test_execute_skips_check_when_tracker_is_none() -> None:
+    """Default backward-compat: no tracker → no fail-fast."""
+    executor = AssignmentExecutor(_build_factories())
+    # Without a real workflow store / task runner this test verifies the
+    # entry-point path doesn't raise on the mcp_busy check.
+    # We use a goal + a mock runner factory so execute() runs through.
+    assignment = _assignment()
+    result = executor.execute(assignment)
+    # Should run through normally (not fail on MCP check)
+    assert result.status == "done"
+
+
+def _build_factories() -> ExecutionFactories:
+    """Shared factory fixture used by MCP-hold tests."""
+    received: list[Task] = []
+
+    class FakeTaskRunner:
+        def run(self, task: Task) -> Task:
+            received.append(task)
+            task.status = "completed"
+            return task
+
+    class FakeMetadataStore:
+        def create_task(
+            self,
+            task: Task,
+            *,
+            source_task_id: str | None = None,
+            source_attempt: int | None = None,
+        ) -> None:
+            pass
+
+    return 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]
+        metadata_store=FakeMetadataStore(),  # type: ignore[arg-type]
+    )
+
+
 def test_unknown_workflow_fails_without_running() -> None:
     class FakeWorkflowStore:
         def get_definition(self, definition_id: str):