fix(ci): isolate runtime caches and Dynamo service listeners

Signed-off-by: functionstackx <47992694+functionstackx@users.noreply.github.com>

Author: functionstackx

scripts/ci/README.md

@@ -7,8 +7,9 @@ SPDX-License-Identifier: Apache-2.0
 
 This lane stages reviewed Dynamo source over SSH and builds/tests it in a one-GPU
 Slurm allocation using Pyxis/Enroot. The login host only stages files and controls
-Slurm. The implementation has offline tests; a successful HIP/Dynamo hardware run,
-local artifact parity, and hosted Actions connectivity still require qualification.
+Slurm. Offline tests validate the controller. Qualify each source/controller
+contract with a successful local build and HIP/Dynamo test run, a fresh local
+allocation reusing that image, and an Actions run reusing the same image.
 
 ## Prerequisites
 
@@ -90,6 +91,19 @@ scheduler's GPU assignment. Images are published as
 `images/<sha256>/image.sqsh`; model snapshots live under `models/<manifest-hash>`.
 No host Docker daemon or login-node compiler is used.
 
+Runtime library caches use per-allocation writable scratch. AITER keeps access to
+the image's prebuilt modules through verified symlinks in its private JIT cache;
+`aiter-cache.json` records those targets. Dynamo's native model-metadata cache
+ignores `XDG_CACHE_HOME`, so only `$HOME/.cache/dynamo/mdc` receives a private
+scratch bind mount inside the test container. The account's real home is not
+mounted, `HOME` is unchanged, and the image and pinned model snapshot stay
+read-only during testing.
+
+This single-node lane binds Dynamo's HTTP, system, request, response and event
+interfaces to loopback. Listener evidence covers the five named service ports
+and every TCP listener owned by the frontend and worker system-endpoint PIDs.
+It does not claim to inventory unrelated engine subprocess sockets.
+
 ## Status, reconnect, and cleanup
 
 Keep the original run key and evidence directory. `status` reports durable state;

scripts/ci/barite.sbatch

@@ -236,10 +236,15 @@ PY
 )
 MODEL_CACHE="$ROOT/models/$MODEL_KEY"
 [[ -d "$MODEL_CACHE" ]]
+# Native MDC caching ignores XDG_CACHE_HOME. Bind only its cache directory;
+# preserve HOME and keep the user's actual home outside the container.
+[[ "${HOME:?}" =~ ^/[A-Za-z0-9_-]+(/[A-Za-z0-9_.-]+)*$ ]]
+[[ "/$HOME/" != *"/../"* && "/$HOME/" != *"/./"* ]]
+mkdir -p "$scratch/mdc-cache"
 export HF_HOME=/models HF_HUB_CACHE=/models/hub HF_HUB_OFFLINE=1 TRANSFORMERS_OFFLINE=1
 run_step "${COMMON[@]}" --container-image="$TEST_SQSH" \
     --container-env=ROCR_VISIBLE_DEVICES,HF_HOME,HF_HUB_CACHE,HF_HUB_OFFLINE,TRANSFORMERS_OFFLINE \
-    --container-mounts="$JOB_SOURCE:/workspace,$RUN_DIR:/results,$MODEL_CACHE:/models:ro" \
+    --container-mounts="$JOB_SOURCE:/workspace,$RUN_DIR:/results,$MODEL_CACHE:/models:ro,$scratch/mdc-cache:$HOME/.cache/dynamo/mdc" \
     --container-workdir=/workspace \
     bash /results/controller/rocm/run-tests.sh --suite aggregate --python /opt/dynamo/venv/bin/python3 --results /results
 python3 "$CONTROLLER/barite_verify.py" workload --run-dir "$RUN_DIR" --runtime-dir "$RUN_DIR" --output "$RUN_DIR/completed.json"

scripts/ci/barite_client.py

@@ -246,7 +246,7 @@ def preflight(connection, args):
 import io,json,pathlib,sys,tarfile
 run=pathlib.Path(sys.argv[1]); cap=64*1024*1024; total=0; omitted=[]
 names={'request.json','source-manifest.json','build-manifest.json','image-manifest.json',
-'model-manifest.json','allocation.json','completed.json','hip.json','import-origins.json',
+'model-manifest.json','aiter-cache.json','allocation.json','completed.json','hip.json','import-origins.json',
 'test-manifest.json','test-summary.json','contract.json','provenance.json','receipt.json','terminal.json','wait-result.json',
 'controller-result.json','monitor.json','heartbeat.json','cancel.json','scheduler-latest.json',
 'submission-intent.json','partition-selection.json','collection.txt','environment.json','wheel-manifest.json',

scripts/ci/barite_verify.py

@@ -361,6 +361,53 @@ def verify_listener_evidence(evidence, request, suite, job_id, started_at):
             and entry["captured_at"] >= started_at,
             "stale or invalid listener process",
         )
