feat(replay): qualify AgentX G2 on existing mocker APIs

.github/actions/bootstrap-buildkit/action.yml

@@ -69,26 +69,30 @@ runs:
     - name: Define remote buildkit builders
       if: inputs.buildkit_worker_addresses != ''
       shell: bash
+      env:
+        BUILDER_NAME: ${{ inputs.builder_name }}
       run: |
         ADDRS="${{ inputs.buildkit_worker_addresses }}"
         IFS=',' read -ra ADDR_LIST <<< "$ADDRS"
         FIRST=true
         for addr in "${ADDR_LIST[@]}"; do
           if $FIRST; then
-            docker buildx create --use --name ${{ inputs.builder_name }} --driver remote "$addr"
+            docker buildx create --use --name "$BUILDER_NAME" --driver remote "$addr"
             FIRST=false
           else
-            docker buildx create --append --name ${{ inputs.builder_name }} --driver remote "$addr"
+            docker buildx create --append --name "$BUILDER_NAME" --driver remote "$addr"
           fi
         done
 
     - name: Create Kubernetes builder for both platforms
       if: inputs.buildkit_worker_addresses == ''
       shell: bash
+      env:
+        BUILDER_NAME: ${{ inputs.builder_name }}
       run: |
         # If the builder already exists, these commands will just set the right configurations to use it
-        echo "🔨 Creating K8s builder '${{ inputs.builder_name }}'."
-        docker buildx create --use --name ${{ inputs.builder_name }} --driver kubernetes --platform=linux/amd64 \
+        echo "🔨 Creating K8s builder."
+        docker buildx create --use --name "$BUILDER_NAME" --driver kubernetes --platform=linux/amd64 \
         '--driver-opt=requests.ephemeral-storage=${{ inputs.ephemeral_storage }}' \
         '--driver-opt=namespace=${{ inputs.namespace }}' \
         '--driver-opt=loadbalance=sticky' \
@@ -99,7 +103,7 @@ runs:
         '--driver-opt="nodeselector=kubernetes.io/arch=amd64,role=dynamo-builder-fallback"' \
         '--driver-opt="tolerations=${{ inputs.tolerations }}"'
 
-        docker buildx create --append --name ${{ inputs.builder_name }} --driver kubernetes --platform=linux/arm64 \
+        docker buildx create --append --name "$BUILDER_NAME" --driver kubernetes --platform=linux/arm64 \
         '--driver-opt=requests.ephemeral-storage=${{ inputs.ephemeral_storage }}' \
         '--driver-opt=namespace=${{ inputs.namespace }}' \
         '--driver-opt=loadbalance=sticky' \
@@ -124,10 +128,12 @@ runs:
     - name: Bootstrap buildkit
       if: inputs.skip_bootstrap != 'true'
       shell: bash
+      env:
+        BUILDER_NAME: ${{ inputs.builder_name }}
       run: |
         echo "Bootstrapping buildkit..."
         for i in 1 2 3; do
-          if docker buildx inspect ${{ inputs.builder_name }} --bootstrap; then
+          if docker buildx inspect --bootstrap -- "$BUILDER_NAME"; then
             echo "Bootstrap succeeded on attempt $i"
             break
           fi

.github/actions/builder-refresher/action.yml

@@ -10,9 +10,8 @@ description: 'Ensure a BuildKit builder is healthy; re-initialize it if the conn
 #      still reachable.  If it succeeds the action exits immediately — no extra
 #      work is done.
 #   2. If the inspect/bootstrap fails the builder is considered stale and
-#      init-dynamo-builder is called with the same flavor/arch/cuda_version that
-#      were used when the builder was first created, effectively re-routing
-#      workers and re-registering the builder from scratch.
+#      init-dynamo-builder is called with the same routing and isolation options
+#      that were used when the builder was first created.
 
 inputs:
   builder_name:
@@ -29,6 +28,10 @@ inputs:
     description: 'CUDA version used on re-init (e.g. 13.0, 13.1). Leave empty for general flavor.'
     required: false
     default: ''
+  fresh_builder:
+    description: 'Preserve an isolated Kubernetes builder when re-initializing'
+    required: false
+    default: 'false'
 
   # Kubernetes fallback passthrough inputs (forwarded to init-dynamo-builder)
   ephemeral_storage:
@@ -66,17 +69,23 @@ runs:
       id: check-health
       continue-on-error: true
       shell: bash
+      env:
+        BUILDER_NAME: ${{ inputs.builder_name }}
       run: |
-        echo "Checking BuildKit builder '${{ inputs.builder_name }}'..."
-        docker buildx inspect "${{ inputs.builder_name }}" --bootstrap
+        echo "Checking BuildKit builder health..."
+        docker buildx inspect --bootstrap -- "$BUILDER_NAME"
         echo "Builder is healthy."
 
     - name: Remove stale builder (if unhealthy)
       if: steps.check-health.outcome == 'failure'
       shell: bash
+      env:
+        BUILDER_NAME: ${{ inputs.builder_name }}
       run: |
