feat(llm): add shadow tap for mirrored request and response streams

lib/bench/AGENTS.md

@@ -11,6 +11,8 @@ Benchmarks and trace-export entrypoints for Dynamo. Hosts:
   request traces for direct replay.
 - `request_trace_to_mooncake` — opt-in export of Dynamo request traces to
   Mooncake replay JSONL.
+- `shadow_consumer` — reads one shadow tap topic and keeps running token and
+  outcome totals; the stand-in shadow for `shadow/demo.sh`.
 - `request_trace_to_satf` — opt-in export of Dynamo request traces to SATF 2.0.
 
 ## Guardrails

lib/bench/Cargo.toml

@@ -73,6 +73,7 @@ router-test-support = [
   "dep:tracing-subscriber",
 ]
 satf = ["dep:dynamo-data-gen"]
+shadow-consumer = ["dep:dynamo-llm", "dep:dynamo-runtime"]
 request-trace-to-mooncake = ["dep:blake3", "dep:dynamo-data-gen", "dep:tempfile"]
 weka-to-agentic-mooncake = [
   "dep:dynamo-data-gen",
@@ -101,6 +102,11 @@ name = "claude_trace_export"
 path = "claude_trace_export.rs"
 required-features = ["claude-trace-export"]
 
+[[bin]]
+name = "shadow_consumer"
+path = "shadow_consumer.rs"
+required-features = ["shadow-consumer"]
+
 [[bin]]
 name = "request_trace_to_satf"
 path = "request_trace_to_satf.rs"
@@ -195,6 +201,8 @@ dynamo-tokens = { workspace = true, optional = true }
 dynamo-data-gen = { workspace = true, optional = true }
 dynamo-mocker = { workspace = true, optional = true }
 dynamo-kv-router = { workspace = true }
+dynamo-llm = { path = "../llm", default-features = false, optional = true }
+dynamo-runtime = { workspace = true, optional = true }
 
 [target.'cfg(target_os = "linux")'.dependencies]
 libc = { workspace = true, optional = true }

lib/bench/README.md

@@ -23,6 +23,7 @@ Build each entrypoint with `--no-default-features` and its matching feature:
 | `claude_trace_export` | `claude-trace-export` |
 | `request_trace_to_mooncake` | `request-trace-to-mooncake` |
 | `request_trace_to_satf` | `satf` |
+| `shadow_consumer` | `shadow-consumer` |
 | `multiturn_bench` | `multiturn` |
 | `offline_replay_bench` | `offline-replay` |
 | `mooncake_bench` | `mooncake` |
@@ -159,6 +160,33 @@ cargo bench --package dynamo-bench --bench offline_replay_bench \
   --kv-bytes-per-token 131072
 ```
 
+## Shadow tap consumer
+
+A shadow tap mirrors the live request stream of a frontend, and optionally the
+responses, to event-plane topics. A shadow deployment reads those topics to
+serve production traffic on separate workers, where a wrong or slow result
+costs users nothing. The tap is configured by the file that
+`DYN_SHADOW_TAP_CONFIG` names; see [`shadow/taps.yaml`](shadow/taps.yaml) for
+the format and `lib/llm/src/shadow/` for the tap itself.
+
+`shadow_consumer` stands in for a shadow deployment. It reads one topic and
+keeps running totals: records, prompt tokens, and for a `request_response` tap
+output tokens and outcomes. It also counts the records the tap dropped
+(`tap_gaps`) and the records the event plane lost (`transport_gaps`).
+
+```bash
+cargo run -p dynamo-bench --no-default-features --features shadow-consumer \
+  --bin shadow_consumer -- --topic kv-history-tracker
+```
+
+It reads the same `DYN_DISCOVERY_BACKEND`, `DYN_FILE_KV`, and
+`DYN_EVENT_PLANE` settings as the frontend, and must run with the same values.
+
+[`shadow/demo.sh`](shadow/demo.sh) runs the whole loop on one machine with no
+etcd and no NATS: a frontend with both example taps, two mocker workers, AIPerf
+load, and one consumer per tap. It then prints the frontend token metrics next
+to the consumer totals so the two can be compared.
+
 ## KV router / sharded indexer benchmarks
 
 See [kv_router/INDEXER_BENCH.md](kv_router/INDEXER_BENCH.md) for trace

lib/bench/shadow/demo.sh

@@ -0,0 +1,80 @@
+#!/usr/bin/env bash
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+# Shadow tap demo: a frontend with two taps, mocker workers, AIPerf load, and
+# one consumer per tap. It needs no etcd and no NATS: discovery is a directory
+# and the event plane is ZMQ.
+#
+# Run from the repository root inside a venv that has `ai-dynamo` and `aiperf`.
+#
+#   lib/bench/shadow/demo.sh [request-count] [concurrency] [isl] [osl]
+#
+# STREAMING= sends non-streaming requests. The taps publish one record per
+# request in both cases.
+#
+# AIPERF_EXTRA adds AIPerf flags, for example a cancellation run:
+#   AIPERF_EXTRA='--request-cancellation-rate 25 --request-cancellation-delay 0.05'
+set -euo pipefail
+
+REQUESTS=${1:-200}
+CONCURRENCY=${2:-16}
+ISL=${3:-512}
+OSL=${4:-32}
+MODEL=${MODEL:-Qwen/Qwen3-0.6B}
+PORT=${PORT:-8000}
+OUT=${OUT:-$(mktemp -d -t shadow-demo.XXXXXX)}
+mkdir -p "$OUT"
+HERE=$(cd "$(dirname "$0")" && pwd)
+
+export DYN_DISCOVERY_BACKEND=file
+export DYN_FILE_KV="$OUT/discovery"
+export DYN_REQUEST_PLANE=tcp
+export DYN_EVENT_PLANE=zmq
+
+cargo build -p dynamo-bench --no-default-features --features shadow-consumer --bin shadow_consumer
+CONSUMER=$(cargo metadata --format-version 1 --no-deps | python3 -c 'import json,sys; print(json.load(sys.stdin)["target_directory"])')/debug/shadow_consumer
+
+pids=()
+cleanup() {
+  kill -INT "${pids[@]}" 2>/dev/null || true
+  wait 2>/dev/null || true
+}
+trap cleanup EXIT
+
+python3 -m dynamo.mocker --model-path "$MODEL" --num-workers 2 --speedup-ratio 10 \
+  --discovery-backend file --request-plane tcp >"$OUT/mocker.log" 2>&1 &
+pids+=($!)
+
+DYN_SHADOW_TAP_CONFIG="$HERE/taps.yaml" python3 -m dynamo.frontend --http-port "$PORT" \
+  --discovery-backend file --request-plane tcp >"$OUT/frontend.log" 2>&1 &
+pids+=($!)
+
+for topic in kv-history-tracker shadow-replay; do
+  "$CONSUMER" --topic "$topic" --summary-path "$OUT/$topic.json" >"$OUT/$topic.log" 2>&1 &
+  pids+=($!)
+done
+
+echo "waiting for the model to register (logs in $OUT)"
+for _ in $(seq 1 120); do
+  if curl -sf "localhost:$PORT/v1/models" | grep -q "$MODEL"; then break; fi
+  sleep 1
+done
+curl -sf "localhost:$PORT/v1/models" | grep -q "$MODEL" || { echo "model never registered"; exit 1; }
+# Subscribers find the tap publishers through discovery; give them a moment.
+sleep 3
+
+aiperf profile --model "$MODEL" --url "localhost:$PORT" --endpoint-type chat ${STREAMING---streaming} \
+  --request-count "$REQUESTS" --concurrency "$CONCURRENCY" \
+  --synthetic-input-tokens-mean "$ISL" --synthetic-input-tokens-stddev 0 \
+  --output-tokens-mean "$OSL" --extra-inputs ignore_eos:true \
+  --artifact-dir "$OUT/aiperf" ${AIPERF_EXTRA:-} >"$OUT/aiperf.log" 2>&1 || echo "aiperf exited non-zero; see $OUT/aiperf.log"
+
+sleep 2
+echo "== frontend metrics"
+curl -s "localhost:$PORT/metrics" | grep -E '^dynamo_frontend_(input|output)_sequence_tokens_(sum|count)|^dynamo_frontend_requests_total' || true
+for topic in kv-history-tracker shadow-replay; do
+  echo "== $topic"
+  cat "$OUT/$topic.json"
+  echo
+done

lib/bench/shadow/taps.yaml

@@ -0,0 +1,27 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+# Example shadow tap config. Point DYN_SHADOW_TAP_CONFIG at this file.
+schema_version: 1
+# Namespace the taps publish in. Consumers subscribe in the same one.
+namespace: dynamo
+
+taps:
+  # Request only, published as soon as the request passes the tap.
+  - name: kv-history-tracker
+    topic: kv-history-tracker
+    capture: request
+    filters: [tokens-only]
+    capacity: 4096
+
+  # Request, the response the primary produced, and the arrival time, in one
+  # record published when the response stream ends.
+  - name: shadow-replay
+    topic: shadow-replay
+    capture: request_response
+    emit: joined
+    filters: [strip-multimodal, strip-embeds]
+    response:
+      tokens: true
+      chunk_timing: false
+    capacity: 1024