+    anchors = [
+        entry
+        for entry in evidence["listeners"]
+        if entry["role"] in ("frontend", "system")
+    ]
+    expanded = evidence["service_listeners"]
+    require(
+        {entry["pid"] for entry in expanded} == {entry["pid"] for entry in anchors},
+        "Dynamo service listener inventory incomplete",
+    )
+    for entry in expanded:
+        owners = [anchor for anchor in anchors if anchor["pid"] == entry["pid"]]
+        require(
+            entry["service_roles"] == sorted({anchor["role"] for anchor in owners}),
+            "Dynamo service listener owner mismatch",
+        )
+        require(
+            all(
+                entry["process_created_at"] == anchor["process_created_at"]
+                and entry["root_pid"] == anchor["root_pid"]
+                for anchor in owners
+            ),
+            "Dynamo service listener process identity mismatch",
+        )
+        require(
+            entry["captured_at"] >= started_at
+            and entry["captured_at"] >= entry["process_created_at"],
+            "stale Dynamo service listener",
+        )
+        require(
+            isinstance(entry["port"], int) and 0 < entry["port"] < 65536,
+            "invalid Dynamo service listener port",
+        )
+        require(
+            ipaddress.ip_address(entry["address"]).is_loopback,
+            f"non-loopback Dynamo service listener: {entry['address']}:{entry['port']}",
+        )
+    for anchor in anchors:
+        require(
+            any(
+                entry["pid"] == anchor["pid"]
+                and entry["port"] == anchor["port"]
+                and entry["address"] == anchor["address"]
+                for entry in expanded
+            ),
+            "named endpoint missing from Dynamo service listener inventory",
+        )
     return evidence
 
 

scripts/ci/rocm/prepare_aiter_cache.py

@@ -0,0 +1,77 @@
+#!/usr/bin/env python3
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+"""Preserve the pinned image's AITER prebuilts in a writable per-job JIT directory."""
+
+import argparse
+import json
+import sys
+from importlib import metadata
+from pathlib import Path
+
+EXPECTED_VERSION = "0.1.19"
+
+
+def seed_prebuilt_modules(source: Path, destination: Path) -> list[dict]:
+    source = source.resolve(strict=True)
+    destination.mkdir(parents=True, exist_ok=True)
+    if destination.is_symlink() or source == destination.resolve():
+        raise ValueError("AITER JIT destination must be a separate writable directory")
+    modules = sorted(source.glob("*.so"))
+    if not modules:
+        raise ValueError("Pinned AITER image has no prebuilt JIT modules")
+    records = []
+    for module in modules:
+        target = module.resolve(strict=True)
+        if not target.is_file() or not target.is_relative_to(source):
+            raise ValueError(
+                f"AITER prebuilt target escapes installed JIT directory: {module}"
+            )
+        link = destination / module.name
+        if link.is_symlink():
+            if link.resolve(strict=True) != target:
+                raise ValueError(f"Unexpected AITER cache symlink: {link}")
+        elif link.exists():
+            raise ValueError(
+                f"AITER cache entry already exists without source provenance: {link}"
+            )
+        else:
+            link.symlink_to(target)
+        records.append(
+            {"name": module.name, "target": str(target), "size": target.stat().st_size}
+        )
+    return records
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--directory", type=Path, required=True)
+    parser.add_argument("--output", type=Path, required=True)
+    args = parser.parse_args()
+    # Metadata lookup does not import aiter: importing it initializes JIT state.
+    distribution = metadata.distribution("amd-aiter")
+    if distribution.version != EXPECTED_VERSION:
+        raise ValueError(
+            f"Expected amd-aiter {EXPECTED_VERSION}, got {distribution.version}"
+        )
+    source = Path(distribution.locate_file("aiter/jit")).resolve(strict=True)
+    if not source.is_relative_to(Path(sys.base_prefix).resolve()):
+        raise ValueError(
+            "AITER prebuilts must come from the installed base interpreter"
+        )
+    modules = seed_prebuilt_modules(source, args.directory)
+    # AITER 0.1.19 unlinks an existing target before copying a rebuilt module.
+    # Thus a genuine required rebuild replaces our link, never the base .so.
+    evidence = {
+        "version": distribution.version,
+        "source": str(source),
+        "directory": str(args.directory),
+        "modules": modules,
+    }
+    args.output.parent.mkdir(parents=True, exist_ok=True)
+    args.output.write_text(json.dumps(evidence, indent=2) + "\n")
+    print(f"Preserved {len(modules)} AITER {distribution.version} prebuilt modules")
+
+
+if __name__ == "__main__":
+    main()

