diff --git a/.agents/skills/serving-bench/SKILL.md b/.agents/skills/serving-bench/SKILL.md
index 4c4f06a24c..192ebf811c 100644
--- a/.agents/skills/serving-bench/SKILL.md
+++ b/.agents/skills/serving-bench/SKILL.md
@@ -30,7 +30,10 @@ router:                             # optional: any router in front of the engin
 storage:                            # optional: a non-GPU sidecar started first (e.g. Mooncake master)
   image: <image>
   serve: mooncake_master --port={{storage_master_port}} --http_metadata_server_port={{storage_meta_port}}
-scenario: agentx                    # agentx | tau2 | livekit | shared-prefix-e2 | hidden-prefix-e3 | synthetic
+arms:                               # optional: engine variants in one run, e.g. base vs speculative (see ARCHITECTURE.md, Arms)
+  base: {}
+  mtp6: {append: "--speculative-config '...'"}
+scenario: agentx                    # agentx | tau2 | livekit | shared-prefix-e2 | hidden-prefix-e3 | synthetic | greenspec
 overrides: {duration: 180}          # optional variant, see the table below
 conc: [4, 8]                        # one benchmark job per value
 ```
@@ -46,6 +49,7 @@ Working examples, one per pattern: `inferencex-e2e/configs/serving/*.yaml`. They
 | `livekit` | LiveKit voice-agent traces, V1.1 warm (dial warm-ups count in TTFT) | 1200 s |
 | `shared-prefix-e2` / `hidden-prefix-e3` | prefix-cache A/B traces (HiCache L2 vs Mooncake L3) | 1800 s |
 | `synthetic` | synthetic ISL/OSL with a shared-prefix cache hit, needs `workload:` | 300 s per point |
+| `greenspec` | speculative-decoding bench from production chat traffic (private; result has `spec_decode.acceptance_length`) | the whole selection, no duration |
 
 Variants (`overrides:`, only these keys):
 
@@ -57,6 +61,7 @@ Variants (`overrides:`, only these keys):
 | sampling instead of greedy | `overrides: {generation: {top_p: 0.95, temperature: 1.0}}` |
 | LiveKit V1 / V2 trace | `overrides: {dataset: livekit-111868-v1}` (or `-v2`) |
 | a different trace idle-gap cap | `overrides: {trace_idle_gap_cap: 1}` |
+| GreenSpec subset, lite, bucket, cap | `scenario: greenspec` + `overrides: {subset: agentic_tools, lite: true, max_tokens: 4096, requests: 200}` (only these keys; no `duration`) |
 
 Synthetic grid, one job per point, all in one run:
 
diff --git a/.agents/skills/serving-runner-setup/SKILL.md b/.agents/skills/serving-runner-setup/SKILL.md
index 527be2851b..05e35b53a4 100644
--- a/.agents/skills/serving-runner-setup/SKILL.md
+++ b/.agents/skills/serving-runner-setup/SKILL.md
@@ -73,7 +73,7 @@ for d in $BASE/gharunner0*/actions-runner; do (cd "$d" && sudo ./svc.sh install
 
 The registration token comes from the settings page or `gh api -X POST repos/vngcloud/InferenceX/actions/runners/registration-token --jq .token` (valid exactly 1 hour, not configurable; mint it right before running `setup.sh`). `setup.sh` prints `All runners configured!` even when registration failed: check that every slot printed its own `[<name>] Done.` line.
 
-No runner `.env` is needed for the datasets: every repo in `datasets/lock.yaml` is public on HF (checked 2026-10-25), and `infx/serving/datasets.py` downloads anonymously when `GREENLAB_HF_TOKEN` is unset. Add `GREENLAB_HF_TOKEN=…` to `actions-runner/.env` only if a private dataset is added to the lock.
+Every `datasets/lock.yaml` entry marked `private: true` (GreenSpec) needs `GREENLAB_HF_TOKEN`; the other repos are public. `infx/serving/datasets.py` downloads anonymously when `GREENLAB_HF_TOKEN` is unset, so a runner that is to run `scenario: greenspec` needs `GREENLAB_HF_TOKEN=…` in `actions-runner/.env`. Other scenarios need no `.env`.
 
 ## 3. Add the cluster record
 
diff --git a/.github/workflows/benchmark-tmpl.yml b/.github/workflows/benchmark-tmpl.yml
index 1ac1b7498c..06c46ce340 100644
--- a/.github/workflows/benchmark-tmpl.yml
+++ b/.github/workflows/benchmark-tmpl.yml
@@ -6,6 +6,8 @@ on:
         required: true
       SERVING_API_KEY:
         required: false
+      GREENLAB_HF_TOKEN:
+        required: false
     inputs:
       serving:
         description: "VNG serving point (SERVING_JSON from infx/serving/to_matrix.py); empty for every other caller"
@@ -100,6 +102,7 @@ env:
   PYTHONPATH: ${{ github.workspace }}/inferencex-e2e
   SERVING_JSON: ${{ inputs.serving }}
   SERVING_API_KEY: ${{ secrets.SERVING_API_KEY }}
+  GREENLAB_HF_TOKEN: ${{ secrets.GREENLAB_HF_TOKEN }}
   INFERENCEX_E2E_ROOT: ${{ github.workspace }}
   PORT: '8888'
   INFMAX_CONTAINER_WORKSPACE: '/workspace'
diff --git a/.github/workflows/gn-benchmark.yml b/.github/workflows/gn-benchmark.yml
index 87293b78f6..e6cf611381 100644
--- a/.github/workflows/gn-benchmark.yml
+++ b/.github/workflows/gn-benchmark.yml
@@ -49,6 +49,7 @@ jobs:
         secrets:
             INFERENCEX_OFFICIAL_RO_HF_TOKEN: ${{ secrets.INFERENCEX_OFFICIAL_RO_HF_TOKEN }}
             SERVING_API_KEY: ${{ matrix.api_key_secret && secrets[matrix.api_key_secret] || '' }}
+            GREENLAB_HF_TOKEN: ${{ secrets.GREENLAB_HF_TOKEN }}
         with:
             config: ${{ toJSON(matrix.config) }}
             serving: ${{ matrix.serving }}
@@ -60,3 +61,24 @@ jobs:
             scenario-type: agentic-coding
             duration: ${{ matrix.config.duration }}
             test-name: ${{ inputs.test-name }}
+
+    summary:
+        name: summary
+        # One table of every point on the run page: throughput, latency, and for speculative decoding
+        # the acceptance length and the speedup over a base arm of the same scenario and concurrency.
+        needs: bench
+        if: ${{ !cancelled() }}
+        runs-on: ubuntu-latest
+        steps:
+            - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
+              with:
+                  persist-credentials: false
+            - uses: astral-sh/setup-uv@20cfd1bf945f4377ade1205e4dbc17946fc9a30d # v10.0.1
+            - uses: actions/download-artifact@3e5f45b2cfb9172054b4087a40e8e0b5a5461e7c # v8.0.1
+              with:
+                  path: results
+                  pattern: bmk_agentic_*
+            - working-directory: inferencex-e2e
+              env:
+                  PYTHONPATH: ${{ github.workspace }}/inferencex-e2e
+              run: uv run --no-project --python 3.12 python -P -m infx.serving.summary "$GITHUB_WORKSPACE/results" >> "$GITHUB_STEP_SUMMARY"
diff --git a/inferencex-e2e/benchmarks/single_node/serving/datasets/lock.yaml b/inferencex-e2e/benchmarks/single_node/serving/datasets/lock.yaml
index d840ce584c..dcf76d53ab 100644
--- a/inferencex-e2e/benchmarks/single_node/serving/datasets/lock.yaml
+++ b/inferencex-e2e/benchmarks/single_node/serving/datasets/lock.yaml
@@ -2,6 +2,7 @@
 # Each entry pins one file at one HF commit; infx/serving/datasets.py downloads it
 # (token: GREENLAB_HF_TOKEN if set) and verifies sha256. `entries` is the dataset's
 # session count, copied into the scenario when a serving YAML picks this dataset.
+# `private: true` marks a repo that needs GREENLAB_HF_TOKEN to read (the runner-setup skill keys off it).
 tau2-cs:
   repo: Green-Lab/customer-service-agent-traces
   revision: 2c793380d8ba45bfde79f9a5c7b5991ed40ab2ea
@@ -42,3 +43,12 @@ livekit-111868-v1.1-warm:
   file: trace.v1.1-append-ideal-warm.jsonl
   sha256: e26542392a202764849c05018f2f8ea3f9574277ffa3400615f0b4df23b0eb2f
   entries: 671
+# GreenSpec v1.1.1: one flat file with every subset (infx.serving.greenspec selects rows from it).
+# Private repo: GREENLAB_HF_TOKEN must be able to read it. The benchmark repo is public, so the
+# greenspec scenario never uploads anything derived from the rows (see infx/serving/greenspec.py).
+greenspec-v1.1.1:
+  repo: Green-Lab/GreenSpec
+  revision: f7ea173f12917bcf1d3b9926a281a167d2b8243d
+  file: greenspec.jsonl
+  sha256: 6a26785d07c235446e57ba34463a1ae9e2dcff938fec6246881492e511fb0f43
+  private: true
diff --git a/inferencex-e2e/benchmarks/single_node/serving/scenarios/greenspec.yaml b/inferencex-e2e/benchmarks/single_node/serving/scenarios/greenspec.yaml
new file mode 100644
index 0000000000..11a43c53e0
--- /dev/null
+++ b/inferencex-e2e/benchmarks/single_node/serving/scenarios/greenspec.yaml
@@ -0,0 +1,29 @@
+# GreenSpec family: the internal speculative-decoding bench cut from production chat traffic
+# (Green-Lab/GreenSpec, one flat greenspec.jsonl per release; datasets/lock.yaml pins the release).
+# infx.serving.greenspec selects the rows (overrides: subset, lite, bucket), builds one request body per
+# row and writes them as the raw_payload file IX_CUSTOM_DATASET_FILE points to; AgentX sends each body
+# as is. Every request streams and completes on its own (no ignore_eos): acceptance only means
+# something on natural output. `requests` is the selection size unless overrides.requests lowers it;
+# the client sets it once it has counted the selection, so the 1 below is only a placeholder.
+schemaVersion: "2.0"
+benchmark:
+  unsafeOverride: true
+  endpoint:
+    type: chat
+    path: /v1/chat/completions
+    streaming: true
+    useServerTokenCount: true
+  datasets:
+  - name: greenspec-v1.1.1
+    type: file
+    format: raw_payload
+    path: ${IX_CUSTOM_DATASET_FILE}
+    sampling: sequential
+    randomSeed: 42
+  phases:
+  - name: profiling
+    kind: profiling
+    type: concurrency
+    requests: 1
+    failedRequestThreshold: 0.1
+randomSeed: 42
diff --git a/inferencex-e2e/configs/serving/gemma4-greenspec-diverse.yaml b/inferencex-e2e/configs/serving/gemma4-greenspec-diverse.yaml
new file mode 100644
index 0000000000..1d7d215f81
--- /dev/null
+++ b/inferencex-e2e/configs/serving/gemma4-greenspec-diverse.yaml
@@ -0,0 +1,40 @@
+# Pattern: speculative decoding against its base, in one run (arms:). GreenSpec, whole diverse subset,
+# Gemma-4-31B FP8-block, vLLM, 1 GPU, CCU 16, MTP draft depth 6. serve: is shared by every arm; an arm adds
+# flags (append) and env. Each arm is its own job and its own engine; the run's summary table lines them up
+# and gives the speedup of each speculative arm over the arm without one.
+# Same engine flags as the earlier SPEED-Bench cells (prefix caching off, 256k context). The run is one
+# request per row of the selection (no duration): the results carry spec_decode.acceptance_length.
+name: gemma4-greenspec-diverse
+runner: cluster:hcm-hgx02
+image: vllm/vllm-openai:v0.30.0
+model: RedHatAI/gemma-4-31B-it-FP8-block
+gpus: 1
+precision: fp8block
+env:
+  NCCL_P2P_LEVEL: NVL
+  VLLM_DISABLE_COMPILE_CACHE: "1"
+  VLLM_ATTENTION_BACKEND: FLASHINFER
+serve: |
+  vllm serve RedHatAI/gemma-4-31B-it-FP8-block
+  --served-model-name RedHatAI/gemma-4-31B-it-FP8-block
+  --trust-remote-code
+  --tensor-parallel-size 1
+  --gpu-memory-utilization 0.92
+  --max-model-len 262144
+  --max-num-seqs 64
+  --max-num-batched-tokens 16384
+  --enable-chunked-prefill
+  --long-prefill-token-threshold 8192
+  --no-enable-prefix-caching
+  --enable-auto-tool-choice
+  --tool-call-parser gemma4
+  --reasoning-parser gemma4
+arms:
+  base: {}
+  mtp6:
+    append: |
+      --speculative-config '{"method": "mtp", "model": "google/gemma-4-31B-it-assistant", "num_speculative_tokens": 6}'
+scenario: greenspec
+overrides:
+  subset: diverse
+conc: [16]
diff --git a/inferencex-e2e/docs/serving/ARCHITECTURE.md b/inferencex-e2e/docs/serving/ARCHITECTURE.md
index 75d18a9801..7f07e9f81d 100644
--- a/inferencex-e2e/docs/serving/ARCHITECTURE.md
+++ b/inferencex-e2e/docs/serving/ARCHITECTURE.md
@@ -214,7 +214,7 @@ that several nodes need belong in a registry, where every node caches its own sq
 
 ## 6. Scenarios
 
-Six base files in `benchmarks/single_node/serving/scenarios/`:
+Seven base files in `benchmarks/single_node/serving/scenarios/`:
 
 | Family | Workload | Dataset |
 |---|---|---|
@@ -224,6 +224,7 @@ Six base files in `benchmarks/single_node/serving/scenarios/`:
 | `shared-prefix-e2` | requests sharing long prefixes | pinned file, Mooncake trace format |
 | `hidden-prefix-e3` | prefixes reused without being visible to the router | pinned file, Mooncake trace format |
 | `synthetic` | generated ISL/OSL with a shared-prefix share | synthetic |
+| `greenspec` | speculative-decoding bench cut from production chat traffic, one request per row | pinned private file (`Green-Lab/GreenSpec`), `raw_payload` format |
 
 Pinned files are listed in `benchmarks/single_node/serving/datasets/lock.yaml` with their URL and
 sha256; the client downloads and checks them before the run.
@@ -233,6 +234,43 @@ A serving YAML picks a family and may set `overrides`. The keys allowed are `dur
 `generation` and `trace_idle_gap_cap`. Any other key fails validation, so cache-bust, warmup and seed
 cannot change without a code review.
 
+`greenspec` takes its own keys instead: `subset` (`diverse`, `agentic_coding`, `agentic_tools`, `llm_task`),
+`lite`, `bucket`, `mode` (`target-default` or `recorded` sampling), `max_tokens` (output cap, default 8192)
+and `requests` (default: the whole selection). It has no duration; the run ends when the selection is sent.
+Three things differ from the other families:
+
+- The dataset is private and the benchmark repo is public, so the client (`infx/serving/greenspec.py`)
+  prints AgentX's console without error text, and before the upload keeps only an allowlist of files in
+  `results/`, strips error messages from the aiperf export, replaces server logs by a stub (the originals
+  stay on the node in `.greenspec-private-logs/`) and fails the job if a `question_id` is still in an
+  upload. The request bodies are written outside `results/` and deleted at the end.
+- The client snapshots the engine's `vllm:spec_decode_*` counters (`sglang:spec_accept_length` on SGLang)
+  before and after the replay and files `spec_decode` (`acceptance_length`, `acceptance_rate`, counts) in
+  the aggregate result (`infx/serving/specdec.py`). A point without speculative decoding has no such block.
+- The runner needs `GREENLAB_HF_TOKEN` in `actions-runner/.env`: the release is not public.
+
+A base arm and a speculative arm are the `arms:` of one serving YAML (below). The `summary` job of
+`gn-benchmark.yml` prints one table of all points of a run on the run page, with the speedup of each
+speculative point over the base point of the same scenario and concurrency.
+
+### Arms
+
+`arms:` runs engine variants of one serving YAML in one run: `serve:` and `env:` are shared, an arm adds
+flags with `append` and overlays `env`. Each arm is its own job on its own engine (the name of the point
+is `<name>-<arm>_c<conc>`), so arms multiply with `conc` and with a workload grid.
+
+```yaml
+serve: vllm serve <model> --tensor-parallel-size 1 ...
+arms:
+  base: {}
+  mtp6:
+    append: |
+      --speculative-config '{"method": "mtp", "model": "<draft>", "num_speculative_tokens": 6}'
+```
+
+Two arms that end up with the same command and env are rejected, and `arms:` does not combine with
+`endpoint:`. The `spec-decoding` field of each point follows its own command (`mtp`, `draft_model` or `none`).
+
 The synthetic family takes `workload:` instead:
 
 ```yaml
diff --git a/inferencex-e2e/infx/bench/agentic/replay.py b/inferencex-e2e/infx/bench/agentic/replay.py
index 8a9dcab9b4..a4df8203dd 100644
--- a/inferencex-e2e/infx/bench/agentic/replay.py
+++ b/inferencex-e2e/infx/bench/agentic/replay.py
@@ -169,8 +169,10 @@ def replay_argv(cfg: ReplayConfig, cli: str | Path) -> list[str]:
     argv += ["--slice-duration", SLICE_DURATION_S]
     argv += ["--no-gpu-telemetry"]
     argv += [
-        "--trajectory-start-min-ratio", TRAJECTORY_START_MIN_RATIO,
-        "--trajectory-start-max-ratio", TRAJECTORY_START_MAX_RATIO,
+        "--trajectory-start-min-ratio",
+        TRAJECTORY_START_MIN_RATIO,
+        "--trajectory-start-max-ratio",
+        TRAJECTORY_START_MAX_RATIO,
     ]
     argv += [
         "--failed-request-threshold",
diff --git a/inferencex-e2e/infx/serving/client.py b/inferencex-e2e/infx/serving/client.py
index 300830d123..4d28348298 100644
--- a/inferencex-e2e/infx/serving/client.py
+++ b/inferencex-e2e/infx/serving/client.py
@@ -13,6 +13,7 @@
 import json
 import os
 import shlex
+import shutil
 import ssl
 import sys
 import urllib.error
@@ -22,11 +23,16 @@
 
 from infx.bench import proc
 from infx.bench.agentic.venv import Runtime, bootstrap
+from infx.serving import greenspec, specdec
 
 OFFLINE = ("HF_HUB_OFFLINE", "TRANSFORMERS_OFFLINE", "HF_DATASETS_OFFLINE")
 DATASET_PATH = ".dataset_path"
 FAILED_REQUEST_THRESHOLD = "0.10"
 COUNT_SOURCE_MARKER = ".token_count"
+SELECTION_DIR = "/tmp/greenspec-selection"  # noqa: S108 -- inside the job container; outside the result dir: request bodies are never uploaded
+PRIVATE_LOGS = (
+    ".greenspec-private-logs"  # beside the aggregate, on the node; not in any upload path
+)
 PROBE_BODY = {  # the shape agentx sends: streaming, usage requested, ignore_eos injected by the scenario
     "messages": [{"role": "user", "content": "Say OK."}],
     "stream": True,
@@ -53,6 +59,10 @@ def datasets_cache(env: Mapping[str, str]) -> dict[str, str]:
     return {"HF_DATASETS_CACHE": f"{hub.rstrip('/')}/.datasets-arrow"}
 
 
+def served_model(env: Mapping[str, str]) -> str:
+    return env.get("SERVED_MODEL_NAME") or env["MODEL"]
+
+
 def authorization(api_key: str) -> str:
     """A key with a space is a full header value ("Basic <b64>"): gateways that
     only accept scheme auth reject "Bearer Basic <b64>"."""
@@ -67,7 +77,7 @@ def agentx_argv(agentx: str, scenario: Path, env: Mapping[str, str]) -> list[str
     migration."""
     argv = [agentx, "profile", "--config", str(scenario)]
     argv += ["--url", env.get("SERVING_REMOTE_URL") or f"http://localhost:{env['PORT']}"]
-    argv += ["--model", env.get("SERVED_MODEL_NAME") or env["MODEL"]]
+    argv += ["--model", served_model(env)]
     argv += ["--tokenizer", env.get("SERVING_TOKENIZER") or env["MODEL"]]
     if key := env.get("SERVING_API_KEY"):
         if " " in key:
@@ -183,6 +193,16 @@ def with_token_source(scenario_yaml: str, source: str) -> str:
     return yaml.safe_dump(doc, sort_keys=False)
 
 
+def with_requests(scenario_yaml: str, requests: int) -> str:
+    """The scenario with its profiling phase set to ``requests`` requests (the size of the selection)."""
+    import yaml  # in the AgentX venv
+
+    doc = yaml.safe_load(scenario_yaml)
+    [phase] = [p for p in doc["benchmark"]["phases"] if p.get("kind") == "profiling"]
+    phase["requests"] = requests
+    return yaml.safe_dump(doc, sort_keys=False)
+
+
 def label_result(path: Path, label: str, token_count_source: str | None = None) -> None:
     """File the aggregate under the legacy scenario label result-report reads first."""
     if not path.exists():
@@ -201,8 +221,16 @@ def prepare(env: Mapping[str, str], runtime: Runtime) -> int:
         from infx.serving.datasets import DatasetError, fetch
 
         try:
-            (result_dir / DATASET_PATH).write_text(str(fetch(env["SERVING_DATASET"])))
-        except DatasetError as error:
+            path = fetch(env["SERVING_DATASET"])
+            if chosen := greenspec.settings(env):
+                selection = Path(SELECTION_DIR) / "selection.jsonl"
+                count = greenspec.write_selection(path, chosen, served_model(env), selection)
+                print(
+                    f"GreenSpec selection {greenspec.label(chosen)}: {count} requests", flush=True
+                )
+                path = selection
+            (result_dir / DATASET_PATH).write_text(str(path))
+        except (DatasetError, greenspec.GreenSpecError) as error:
             print(f"ERROR {error}", file=sys.stderr)
             return 1
     if url := env.get("SERVING_REMOTE_URL"):
@@ -237,7 +265,8 @@ def run(env: Mapping[str, str], runtime: Runtime) -> int:
     scenario = result_dir / "scenario.yaml"
     token_marker = result_dir / COUNT_SOURCE_MARKER
     source = token_marker.read_text().strip() if token_marker.exists() else None
-    scenario.write_text(with_token_source(env["SERVING_SCENARIO_YAML"], source or "server"))
+    chosen = greenspec.settings(env)
+    scenario_yaml = with_token_source(env["SERVING_SCENARIO_YAML"], source or "server")
     child = online(env) | {
         "AIPERF_DATASET_CONFIGURATION_TIMEOUT": "1800",
         "AIPERF_SERVICE_PROFILE_CONFIGURE_TIMEOUT": "1800",
@@ -251,9 +280,20 @@ def run(env: Mapping[str, str], runtime: Runtime) -> int:
         child["AIPERF_HTTP_SSL_VERIFY"] = "false"  # aiperf's own switch for self-signed gateways
     if (marker := result_dir / DATASET_PATH).exists():
         child["IX_CUSTOM_DATASET_FILE"] = marker.read_text().strip()
+    if chosen:
+        with open(child["IX_CUSTOM_DATASET_FILE"], encoding="utf-8") as selection:
+            size = sum(1 for _ in selection)
+        scenario_yaml = with_requests(scenario_yaml, min(chosen.requests or size, size))
+    scenario.write_text(scenario_yaml)
     argv = agentx_argv(str(runtime.aiperf), scenario, env)
     (result_dir / "benchmark_command.txt").write_text(shlex.join(redacted(argv)) + "\n")
-    replay_rc = proc.tee(argv, result_dir / "benchmark.log", child)
+    metrics_urls = [url for url in env.get("AIPERF_SERVER_METRICS_URLS", "").split(",") if url]
+    if chosen:
+        specdec.save(result_dir / "spec_metrics_before.json", specdec.snapshot(metrics_urls))
+    replay = greenspec.tee_filtered if chosen else proc.tee
+    replay_rc = replay(argv, result_dir / "benchmark.log", child)
+    if chosen:
+        specdec.save(result_dir / "spec_metrics_after.json", specdec.snapshot(metrics_urls))
     python = str(runtime.python)
     if env.get("SERVING_SEARCH") == "1":
         return replay_rc or proc.call([python, "-m", "infx.serving.sweep_result"], child)
@@ -280,12 +320,39 @@ def run(env: Mapping[str, str], runtime: Runtime) -> int:
         ],
         child,
     )
-    label_result(
-        output_dir / f"{env['RESULT_FILENAME']}.json",
-        env["SERVING_SCENARIO_LABEL"],
-        token_count_source=source,
+    aggregate = output_dir / f"{env['RESULT_FILENAME']}.json"
+    label_result(aggregate, env["SERVING_SCENARIO_LABEL"], token_count_source=source)
+    guard_rc = 0
+    if chosen:
+        spec = specdec.summarize(
+            result_dir / "spec_metrics_before.json", result_dir / "spec_metrics_after.json"
+        )
+        if spec is None:
+            print(
+                "WARNING no speculative-decoding series on the engine metrics; the result has no spec_decode block",
+                file=sys.stderr,
+            )
+        else:
+            specdec.merge(aggregate, spec)
+        guard_rc = guard(env, result_dir, aggregate, output_dir)
+    return replay_rc or aggregate_rc or validate_rc or guard_rc
+
+
+def guard(env: Mapping[str, str], result_dir: Path, aggregate: Path, output_dir: Path) -> int:
+    """Prune, redact and scan what a greenspec job uploads (the benchmark repo is public); 1 on a problem."""
+    from infx.serving.datasets import fetch
+
+    shutil.rmtree(SELECTION_DIR, ignore_errors=True)
+    problems = greenspec.guard(
+        fetch(env["SERVING_DATASET"]),
+        result_dir,
+        [aggregate],
+        [output_dir / "server.log"],
+        keep_log_dir=output_dir / PRIVATE_LOGS,
     )
-    return replay_rc or aggregate_rc or validate_rc
+    for problem in problems:
+        print(f"GreenSpec guard: {problem}", file=sys.stderr)
+    return 1 if problems else 0
 
 
 def main(argv: list[str] | None = None) -> int:
diff --git a/inferencex-e2e/infx/serving/datasets.py b/inferencex-e2e/infx/serving/datasets.py
index 57977b2023..1fb3da4086 100644
--- a/inferencex-e2e/infx/serving/datasets.py
+++ b/inferencex-e2e/infx/serving/datasets.py
@@ -37,6 +37,7 @@ class LockEntry:
     file: str
     sha256: str
     entries: int | None = None
+    private: bool = False
 
 
 def load_lock(path: Path = LOCK) -> dict[str, LockEntry]:
@@ -50,15 +51,24 @@ def load_lock(path: Path = LOCK) -> dict[str, LockEntry]:
     lock = {}
     for name, entry in raw.items():
         keys = set(entry) if isinstance(entry, dict) else set()
-        if not set(FIELDS) <= keys or keys - {*FIELDS, "entries"}:
-            raise DatasetError(f"lock: {name} needs exactly {', '.join(FIELDS)} (and optional entries)")
+        if not set(FIELDS) <= keys or keys - {*FIELDS, "entries", "private"}:
+            raise DatasetError(
+                f"lock: {name} needs exactly {', '.join(FIELDS)} (and optional entries, private)"
+            )
         for field, pattern in FIELDS.items():
             if not re.fullmatch(pattern, str(entry[field])):
-                raise DatasetError(f"lock: {name}.{field} {entry[field]!r} does not match {pattern}")
+                raise DatasetError(
+                    f"lock: {name}.{field} {entry[field]!r} does not match {pattern}"
+                )
         entries = entry.get("entries")
-        if entries is not None and (isinstance(entries, bool) or not isinstance(entries, int) or entries < 1):
+        if entries is not None and (
+            isinstance(entries, bool) or not isinstance(entries, int) or entries < 1
+        ):
             raise DatasetError(f"lock: {name}.entries must be a positive integer")
-        lock[name] = LockEntry(name, *(str(entry[field]) for field in FIELDS), entries)
+        private = entry.get("private", False)
+        if not isinstance(private, bool):
+            raise DatasetError(f"lock: {name}.private must be true or false")
+        lock[name] = LockEntry(name, *(str(entry[field]) for field in FIELDS), entries, private)
     return lock
 
 
@@ -80,10 +90,19 @@ def fetch(name: str, lock: dict[str, LockEntry] | None = None) -> Path:
     from huggingface_hub.errors import HfHubHTTPError, RepositoryNotFoundError
 
     where = f"{entry.repo}@{entry.revision}"
-    unreadable = f"dataset {name}: {where} not readable; GREENLAB_HF_TOKEN is {'set' if token else 'unset'}"
+    unreadable = (
+        f"dataset {name}: {where} not readable; GREENLAB_HF_TOKEN is {'set' if token else 'unset'}"
+    )
     try:
-        path = Path(hf_hub_download(entry.repo, entry.file, revision=entry.revision,
-                                    repo_type="dataset", token=token or False))
+        path = Path(
+            hf_hub_download(
+                entry.repo,
+                entry.file,
+                revision=entry.revision,
+                repo_type="dataset",
+                token=token or False,
+            )
+        )
     except RepositoryNotFoundError:  # includes GatedRepoError
         raise DatasetError(unreadable) from None
     except HfHubHTTPError as error:
@@ -92,11 +111,14 @@ def fetch(name: str, lock: dict[str, LockEntry] | None = None) -> Path:
         raise
     actual = sha256(path)
     if actual != entry.sha256:
-        raise DatasetError(f"dataset {name}: sha256 mismatch for {where}/{entry.file}: "
-                           f"expected {entry.sha256}, got {actual}")
+        raise DatasetError(
+            f"dataset {name}: sha256 mismatch for {where}/{entry.file}: "
+            f"expected {entry.sha256}, got {actual}"
+        )
     if os.environ.get("RESULT_DIR"):
         Path(os.environ["RESULT_DIR"], "dataset.txt").write_text(
-            f"{name} {where} file={entry.file} sha256={actual} path={path}\n")
+            f"{name} {where} file={entry.file} sha256={actual} path={path}\n"
+        )
     return path
 
 
diff --git a/inferencex-e2e/infx/serving/greenspec.py b/inferencex-e2e/infx/serving/greenspec.py
new file mode 100644
index 0000000000..f4ceac121f
--- /dev/null
+++ b/inferencex-e2e/infx/serving/greenspec.py
@@ -0,0 +1,454 @@
+"""GreenSpec on the serving client: subset selection, request payloads, and the artifact guard.
+
+GreenSpec is an internal speculative-decoding bench cut from production chat traffic. A release
+ships every subset in one flat ``greenspec.jsonl``: each line is a fixed set of metadata keys
+(``META_KEYS``) followed by a complete chat-completions body with no ``model`` and no ``stream``.
+The ``greenspec`` scenario replays one selection of it through AgentX as a ``raw_payload`` file:
+
+``settings``   the selection and payload options a serving YAML asks for (overrides:).
+``write_selection``  select the rows and write each as a ready request body.
+``tee_filtered``     run AgentX with its console passed through ``ConsoleFilter``: no text that can
+                     echo server error text, each progress bar at most once per interval.
+``guard``      run after the client, before any upload. The benchmark repo is public and its
+               artifacts are downloadable, so nothing derived from row content may leave the
+               runner: keep only ``ALLOWLIST`` in the result dir, strip error messages from the
+               aiperf export (a server error can echo request content), withhold the server logs,
+               then fail if any ``question_id`` of the release, or any file over ``MAX_BYTES``,
+               is still in what gets uploaded.
+"""
+
+from __future__ import annotations
+
+import json
+import re
+import shutil
+import subprocess
+import sys
+import time
+from collections.abc import Callable, Iterator, Mapping, Sequence
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+FAMILY = "greenspec"
+META_KEYS = ("question_id", "category", "bucket", "lite", "isl", "osl", "thinking_src", "recorded")
+MAX_TOKEN_KEYS = ("max_tokens", "max_completion_tokens")
+SUBSETS = ("diverse", "agentic_coding", "agentic_tools", "llm_task")
+MODES = ("target-default", "recorded")
+OVERRIDES = ("subset", "lite", "bucket", "mode", "max_tokens", "requests")
+DEFAULT_MAX_TOKENS = 8192
+BUCKET = re.compile(r"[a-z_]+__[0-9a-z+-]+")
+# Relative to the result dir; everything else is removed before the upload.
+ALLOWLIST = (
+    "aiperf_artifacts/profile_export_aiperf.json",
+    "aiperf_artifacts/server_metrics_export.json",  # engine counters and gauges, nothing of a row
+    "benchmark_command.txt",
+    "benchmark.log",
+    "scenario.yaml",
+    "serving.yaml",
+    "dataset.txt",
+    "spec_metrics_before.json",
+    "spec_metrics_after.json",
+)
+MAX_BYTES = 16 * 1024 * 1024
+# aiperf's tqdm bars: "{desc}: {n}/{total} |{bar}| {pct}% [{elapsed}<{remaining}]", or
+# "{desc}: {pct}% |{bar}| [{elapsed}<{remaining}]" when the total is unknown.
+BAR_RE = re.compile(
+    r"(?P<desc>[A-Za-z][A-Za-z _-]{0,30}): +(?:(?P<n>[\d,]+)/(?P<total>[\d,]+) +)?"
+    r"(?:(?P<pct0>\d+)% +)?\|[^|]*\| *(?:(?P<pct>\d+)% +)?"
+    r"\[(?P<elapsed>[\d:]+)<(?P<eta>[\d:?]+)\]"
+)
+# aiperf's log record: "14:27:38.612 INFO     message (file.py:64)".
+LOG_RE = re.compile(r"\d\d:\d\d:\d\d\.\d{3} +([A-Z]+) +")
+ANSI_RE = re.compile(r"\x1b\[[0-9;?]*[A-Za-z]")
+WITHHELD_LEVELS = ("ERROR", "CRITICAL")
+SERVER_LOG_STUB = (
+    "server log withheld: GreenSpec rows are internal and a server log can echo them\n"
+)
+PROGRESS_INTERVAL = 60.0
+
+
+class GreenSpecError(ValueError):
+    """A selection problem, as one line."""
+
+
+@dataclass(frozen=True)
+class Settings:
+    subset: str = "diverse"
+    lite: bool = False
+    bucket: str | None = None
+    mode: str = "target-default"
+    max_tokens: int = DEFAULT_MAX_TOKENS
+    requests: int | None = None  # None: the whole selection
+
+    @property
+    def recorded(self) -> bool:
+        return self.mode == "recorded"
+
+
+def normalize(overrides: Mapping[str, Any]) -> Settings:
+    """The overrides of a greenspec serving YAML as ``Settings``; one line per problem."""
+    if unknown := sorted(set(overrides) - set(OVERRIDES)):
+        raise GreenSpecError(
+            f"overrides: greenspec takes {', '.join(OVERRIDES)}, not {', '.join(unknown)}"
+        )
+    subset = overrides.get("subset", "diverse")
+    if subset not in SUBSETS:
+        raise GreenSpecError(f"overrides.subset: {subset!r} is not one of {', '.join(SUBSETS)}")
+    lite = overrides.get("lite", False)
+    if not isinstance(lite, bool):
+        raise GreenSpecError(f"overrides.lite: true or false, got {lite!r}")
+    bucket = overrides.get("bucket")
+    if bucket is not None and not (isinstance(bucket, str) and BUCKET.fullmatch(bucket)):
+        raise GreenSpecError(
+            f"overrides.bucket: <category>__<range> like llm_task__16k+, got {bucket!r}"
+        )
+    mode = overrides.get("mode", "target-default")
+    if mode not in MODES:
+        raise GreenSpecError(f"overrides.mode: {mode!r} is not one of {', '.join(MODES)}")
+    max_tokens = overrides.get("max_tokens", DEFAULT_MAX_TOKENS)
+    requests = overrides.get("requests")
+    for field, value in (("max_tokens", max_tokens), ("requests", requests)):
+        if value is not None and (
+            isinstance(value, bool) or not isinstance(value, int) or value < 1
+        ):
+            raise GreenSpecError(f"overrides.{field}: a positive integer, got {value!r}")
+    return Settings(subset, lite, bucket, mode, max_tokens, requests)
+
+
+def settings(env: Mapping[str, str]) -> Settings | None:
+    """The job's ``Settings`` when its serving YAML runs the greenspec scenario, else None."""
+    import yaml  # not imported at the top: the client's first pass runs before the AgentX venv exists
+
+    try:
+        raw = yaml.safe_load(env.get("SERVING_YAML") or "")
+    except yaml.YAMLError:
+        return None
+    if not isinstance(raw, dict) or raw.get("scenario") != FAMILY:
+        return None
+    return normalize(raw.get("overrides") or {})
+
+
+def label(chosen: Settings) -> str:
+    """The result label of a selection: greenspec-diverse, greenspec-agentic_tools-lite-recorded-mt4096-n200, ..."""
+    parts = [FAMILY, chosen.subset]
+    if chosen.bucket:
+        parts.append(chosen.bucket.replace("__", "_").replace("+", "plus"))
+    if chosen.lite:
+        parts.append("lite")
+    if chosen.recorded:
+        parts.append("recorded")
+    if chosen.max_tokens != DEFAULT_MAX_TOKENS:
+        parts.append(f"mt{chosen.max_tokens}")
+    if chosen.requests:
+        parts.append(f"n{chosen.requests}")
+    return "-".join(parts)
+
+
+def iter_rows(path: Path) -> Iterator[dict[str, Any]]:
+    with open(path, encoding="utf-8") as f:
+        for line in f:
+            if line.strip():
+                yield json.loads(line)
+
+
+def keep(row: dict[str, Any], chosen: Settings) -> bool:
+    return (
+        (chosen.subset == "diverse" or row["category"] == chosen.subset)
+        and (not chosen.lite or row["lite"])
+        and (chosen.bucket is None or row["bucket"] == chosen.bucket)
+    )
+
+
+def select(path: Path, chosen: Settings) -> list[dict[str, Any]]:
+    return [row for row in iter_rows(path) if keep(row, chosen)]
+
+
+SCHEMA_MAPS = ("properties", "patternProperties", "$defs", "definitions", "dependentSchemas")
+SCHEMA_LISTS = ("allOf", "anyOf", "oneOf", "prefixItems", "items")
+SCHEMA_ONES = (
+    "items",
+    "additionalProperties",
+    "additionalItems",
+    "contains",
+    "propertyNames",
+    "not",
+    "if",
+    "then",
+    "else",
+    "unevaluatedItems",
+    "unevaluatedProperties",
+)
+
+
+def fix_required(schema: Any) -> Any:
+    """A copy of a JSON schema whose ``required`` keyword is a list wherever it was an empty dict.
+
+    Some releases (v1.1.0) serialized "no required fields" as ``{}``. vLLM accepts it, SGLang
+    rejects the request with a 400, and the chat template prints nothing for either. Only the
+    keyword is rewritten: a property *named* ``required`` lives under ``properties`` and is a
+    sub-schema, so the walk goes by schema structure and leaves names, enum, const and default alone.
+    """
+    if not isinstance(schema, dict):
+        return schema
+    out = dict(schema)
+    if out.get("required") == {}:
+        out["required"] = []
+    for key in SCHEMA_MAPS:
+        if isinstance(out.get(key), dict):
+            out[key] = {name: fix_required(sub) for name, sub in out[key].items()}
+    for key in SCHEMA_LISTS:
+        if isinstance(out.get(key), list):
+            out[key] = [fix_required(sub) for sub in out[key]]
+    for key in SCHEMA_ONES:
+        if isinstance(out.get(key), dict):
+            out[key] = fix_required(out[key])
+    return out
+
+
+def fix_tools(tools: Any) -> Any:
+    """``tools`` with every function's ``parameters`` run through :func:`fix_required`."""
+    if not isinstance(tools, list):
+        return tools
+    fixed = []
+    for tool in tools:
+        function = tool.get("function") if isinstance(tool, dict) else None
+        if isinstance(function, dict) and "parameters" in function:
+            tool = {
+                **tool,
+                "function": {**function, "parameters": fix_required(function["parameters"])},
+            }
+        fixed.append(tool)
+    return fixed
+
+
+def payload(row: dict[str, Any], model: str, chosen: Settings) -> dict[str, Any]:
+    """One row -> the request body the client sends for it.
+
+    Target-default keeps messages, tools and the like and lets the served model choose its own
+    sampling; recorded also replays the production sampling. The output cap only ever lowers the
+    row's own cap.
+    """
+    body = {k: v for k, v in row.items() if k not in META_KEYS}
+    if "tools" in body:
+        body["tools"] = fix_tools(body["tools"])
+    if chosen.recorded:
+        body.update(row.get("recorded") or {})
+    capped = [k for k in MAX_TOKEN_KEYS if k in body]
+    if capped:
+        for k in capped:
+            body[k] = min(body[k], chosen.max_tokens)
+    else:
+        body["max_tokens"] = chosen.max_tokens
+    return {"model": model, **body, "stream": True, "stream_options": {"include_usage": True}}
+
+
+def write_selection(dataset: Path, chosen: Settings, model: str, out: Path) -> int:
+    """Write the selected rows as one request body per line to ``out``; return how many."""
+    rows = select(dataset, chosen)
+    if not rows:
+        raise GreenSpecError(f"selection {label(chosen)}: no row of the release matches")
+    out.parent.mkdir(parents=True, exist_ok=True)
+    with open(out, "w", encoding="utf-8") as f:
+        f.writelines(
+            json.dumps(payload(row, model, chosen), ensure_ascii=False) + "\n" for row in rows
+        )
+    return len(rows)
+
+
+class ConsoleFilter:
+    """aiperf's console for the job log, minus what can echo server error text.
+
+    The Error Summary table is dropped (its Message column is the raw server response) and
+    replaced by its codes and counts; an ERROR or CRITICAL record keeps its timestamp and level and
+    loses its text and continuation lines (a traceback, wrapped text). An unterminated table stays
+    dropped. Each progress bar is rebuilt from its numbers and printed at most once per
+    ``interval`` seconds, plus once at 100%.
+    """
+
+    def __init__(
+        self,
+        write: Callable[[str], None],
+        interval: float,
+        clock: Callable[[], float] = time.monotonic,
+    ) -> None:
+        self._write = write
+        self._interval = interval
+        self._clock = clock
+        self._bars: dict[str, tuple[float, str]] = {}
+        self._in_table = False
+        self._dropping = False
+        self._errors: list[str] = []
+
+    def feed(self, segment: str) -> None:
+        text = ANSI_RE.sub("", segment).rstrip()
+        if not text:  # tqdm's \r next to a record's \n leaves empty segments
+            return
+        bar = BAR_RE.match(text.lstrip())
+        if bar:
+            self._dropping = False  # a bar ends the continuation of an ERROR record
+            self._bar(bar)
+            # tqdm writes no newline, so a log record can follow on the same line.
+            text = text.lstrip()[bar.end() :]
+            if not text.strip():
+                return
+        if self._in_table:
+            self._table_line(text)
+            return
+        log = LOG_RE.match(text)
+        if log:
+            self._dropping = log.group(1) in WITHHELD_LEVELS
+            if self._dropping:
+                self._write(f"{text[: log.end()]}<withheld: may echo server error text>")
+                return
+        elif "Error Summary" in text:
+            self._in_table, self._dropping = True, False
+            return
+        elif self._dropping:
+            return
+        self._write(text)
+
+    def close(self) -> None:
+        if self._in_table:
+            self._in_table = False
+            self._report_errors()
+
+    def _bar(self, m: re.Match[str]) -> None:
+        desc, pct = m["desc"].strip(), m["pct"] or m["pct0"]
+        now, last = self._clock(), self._bars.get(desc)
+        if last and (last[1] == pct == "100" or (pct != "100" and now - last[0] < self._interval)):
+            return
+        self._bars[desc] = (now, pct)
+        count = f" {m['n']}/{m['total']}" if m["n"] else ""
+        self._write(
+            f"GreenSpec progress: {desc}{count} ({pct}%) elapsed {m['elapsed']} eta {m['eta']}"
+        )
+
+    def _table_line(self, text: str) -> None:
+        if text.lstrip().startswith(("└", "╰")):
+            self._in_table = False
+            self._report_errors()
+            return
+        cells = [c.strip() for c in text.split("│")]
+        if len(cells) >= 4 and (cells[1].isdigit() or cells[1] == "N/A"):
+            count = cells[-2].replace(",", "")
+            if count.isdigit():
+                self._errors.append(f"{cells[1]} x{count}")
+
+    def _report_errors(self) -> None:
+        listed = ", ".join(self._errors) or "no rows parsed"
+        self._write(
+            f"GreenSpec: aiperf error summary withheld (messages can echo requests): {listed}"
+        )
+        self._errors = []
+
+
+def filter_console(stream: Any, write: Callable[[str], None], interval: float) -> None:
+    """Feed a binary stream to ``ConsoleFilter`` one \\r- or \\n-delimited segment at a time.
+
+    A segment the filter fails on is dropped rather than ending the read: aiperf writes into this
+    pipe and would die of a broken pipe.
+    """
+    console = ConsoleFilter(write, interval)
+
+    def feed(raw: bytes) -> None:
+        try:
+            console.feed(raw.decode("utf-8", errors="replace"))
+        except Exception:  # noqa: BLE001
+            write("GreenSpec: console line dropped (filter error)")
+
+    pending = b""
+    while chunk := stream.read1(65536):
+        *segments, pending = re.split(rb"[\r\n]", pending + chunk)
+        for segment in segments:
+            feed(segment)
+    if pending:
+        feed(pending)
+    console.close()
+
+
+def tee_filtered(
+    argv: Sequence[str], log: Path, env: Mapping[str, str], interval: float = PROGRESS_INTERVAL
+) -> int:
+    """``argv 2>&1 | filter | tee log``: the job log and ``log`` get the filtered console only."""
+    from infx.bench import proc
+
+    proc.echo(argv, env)
+    with log.open("w", encoding="utf-8") as sink:
+        try:
+            child = subprocess.Popen(
+                argv, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, env=env
+            )
+        except OSError as error:
+            print(f"ERROR: cannot run {argv[0]}: {error}", file=sys.stderr, flush=True)
+            return 127
+
+        def write(line: str) -> None:
+            for stream in (sys.stdout, sink):
+                print(line, file=stream, flush=True)
+
+        with child:
+            if child.stdout is not None:
+                filter_console(child.stdout, write, interval)
+    return proc.status(child.returncode)
+
+
+def _redact_errors(export_path: Path) -> None:
+    export = json.loads(export_path.read_text())
+    for entry in export.get("error_summary") or []:
+        details = entry.get("error_details")
+        if isinstance(details, dict) and "message" in details:
+            details["message"] = "<withheld>"
+    export_path.write_text(json.dumps(export, indent=2))
+
+
+def guard(
+    dataset: Path,
+    result_dir: Path,
+    uploads: Sequence[Path],
+    server_logs: Sequence[Path],
+    keep_log_dir: Path | None = None,
+) -> list[str]:
+    """Prune, redact and scan everything a GreenSpec job uploads. Returns the problems.
+
+    ``result_dir`` keeps only ``ALLOWLIST``; ``uploads`` (the aggregate result) are scanned as they
+    are; each of ``server_logs`` is moved to ``keep_log_dir`` (node-local) if given and replaced by a stub.
+    """
+    if result_dir.is_dir():
+        for p in sorted(result_dir.rglob("*"), reverse=True):
+            rel = p.relative_to(result_dir).as_posix()
+            if p.is_dir():
+                if not any(p.iterdir()):
+                    p.rmdir()
+            elif rel not in ALLOWLIST:
+                if p.parent == result_dir and p.suffix == ".log" and keep_log_dir is not None:
+                    keep_log_dir.mkdir(parents=True, exist_ok=True)
+                    shutil.move(str(p), str(keep_log_dir / p.name))
+                else:
+                    p.unlink()
+        export = result_dir / "aiperf_artifacts" / "profile_export_aiperf.json"
+        if export.is_file():
+            _redact_errors(export)
+    for log in server_logs:
+        if log.is_file():
+            if keep_log_dir is not None:
+                keep_log_dir.mkdir(parents=True, exist_ok=True)
+                shutil.move(str(log), str(keep_log_dir / log.name))
+            # Swap in a new inode: the server may still hold the log open, and a truncate
+            # would let its next writes land in the stub (after a hole of NULs).
+            tmp = log.with_name(log.name + ".stub")
+            tmp.write_text(SERVER_LOG_STUB)
+            tmp.replace(log)
+    qids = {r["question_id"] for r in iter_rows(dataset)}
+    files = [p for p in result_dir.rglob("*") if p.is_file()] if result_dir.is_dir() else []
+    files += [p for p in uploads if p.is_file()]
+    problems = []
+    for p in files:
+        size = p.stat().st_size
+        if size > MAX_BYTES:
+            problems.append(f"{p}: {size} bytes > {MAX_BYTES}")
+            continue
+        text = p.read_text(errors="replace")
+        if any(q in text for q in qids):
+            problems.append(f"{p}: contains a GreenSpec question_id")
+    return problems
diff --git a/inferencex-e2e/infx/serving/render.py b/inferencex-e2e/infx/serving/render.py
index 17cda9f160..3b2db6a302 100644
--- a/inferencex-e2e/infx/serving/render.py
+++ b/inferencex-e2e/infx/serving/render.py
@@ -117,15 +117,20 @@ def _render_router(spec: ServingSpec, engine_argv: list[str]) -> dict[str, str]:
                 "add a router block running python3 -m dynamo.frontend"
             )
         return {
-            "router-image": "", "router-cmd": "", "router-env": "",
-            "router-profile": "", "router-metrics-port": "",
+            "router-image": "",
+            "router-cmd": "",
+            "router-env": "",
+            "router-profile": "",
+            "router-metrics-port": "",
         }
     prefix_env, argv = _pop_env_prefix(split_serve(spec.router.serve, "router.serve"))
     if not argv:
         raise ServingError("router.serve: empty command")
     is_dynamo_router = _python_module(argv) == "dynamo.frontend"
     if is_dynamo_router and not is_dynamo_engine:
-        raise ServingError("serve: python3 -m dynamo.frontend needs a python3 -m dynamo.sglang engine")
+        raise ServingError(
+            "serve: python3 -m dynamo.frontend needs a python3 -m dynamo.sglang engine"
+        )
     if is_dynamo_engine and not is_dynamo_router:
         raise ServingError(
             "router.serve: a python3 -m dynamo.sglang engine needs python3 -m dynamo.frontend as its router"
@@ -144,7 +149,9 @@ def _render_router(spec: ServingSpec, engine_argv: list[str]) -> dict[str, str]:
         "router-cmd": shlex.join(argv),
         "router-env": "\n".join(f"{key}={value}" for key, value in env.items()),
         "router-profile": "dynamo" if is_dynamo_router else "generic",
-        "router-metrics-port": "" if spec.router.metrics_port is None else str(spec.router.metrics_port),
+        "router-metrics-port": ""
+        if spec.router.metrics_port is None
+        else str(spec.router.metrics_port),
     }
 
 
diff --git a/inferencex-e2e/infx/serving/scenarios.py b/inferencex-e2e/infx/serving/scenarios.py
index e1fb57e917..c76b23abb1 100644
--- a/inferencex-e2e/infx/serving/scenarios.py
+++ b/inferencex-e2e/infx/serving/scenarios.py
@@ -13,11 +13,20 @@
 
 import yaml
 
+from infx.serving import greenspec
 from infx.serving.datasets import LockEntry, load_lock
 from infx.serving.slo import SLOError, parse_slo
 
 SCENARIOS_DIR = Path(__file__).resolve().parents[2] / "benchmarks/single_node/serving/scenarios"
-FAMILIES = ("agentx", "tau2", "livekit", "shared-prefix-e2", "hidden-prefix-e3", "synthetic")
+FAMILIES = (
+    "agentx",
+    "tau2",
+    "livekit",
+    "shared-prefix-e2",
+    "hidden-prefix-e3",
+    "synthetic",
+    "greenspec",
+)
 OVERRIDE_KEYS = (
     "duration",
     "dataset",
@@ -91,25 +100,32 @@ def _positive(field: str, value: Any) -> float:
     return float(value)
 
 
-def _apply(doc: dict[str, Any], family: str, overrides: dict[str, Any],
-           lock: dict[str, LockEntry]) -> list[str]:
+def _apply(
+    doc: dict[str, Any], family: str, overrides: dict[str, Any], lock: dict[str, LockEntry]
+) -> list[str]:
     """Apply ``overrides`` to ``doc`` in place; return the label suffixes, in a fixed order."""
     if unknown := sorted(set(overrides) - set(OVERRIDE_KEYS)):
-        raise ScenarioError(f"overrides: unknown key(s) {', '.join(unknown)}; allowed: {', '.join(OVERRIDE_KEYS)}")
+        raise ScenarioError(
+            f"overrides: unknown key(s) {', '.join(unknown)}; allowed: {', '.join(OVERRIDE_KEYS)}"
+        )
     dataset = doc["benchmark"]["datasets"][0]
     suffixes: list[str] = []
     if "dataset" in overrides:
         value = overrides["dataset"]
         if dataset["type"] == "public":
             if not (isinstance(value, str) and value.startswith(PUBLIC_DATASET)):
-                raise ScenarioError(f"overrides.dataset: {family} takes a public {PUBLIC_DATASET}* dataset, got {value!r}")
+                raise ScenarioError(
+                    f"overrides.dataset: {family} takes a public {PUBLIC_DATASET}* dataset, got {value!r}"
+                )
             dataset["dataset"] = value
             if value != PUBLIC_DATASET:
                 suffixes.append("-" + value.removeprefix(PUBLIC_DATASET).lstrip("_"))
         else:
             entry = lock.get(value) if isinstance(value, str) else None
             if entry is None:
-                raise ScenarioError(f"overrides.dataset: {value!r} is not in the dataset lock; available: {', '.join(lock)}")
+                raise ScenarioError(
+                    f"overrides.dataset: {value!r} is not in the dataset lock; available: {', '.join(lock)}"
+                )
             if entry.entries is None:
                 raise ScenarioError(f"overrides.dataset: lock entry {value} has no entries count")
             dataset["name"], dataset["entries"] = value, entry.entries
@@ -163,7 +179,9 @@ def _synthetic(point: Point, search: bool, slo: str | None) -> dict[str, Any]:
     if point.osl not in OSL:
         raise ScenarioError(f"workload.osl: {point.osl!r} is not one of {', '.join(OSL)}")
     if point.cache not in CACHE:
-        raise ScenarioError(f"workload.cache: {point.cache!r} is not one of {', '.join(map(str, CACHE))}")
+        raise ScenarioError(
+            f"workload.cache: {point.cache!r} is not one of {', '.join(map(str, CACHE))}"
+        )
     if slo and not search:
         raise ScenarioError("slo: only applies with workload.search: true")
     isl = ISL[point.isl]
@@ -182,10 +200,15 @@ def _synthetic(point: Point, search: bool, slo: str | None) -> dict[str, Any]:
         doc["sweep"] = {
             "type": "adaptive_search",
             "planner": "smooth_isotonic",
-            "objectives": [{"metric": "output_token_throughput", "stat": "avg", "direction": "maximize"}],
-            "search_space": [{"path": "phases.profiling.concurrency", "lo": 1, "hi": 512, "kind": "int"}],
+            "objectives": [
+                {"metric": "output_token_throughput", "stat": "avg", "direction": "maximize"}
+            ],
+            "search_space": [
+                {"path": "phases.profiling.concurrency", "lo": 1, "hi": 512, "kind": "int"}
+            ],
             "sla_filters": [
-                {"metric_tag": s.metric, "stat": s.stat, "op": s.op, "threshold": s.threshold} for s in filters
+                {"metric_tag": s.metric, "stat": s.stat, "op": s.op, "threshold": s.threshold}
+                for s in filters
             ],
             "max_iterations": 16,
             "sla_precision": "normal",  # 1000 requests/probe cap
@@ -193,17 +216,49 @@ def _synthetic(point: Point, search: bool, slo: str | None) -> dict[str, Any]:
     return doc
 
 
-def render(family: str, overrides: dict[str, Any] | None = None, point: Point | None = None,
-           search: bool = False, slo: str | None = None,
-           lock: dict[str, LockEntry] | None = None) -> Rendered:
+def _greenspec(
+    overrides: dict[str, Any],
+    point: Point | None,
+    search: bool,
+    slo: str | None,
+    lock: dict[str, LockEntry],
+) -> Rendered:
+    """The greenspec scenario with its selection; the run length is the selection, so there is no duration."""
+    if point is not None or search or slo:
+        raise ScenarioError("workload/slo: only for scenario synthetic, not greenspec")
+    try:
+        chosen = greenspec.normalize(overrides)
+    except greenspec.GreenSpecError as error:
+        raise ScenarioError(str(error)) from None
+    doc = load_base("greenspec")
+    name = doc["benchmark"]["datasets"][0]["name"]
+    if name not in lock:
+        raise ScenarioError(f"scenario: dataset {name} of greenspec is not in the dataset lock")
+    if chosen.requests:
+        _profiling(doc)["requests"] = chosen.requests
+    return Rendered(doc, greenspec.label(chosen), "file", name, False, 0.0)
+
+
+def render(
+    family: str,
+    overrides: dict[str, Any] | None = None,
+    point: Point | None = None,
+    search: bool = False,
+    slo: str | None = None,
+    lock: dict[str, LockEntry] | None = None,
+) -> Rendered:
     """The final AIPerf config of ``family`` with its variant applied."""
     overrides = dict(overrides or {})
     lock = load_lock() if lock is None else lock
+    if family == "greenspec":
+        return _greenspec(overrides, point, search, slo, lock)
     if family == "synthetic":
         if point is None:
             raise ScenarioError("workload: synthetic needs isl, osl and cache")
         if extra := sorted(set(overrides) - set(SYNTHETIC_OVERRIDES)):
-            raise ScenarioError(f"overrides: synthetic only takes {', '.join(SYNTHETIC_OVERRIDES)}, not {', '.join(extra)}")
+            raise ScenarioError(
+                f"overrides: synthetic only takes {', '.join(SYNTHETIC_OVERRIDES)}, not {', '.join(extra)}"
+            )
         doc = _synthetic(point, search, slo)
         suffixes = _apply(doc, family, overrides, lock)
         label = f"{'sweep' if search else 'fixed'}-{point.label}{''.join(suffixes)}"
diff --git a/inferencex-e2e/infx/serving/schema.py b/inferencex-e2e/infx/serving/schema.py
index ab6a8c4a6b..90a7b771f6 100644
--- a/inferencex-e2e/infx/serving/schema.py
+++ b/inferencex-e2e/infx/serving/schema.py
@@ -71,6 +71,7 @@
     "env",
     "router",
     "storage",
+    "arms",
 )
 
 
@@ -124,6 +125,20 @@ def _env_strings(cls, value: Any) -> Any:
         return _coerce_env(value)
 
 
+class ArmSpec(BaseModel):
+    """One engine variant of a serving YAML: ``serve:`` plus ``append``, ``env:`` overlaid by ``env``."""
+
+    model_config = ConfigDict(extra="forbid")
+
+    append: str = ""
+    env: dict[str, str] = Field(default_factory=dict)
+
+    @field_validator("env", mode="before")
+    @classmethod
+    def _env_strings(cls, value: Any) -> Any:
+        return _coerce_env(value)
+
+
 class EndpointSpec(BaseModel):
     """An endpoint we do not launch: an OpenAI-compatible URL, its served model and its key."""
 
@@ -205,8 +220,15 @@ class ServingSpec(BaseModel):
     endpoint: EndpointSpec | None = None
     router: RouterSpec | None = None
     storage: StorageSpec | None = None
+    arms: dict[str, ArmSpec] | None = Field(default=None, min_length=1)
     scenario: Literal[
-        "agentx", "tau2", "livekit", "shared-prefix-e2", "hidden-prefix-e3", "synthetic"
+        "agentx",
+        "tau2",
+        "livekit",
+        "shared-prefix-e2",
+        "hidden-prefix-e3",
+        "synthetic",
+        "greenspec",
     ]
     overrides: dict[str, Any] = Field(default_factory=dict)
     workload: Workload | None = None
@@ -223,6 +245,14 @@ def _scalar_conc(cls, value: Any) -> Any:
     def _env_strings(cls, value: Any) -> Any:
         return _coerce_env(value)
 
+    @field_validator("arms")
+    @classmethod
+    def _arm_names(cls, value: dict[str, ArmSpec] | None) -> dict[str, ArmSpec] | None:
+        for name in value or {}:
+            if not re.fullmatch(NAME_PATTERN, name):
+                raise ValueError(f"{name!r}: arm names use lowercase letters, digits, . and -")
+        return value
+
 
 def _format_validation_error(error: ValidationError) -> str:
     models = {
@@ -271,6 +301,24 @@ def runner_gpus(runner: str, inventory: RunnerInventory) -> int:
     return min(gpus)
 
 
+def arm_specs(spec: ServingSpec) -> list[tuple[str | None, ServingSpec]]:
+    """The engine configurations of a serving YAML: one per arm (``serve:`` + the arm's append), or just the YAML."""
+    if spec.arms is None:
+        return [(None, spec)]
+    out = []
+    for name, arm in spec.arms.items():
+        serve = f"{spec.serve.rstrip()}\n{arm.append}" if spec.serve and arm.append else spec.serve
+        out.append(
+            (
+                name,
+                spec.model_copy(
+                    update={"serve": serve, "env": {**spec.env, **arm.env}, "arms": None}
+                ),
+            )
+        )
+    return out
+
+
 def variants(spec: ServingSpec) -> list[Rendered]:
     """Every scenario variant the YAML asks for (one per workload point, or exactly one)."""
     try:
@@ -441,7 +489,15 @@ def load_serving(text: str, inventory: RunnerInventory | None = None) -> Serving
         ]
         if missing:
             raise ServingError("\n".join(missing))
-        problems += _engine_problems(spec)
+        seen: dict[tuple[str, tuple[tuple[str, str], ...]], str] = {}
+        for name, derived in arm_specs(spec):
+            problems += [
+                f"arms.{name}.{line}" if name else line for line in _engine_problems(derived)
+            ]
+            key = (" ".join(derived.serve.split()), tuple(sorted(derived.env.items())))
+            if name and key in seen:
+                problems.append(f"arms.{name}: the same engine command and env as arms.{seen[key]}")
+            seen[key] = name or ""
     if len(set(spec.conc)) != len(spec.conc):
         problems.append("conc: duplicate values")
     try:
diff --git a/inferencex-e2e/infx/serving/specdec.py b/inferencex-e2e/infx/serving/specdec.py
new file mode 100644
index 0000000000..5a62a4860d
--- /dev/null
+++ b/inferencex-e2e/infx/serving/specdec.py
@@ -0,0 +1,112 @@
+"""Speculative-decoding acceptance of one serving point, shared by every scenario that measures it.
+
+The client snapshots the engines' Prometheus counters before and after the replay and files the
+delta under ``spec_decode`` in the aggregate result:
+
+    acceptance_length = 1 + accepted tokens / draft events     (vLLM counters; independent of depth)
+    acceptance_rate   = accepted tokens / drafted tokens
+    draft_depth       = drafted tokens / draft events           (num_speculative_tokens)
+
+SGLang only exports a running-average gauge (``sglang:spec_accept_length``), recorded as read after
+the run; it is 0.0 on an engine without speculative decoding, which counts as no series. A point without speculative decoding has no such series: ``summarize`` returns None and the
+aggregate stays untouched.
+"""
+
+from __future__ import annotations
+
+import json
+import math
+import re
+import sys
+import urllib.request
+from collections.abc import Iterable
+from pathlib import Path
+from typing import Any
+
+VLLM_COUNTERS = {
+    "accepted_tokens": "vllm:spec_decode_num_accepted_tokens_total",
+    "draft_events": "vllm:spec_decode_num_drafts_total",
+    "drafted_tokens": "vllm:spec_decode_num_draft_tokens_total",
+}
+SGLANG_ACCEPT_LENGTH = "sglang:spec_accept_length"
+WANTED = {*VLLM_COUNTERS.values(), SGLANG_ACCEPT_LENGTH}
+_SAMPLE = re.compile(r"^(?P<name>[A-Za-z_:][\w:]*)(?:\{[^}]*\})?\s+(?P<value>\S+)")
+
+
+def parse_prometheus(text: str) -> dict[str, float]:
+    """Sum every sample of each metric across label sets (e.g. per-engine series)."""
+    totals: dict[str, float] = {}
+    for line in text.splitlines():
+        match = _SAMPLE.match(line)
+        if match is None:
+            continue
+        try:
+            value = float(match["value"])
+        except ValueError:
+            continue
+        if math.isfinite(value):
+            totals[match["name"]] = totals.get(match["name"], 0.0) + value
+    return totals
+
+
+def snapshot(urls: Iterable[str], timeout: float = 10.0) -> dict[str, float]:
+    """The speculative-decoding series summed over ``urls``; an unreachable URL or a missing series is skipped."""
+    out: dict[str, float] = {}
+    for url in urls:
+        try:
+            with urllib.request.urlopen(url, timeout=timeout) as response:  # noqa: S310 -- the job's own engine ports
+                totals = parse_prometheus(response.read().decode())
+        except (OSError, ValueError) as error:
+            print(f"WARNING spec-decode metrics: {url} unreadable ({error})", file=sys.stderr)
+            continue
+        for name, value in totals.items():
+            if name in WANTED:
+                out[name] = out.get(name, 0.0) + value
+    return out
+
+
+def acceptance(before: dict[str, float], after: dict[str, float]) -> dict[str, Any] | None:
+    """Acceptance over the run from two snapshots, or None when the engine reports no speculative decoding."""
+    # SGLang exports the gauge on every engine, 0.0 when the engine does not speculate.
+    sglang_length = after.get(SGLANG_ACCEPT_LENGTH, 0.0)
+    if not {*before, *after} & set(VLLM_COUNTERS.values()) and sglang_length <= 0:
+        return None
+    delta = {
+        key: round(after.get(name, 0.0) - before.get(name, 0.0))
+        for key, name in VLLM_COUNTERS.items()
+    }
+    result: dict[str, Any] = {f"spec_{key}": value for key, value in delta.items()}
+    result["acceptance_length"] = (
+        1 + delta["accepted_tokens"] / delta["draft_events"] if delta["draft_events"] > 0 else None
+    )
+    result["draft_depth"] = (
+        round(delta["drafted_tokens"] / delta["draft_events"])
+        if delta["draft_events"] > 0
+        else None
+    )
+    result["acceptance_rate"] = (
+        delta["accepted_tokens"] / delta["drafted_tokens"] if delta["drafted_tokens"] > 0 else None
+    )
+    if result["acceptance_length"] is None and sglang_length > 0:
+        result["acceptance_length"] = sglang_length
+    return result
+
+
+def save(path: Path, snap: dict[str, float]) -> None:
+    path.write_text(json.dumps(snap, indent=2))
+
+
+def summarize(before_path: Path, after_path: Path) -> dict[str, Any] | None:
+    """``acceptance`` of the two snapshot files the client wrote; None when either is missing."""
+    if not (before_path.is_file() and after_path.is_file()):
+        return None
+    return acceptance(json.loads(before_path.read_text()), json.loads(after_path.read_text()))
+
+
+def merge(aggregate_path: Path, spec: dict[str, Any]) -> None:
+    """Add ``spec`` as ``spec_decode`` to the aggregate result, if the aggregate exists."""
+    if not aggregate_path.is_file():
+        return
+    result = json.loads(aggregate_path.read_text())
+    result["spec_decode"] = spec
+    aggregate_path.write_text(json.dumps(result, indent=2))
diff --git a/inferencex-e2e/infx/serving/summary.py b/inferencex-e2e/infx/serving/summary.py
new file mode 100644
index 0000000000..9db3d04d7a
--- /dev/null
+++ b/inferencex-e2e/infx/serving/summary.py
@@ -0,0 +1,121 @@
+"""Markdown table of the points of one gn-benchmark run, for the run's summary page.
+
+    python -P -m infx.serving.summary <dir>  >> "$GITHUB_STEP_SUMMARY"
+
+``<dir>`` holds the aggregate result JSONs the points uploaded (any depth). One row per point; when a
+(scenario, concurrency) has a speculative arm and a base arm (spec_decoding none), the speculative
+row also carries the speedup over the base. Only aggregate numbers are printed: nothing of a request.
+"""
+
+from __future__ import annotations
+
+import json
+import sys
+from pathlib import Path
+from typing import Any
+
+HEADERS = (
+    "scenario",
+    "spec",
+    "conc",
+    "requests ok/total",
+    "output tok/s",
+    "TTFT p50 / p95 (ms)",
+    "TPOT mean (ms)",
+    "acceptance length",
+    "accept rate",
+    "speedup",
+)
+DASH = "-"
+OUTPUT_TPS = ("request_metrics", "throughput", "output", "tokens_per_second")
+
+
+def dig(doc: Any, *path: str) -> Any:
+    for key in path:
+        if not isinstance(doc, dict) or key not in doc:
+            return None
+        doc = doc[key]
+    return doc
+
+
+def load(directory: Path) -> list[dict[str, Any]]:
+    """The aggregate results under ``directory``, ordered for display."""
+    results = []
+    for path in sorted(directory.rglob("*.json")):
+        try:
+            doc = json.loads(path.read_text())
+        except (OSError, ValueError):
+            continue
+        if isinstance(doc, dict) and "request_metrics" in doc and "scenario" in doc:
+            results.append(doc)
+    return sorted(
+        results,
+        key=lambda r: (str(r["scenario"]), int(r.get("conc") or 0), str(r.get("spec_decoding"))),
+    )
+
+
+def number(value: Any, digits: int = 1, scale: float = 1.0) -> str:
+    return (
+        f"{value * scale:,.{digits}f}"
+        if isinstance(value, int | float) and not isinstance(value, bool)
+        else DASH
+    )
+
+
+def speedups(results: list[dict[str, Any]]) -> dict[int, float]:
+    """Output-throughput ratio of each speculative result over the base result of its scenario and concurrency."""
+    base = {
+        (r["scenario"], r.get("conc")): dig(r, *OUTPUT_TPS)
+        for r in results
+        if r.get("spec_decoding", "none") == "none"
+    }
+    out = {}
+    for index, result in enumerate(results):
+        denominator = base.get((result["scenario"], result.get("conc")))
+        numerator = dig(result, *OUTPUT_TPS)
+        if result.get("spec_decoding", "none") != "none" and denominator and numerator:
+            out[index] = numerator / denominator
+    return out
+
+
+def row(result: dict[str, Any], speedup: float | None) -> list[str]:
+    metrics = result["request_metrics"]
+    spec = result.get("spec_decode") or {}
+    ok, total = result.get("num_requests_successful"), result.get("num_requests_total")
+    return [
+        str(result["scenario"]),
+        str(result.get("spec_decoding", "none"))
+        + (f"x{spec['draft_depth']}" if spec.get("draft_depth") else ""),
+        str(result.get("conc", DASH)),
+        f"{ok}/{total}" if ok is not None and total is not None else DASH,
+        number(dig(result, *OUTPUT_TPS)),
+        f"{number(dig(metrics, 'latency', 'ttft', 'p50'), 0, 1000)} / {number(dig(metrics, 'latency', 'ttft', 'p95'), 0, 1000)}",
+        number(dig(metrics, "latency", "tpot", "mean"), 1, 1000),
+        number(spec.get("acceptance_length"), 2),
+        number(spec.get("acceptance_rate"), 1, 100)
+        + ("%" if spec.get("acceptance_rate") is not None else ""),
+        f"{speedup:.2f}x" if speedup else DASH,
+    ]
+
+
+def table(results: list[dict[str, Any]]) -> str:
+    if not results:
+        return "No aggregate result was uploaded by this run.\n"
+    ratios = speedups(results)
+    lines = ["| " + " | ".join(HEADERS) + " |", "|" + "|".join(["---"] * len(HEADERS)) + "|"]
+    lines += ["| " + " | ".join(row(r, ratios.get(i))) + " |" for i, r in enumerate(results)]
+    return "\n".join(lines) + "\n"
+
+
+def main(argv: list[str] | None = None) -> int:
+    argv = sys.argv[1:] if argv is None else argv
+    if len(argv) != 1:
+        print("usage: python -m infx.serving.summary <dir>", file=sys.stderr)
+        return 2
+    print("### Results\n")
+    print(table(load(Path(argv[0]))))
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())
diff --git a/inferencex-e2e/infx/serving/to_matrix.py b/inferencex-e2e/infx/serving/to_matrix.py
index a080edfde3..25ac7dc6dc 100644
--- a/inferencex-e2e/infx/serving/to_matrix.py
+++ b/inferencex-e2e/infx/serving/to_matrix.py
@@ -15,12 +15,28 @@
 from pathlib import Path
 
 from infx.clusters import RunnerInventory
-from infx.serving.render import CLIENT_IMAGE, render, render_remote
+from infx.serving.render import CLIENT_IMAGE, _flag_value, render, render_remote, split_serve
 from infx.serving.scenarios import Rendered
-from infx.serving.schema import ServingError, ServingSpec, load_serving, variants
+from infx.serving.schema import ServingError, ServingSpec, arm_specs, load_serving, variants
 
 
-def _row(spec: ServingSpec, framework: str, conc: int, duration: int, exp_name: str) -> dict:
+def spec_decoding(serve_cmd: str) -> str:
+    """The row's spec-decoding value: mtp for an MTP draft, draft_model for any other, none without one."""
+    argv = split_serve(serve_cmd) if serve_cmd else []
+    if config := _flag_value(argv, "--speculative-config"):  # vLLM
+        try:
+            method = json.loads(config).get("method", "")
+        except (ValueError, AttributeError):
+            method = ""
+        return "mtp" if method == "mtp" else "draft_model"
+    if algorithm := _flag_value(argv, "--speculative-algorithm"):  # SGLang
+        return "mtp" if algorithm.upper() in ("NEXTN", "MTP") else "draft_model"
+    return "none"
+
+
+def _row(
+    spec: ServingSpec, framework: str, conc: int, duration: int, exp_name: str, serve_cmd: str
+) -> dict:
     remote = spec.endpoint
     return {
         "image": CLIENT_IMAGE if remote else spec.image, "model": remote.model if remote else spec.model,
@@ -29,7 +45,7 @@ def _row(spec: ServingSpec, framework: str, conc: int, duration: int, exp_name:
         # tp stays 1 for a remote point: upstream derives GPU_COUNT from it and divides by it;
         # the serving driver, not GPU_COUNT, decides that a remote job takes no GPU.
         "tp": 1 if remote else spec.gpus, "pp": 1, "dcp-size": 1, "pcp-size": 1, "ep": 1, "dp-attn": False,
-        "spec-decoding": "none", "conc": conc, "kv-offloading": "none", "total-cpu-dram-gb": 0,
+        "spec-decoding": spec_decoding(serve_cmd), "conc": conc, "kv-offloading": "none", "total-cpu-dram-gb": 0,
         "duration": duration, "exp-name": exp_name, "scenario-type": "agentic-coding",
     }  # fmt: skip
 
@@ -71,27 +87,45 @@ def _secret(spec: ServingSpec) -> str:
 
 def build(text: str, inventory: RunnerInventory | None = None) -> dict:
     spec = load_serving(text, inventory)
-    rendered = render_remote() if spec.endpoint else render(spec)
     chosen = variants(spec)
+    arms = arm_specs(spec)
     multi = len(chosen) > 1
     points = []
-    for variant in chosen:
-        for conc in spec.conc:
-            if multi:
-                exp_name = f"{spec.name}-{variant.label}"
-            else:
-                exp_name = f"{spec.name}_search" if variant.search else f"{spec.name}_c{conc}"
-            row = _row(spec, rendered["framework"], conc, int(variant.duration), exp_name)
-            points.append(
-                {
-                    "config": row,
-                    "serving": _serving(spec, rendered, variant, text),
-                    "api_key_secret": _secret(spec),
-                }
-            )
+    for arm, engine in arms:
+        rendered = render_remote() if spec.endpoint else render(engine)
+        stem = f"{spec.name}-{arm}" if arm else spec.name
+        for variant in chosen:
+            for conc in spec.conc:
+                if multi:
+                    exp_name = f"{stem}-{variant.label}"
+                else:
+                    exp_name = f"{stem}_search" if variant.search else f"{stem}_c{conc}"
+                row = _row(
+                    spec,
+                    rendered["framework"],
+                    conc,
+                    int(variant.duration),
+                    exp_name,
+                    rendered["serve-cmd"],
+                )
+                points.append(
+                    {
+                        "config": row,
+                        "serving": _serving(engine, rendered, variant, text),
+                        "api_key_secret": _secret(spec),
+                    }
+                )
     what = f"{len(chosen)} points" if multi else chosen[0].label
+    if spec.arms:
+        what += f" x {len(arms)} arms ({','.join(spec.arms)})"
     durations = {int(variant.duration) for variant in chosen}
-    seconds = f"{durations.pop()}s" if len(durations) == 1 else "mixed"
+    seconds = (
+        "mixed"
+        if len(durations) > 1
+        else f"{durations.pop()}s"
+        if durations != {0}
+        else "whole selection"
+    )  # greenspec has no duration
     concs = ",".join(str(conc) for conc in spec.conc)
     return {
         "points": points,
diff --git a/inferencex-e2e/infx/tests/serving/test_client.py b/inferencex-e2e/infx/tests/serving/test_client.py
index ce93bc1271..e30daea8b5 100644
--- a/inferencex-e2e/infx/tests/serving/test_client.py
+++ b/inferencex-e2e/infx/tests/serving/test_client.py
@@ -15,6 +15,7 @@
     online,
     probe,
     redacted,
+    with_requests,
     with_token_source,
 )
 
@@ -265,3 +266,33 @@ def test_datasets_arrow_cache_lives_beside_the_node_hub_cache():
         datasets_cache({"HF_HUB_CACHE": "/mnt/hf", "HF_DATASETS_CACHE": "/x"}) == {}
     )  # caller's choice wins
     assert datasets_cache({}) == {}  # no node cache mounted: keep the library default
+
+
+def test_with_requests_sets_the_profiling_phase_only():
+    scenario = yaml.safe_dump({"benchmark": {"phases": [
+        {"name": "warmup", "kind": "warmup", "requests": 4},
+        {"name": "profiling", "kind": "profiling", "requests": 1},
+    ]}})
+    phases = yaml.safe_load(with_requests(scenario, 777))["benchmark"]["phases"]
+    assert [phase["requests"] for phase in phases] == [4, 777]
+
+
+def test_guard_scans_the_release_and_prunes_the_result_dir(tmp_path, monkeypatch):
+    from infx.serving import client
+
+    release = tmp_path / "greenspec.jsonl"
+    release.write_text(json.dumps({"question_id": "m/secret", "category": "llm_task"}) + "\n")
+    monkeypatch.setattr("infx.serving.datasets.fetch", lambda name: release)
+    monkeypatch.setattr(client, "SELECTION_DIR", str(tmp_path / "selection"))
+    (tmp_path / "selection").mkdir()
+    result_dir, output = tmp_path / "results", tmp_path / "out"
+    result_dir.mkdir(), output.mkdir()
+    (result_dir / "server.log").write_text("engine output")
+    (result_dir / "scenario.yaml").write_text("benchmark: {}")
+    aggregate = output / "agg.json"
+    aggregate.write_text("{}")
+    assert client.guard({"SERVING_DATASET": "greenspec-v1.1.1"}, result_dir, aggregate, output) == 0
+    assert not (tmp_path / "selection").exists() and not (result_dir / "server.log").exists()
+    assert (output / client.PRIVATE_LOGS / "server.log").read_text() == "engine output"
+    aggregate.write_text('{"note": "m/secret"}')
+    assert client.guard({"SERVING_DATASET": "greenspec-v1.1.1"}, result_dir, aggregate, output) == 1
diff --git a/inferencex-e2e/infx/tests/serving/test_datasets.py b/inferencex-e2e/infx/tests/serving/test_datasets.py
index 2fc707915c..ceaffdf449 100644
--- a/inferencex-e2e/infx/tests/serving/test_datasets.py
+++ b/inferencex-e2e/infx/tests/serving/test_datasets.py
@@ -7,14 +7,13 @@
 from infx.serving.datasets import DatasetError, LockEntry, fetch, load_lock
 
 
-def test_lock_pins_six_datasets_with_entries():
-    lock = load_lock()
-    assert set(lock) == {
-        "tau2-cs", "shared-prefix-e2", "hidden-prefix-e3",
-        "livekit-111868-v1", "livekit-111868-v2", "livekit-111868-v1.1-warm",
-    }
-    assert lock["livekit-111868-v1.1-warm"].entries == 671
-    assert lock["livekit-111868-v1"].entries == 1217
+def test_lock_private_flag_defaults_to_false(tmp_path):
+    body = "x: {repo: a/b, revision: " + "a" * 40 + ", file: f, sha256: " + "b" * 64 + "}\n"
+    path = tmp_path / "lock.yaml"
+    path.write_text(body)
+    assert load_lock(path)["x"].private is False
+    path.write_text(body.replace("}\n", ", private: true}\n"))
+    assert load_lock(path)["x"].private is True
 
 
 @pytest.mark.parametrize("body, message", [
@@ -23,6 +22,8 @@ def test_lock_pins_six_datasets_with_entries():
     # all-digit hashes would load as YAML ints, hence a/b
     ("x: {repo: a/b, revision: " + "a" * 40 + ", file: f, sha256: " + "b" * 64 + ", entries: 0}\n",
      "x.entries must be a positive integer"),
+    ("x: {repo: a/b, revision: " + "a" * 40 + ", file: f, sha256: " + "b" * 64 + ", private: yes-please}\n",
+     "x.private must be true or false"),
 ])
 def test_bad_lock_is_one_line_error(tmp_path, body, message):
     path = tmp_path / "lock.yaml"
diff --git a/inferencex-e2e/infx/tests/serving/test_greenspec.py b/inferencex-e2e/infx/tests/serving/test_greenspec.py
new file mode 100644
index 0000000000..662004fc19
--- /dev/null
+++ b/inferencex-e2e/infx/tests/serving/test_greenspec.py
@@ -0,0 +1,279 @@
+import json
+from pathlib import Path
+
+import pytest
+
+from infx.serving.greenspec import (
+    ALLOWLIST,
+    MAX_BYTES,
+    META_KEYS,
+    SERVER_LOG_STUB,
+    ConsoleFilter,
+    GreenSpecError,
+    Settings,
+    fix_required,
+    fix_tools,
+    guard,
+    label,
+    normalize,
+    payload,
+    select,
+    settings,
+    write_selection,
+)
+
+
+def _row(qid: str, category: str, bucket: str, lite: bool, **body) -> dict:
+    return {
+        "question_id": qid, "category": category, "bucket": bucket, "lite": lite, "isl": 100, "osl": 10,
+        "thinking_src": "none", "recorded": {"temperature": 0.6, "max_completion_tokens": 4096},
+        "messages": [{"role": "user", "content": f"q {qid}"}], **body,
+    }
+
+
+ROWS = [
+    _row("m/aaaa", "agentic_tools", "agentic_tools__0-4k", True, tools=[{"type": "function"}]),
+    _row("m/bbbb", "agentic_tools", "agentic_tools__4k-16k", False),
+    _row("m/cccc", "llm_task", "llm_task__16k+", True),
+]
+
+
+@pytest.fixture
+def dataset(tmp_path: Path) -> Path:
+    path = tmp_path / "greenspec.jsonl"
+    path.write_text("".join(json.dumps(r) + "\n" for r in ROWS))
+    return path
+
+
+@pytest.mark.parametrize("chosen, expected", [
+    (Settings(), ["m/aaaa", "m/bbbb", "m/cccc"]),
+    (Settings(lite=True), ["m/aaaa", "m/cccc"]),
+    (Settings(subset="agentic_tools"), ["m/aaaa", "m/bbbb"]),
+    (Settings(subset="agentic_tools", lite=True), ["m/aaaa"]),
+    (Settings(bucket="llm_task__16k+"), ["m/cccc"]),
+])
+def test_select(dataset, chosen, expected):
+    assert [r["question_id"] for r in select(dataset, chosen)] == expected
+
+
+@pytest.mark.parametrize("overrides, message", [
+    ({"subset": "coding"}, "overrides.subset"),
+    ({"lite": 1}, "overrides.lite"),
+    ({"bucket": "16k"}, "overrides.bucket"),
+    ({"mode": "greedy"}, "overrides.mode"),
+    ({"max_tokens": 0}, "overrides.max_tokens"),
+    ({"requests": True}, "overrides.requests"),
+    ({"duration": 180}, "greenspec takes"),
+])
+def test_bad_overrides(overrides, message):
+    with pytest.raises(GreenSpecError, match=message):
+        normalize(overrides)
+
+
+def test_label_names_every_non_default():
+    assert label(Settings()) == "greenspec-diverse"
+    chosen = normalize({"subset": "agentic_tools", "lite": True, "mode": "recorded", "max_tokens": 4096, "requests": 200})
+    assert label(chosen) == "greenspec-agentic_tools-lite-recorded-mt4096-n200"
+    assert label(normalize({"bucket": "llm_task__16k+"})) == "greenspec-diverse-llm_task_16kplus"
+
+
+def test_settings_come_from_the_serving_yaml_of_the_job():
+    yaml_text = "name: x\nscenario: greenspec\noverrides: {subset: llm_task, lite: true}\nconc: 4\n"
+    assert settings({"SERVING_YAML": yaml_text}) == Settings(subset="llm_task", lite=True)
+    assert settings({"SERVING_YAML": "name: x\nscenario: agentx\nconc: 4\n"}) is None
+    assert settings({}) is None
+
+
+def test_payload_target_default_keeps_the_row_and_caps_output():
+    body = payload(ROWS[0], "served", Settings())
+    assert list(body)[0] == "model" and body["model"] == "served"
+    assert not set(META_KEYS) & set(body)
+    assert body["max_tokens"] == 8192 and "temperature" not in body
+    assert body["stream"] is True and body["stream_options"] == {"include_usage": True}
+    assert body["tools"] == [{"type": "function"}]
+
+
+def test_payload_recorded_replays_sampling_and_the_cap_only_lowers():
+    low = payload(ROWS[0], "m", Settings(mode="recorded", max_tokens=2048))
+    assert low["max_completion_tokens"] == 2048 and "max_tokens" not in low
+    high = payload(ROWS[0], "m", Settings(mode="recorded"))
+    assert high["max_completion_tokens"] == 4096 and high["temperature"] == 0.6
+
+
+def test_fix_required_rewrites_the_keyword_at_every_schema_level():
+    schema = {
+        "type": "object",
+        "required": {},
+        "properties": {
+            "a": {"type": "object", "required": {}, "properties": {"b": {"required": {}}}},
+            "c": {"type": "array", "items": {"type": "object", "required": {}}},
+            "d": {"anyOf": [{"required": {}}, {"type": "null"}]},
+        },
+        "$defs": {"e": {"required": {}}},
+    }
+    fixed = fix_required(schema)
+    assert fixed["required"] == []
+    assert fixed["properties"]["a"]["required"] == []
+    assert fixed["properties"]["a"]["properties"]["b"]["required"] == []
+    assert fixed["properties"]["c"]["items"]["required"] == []
+    assert fixed["properties"]["d"]["anyOf"][0]["required"] == []
+    assert fixed["$defs"]["e"]["required"] == []
+    assert schema["required"] == {}  # the input is not mutated
+
+
+def test_fix_required_leaves_a_property_named_required_and_other_values_alone():
+    schema = {
+        "type": "object",
+        "required": ["x"],
+        "properties": {
+            "required": {},
+            "x": {"type": "string", "enum": [{}], "default": {"required": {}}, "const": {"required": {}}},
+        },
+    }
+    assert fix_required(schema) == schema
+
+
+def test_fix_tools_only_touches_function_parameters():
+    tools = [
+        {"type": "function", "function": {"name": "f", "parameters": {"type": "object", "required": {}}}},
+        {"type": "function", "function": {"name": "g"}},
+        {"type": "function"},
+    ]
+    fixed = fix_tools(tools)
+    assert fixed[0]["function"]["parameters"]["required"] == []
+    assert fixed[1:] == tools[1:]
+    assert fix_tools(None) is None
+
+
+def test_payload_sends_valid_tool_schemas():
+    row = {**ROWS[0], "tools": [{"type": "function", "function": {"name": "f", "parameters": {"required": {}}}}]}
+    assert payload(row, "m", Settings())["tools"][0]["function"]["parameters"] == {"required": []}
+
+
+def test_write_selection_one_body_per_line(dataset, tmp_path):
+    out = tmp_path / "sel" / "selection.jsonl"
+    assert write_selection(dataset, Settings(subset="agentic_tools", lite=True), "served", out) == 1
+    assert [json.loads(line) for line in out.read_text().splitlines()] == [payload(ROWS[0], "served", Settings())]
+
+
+def test_an_empty_selection_is_an_error(dataset, tmp_path):
+    with pytest.raises(GreenSpecError, match="no row of the release matches"):
+        write_selection(dataset, Settings(bucket="llm_task__0-1k"), "m", tmp_path / "out.jsonl")
+
+
+CONSOLE = (
+    "14:27:38.612 INFO     Starting AIPerf System (_single_run.py:64)\n"
+    "Profiling:   0% |          | [00:00<?]\r"
+    "Profiling:  40% |####      | [00:20<00:30]\r"
+    "Profiling:  90% |######### | [00:50<00:05]14:27:46.424 WARNING  OSL mismatch: got 18 tokens\n"
+    "14:27:46.500 ERROR    Error in SSE response: ValueError('secret prompt m/aaaa')\n"
+    "Traceback (most recent call last):\n"
+    "  ValueError: secret prompt m/aaaa\n"
+    "\x1b[32mProfiling: 100% |##########| [00:55<00:00]\x1b[0m\r"
+    "Processing Records: 40/40 |##########| 100% [00:01<00:00]\n"
+    "                     NVIDIA AIPerf | Error Summary                     \n"
+    "┏━━━━━━┳━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━┓\n"
+    "┃ Code ┃        Type ┃ Message                      ┃ Count ┃\n"
+    "┡━━━━━━╇━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━┩\n"
+    "│  400 │ Bad Request │ {\"input_value\": \"secret m/b │ 1,203 │\n"
+    "│      │             │ bbb\"}                        │       │\n"
+    "│  N/A │         N/A │ timeout                      │     2 │\n"
+    "└──────┴─────────────┴──────────────────────────────┴───────┘\n"
+    "            NVIDIA AIPerf | LLM Metrics            \n"
+    "│ Request Latency (ms) │ 12.00 │\n"
+)
+
+
+def _filtered(text: str, interval: float) -> list[str]:
+    out: list[str] = []
+    ticks = iter(range(0, 1000, 10))  # each bar line is 10 s after the previous one
+    console = ConsoleFilter(out.append, interval, clock=lambda: next(ticks))
+    for segment in text.replace("\r", "\n").split("\n"):
+        console.feed(segment)
+    console.close()
+    return out
+
+
+def test_console_withholds_error_text_and_throttles_bars():
+    out = _filtered(CONSOLE, interval=15)
+    assert "secret" not in "\n".join(out) and "aaaa" not in "\n".join(out)
+    assert out == [
+        "14:27:38.612 INFO     Starting AIPerf System (_single_run.py:64)",
+        "GreenSpec progress: Profiling (0%) elapsed 00:00 eta ?",
+        "GreenSpec progress: Profiling (90%) elapsed 00:50 eta 00:05",
+        "14:27:46.424 WARNING  OSL mismatch: got 18 tokens",
+        "14:27:46.500 ERROR    <withheld: may echo server error text>",
+        "GreenSpec progress: Profiling (100%) elapsed 00:55 eta 00:00",
+        "GreenSpec progress: Processing Records 40/40 (100%) elapsed 00:01 eta 00:00",
+        "GreenSpec: aiperf error summary withheld (messages can echo requests): 400 x1203, N/A x2",
+        "            NVIDIA AIPerf | LLM Metrics",
+        "│ Request Latency (ms) │ 12.00 │",
+    ]
+
+
+def test_console_keeps_an_unterminated_error_table_dropped():
+    out = _filtered(CONSOLE[: CONSOLE.index("│  N/A")], interval=0)
+    assert "secret" not in "\n".join(out)
+    assert out[-1] == "GreenSpec: aiperf error summary withheld (messages can echo requests): 400 x1203"
+
+
+def _result_dir(tmp_path: Path) -> Path:
+    result = tmp_path / "results"
+    (result / "aiperf_artifacts").mkdir(parents=True)
+    for name in ALLOWLIST:
+        (result / name).write_text("{}")
+    export = {"error_summary": [{"count": 1, "error_details": {"code": 400, "message": "x"}}]}
+    (result / "aiperf_artifacts/profile_export_aiperf.json").write_text(json.dumps(export))
+    (result / "aiperf_artifacts/inputs.json").write_text("prompts")
+    (result / "aiperf_artifacts/profile_export.jsonl").write_text("{}")
+    (result / ".dataset_path").write_text("/tmp/greenspec-selection/selection.jsonl")
+    (result / "server.log").write_text("engine output")
+    (result / "gpu_distribution.json").write_text("{}")
+    return result
+
+
+def test_guard_prunes_redacts_and_keeps_server_logs_on_the_node(dataset, tmp_path):
+    result, keep = _result_dir(tmp_path), tmp_path / "kept"
+    outside = tmp_path / "outside-server.log"
+    outside.write_text("engine output")
+    assert guard(dataset, result, [], [outside], keep_log_dir=keep) == []
+    assert sorted(p.relative_to(result).as_posix() for p in result.rglob("*") if p.is_file()) == sorted(ALLOWLIST)
+    redacted = json.loads((result / "aiperf_artifacts/profile_export_aiperf.json").read_text())
+    assert redacted["error_summary"][0]["error_details"] == {"code": 400, "message": "<withheld>"}
+    assert (keep / "server.log").read_text() == "engine output"
+    assert outside.read_text() == SERVER_LOG_STUB and (keep / "outside-server.log").read_text() == "engine output"
+
+
+def test_guard_stub_survives_a_server_still_writing(dataset, tmp_path):
+    log = tmp_path / "server.log"
+    log.write_text("engine output")
+    with log.open("r+") as server:  # like `> server.log`: no O_APPEND
+        server.seek(0, 2)
+        assert guard(dataset, tmp_path / "none", [], [log]) == []
+        server.write("late engine output")
+    assert log.read_text() == SERVER_LOG_STUB
+
+
+def test_guard_fails_on_a_row_id_or_a_big_file(dataset, tmp_path):
+    result = tmp_path / "results"
+    result.mkdir()
+    (result / "scenario.yaml").write_text("note: m/bbbb")
+    big = tmp_path / "aggregate.json"
+    big.write_text("x" * (MAX_BYTES + 1))
+    problems = guard(dataset, result, [big], [])
+    assert any("question_id" in p for p in problems)
+    assert any("bytes" in p for p in problems)
+
+
+def test_tee_filtered_writes_the_filtered_console_to_stdout_and_the_log(tmp_path, capsys):
+    import sys
+
+    from infx.serving.greenspec import tee_filtered
+
+    script = ("import sys; sys.stdout.write("
+              "'14:27:46.500 ERROR    Error in SSE response: secret prompt\\nTraceback x\\nplain line\\n'); sys.exit(3)")
+    log = tmp_path / "benchmark.log"
+    assert tee_filtered([sys.executable, "-c", script], log, {}) == 3
+    out = capsys.readouterr().out
+    assert "secret" not in out and "secret" not in log.read_text()
+    assert log.read_text().splitlines() == ["14:27:46.500 ERROR    <withheld: may echo server error text>"]
diff --git a/inferencex-e2e/infx/tests/serving/test_scenarios.py b/inferencex-e2e/infx/tests/serving/test_scenarios.py
index 0b631b2973..83836cac40 100644
--- a/inferencex-e2e/infx/tests/serving/test_scenarios.py
+++ b/inferencex-e2e/infx/tests/serving/test_scenarios.py
@@ -87,6 +87,27 @@ def test_bad_variants_fail_with_one_line(args, message):
         render(*args)
 
 
+def test_greenspec_renders_the_selection_and_has_no_duration():
+    rendered = render("greenspec", {"subset": "agentic_tools", "lite": True, "requests": 50})
+    dataset = rendered.doc["benchmark"]["datasets"][0]
+    assert (rendered.kind, rendered.dataset, rendered.label) == ("file", "greenspec-v1.1.1", "greenspec-agentic_tools-lite-n50")
+    assert dataset["format"] == "raw_payload" and dataset["path"] == "${IX_CUSTOM_DATASET_FILE}"
+    [phase] = rendered.doc["benchmark"]["phases"]
+    assert phase["requests"] == 50 and "duration" not in phase and rendered.duration == 0.0
+    assert "ignore_eos" not in rendered.doc["benchmark"]["endpoint"].get("extra", {})
+    assert render("greenspec").label == "greenspec-diverse"
+
+
+@pytest.mark.parametrize("args, message", [
+    (("greenspec", {"subset": "coding"}), "overrides.subset"),
+    (("greenspec", {"duration": 180}), "greenspec takes"),
+    (("greenspec", {}, Point("8k", "1k", 0)), "workload/slo: only for scenario synthetic"),
+])
+def test_bad_greenspec_variants_fail_with_one_line(args, message):
+    with pytest.raises(ScenarioError, match=message):
+        render(*args)
+
+
 def test_warmup_requests_per_lane_override():
     def profiling(doc):
         return next(p for p in doc["benchmark"]["phases"] if p.get("kind") == "profiling")
diff --git a/inferencex-e2e/infx/tests/serving/test_schema.py b/inferencex-e2e/infx/tests/serving/test_schema.py
index 7ba3ab8640..0686a04413 100644
--- a/inferencex-e2e/infx/tests/serving/test_schema.py
+++ b/inferencex-e2e/infx/tests/serving/test_schema.py
@@ -214,3 +214,44 @@ def test_engine_yaml_without_image_says_what_is_missing():
     assert "image: required (or use endpoint: for a remote endpoint)" in errors(
         text + "scenario: agentx\nconc: 8\n"
     )
+
+
+ARMS = """\
+arms:
+  base: {}
+  mtp:
+    append: |
+      --speculative-config '{"method": "mtp", "num_speculative_tokens": 4}'
+    env: {X: "1"}
+scenario: agentx
+conc: 4
+"""
+
+
+def test_arms_are_engine_variants_of_the_shared_serve():
+    from infx.serving.schema import arm_specs
+
+    spec = load_serving(yaml_of(ARMS))
+    arms = dict(arm_specs(spec))
+    assert list(arms) == ["base", "mtp"]
+    assert arms["base"].serve == spec.serve and "--speculative-config" not in arms["base"].serve
+    assert arms["mtp"].serve.startswith(spec.serve.rstrip()) and "--speculative-config" in arms["mtp"].serve
+    assert arms["mtp"].env == {"X": "1"} and arms["base"].env == {}
+    assert arm_specs(load_serving(yaml_of("scenario: agentx\nconc: 4\n"))) == [(None, load_serving(yaml_of("scenario: agentx\nconc: 4\n")))]
+
+
+@pytest.mark.parametrize("extra, message", [
+    ("arms: {base: {}, other: {}}\nscenario: agentx\nconc: 4\n", "arms.other: the same engine command and env as arms.base"),
+    ("arms: {Base: {}}\nscenario: agentx\nconc: 4\n", "arm names use lowercase"),
+    ("arms: {base: {extra: 1}}\nscenario: agentx\nconc: 4\n", "unknown key"),
+    ("arms: {}\nscenario: agentx\nconc: 4\n", "arms"),
+    ("arms: {a: {append: --numa-node 0}}\nscenario: agentx\nconc: 4\n", "arms.a.serve: drop --numa-node"),
+])
+def test_bad_arms_fail_with_one_line_each(extra, message):
+    assert message in errors(yaml_of(extra))
+
+
+def test_arms_do_not_combine_with_a_remote_endpoint():
+    text = ("name: r\nrunner: cluster:hcm-hgx01\nendpoint: {url: 'http://h:8000/v1', model: m}\n"
+            "arms: {a: {}}\nscenario: agentx\nconc: 4\n")
+    assert "arms: not used with endpoint:" in errors(text)
diff --git a/inferencex-e2e/infx/tests/serving/test_specdec.py b/inferencex-e2e/infx/tests/serving/test_specdec.py
new file mode 100644
index 0000000000..812d894ec9
--- /dev/null
+++ b/inferencex-e2e/infx/tests/serving/test_specdec.py
@@ -0,0 +1,58 @@
+import json
+
+import pytest
+
+from infx.serving import specdec
+
+TEXT = "\n".join([
+    "# HELP vllm:spec_decode_num_drafts_total drafts",
+    "# TYPE vllm:spec_decode_num_drafts_total counter",
+    'vllm:spec_decode_num_drafts_total{engine="0",model_name="m"} 1.5e+03',
+    'vllm:spec_decode_num_drafts_total{engine="1",model_name="m"} 500',
+    "vllm:spec_decode_num_accepted_tokens_total 4000",
+    "vllm:spec_decode_num_draft_tokens_total 12000",
+    "vllm:num_requests_running 3",
+    "vllm:bad nan",
+])
+
+
+def test_parse_sums_label_sets_and_skips_non_finite():
+    totals = specdec.parse_prometheus(TEXT)
+    assert totals["vllm:spec_decode_num_drafts_total"] == 2000
+    assert "vllm:bad" not in totals
+
+
+def test_acceptance_is_one_plus_accepted_per_draft_event():
+    before = {"vllm:spec_decode_num_accepted_tokens_total": 100.0, "vllm:spec_decode_num_drafts_total": 50.0,
+              "vllm:spec_decode_num_draft_tokens_total": 300.0}
+    after = {"vllm:spec_decode_num_accepted_tokens_total": 1100.0, "vllm:spec_decode_num_drafts_total": 550.0,
+             "vllm:spec_decode_num_draft_tokens_total": 3300.0}
+    result = specdec.acceptance(before, after)
+    assert result["acceptance_length"] == pytest.approx(3.0)
+    assert result["acceptance_rate"] == pytest.approx(1000 / 3000)
+    assert result["spec_draft_events"] == 500 and result["draft_depth"] == 6
+
+
+def test_no_series_means_no_speculative_decoding():
+    assert specdec.acceptance({}, {}) is None
+    assert specdec.acceptance({"vllm:num_requests_running": 1.0}, {}) is None
+
+
+def test_sglang_gauge_is_read_after_the_run():
+    result = specdec.acceptance({}, {"sglang:spec_accept_length": 2.4})
+    assert result["acceptance_length"] == 2.4
+
+
+def test_sglang_gauge_of_zero_is_no_speculative_decoding():
+    assert specdec.acceptance({"sglang:spec_accept_length": 0.0}, {"sglang:spec_accept_length": 0.0}) is None
+
+
+def test_summarize_and_merge(tmp_path):
+    before, after, agg = tmp_path / "b.json", tmp_path / "a.json", tmp_path / "agg.json"
+    assert specdec.summarize(before, after) is None
+    specdec.save(before, {"vllm:spec_decode_num_drafts_total": 0.0})
+    specdec.save(after, {"vllm:spec_decode_num_drafts_total": 10.0, "vllm:spec_decode_num_accepted_tokens_total": 20.0})
+    agg.write_text(json.dumps({"scenario": "x"}))
+    specdec.merge(agg, specdec.summarize(before, after))
+    assert json.loads(agg.read_text())["spec_decode"]["acceptance_length"] == pytest.approx(3.0)
+    specdec.merge(tmp_path / "missing.json", {})  # no aggregate, nothing to do
diff --git a/inferencex-e2e/infx/tests/serving/test_summary.py b/inferencex-e2e/infx/tests/serving/test_summary.py
new file mode 100644
index 0000000000..5293ebb19c
--- /dev/null
+++ b/inferencex-e2e/infx/tests/serving/test_summary.py
@@ -0,0 +1,34 @@
+import json
+
+from infx.serving.summary import load, main, table
+
+
+def _result(spec: str, tput: float, **extra) -> dict:
+    return {
+        "scenario": "greenspec-diverse", "spec_decoding": spec, "conc": 16, "num_requests_successful": 99,
+        "num_requests_total": 100,
+        "request_metrics": {"throughput": {"output": {"tokens_per_second": tput}},
+                            "latency": {"ttft": {"p50": 0.2, "p95": 1.5}, "tpot": {"mean": 0.02}}},
+        **extra,
+    }
+
+
+def test_speculative_row_gets_acceptance_and_speedup_over_the_base(tmp_path):
+    (tmp_path / "bmk_a").mkdir()
+    (tmp_path / "bmk_a/base.json").write_text(json.dumps(_result("none", 100.0)))
+    (tmp_path / "bmk_a/mtp.json").write_text(json.dumps(
+        _result("mtp", 180.0, spec_decode={"acceptance_length": 3.25, "acceptance_rate": 0.375, "draft_depth": 6})))
+    (tmp_path / "bmk_a/other.json").write_text(json.dumps({"hello": 1}))  # not an aggregate
+    lines = table(load(tmp_path)).splitlines()
+    assert len(lines) == 4  # header, rule, two points
+    assert "| greenspec-diverse | mtpx6 | 16 | 99/100 | 180.0 | 200 / 1,500 | 20.0 | 3.25 | 37.5% | 1.80x |" in lines
+    assert "| greenspec-diverse | none | 16 | 99/100 | 100.0 | 200 / 1,500 | 20.0 | - | - | - |" in lines
+
+
+def test_a_missing_number_is_a_dash_and_an_empty_dir_says_so(tmp_path, capsys):
+    (tmp_path / "x.json").write_text(json.dumps({"scenario": "s", "request_metrics": {}}))
+    assert table(load(tmp_path)).count(" - ") >= 6
+    empty = tmp_path / "empty"
+    empty.mkdir()
+    assert main([str(empty)]) == 0
+    assert "No aggregate result" in capsys.readouterr().out
diff --git a/inferencex-e2e/infx/tests/serving/test_to_matrix.py b/inferencex-e2e/infx/tests/serving/test_to_matrix.py
index 39be43008b..94bb82a9a8 100644
--- a/inferencex-e2e/infx/tests/serving/test_to_matrix.py
+++ b/inferencex-e2e/infx/tests/serving/test_to_matrix.py
@@ -1,9 +1,10 @@
 import json
 
+import pytest
 import yaml
 
 from infx.serving import scenarios
-from infx.serving.to_matrix import build, main
+from infx.serving.to_matrix import build, main, spec_decoding
 
 ENGINE = """\
 name: qwen-tp1
@@ -86,3 +87,27 @@ def test_engine_point_carries_no_remote_fields():
     point = build(ENGINE + "scenario: tau2\nconc: 4\n")["points"][0]
     serving = json.loads(point["serving"])
     assert (serving["remote_url"], serving["tokenizer"], point["api_key_secret"]) == ("", "", "")
+
+
+@pytest.mark.parametrize("serve, expected", [
+    ("", "none"),
+    ("vllm serve m --tensor-parallel-size 1", "none"),
+    ("vllm serve m --speculative-config '{\"method\":\"mtp\",\"num_speculative_tokens\":6}'", "mtp"),
+    ("vllm serve m --speculative-config '{\"method\":\"eagle3\",\"model\":\"d\"}'", "draft_model"),
+    ("python3 -m sglang.launch_server --speculative-algorithm NEXTN", "mtp"),
+    ("python3 -m sglang.launch_server --speculative-algorithm EAGLE3", "draft_model"),
+])
+def test_spec_decoding_follows_the_serve_command(serve, expected):
+    assert spec_decoding(serve) == expected
+
+
+def test_arms_are_one_job_each_with_their_own_engine_in_one_run():
+    text = (ENGINE + "arms:\n  base: {}\n  mtp:\n    append: |\n      --speculative-config '{\"method\": \"mtp\", \"num_speculative_tokens\": 4}'\n"
+            "scenario: agentx\nconc: [8]\n")
+    result = build(text)
+    names = [p["config"]["exp-name"] for p in result["points"]]
+    assert names == ["qwen-tp1-base_c8", "qwen-tp1-mtp_c8"]
+    assert [p["config"]["spec-decoding"] for p in result["points"]] == ["none", "mtp"]
+    commands = [json.loads(p["serving"])["serve_cmd"] for p in result["points"]]
+    assert "--speculative-config" not in commands[0] and "--speculative-config" in commands[1]
+    assert "2 arms (base,mtp)" in result["title"]