-        echo "::warning::Builder '${{ inputs.builder_name }}' is unhealthy. Removing and re-initializing..."
-        docker buildx rm "${{ inputs.builder_name }}" || true
+        echo "::warning::Builder is unhealthy. Removing and re-initializing..."
+        # Other image jobs may still use the same run-scoped Kubernetes pods.
+        # Only the workflow cleanup job should delete those shared daemons.
+        docker buildx rm --keep-daemon -- "$BUILDER_NAME" || true
 
     - name: Re-initialize builder (if unhealthy)
       if: steps.check-health.outcome == 'failure'
@@ -86,6 +95,7 @@ runs:
         flavor: ${{ inputs.flavor }}
         arch: ${{ inputs.arch }}
         cuda_version: ${{ inputs.cuda_version }}
+        fresh_builder: ${{ inputs.fresh_builder }}
         ephemeral_storage: ${{ inputs.ephemeral_storage }}
         namespace: ${{ inputs.namespace }}
         replicas: ${{ inputs.replicas }}

.github/workflows/pr.yaml

@@ -1957,19 +1957,29 @@ jobs:
     name: Clean K8s builder if exists
     runs-on: prod-default-small-v2
     if: always()
+    # Wait for every builder consumer, including jobs that can fall back to
+    # the same K8s pair. Test-only jobs use published images, not the builder.
     needs:
       - changed-files
       - operator
       - power-agent
       - sidecar-build
       - planner-test
       - planner-copy-to-acr
+      - frontend-build
+      - triton-build
+      - vllm-dev-build
+      - vllm-local-dev-build
       - vllm-copy-to-acr
       - vllm-multi-gpu-test
       - vllm-efa-build
+      - sglang-dev-build
+      - sglang-local-dev-build
       - sglang-copy-to-acr
       - sglang-multi-gpu-test
       - sglang-efa-build
+      - trtllm-dev-build
+      - trtllm-local-dev-build
       - trtllm-copy-to-acr
       - trtllm-multi-gpu-test
       - trtllm-efa-build

.github/workflows/shared-build-image.yml

@@ -439,6 +439,7 @@ jobs:
           builder_name: ${{ inputs.builder_name }}
           flavor: ${{ steps.calculate-builder-flavor.outputs.builder_flavor }}
           arch: ${{ inputs.platform }}
+          fresh_builder: ${{ inputs.fresh_builder }}
           namespace: ${{ vars.BUILDKIT_NAMESPACE }}
       - name: Generate Dockerfile
         id: gen-dockerfile
@@ -570,6 +571,7 @@ jobs:
           flavor: ${{ steps.calculate-builder-flavor.outputs.builder_flavor }}
           arch: ${{ inputs.platform }}
           cuda_version: ${{ matrix.cuda_version }}
+          fresh_builder: ${{ inputs.fresh_builder }}
           namespace: ${{ vars.BUILDKIT_NAMESPACE }}
       - name: Compliance extract
         if: ${{ (inputs.compliance_only || inputs.inline_compliance) && inputs.target != 'dev' && inputs.target != 'local-dev' }}
@@ -614,6 +616,7 @@ jobs:
           flavor: ${{ steps.calculate-builder-flavor.outputs.builder_flavor }}
           arch: ${{ inputs.platform }}
           cuda_version: ${{ matrix.cuda_version }}
+          fresh_builder: ${{ inputs.fresh_builder }}
           namespace: ${{ vars.BUILDKIT_NAMESPACE }}
       - name: Build and Push Test Image
         if: ${{ !inputs.compliance_only && inputs.target != 'dev' && inputs.target != 'local-dev' && inputs.target != 'frontend' }}

components/src/dynamo/replay/api.py

@@ -78,6 +78,9 @@ class _CommonReplayOptions(TypedDict, total=False):
 
 class _TraceReplayOptions(_CommonReplayOptions, total=False):
     agentic_lanes: int | None
+    agentic_snapshot: dict[str, Any] | None
+    agentic_warmup: bool
+    agentic_profile: dict[str, Any] | None
     execution_model: str | None
     weka_nested_timestamp_basis: Literal["auto", "absolute", "relative"] | None
     trace_block_size: int | None