scripts/ci/rocm/pytest_driver.py

@@ -48,6 +48,31 @@ def owned_listeners(root_pid, expected_ports):
     return records
 
 
+def service_listeners(services):
+    """Inspect every listener on the PIDs owning the two proven service ports."""
+    records = []
+    for pid in sorted({entry["pid"] for entry in services}):
+        anchors = [entry for entry in services if entry["pid"] == pid]
+        process = psutil.Process(pid)
+        created = process.create_time()
+        if any(entry["process_created_at"] != created for entry in anchors):
+            raise ValueError("Dynamo service PID was reused during listener capture")
+        for connection in process.net_connections(kind="tcp"):
+            if connection.status == psutil.CONN_LISTEN:
+                records.append(
+                    {
+                        "pid": pid,
+                        "process_created_at": created,
+                        "root_pid": anchors[0]["root_pid"],
+                        "captured_at": time.time(),
+                        "service_roles": sorted({entry["role"] for entry in anchors}),
+                        "port": connection.laddr.port,
+                        "address": connection.laddr.ip,
+                    }
+                )
+    return records
+
+
 def main():
     suite, directory = sys.argv[1:]
     result = Path(directory)
@@ -123,6 +148,7 @@ def capture_services(root_pid):
                     for entry in evidence["listeners"]
                     if entry["role"] not in ("frontend", "system")
                 ] + services
+                evidence["service_listeners"] = service_listeners(services)
                 evidence["status"] = "passed"
                 atomic_json(path, evidence)
                 verify_listener_evidence(

scripts/ci/rocm/run-tests.sh

@@ -15,9 +15,13 @@ export XDG_CONFIG_HOME="$cache_root/config"
 export VLLM_CONFIG_ROOT="$cache_root/config/vllm"
 export TORCHINDUCTOR_CACHE_DIR="$cache_root/torchinductor"
 export TRITON_CACHE_DIR="$cache_root/triton"
+export CUPY_CACHE_DIR="$cache_root/cupy"
 export TORCH_EXTENSIONS_DIR="$cache_root/torch-extensions"
 export VLLM_CACHE_ROOT="$cache_root/vllm"
 export TMPDIR="$cache_root/tmp"
 export HF_HOME=/models HF_HUB_CACHE=/models/hub
 export HF_HUB_OFFLINE=1 TRANSFORMERS_OFFLINE=1
+export AITER_JIT_DIR="$cache_root/aiter"
+/opt/dynamo/venv/bin/python3 -I "$(dirname "$0")/prepare_aiter_cache.py" \
+    --directory "$AITER_JIT_DIR" --output /results/aiter-cache.json
 exec /opt/dynamo/venv/bin/python3 -I "$(dirname "$0")/run_tests.py" "$@"

scripts/ci/rocm/run_tests.py

@@ -37,6 +37,10 @@ def main():
         DYNAMO_CI_SERVICE_HOST="127.0.0.1",
         DYN_HTTP_HOST="127.0.0.1",
         DYN_SYSTEM_HOST="127.0.0.1",
+        # All cooperating Dynamo services share this one-node allocation.
+        DYN_TCP_RPC_HOST="127.0.0.1",
+        DYN_TCP_RESPONSE_STREAM_HOST="127.0.0.1",
+        DYN_EVENT_PLANE_HOST="127.0.0.1",
     )
     contract = json.loads(Path(__file__).with_name("contract.json").read_text())
     request = json.loads((result / "request.json").read_text())

scripts/ci/tests/test_barite_aiter.py

