GPUs follow the engine's own CUDA_VISIBLE_DEVICES

Its entries are the cards present, and a local child's allocation is
named through them, so under =2,3 card 0 is device 2. Remote workers
keep mapping against their own mask.

backend/fluksio/flow/controller.py

@@ -979,7 +979,7 @@ class FlowController:
             with placer.claim(  # type: ignore[union-attr]
                 wanted, device=device, policy=policy, node=node_id, run=run_id
             ) as (target, allocation):
-                env = derive_env(wanted, allocation)
+                env = derive_env(wanted, allocation, local=target.worker is None)
                 if target.worker is None:
                     return pool.for_env(env).run(  # type: ignore[union-attr]
                         owner,

backend/fluksio/flow/resources.py

@@ -96,15 +96,32 @@ def warn_oversubscribed(limits: dict[str, int]) -> None:
         )
 
 
+def visible_gpus() -> list[str] | None:
+    """The cards the engine's own ``CUDA_VISIBLE_DEVICES`` lists, if it is set.
+
+    Indices, UUIDs or MIG instances, kept as written: a child is handed the
+    same names, so whatever the driver took here it takes there.
+    """
+    mask = os.environ.get("CUDA_VISIBLE_DEVICES")
+    if mask is None:
+        return None
+    return [part.strip() for part in mask.split(",") if part.strip()]
+
+
 def machine_gpus() -> int:
     """Cards a node may be given here, when nobody said.
 
-    The device nodes the NVIDIA driver creates, counted — no vendor tool, so
+    The engine's own ``CUDA_VISIBLE_DEVICES`` where it has one, since a card
+    the process can see but was not given is not its to hand out. Otherwise
+    the device nodes the NVIDIA driver creates, counted — no vendor tool, so
     the one dependency does not become two. It answers for the case that
     actually bit people (a box with cards, an engine told nothing, every GPU
     node clamped to zero and running at once) and for nothing else: another
     vendor, or a card this process cannot see, is still `--gpus N`.
     """
+    visible = visible_gpus()
+    if visible is not None:
+        return len(visible)
     return len(glob.glob("/dev/nvidia[0-9]*"))
 
 
@@ -229,7 +246,9 @@ class ResourceAccountant:
 # -----------------------------------------------------------------------------
 
 
-def derive_env(wanted: Resources, allocation: Allocation) -> dict[str, str]:
+def derive_env(
+    wanted: Resources, allocation: Allocation, local: bool = True
+) -> dict[str, str]:
     """The environment a worker running this node is started with.
 
     Precedence is engine environment, then what the allocation implies, then
@@ -240,10 +259,19 @@ def derive_env(wanted: Resources, allocation: Allocation) -> dict[str, str]:
     contents depend on the version installed, so writing it here would silently
     replace whatever the author had put there. It travels in ``resources.env``,
     where it is visible.
+
+    ``local`` is whether the allocation is this machine's. A child's
+    ``CUDA_VISIBLE_DEVICES`` replaces the engine's rather than narrowing it, so
+    here a card is named the way the engine's own list names it: under ``=2,3``
+    card 0 is device 2. A remote worker does the same against its own list.
     """
     env = {var: str(allocation.cpus) for var in THREAD_VARS}
     if wanted.gpus:
-        env["CUDA_VISIBLE_DEVICES"] = ",".join(str(index) for index in allocation.gpus)
+        names = (visible_gpus() if local else None) or []
+        env["CUDA_VISIBLE_DEVICES"] = ",".join(
+            names[index] if index < len(names) else str(index)
+            for index in allocation.gpus
+        )
     env.update(wanted.env)
     return env
 

backend/tests/flow/test_resources.py

@@ -33,6 +33,7 @@ def test_the_cards_are_counted_from_the_devices_the_driver_makes(monkeypatch):
     """
     from fluksio.flow import resources
 
+    monkeypatch.delenv("CUDA_VISIBLE_DEVICES", raising=False)
     monkeypatch.setattr(
         resources.glob,
         "glob",
@@ -150,12 +151,37 @@ def test_the_share_becomes_the_thread_limit():
     assert "CUDA_VISIBLE_DEVICES" not in env
 
 
-def test_a_gpu_node_is_told_which_card_is_its():
+def test_a_gpu_node_is_told_which_card_is_its(monkeypatch):
+    monkeypatch.delenv("CUDA_VISIBLE_DEVICES", raising=False)
     env = derive_env(Resources(cpus=1, gpus=2), Allocation(cpus=1, gpus=(1, 3)))
 
     assert env["CUDA_VISIBLE_DEVICES"] == "1,3"
 
 
+@pytest.mark.parametrize(
+    ("mask", "cards", "named"),
+    [(None, 2, "0"), ("2,3", 2, "2"), ("GPU-a1b2, MIG-c3d4", 2, "GPU-a1b2")],
+)
+def test_the_engines_own_mask_is_the_cards_it_has(monkeypatch, mask, cards, named):
+    """Under ``CUDA_VISIBLE_DEVICES=2,3`` card 0 is device 2, not device 0.
+
+    A child's mask replaces the engine's, so an unmapped 0 would be a card
+    the engine was never given. A remote worker maps against its own mask.
+    """
+    from fluksio.flow import resources
+
+    monkeypatch.setattr(resources.glob, "glob", lambda pattern: ["/dev/nvidia0"] * 2)
+    if mask is None:
+        monkeypatch.delenv("CUDA_VISIBLE_DEVICES", raising=False)
+    else:
+        monkeypatch.setenv("CUDA_VISIBLE_DEVICES", mask)
+    wanted, allocation = Resources(cpus=1, gpus=1), Allocation(cpus=1, gpus=(0,))
+
+    assert machine_gpus() == cards
+    assert derive_env(wanted, allocation)["CUDA_VISIBLE_DEVICES"] == named
+    assert derive_env(wanted, allocation, local=False)["CUDA_VISIBLE_DEVICES"] == "0"
+
+
 def test_what_the_node_asked_for_wins():
     """A declaration outranks what the allocation would imply — it is deliberate."""
     wanted = Resources(