@@ -240,6 +243,9 @@ def run_trace_replay(
     num_decode_workers=1,
     replay_concurrency=None,
     agentic_lanes=None,
+    agentic_snapshot=None,
+    agentic_warmup=False,
+    agentic_profile=None,
     replay_mode="offline",
     router_mode="round_robin",
     arrival_speedup_ratio=1.0,
@@ -276,6 +282,10 @@ def run_trace_replay(
     on the response path. ``None`` or ``0`` keeps synchronous updates. Offline
     KV-router replay only.
 
+    ``agentic_snapshot``, ``agentic_warmup`` and ``agentic_profile`` use AISimulate
+    workload controls on the existing offline trace path. Native validation owns
+    their schemas and lifecycle constraints.
+
     Pass ``TelemetryOptions`` to enable policy-neutral sampling; omitting it
     leaves telemetry disabled. Callbacks and JSONL writes run synchronously on
     the replay loop, so their latency contributes to replay wall time. The
@@ -311,6 +321,9 @@ def run_trace_replay(
         "num_decode_workers": num_decode_workers,
         "replay_concurrency": replay_concurrency,
         "agentic_lanes": agentic_lanes,
+        "agentic_snapshot": agentic_snapshot,
+        "agentic_warmup": agentic_warmup,
+        "agentic_profile": agentic_profile,
         "replay_mode": replay_mode,
         "router_mode": router_mode,
         "arrival_speedup_ratio": arrival_speedup_ratio,

components/src/dynamo/replay/simulation.py

@@ -87,6 +87,15 @@ def capabilities(self) -> RunnerCapabilities:
                 "weka",
             ),
             supports_agentic_lanes=engine_capabilities.supports_agentic_lanes,
+            supports_agentic_host_offload=(
+                engine_capabilities.supports_agentic_host_offload
+            ),
+            # The composed mocker path preserves HBM-only MTP and decode-speedup
+            # support. The shared G2 gate and native guard reject that combination.
+            supports_agentic_speculative_decoding=True,
+            supports_agentic_snapshots=engine_capabilities.supports_agentic_snapshots,
+            supports_agentic_warmup=engine_capabilities.supports_agentic_warmup,
+            supports_agentic_profile=engine_capabilities.supports_agentic_profile,
             # Full AgentX runtime conformance remains a separate checkpoint.
             # Keep the public runner honest about the narrower integration here.
             supported_agentic_topologies=tuple(
@@ -169,8 +178,10 @@ def run(
         metrics, metadata = self._normalize_report(report, output_requirements)
         trace_format = spec.workload.get("trace_format")
         agentic_lanes = spec.workload.get("agentic_lanes")
-        if trace_format in {"weka", "agentic_mooncake"} or (
-            trace_format == "dynamo" and agentic_lanes is not None
+        if (
+            "agentic_graph" in metadata
+            or trace_format in {"weka", "agentic_mooncake"}
+            or (trace_format == "dynamo" and agentic_lanes is not None)
         ):
             metadata.update(
                 agentic_qualification=self.capabilities.agentic_qualification,
@@ -311,9 +322,16 @@ def _execution_target_model(spec: ReplaySpec) -> str | None:
                 candidates += [config.get("model_path"), config.get("model")]
         if isinstance(raw_engine_args, Mapping):
             candidates.append(raw_engine_args.get("aic_model_path"))
-            ais_config = raw_engine_args.get("ais_perf_config")
-            if isinstance(ais_config, Mapping):
-                candidates.append(ais_config.get("model"))
+            rank = raw_engine_args.get("rank", raw_engine_args)
+            if isinstance(rank, Mapping):
+                ais_config = rank.get("ais_perf_config")
+                if isinstance(ais_config, Mapping):
+                    candidates.append(ais_config.get("model"))
+                timing = rank.get("timing_model")
+                if isinstance(timing, Mapping) and isinstance(
+                    timing.get("config"), Mapping
+                ):
+                    candidates.append(timing["config"].get("model"))
         for model in candidates:
             if isinstance(model, str) and model.strip():
                 return model.strip()
@@ -366,6 +384,11 @@ def _run_trace(
         if not isinstance(trace_format, str):
             raise TypeError("trace workload requires a string trace_format")
         agentic_lanes = spec.workload.get("agentic_lanes")
+        agentic_options = {
+            name: spec.workload[name]
+            for name in ("agentic_snapshot", "agentic_warmup", "agentic_profile")
+            if name in spec.workload
+        }
         trace_block_size = spec.workload.get("trace_block_size")
         # Weka and Dynamo traces carry their source block size in the trace
         # metadata. Leave it unset so AISimulate can validate and use that
@@ -382,6 +405,7 @@ def _run_trace(
                 trace_block_size=trace_block_size,
                 max_sim_time_ms=spec.workload.get("max_sim_time_ms"),
                 agentic_lanes=agentic_lanes,
+                **agentic_options,
                 execution_model=execution_model,
                 extra_engine_args=self._engine_args(deployment, "aggregated"),
                 num_workers=deployment.num_workers,
@@ -396,6 +420,7 @@ def _run_trace(
             trace_block_size=trace_block_size,
             max_sim_time_ms=spec.workload.get("max_sim_time_ms"),
             agentic_lanes=agentic_lanes,
+            **agentic_options,
             execution_model=execution_model,
             prefill_engine_args=self._engine_args(deployment, "prefill"),
             decode_engine_args=self._engine_args(deployment, "decode"),
@@ -516,9 +541,15 @@ def _normalize_report(
         else:
             trace_report = dict(report)
 
-        for name in ("agentic_graph", "agentic_model_projection"):
+        for name in (
+            "agentic_graph",
+            "agentic_model_projection",
+            "agentic_snapshots",
+            "agentic_phases",
+            "agentic_profile",
+        ):
             value = trace_report.get(name)
-            if isinstance(value, dict):
+            if isinstance(value, (dict, list)):
                 metadata[name] = value
         resolved_weka_basis = trace_report.get("weka_nested_timestamp_basis")
         if isinstance(resolved_weka_basis, str):