@@ -0,0 +1,123 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+"""Exercise AITER cache seeding without importing AITER or accessing a GPU."""
+
+import importlib.util
+import json
+import tempfile
+import unittest
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import patch
+
+
+class AiterCacheTests(unittest.TestCase):
+    def setUp(self):
+        temporary = tempfile.TemporaryDirectory()
+        self.addCleanup(temporary.cleanup)
+        self.root = Path(temporary.name).resolve()
+        self.source = self.root / "base/aiter/jit"
+        self.source.mkdir(parents=True)
+        self.module = self.source / "module.so"
+        self.module.write_bytes(b"pinned prebuilt")
+        self.destination = self.root / "job/cache"
+        path = Path(__file__).resolve().parents[1] / "rocm/prepare_aiter_cache.py"
+        spec = importlib.util.spec_from_file_location("barite_aiter_test", path)
+        self.helper = importlib.util.module_from_spec(spec)
+        spec.loader.exec_module(self.helper)
+
+    def test_seed_is_idempotent_and_rebuild_does_not_modify_source(self):
+        records = self.helper.seed_prebuilt_modules(self.source, self.destination)
+        self.assertEqual(
+            records, self.helper.seed_prebuilt_modules(self.source, self.destination)
+        )
+        link = self.destination / "module.so"
+        self.assertTrue(link.is_symlink())
+        self.assertEqual(link.resolve(), self.module)
+        self.assertEqual(
+            records,
+            [
+                {
+                    "name": "module.so",
+                    "target": str(self.module),
+                    "size": len(b"pinned prebuilt"),
+                }
+            ],
+        )
+        # AITER 0.1.19 removes its output before installing a rebuilt extension.
+        link.unlink()
+        link.write_bytes(b"rebuilt for this job")
+        self.assertEqual(self.module.read_bytes(), b"pinned prebuilt")
+        self.assertEqual(link.read_bytes(), b"rebuilt for this job")
+        with self.assertRaisesRegex(ValueError, "without source provenance"):
+            self.helper.seed_prebuilt_modules(self.source, self.destination)
+        self.assertEqual(link.read_bytes(), b"rebuilt for this job")
+
+    def test_escaping_source_symlink_is_rejected(self):
+        outside = self.root / "outside.so"
+        outside.write_bytes(b"untrusted")
+        (self.source / "escape.so").symlink_to(outside)
+        with self.assertRaisesRegex(ValueError, "escapes installed"):
+            self.helper.seed_prebuilt_modules(self.source, self.destination)
+        self.assertFalse((self.destination / "escape.so").exists())
+
+    def test_conflicting_destination_and_empty_source_are_rejected(self):
+        for destination in (self.source, self.root / "alias"):
+            if destination != self.source:
+                destination.symlink_to(self.root / "base", target_is_directory=True)
+            with (
+                self.subTest(destination=destination),
+                self.assertRaisesRegex(ValueError, "separate writable"),
+            ):
+                self.helper.seed_prebuilt_modules(self.source, destination)
+        self.destination.mkdir(parents=True)
+        wrong = self.root / "wrong.so"
+        wrong.write_bytes(b"wrong")
+        (self.destination / "module.so").symlink_to(wrong)
+        with self.assertRaisesRegex(ValueError, "Unexpected AITER cache symlink"):
+            self.helper.seed_prebuilt_modules(self.source, self.destination)
+        self.module.unlink()
+        with self.assertRaisesRegex(ValueError, "no prebuilt"):
+            self.helper.seed_prebuilt_modules(self.source, self.destination)
+
+    def test_metadata_gate_and_evidence(self):
+        output = self.root / "evidence.json"
+        distribution = SimpleNamespace(
+            version="0.1.19", locate_file=lambda _: self.source
+        )
+        with (
+            patch.object(
+                self.helper.sys,
+                "argv",
+                [
+                    "prepare",
+                    "--directory",
+                    str(self.destination),
+                    "--output",
+                    str(output),
+                ],
+            ),
+            patch.object(self.helper.sys, "base_prefix", str(self.root / "base")),
+            patch.object(
+                self.helper.metadata, "distribution", return_value=distribution
+            ),
+            patch("builtins.print"),
+        ):
+            self.helper.main()
+            evidence = json.loads(output.read_text())
+            self.assertEqual(evidence["source"], str(self.source))
+            self.assertEqual(evidence["version"], "0.1.19")
+            self.assertEqual(len(evidence["modules"]), 1)
+            distribution.version = "0.1.20"
+            with self.assertRaisesRegex(ValueError, "Expected amd-aiter"):
+                self.helper.main()
+            distribution.version = "0.1.19"
+            with (
+                patch.object(self.helper.sys, "base_prefix", str(self.root / "other")),
+                self.assertRaisesRegex(ValueError, "installed base interpreter"),
+            ):
+                self.helper.main()
+
+
+if __name__ == "__main__":
+    unittest.main()