fix(azure): isolate worker slot conflicts

bin/fm-worker-lifecycle.py

@@ -1465,6 +1465,28 @@ def inventory_by_slot(inventory):
     return {worker["slot"]: worker for worker in inventory["workers"]}
 
 
+def inventory_conflict_slots(inventory):
+    """Return conflicts that the provider bound to one exact worker slot.
+
+    A conflict in one disposable compartment must not freeze unrelated slots.
+    An unscoped or malformed conflict still refuses globally because there is
+    no exact boundary within which the controller can safely contain it.
+    """
+    slots = set()
+    for conflict in inventory.get("conflicts", []):
+        slot = conflict.get("slot") if isinstance(conflict, dict) else None
+        if (
+            not isinstance(slot, int)
+            or isinstance(slot, bool)
+            or not 1 <= slot <= MAX_WORKERS
+        ):
+            raise LifecycleError(
+                "provider conflict is not bound to one exact worker slot"
+            )
+        slots.add(slot)
+    return slots
+
+
 def resource_identity(resource):
     return {
         "id": resource.get("id"),
@@ -2545,6 +2567,7 @@ def refresh_classifications(state, inventory, now=None):
 def choose_free_slot(env, state, inventory):
     occupied = set(int(slot) for slot in state["workers"])
     occupied.update(worker["slot"] for worker in inventory["workers"])
+    occupied.update(inventory_conflict_slots(inventory))
     for slot in range(1, env["max_workers"] + 1):
         if slot not in occupied:
             return slot
@@ -2605,9 +2628,7 @@ def service_worker_for_key(state, key):
 def next_reconcile_action(env, state, inventory, now=None):
     now = now or now_utc()
     cloud = inventory_by_slot(inventory)
-    conflicts = inventory.get("conflicts", [])
-    if conflicts:
-        raise LifecycleError("provider found same-fleet worker-name conflicts; unrelated resources were not adopted")
+    conflicted_slots = inventory_conflict_slots(inventory)
     # The first planning pass of a new UTC day snapshots the day's spend
     # baseline, whether or not any new compute is wanted; the caller's save
     # makes it durable.
@@ -2627,6 +2648,12 @@ def next_reconcile_action(env, state, inventory, now=None):
             # shield, and the post-convergence drain owns the replay.
             continue
         worker = state["workers"][slot_key]
+        if worker["slot"] in conflicted_slots:
+            worker["last_classification"] = "retained-for-investigation"
+            worker["classification_note"] = (
+                "provider reported a conflict inside this exact worker slot"
+            )
+            continue
         current = cloud.get(worker["slot"])
         classification, note = classify_worker(worker, current, now=now)
         worker["last_classification"] = classification
@@ -2686,10 +2713,7 @@ def next_service_reconcile_action(env, state, inventory, task, generation, now=N
     provider action must never become work the caller synchronously owns.
     """
     now = now or now_utc()
-    if inventory.get("conflicts"):
-        raise LifecycleError(
-            "provider found same-fleet worker-name conflicts; unrelated resources were not adopted"
-        )
+    conflicted_slots = inventory_conflict_slots(inventory)
     key = request_key(task, generation)
     item = state["queue"].get(key)
     if item is None or item.get("role") != "no-mistakes":
@@ -2714,6 +2738,12 @@ def next_service_reconcile_action(env, state, inventory, task, generation, now=N
         raise LifecycleError("service reconcile task has no exact durable worker owner")
     if worker.get("role") != "no-mistakes":
         raise LifecycleError("service reconcile worker role differs")
+    if worker["slot"] in conflicted_slots:
+        raise LifecycleError(
+            "provider found a conflict inside service worker slot {}; the exact slot was not adopted".format(
+                worker["slot"]
+            )
+        )
     if slot_key in (state.get("pending_actions") or {}):
         # The command drains this exact claim before planning. Seeing it here
         # means another process still owns the slot lease; wait without

docs/azure-workers.md

@@ -331,6 +331,7 @@ Every live read classifies each controller-owned slot into exactly one operator
 - `retained-for-investigation` means identity, state, or ownership is missing, conflicting, or unlanded and nothing may be deleted or duplicated.
 
 The Azure adapter inventories only names in the reviewed resource group and separates same-name foreign owner or generation conflicts.
+An exact-slot conflict quarantines that compartment while unrelated slots continue reconciling; an unscoped or malformed conflict still refuses the fleet because it has no safe containment boundary.
 It never adopts unrelated subscription resources.
 The monitor extension, bootstrap/execute Run Commands, and TTL schedule use their exact ARM resource IDs as identity; Azure `provisioningState` is mutable lifecycle state, not identity.
 Create, execute, and steer still require ready monitor/bootstrap children, while deallocation and exact cleanup tolerate Azure's ordinary post-deallocation state transition and retain the resource-ID, parent-VM, tag, script, and digest fences.
@@ -408,7 +409,7 @@ The guest verifies the runtime's exact file inventory, runs the role command wit
 The wrapper verifies the semantic bytes and head binding before writing the controller-facing result; a process exit, missing outcome, malformed outcome, or changed read-only head is a failed result, never `CLEAR` by inference.
 Repair results return one digest-bound single-ref bundle whose head must descend from the requested head, while review and test return no code bundle and must keep the exact requested head.
 The wrapper records a retryable local candidate before cleanup, releases through `service-complete` only after the lifecycle owns the exact execution result, and replays the candidate after a lost response instead of executing the step again.
-Admission, execute recovery, and cleanup use `service-reconcile`, which advances only the caller's exact task generation or replays that task's own pending slot claim rather than converging unrelated fleet work. Once a service task owns a slot, its execution and cleanup inventory reads that slot's deterministic Azure object paths in a bounded parallel batch; it neither lists the resource group nor expands peer children. Queued admission retains the whole-fleet quota, spend, and conflict census.
+Admission, execute recovery, and cleanup use `service-reconcile`, which advances only the caller's exact task generation or replays that task's own pending slot claim rather than converging unrelated fleet work. Once a service task owns a slot, its execution and cleanup inventory reads that slot's deterministic Azure object paths in a bounded parallel batch; it neither lists the resource group nor expands peer children. Queued admission retains the whole-fleet quota, spend, and conflict census, excludes every conflicted slot from placement, and continues through conflicts bound to other exact slots.
 The guest supervisor marks a no-mistakes Azure execution as the already-isolated test boundary, so the repository test command runs the focused service suite directly instead of recursively provisioning the general validation fleet or a Herdr lab.
 The root-owned supervisor stages the job, then runs the no-mistakes process as the dedicated non-root `fmworker` user with no supplementary groups; the sealed runtime remains root-owned and read-only while the exact repository and projected account are writable only by that service identity.
 `bin/fm-azure-service-test-scope.py` owns that focused inventory and the narrow source set eligible for focused pull-request CI; an empty, mixed, or unknown diff and every push to `main` retain the complete behavior suite.

tests/fm-worker-lifecycle.test.sh

@@ -58,7 +58,10 @@ env = {
     "max_workers": 4, "deployment_generation": "dep", "owner": "owner",
     "subscription": "subscription",
 }
-inventory = {"metrics": {}, "workers": [], "capacity_reservations": [], "conflicts": []}
+inventory = {
+    "metrics": {}, "workers": [], "capacity_reservations": [],
+    "conflicts": [{"slot": 1, "kind": "run-command", "reason": "unrelated"}],
+}
 module.roll_daily_baseline = lambda *_args, **_kwargs: None
 module.active_count = lambda *_args, **_kwargs: 1
 module.choose_free_slot = lambda *_args, **_kwargs: 2
@@ -89,6 +92,18 @@ action = module.next_service_reconcile_action(
 )
 assert action["type"] == "create" and admitted == [("service", "generation", 2)], action
 
+# Conflict isolation is exact. A malformed provider conflict with no slot
+# remains a global refusal because the controller cannot contain it safely.
+try:
+    module.next_service_reconcile_action(
+        env, state, dict(inventory, conflicts=[{"kind": "run-command"}]),
+        "service", "generation",
+    )
+except module.LifecycleError as exc:
+    assert "not bound to one exact worker slot" in str(exc), exc
+else:
+    raise AssertionError("an unscoped provider conflict was treated as isolated")
+
 def worker(task, generation, slot, status):
     key = module.request_key(task, generation)
     item = {
@@ -208,6 +223,18 @@ action = module.next_service_reconcile_action(
 )
 assert action["type"] == "deallocate" and action["slot"] == 2, action
 
+# A conflict on the target slot still refuses that exact service mutation.
+try:
+    module.next_service_reconcile_action(
+        env, state,
+        dict(inventory, conflicts=[{"slot": 2, "kind": "run-command"}]),
+        "service", "generation",
+    )
+except module.LifecycleError as exc:
+    assert "inside service worker slot 2" in str(exc), exc
+else:
+    raise AssertionError("a conflicted service slot was adopted")
+
 # The operator's ordinary reconciler keeps its existing fleet-wide policy.
 # It still selects a released ordinary worker before queued admission.
 ordinary = dict(target)
@@ -226,7 +253,11 @@ ordinary_state = {
     "workers": {"1": ordinary}, "pending_actions": {},
 }
 module.classify_worker = lambda *_args, **_kwargs: ("assigned", "released ordinary")
-action = module.next_reconcile_action(env, ordinary_state, inventory)
+ordinary_inventory = dict(
+    inventory,
+    conflicts=[{"slot": 2, "kind": "run-command", "reason": "unrelated"}],
+)
+action = module.next_reconcile_action(env, ordinary_state, ordinary_inventory)
 assert action["type"] == "deallocate" and action["slot"] == 1, action
 PY
   pass "exact service admission, recovery, and cleanup ignore unrelated slow fleet work"
@@ -8326,6 +8357,26 @@ def build_assigned(slot):
     }
     return build_state, worker, item, cloud, inventory
 
+# --- one exact conflicted compartment cannot freeze unrelated cleanup or be
+# selected for fresh admission. The conflicting slot itself stays parked.
+cstate, cworker, citem, ccloud, cinventory = build_assigned(2)
+cworker["release_proof"] = {"proof_digest": "c" * 64}
+cinventory["conflicts"] = [
+    {"slot": 1, "kind": "run-command", "reason": "undeclared child"}
+]
+action = module.next_reconcile_action(penv, cstate, cinventory, now=T0)
+assert action["type"] == "deallocate" and action["slot"] == 2, action
+assert module.choose_free_slot(
+    dict(penv, max_workers=4), {"workers": {"1": {}}},
+    {"workers": [], "conflicts": [{"slot": 2}]},
+) == 3
+cinventory["conflicts"] = [
+    {"slot": 2, "kind": "run-command", "reason": "undeclared child"}
+]
+assert module.next_reconcile_action(penv, cstate, cinventory, now=T0) is None
+assert cworker["last_classification"] == "retained-for-investigation"
+assert "inside this exact worker slot" in cworker["classification_note"]
+
 # --- idle deallocate: provable end-of-task signals only, at the threshold and
 # never before, never with a claim, never on already-dark compute.
 istate, worker, item, cloud, inventory = build_assigned(1)