dmux · Oct 27, 2026 · Oct 26, 2026 · Oct 27, 2026
diff --git a/src/minilake/services/clusters.py b/src/minilake/services/clusters.py
@@ -46,6 +46,7 @@
 router = APIRouter(prefix="/api/2.1/clusters", tags=["clusters"])
 
 _state: Dict[str, Any] = {"clusters": {}}
+_transition_tokens: Dict[str, str] = {}
 
 
 def _validate_references(policy_id: Any, instance_pool_id: Any) -> None:
@@ -81,9 +82,19 @@ def _get_or_404(cluster_id: str) -> dict:
     return cluster
 
 
-async def _transition(cluster_id: str, delay_seconds: float, target_state: str) -> None:
-    """Background transition: flips a cluster's state after a real delay."""
+def _schedule_transition(cluster_id: str, delay_seconds: float, target_state: str) -> None:
+    """Schedule a transition that supersedes any earlier operation for the cluster."""
+    token = str(uuid.uuid4())
+    _transition_tokens[cluster_id] = token
+    asyncio.create_task(_transition(cluster_id, token, delay_seconds, target_state))
+
+
+async def _transition(cluster_id: str, token: str, delay_seconds: float, target_state: str) -> None:
+    """Apply a delayed transition only if no newer lifecycle operation replaced it."""
     await asyncio.sleep(delay_seconds)
+    if _transition_tokens.get(cluster_id) != token:
+        return
+    _transition_tokens.pop(cluster_id, None)
     cluster = _state["clusters"].get(cluster_id)
     if cluster is None:
         return
@@ -131,7 +142,7 @@ async def create_cluster(req: CreateClusterRequest) -> CreateClusterResponse:
         "default_tags": {},
     }
     _state["clusters"][cluster_id] = cluster
-    asyncio.create_task(_transition(cluster_id, settings.cluster_start_delay_seconds, "RUNNING"))
+    _schedule_transition(cluster_id, settings.cluster_start_delay_seconds, "RUNNING")
     logger.info(f"Created cluster: {cluster_id} ({req.cluster_name})")
     return CreateClusterResponse(cluster_id=cluster_id)
 
@@ -238,7 +249,7 @@ async def start_cluster(req: ClusterIdRequest) -> ClusterInfo:
     cluster["state"] = "PENDING"
     cluster["state_message"] = "Starting cluster"
     cluster["terminated_time"] = None
-    asyncio.create_task(_transition(req.cluster_id, settings.cluster_start_delay_seconds, "RUNNING"))
+    _schedule_transition(req.cluster_id, settings.cluster_start_delay_seconds, "RUNNING")
     return _to_info(cluster)
 
 
@@ -250,7 +261,7 @@ async def terminate_cluster(req: ClusterIdRequest) -> ClusterInfo:
         return _to_info(cluster)
     cluster["state"] = "TERMINATING"
     cluster["state_message"] = "Terminating cluster"
-    asyncio.create_task(_transition(req.cluster_id, settings.cluster_terminate_delay_seconds, "TERMINATED"))
+    _schedule_transition(req.cluster_id, settings.cluster_terminate_delay_seconds, "TERMINATED")
     return _to_info(cluster)
 
 
@@ -259,6 +270,7 @@ async def permanent_delete_cluster(req: ClusterIdRequest) -> dict:
     """Permanently remove a cluster's record (unlike /delete, no history kept)."""
     _get_or_404(req.cluster_id)
     del _state["clusters"][req.cluster_id]
+    _transition_tokens.pop(req.cluster_id, None)
     return {}
 
 
@@ -269,7 +281,7 @@ async def restart_cluster(req: RestartClusterRequest) -> ClusterInfo:
     cluster["state"] = "RESTARTING"
     cluster["state_message"] = "Restarting cluster"
     cluster["last_restarted_time"] = int(time.time() * 1000)
-    asyncio.create_task(_transition(req.cluster_id, settings.cluster_start_delay_seconds, "RUNNING"))
+    _schedule_transition(req.cluster_id, settings.cluster_start_delay_seconds, "RUNNING")
     return _to_info(cluster)
 
 
@@ -283,7 +295,7 @@ async def resize_cluster(req: ResizeClusterRequest) -> ClusterInfo:
         cluster["autoscale"] = req.autoscale
     cluster["state"] = "RESIZING"
        cluster["state_message"] = "Resizing cluster"
-    asyncio.create_task(_transition(req.cluster_id, settings.cluster_start_delay_seconds, "RUNNING"))
+    _schedule_transition(req.cluster_id, settings.cluster_start_delay_seconds, "RUNNING")
     return _to_info(cluster)
 
 
@@ -371,3 +383,4 @@ def restore_state(data: Dict[str, Any]) -> None:
 async def reset() -> None:
     global _state
     _state = {"clusters": {}}
+    _transition_tokens.clear()
diff --git a/tests/test_clusters.py b/tests/test_clusters.py
@@ -3,6 +3,8 @@
 until RUNNING/TERMINATED, exercising real polling logic against minilake's
 timed state transitions, not an instant canned response)."""
 
+import os
+import time
 from uuid import uuid4
 
 import pytest
@@ -30,6 +32,27 @@ def test_cluster_create_reaches_running(workspace_client: WorkspaceClient):
     print(f"✓ Cluster reached RUNNING for real: {cluster.cluster_id}")
 
 
+@pytest.mark.crud
+def test_terminating_cluster_cancels_pending_start(workspace_client: WorkspaceClient):
+    """A create transition must not undo a later termination request."""
+    created = workspace_client.clusters.create(
+        spark_version=_spark_version(),
+        node_type_id="Standard_DS3_v2",
+        num_workers=1,
+        cluster_name=f"cluster_{uuid4().hex[:6]}",
+    )
+    cluster_id = created.response.cluster_id
+
+    workspace_client.clusters.delete(cluster_id=cluster_id)
+
+    start_delay = float(os.getenv("MINILAKE_CLUSTER_START_DELAY", "1"))
+    terminate_delay = float(os.getenv("MINILAKE_CLUSTER_TERMINATE_DELAY", "0.5"))
+    time.sleep(max(start_delay, terminate_delay) + 0.5)
+
+    cluster = workspace_client.clusters.get(cluster_id=cluster_id)
+    assert cluster.state == State.TERMINATED
+
+
 @pytest.mark.crud
 def test_cluster_get_matches_created_config(workspace_client: WorkspaceClient):
     """Test: GET returns the same config used at creation."""