Add full-volume fiber tracing and scalable chunk-cache scheduling (#2359)

* Fix fiber normal export, accumulator scheduling and compact pyramid execution

* Preserve pyramid policies and legacy stage order in compact output

* Default fiber volume inference to native network output resolution

* Cache-block normal encoding without changing output bytes

* Keep inference coordination nonblocking and use chunk-local mmap accumulation

* Add Qualzz AFV inference with managed GPUs and paired probability Zarr arrays

* Establish whole-volume chunk tracing and queued loading baseline

* Add reusable approved fiber test and benchmark runner

* Accelerate chunk ROI reads and reuse decoded buffers

Reduce keyed-reader and queue wakeups, keep batch scheduling internal, reuse final-reader-released decode allocations within the shared budget, and use blocked ROI layout conversion. Add codec/ownership/ROI regressions and standalone multi-channel reader benchmarks with documented results.

* Record matched tracing throughput after decoder acceleration

* Preserve fast cache-backed fiber tracing and evaluated route policies

Unify scalar-equivalent channel sampling over retained decoded chunks, with shared addressing and optional profiling. Include accumulated tracing eligibility, reverse convergence, chunk ownership and export changes with regression coverage and benchmark documentation. Planning and local experiment artifacts are excluded.

* Extract reusable fiber volume coverage and collision policies

* Add resumable native continuous volume fiber tracing

* Prevent shared chunk-cache eviction stalls and stranded trace waiters

* Record chosen beam-node observations in fiber trace commits

* Persist extensible point and segment attributes in volume fibers

* Register attributed fiber pilot and output audit workflows

* Allow managed commands to accept partial GPU allocations

Author: hendrikschilling

gpu_manager/README.md

@@ -217,6 +217,17 @@ grants without starting workers and retries registration until the requested
 allocation is available. This is opt-in; it can wait indefinitely and each new
 registration joins the FIFO queue anew.
 
+For commands that can size their workers at startup, `accept_partial=True`
+instead accepts any nonempty granted GPU subset and positive CPU count within
+the request. Each admitted GPU must still have its requested VRAM budget.
+This conflicts with `wait_for_exact`. The child receives only admitted GPUs in
+`CUDA_VISIBLE_DEVICES` (in requested order), and the actual allocation as JSON
+in `VILLA_GPU_MANAGER_GRANT` (`cpu_cores`, `gpus` mapping UUIDs to bytes).
+The child must size CPU/GPU workers to this allocation. It remains fixed for
+the cohort lifetime; this is admission flexibility, not live worker resizing.
+Ownership checks and join-before-release cleanup remain unchanged. The default
+continues to require the complete requested allocation.
+
 `ProcessOwner` provides a separate ownership root for sibling cohorts managed
 by one coordinator. Pass its identity as `Client(registration_root=...)`,
 track each worker session before authorizing it, and retire it only after joining

gpu_manager/src/villa_gpu_manager/exclusive.py

@@ -201,7 +201,7 @@ class ExclusiveTask:
     """
     def __init__(self, path, grant, *, label, priority=0, poll_seconds=.1,
                  client_factory=Client, stop_event=None, cleanup_timeout=30.0, exclusive=True,
-                 wait_for_exact=False, exact_retry_seconds=5.0):
+                 wait_for_exact=False, exact_retry_seconds=5.0, accept_partial=False):
         if type(exclusive) is not bool:
             raise ValueError("exclusive must be boolean")
         if type(priority) is not int or poll_seconds <= 0 or cleanup_timeout <= 0:
@@ -211,6 +211,7 @@ def __init__(self, path, grant, *, label, priority=0, poll_seconds=.1,
             raise ValueError("exclusive task requires positive GPU budgets and CPUs")
         sizes = list(grant["gpus"].values())
         self.grant = deepcopy(grant)
+        self.requested_grant = deepcopy(grant)
         # The locked scheduler scales CPU promises with the sum of GPU bytes.
         # Compensate for heterogeneous full-device capacities so the resulting
         # CPU grant is exact too (ceil in cores_for cancels this floor).
@@ -225,6 +226,9 @@ def __init__(self, path, grant, *, label, priority=0, poll_seconds=.1,
         if type(wait_for_exact) is not bool or not math.isfinite(exact_retry_seconds) or exact_retry_seconds <= 0:
             raise ValueError("boolean exact waiting and positive retry interval required")
         self.wait_for_exact, self.exact_retry_seconds = wait_for_exact, exact_retry_seconds
+        if type(accept_partial) is not bool or (accept_partial and wait_for_exact):
+            raise ValueError("accept_partial must be boolean and conflicts with wait_for_exact")
+        self.accept_partial = accept_partial
         self.cohort = None
         self.epoch = self.generation = None
         self.closed = False
@@ -273,7 +277,17 @@ def __enter__(self):
                     last = status
                 time.sleep(self.poll_seconds)
                 task = self.client.poll()
-            if task["granted"] != self.grant:
+            if self.accept_partial:
+                actual = task["granted"]
+                requested = self.requested_grant
+                if (type(actual["cpu_cores"]) is not int or
+                    not 0 < actual["cpu_cores"] <= requested["cpu_cores"] or
+                    not actual["gpus"] or
+                    any(gpu not in requested["gpus"] or size != requested["gpus"][gpu]
+                        for gpu, size in actual["gpus"].items())):
+                    raise RuntimeError("invalid partial grant: requires positive bounded CPUs and requested GPU memory")
+                self.grant = deepcopy(actual)
+            elif task["granted"] != self.grant:
                 raise RuntimeError("manager granted only part of the requested CPU/devices/VRAM; not launching")
             self.epoch, self.generation = self.client.epoch, task["generation"]
             print(f"gpu-manager task={self.client.task_id} exclusive={self.request['exclusive']} grant={self.grant}", flush=True)
@@ -358,7 +372,7 @@ def run_exclusive(command, *, socket, gpu_uuids, priority=100, label="fiber-eval
                   environment=None, poll_interval=1.0, stop_event=None,
                   client_factory=Client, cleanup_timeout=30.0, before_authorize=None,
                   exclusive=True, cpu_cores=None, vram_bytes_per_gpu=None,
-                  wait_for_exact=False, exact_retry_seconds=5.0):
+                  wait_for_exact=False, exact_retry_seconds=5.0, accept_partial=False):
     """Gated command adapter around ExclusiveTask; no unmanaged fallback.
 
     The caller may bootstrap the shared manager before calling. An independent
@@ -380,14 +394,16 @@ def run_exclusive(command, *, socket, gpu_uuids, priority=100, label="fiber-eval
     lease = ExclusiveTask(socket, grant, label=label, priority=priority,
         poll_seconds=poll_interval, stop_event=stop_event, client_factory=client_factory,
         cleanup_timeout=cleanup_timeout, exclusive=exclusive,
-        wait_for_exact=wait_for_exact, exact_retry_seconds=exact_retry_seconds)
+        wait_for_exact=wait_for_exact, exact_retry_seconds=exact_retry_seconds,
+        accept_partial=accept_partial)
     outcome = "failed"
     try:
         lease.__enter__()
         if before_authorize is not None:
             before_authorize()
         env = dict(os.environ if environment is None else environment)
-        env["CUDA_VISIBLE_DEVICES"] = ",".join(gpu_uuids)
+        env["CUDA_VISIBLE_DEVICES"] = ",".join(gpu for gpu in gpu_uuids if gpu in lease.grant["gpus"])
+        env["VILLA_GPU_MANAGER_GRANT"] = json.dumps(lease.grant, sort_keys=True)
         lease.cohort = child = GatedCommand(command, env)
         lease.authorize()
         while child.process.poll() is None:

gpu_manager/tests/test_exclusive.py

@@ -1,5 +1,6 @@
 """Synthetic GPU promises and real CPU-only gated subprocesses; no CUDA."""
 import os
+import json
 from pathlib import Path
 import sys
 import threading
@@ -89,6 +90,66 @@ def test_fixed_nonexclusive_command_shares_inventory(tmp_path):
     assert e.tasks[donor["task"]["task_id"]]["phase"] == "applied"
 
 
+def test_partial_admission_exports_actual_resources_and_joins_before_release(tmp_path):
+    e = engine()
+    def donor_identity(op, body):
+        if op == "register":
+            body["root"] = dict(pid=1, start_us=1, boot_us=1)
+    donor = memory_client(e, donor_identity)("unused")
+    donor.connect()
+    donor.register("occupied", dict(max_cpu_cores=4, max_gpus=1,
+        min_vram_bytes_per_gpu=100, max_vram_bytes_per_gpu=100,
+        allowed_gpu_uuids=["GPU-b"], priority=0, exclusive=False),
+        pausable=False, adaptable=False)
+    donor.report("applied", donor.poll()["granted"])
+    released=[]
+    def hook(op, body):
+        if op == "report" and body["phase"] == "released":
+            task=e.tasks[body["task_id"]]
+            assert all(process_state(p, os.getuid()) == "dead" for p in task["processes"][1:])
+            released.append(True)
+    marker=tmp_path / "actual.json"
+    command=[sys.executable,"-c",
+        "import os,json,pathlib,sys; pathlib.Path(sys.argv[1]).write_text(json.dumps(dict(visible=os.environ['CUDA_VISIBLE_DEVICES'],grant=json.loads(os.environ['VILLA_GPU_MANAGER_GRANT']))))",
+        str(marker)]
+    assert run_exclusive(command,socket="unused",gpu_uuids=["GPU-b","GPU-a"],
+        client_factory=memory_client(e,hook),exclusive=False,cpu_cores=8,
+        vram_bytes_per_gpu=40,accept_partial=True,poll_interval=.01) == 0
+    result=json.loads(marker.read_text())
+    assert result == {"visible":"GPU-a","grant":{"cpu_cores":4,"gpus":{"GPU-a":40}}}
+    task=max(e.tasks.values(),key=lambda t:t["order"])
+    assert task["terminal"] and task["granted"] == empty() and released
+    assert not task["adaptable"] and not task["pausable"]
+    assert donor.poll()["phase"] == "applied"
+
+
+@pytest.mark.parametrize("kind", ["empty", "foreign", "memory", "cpu_zero", "cpu_excess"])
+def test_invalid_partial_grants_do_not_launch(tmp_path,kind):
+    e=engine();base=memory_client(e)
+    class InvalidGrant(base):
+        def register(self,*args,**kwargs):
+            task=super().register(*args,**kwargs)
+            if kind=="empty":task["granted"]["gpus"]={}
+            elif kind=="foreign":task["granted"]["gpus"]={"GPU-foreign":40}
+            elif kind=="memory":task["granted"]["gpus"]["GPU-a"]=39
+            elif kind=="cpu_zero":task["granted"]["cpu_cores"]=0
+            elif kind=="cpu_excess":task["granted"]["cpu_cores"]=9
+            return task
+    marker=tmp_path/"forbidden"
+    with pytest.raises(RuntimeError,match="invalid partial grant"):
+        run_exclusive(marker_command(marker),socket="unused",gpu_uuids=["GPU-a","GPU-b"],
+            exclusive=False,cpu_cores=8,vram_bytes_per_gpu=40,accept_partial=True,
+            client_factory=InvalidGrant,poll_interval=.01)
+    assert not marker.exists()
+
+
+@pytest.mark.parametrize("partial,exact", [(True,True),(1,False)])
+def test_partial_admission_arguments(tmp_path,partial,exact):
+    with pytest.raises(ValueError,match="accept_partial"):
+        run_exclusive(marker_command(tmp_path/"forbidden"),socket="unused",gpu_uuids=["GPU-a"],
+            client_factory=memory_client(engine()),accept_partial=partial,wait_for_exact=exact)
+
+
 def test_nonexclusive_requires_explicit_budgets(tmp_path):
     with pytest.raises(ValueError, match="explicit CPU"):
         run_exclusive(marker_command(tmp_path / "no"), socket="unused", gpu_uuids=["GPU-a"],

lasagna/README.md

@@ -48,6 +48,10 @@ modules; direct and managed inference do not require `PYTHONPATH`.
 
 ### Inference manager
 
+For Qualzz AFV V2 models and plain V/H probability Zarr output, see
+[Qualzz AFV inference](docs/qualzz_afv_inference.md). This uses the same tiled
+engine and GPU manager without writing Lasagna-schema metadata.
+
 `las_manager` provides configuration, discovery, prefetch, and durable tmux
 execution for managed Fiber and Lasagna inference. Initialize its XDG configuration, fill
 in `cache_dir`, `output_dir`, `venv`, and one or more `snapshot_dirs`, then: