feat(serving): open-loop LiveKit V1.1-warm scenario (calls arrive by the clock)

.gitignore

@@ -1,3 +1,4 @@
 **/__pycache__/**
 **/.coverage
 experimental/multiturn/vllm_benchmark/results/
+.openloop/

benchmarks/benchmark_lib.sh

@@ -3583,7 +3583,9 @@ build_replay_cmd_yaml() {
     REPLAY_CMD+=" --tokenizer $MODEL"
     if [ "$SERVING_SEARCH" != "1" ]; then
         check_env_vars CONC
-        REPLAY_CMD+=" --concurrency $CONC"
+        # open-loop scenarios: CONC is the target number of concurrent calls (it sets the arrival rate in
+        # openloop_trace); --concurrency only caps open sessions, so it is set well above it there
+        REPLAY_CMD+=" --concurrency ${IX_CLIENT_CONCURRENCY:-$CONC}"
     fi
     # else: adaptive_search sweeps the concurrency itself (scenario YAML's sweep:
     # block; SLOs come from sweep.sla_filters, optionally overridden above).
@@ -3601,6 +3603,25 @@ build_replay_cmd_yaml() {
     REPLAY_CMD+=" --output-artifact-dir $result_dir/aiperf_artifacts"
 }
 
+# Open-loop scenarios (scenarios/*-openloop.yaml): rewrite the fetched LiveKit trace so calls arrive by the clock
+# at CONC / call duration per second (utils/serving/livekit_openloop.py), point IX_CUSTOM_DATASET_FILE at it and
+# lift aiperf's session cap above CONC. Settings: IX_OPENLOOP_{CALL_DURATION_S,RAMP_S,TAIL_S,SEED} (serving-YAML
+# env), defaults in the script. The trace goes to node-local disk; openloop.json (arrivals, scoring window) to
+# the run dir for utils/serving/livekit_ttft.py.
+openloop_trace() {
+    check_env_vars INFMAX_CONTAINER_WORKSPACE AIPERF_PYTHON IX_CUSTOM_DATASET_FILE CONC SERVING_SCENARIO
+    local result_dir="$1" dir="${IX_OPENLOOP_DIR:-$INFMAX_CONTAINER_WORKSPACE/.openloop}" derived
+    mkdir -p "$dir"
+    derived="$dir/${SERVING_SCENARIO}-c${CONC}.jsonl"
+    "$AIPERF_PYTHON" "$INFMAX_CONTAINER_WORKSPACE/utils/serving/livekit_openloop.py" \
+        "$IX_CUSTOM_DATASET_FILE" "$derived" "$result_dir/openloop.json" --ccu "$CONC" \
+        --scenario "$INFMAX_CONTAINER_WORKSPACE/scenarios/${SERVING_SCENARIO}.yaml" \
+        ${IX_OPENLOOP_CALL_DURATION_S:+--call-duration-s "$IX_OPENLOOP_CALL_DURATION_S"} \
+        ${IX_OPENLOOP_RAMP_S:+--ramp-s "$IX_OPENLOOP_RAMP_S"} ${IX_OPENLOOP_TAIL_S:+--tail-s "$IX_OPENLOOP_TAIL_S"} \
+        ${IX_OPENLOOP_SEED:+--seed "$IX_OPENLOOP_SEED"} || return 1
+    export IX_CUSTOM_DATASET_FILE="$derived" IX_CLIENT_CONCURRENCY=$(( CONC * 4 ))
+}
+
 # Exit 0 when <name> is a lock-pinned file dataset (scenarios/datasets/lock.yaml),
 # 1 when it is not, 2 when the lock cannot be read.
 is_custom_dataset() {

benchmarks/single_node/agentic/serve_generic.sh

@@ -66,6 +66,10 @@ else
         *) echo "ERROR: cannot tell whether $SERVING_LOADER is a lock-pinned dataset (exit $custom_rc)" >&2; exit 1 ;;
     esac
 fi
+# open-loop scenarios replay the fetched trace with calls arriving by the clock (benchmark_lib.sh openloop_trace)
+if [[ "$SERVING_SCENARIO" == *-openloop ]]; then
+    openloop_trace "$RESULT_DIR"
+fi
 # the user's env may set HF_HUB_OFFLINE=1 for the engine; the pre-download must reach the Hub
 HF_HUB_OFFLINE= TRANSFORMERS_OFFLINE= HF_DATASETS_OFFLINE= "$AIPERF_HF_CLI" download "$MODEL"
 

scenarios/custom-agentic-livekit111868-v1.1-warm-openloop.yaml

@@ -0,0 +1,44 @@
+# custom-agentic family, open loop: the V1.1-warm LiveKit 111868 trace (V1 append-ideal restricted to the 671 rooms
+# with >= 4 real requests, a dial warm-up row before each room's first turn; HF README) with calls arriving by the clock instead of when a lane frees up, so every server receives the same calls per hour
+# at the same concurrency. conc = target mean concurrent calls. Before aiperf runs, serve_generic.sh rewrites the
+# trace with utils/serving/livekit_openloop.py: calls open as a seeded Poisson process at conc / 127 s per second
+# (127 s = the trace's mean call length at production reply speed), 120 s linear ramp, arrivals stop 60 s before
+# the end so the last scored calls can finish. Inside a call each request keeps its logged gap after the previous
+# reply ends. Override with serving-YAML env: IX_OPENLOOP_CALL_DURATION_S, IX_OPENLOOP_RAMP_S, IX_OPENLOOP_TAIL_S,
+# IX_OPENLOOP_SEED. Score with utils/serving/livekit_ttft.py: it keeps the calls opened in [ramp, duration - tail)
+# (openloop.json in the run dir) and drops the dial warm-ups.
+# Client policy: a request with no answer after 10 s is abandoned (the agent's side-request timeout; agent turns
+# time out at 5 s, counted by the scorer).
+schemaVersion: "2.0"
+benchmark:
+  unsafeOverride: true
+  endpoint:
+    type: chat
+    path: /v1/chat/completions
+    streaming: true
+    useServerTokenCount: true
+    timeout: 10.0
+  datasets:
+  - name: livekit-111868-v1.1-warm
+    type: file
+    format: mooncake_trace
+    path: ${IX_CUSTOM_DATASET_FILE}
+    # fixed_schedule sends each call at its timestamp, so file order does not matter; calls are drawn by the
+    # rewrite's seed
+    sampling: sequential
+  phases:
+  - name: profiling
+    kind: profiling
+    type: fixed_schedule
+    autoOffset: true
+    # 120 s ramp + 1800 s scored + 60 s tail. Hard stop: requests of calls still running then are not sent.
+    duration: 1980.0
+    gracePeriod: 30.0
+    failedRequestThreshold: 0.1
+  artifacts:
+    sliceDuration: 1.0
+  tokenizer:
+    trustRemoteCode: true
+  runtime:
+    statsInterval: 30.0
+randomSeed: 42

scenarios/datasets/lock.yaml

@@ -29,3 +29,15 @@ livekit-111868-v2:
   revision: 7e36287967360851a1b05ba05730ff1d256b60c7
   file: trace.v2-observed.jsonl
   sha256: 190af677563ea782d060456bbc44f92e802e1d63e4322b147ef0290a49efd4f3
+# V1.1-warm and its per-request plan (from bench/gemma4-livekit-dyn-swa-arms); the plan is read by
+# utils/serving/livekit_ttft.py as <dataset>-plan.
+livekit-111868-v1.1-warm:
+  repo: Green-Lab/2026-10-16-111868-agentic-traces
+  revision: 6e0934c957565a42d52cafac096ec758a4302355
+  file: trace.v1.1-append-ideal-warm.jsonl
+  sha256: e26542392a202764849c05018f2f8ea3f9574277ffa3400615f0b4df23b0eb2f
+livekit-111868-v1.1-warm-plan:
+  repo: Green-Lab/2026-10-16-111868-agentic-traces
+  revision: 6e0934c957565a42d52cafac096ec758a4302355
+  file: plan.v1.1-append-ideal-warm.csv
+  sha256: fc5f69bc810a483905adc7d0bcd40258cf78bae869465ee9f19d3261217dc391

serving/gemma4-livekit-1gpu-v11-openloop.yaml

@@ -0,0 +1,30 @@
+# gemma4-livekit-1gpu-v1 (Gemma-4-31B FP8-block, prod vLLM flags minus MTP, 1 GPU) on the open-loop V1.1-warm
+# scenario: calls arrive by the clock, so a faster server gets the same calls per hour instead of more (scenario
+# header). Score each conc with: python utils/serving/livekit_ttft.py <run artifacts> (window from openloop.json).
+name: gemma4-livekit-1gpu-v11-openloop
+runner: hardware-hcm-8x_02
+image: vllm/vllm-openai:v0.25.1
+model: RedHatAI/gemma-4-31B-it-FP8-block
+gpus: 1
+precision: fp8
+env:
+  VLLM_DISABLE_COMPILE_CACHE: "1"
+  NCCL_P2P_LEVEL: NVL
+serve: |
+  vllm serve RedHatAI/gemma-4-31B-it-FP8-block
+  --served-model-name RedHatAI/gemma-4-31B-it-FP8-block
+  --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
+  --kv-cache-metrics
+  --enable-prompt-tokens-details
+  --enable-auto-tool-choice
+  --tool-call-parser gemma4
+  --reasoning-parser gemma4
+scenario: custom-agentic-livekit111868-v1.1-warm-openloop
+# target mean concurrent calls (calls arrive by the clock at conc / 127 s per second)
+conc: [8, 16, 32]

utils/serving/livekit_openloop.py

@@ -0,0 +1,150 @@
+"""Open-loop LiveKit trace: calls arrive by the clock, not when a lane frees up.
+
+    python utils/serving/livekit_openloop.py <trace.jsonl> <out.jsonl> <openloop.json> --ccu N
+        [--call-duration-s 127] [--duration-s T | --scenario scenarios/<name>.yaml] [--ramp-s 120] [--tail-s 60]
+        [--seed 42]
+
+The lane replay (custom-agentic-livekit111868-*) starts a lane's next call the moment its previous one ends, and
+each turn the logged gap after the previous reply ends, so a faster server shortens every call and receives more
+calls per hour at the same concurrency. Here the load is a parameter: calls open as a Poisson process at rate
+CCU / call_duration_s per second (call_duration_s = the trace's mean call length at production reply speed), ramping
+linearly from zero over the first ramp_s seconds, then steady until duration_s - tail_s. Every server sees the same
+arrivals for the same seed.
+
+Each arrival takes the next call of a seeded shuffle of the trace. Its first request gets the arrival `timestamp`
+(aiperf fixed_schedule); later requests keep their logged `delay` (sent that long after the previous reply ends).
+When more arrivals than calls are needed the shuffle cycles: the k-th play of a call (k >= 2) gets session id
+`<room>~k` and a `play<k> ` marker at the start of its first user message, so the call's own context is cold again
+while the shared system prompt still hits, as for a new caller. livekit_ttft.py strips the suffix to find the plan.
+
+openloop.json records the arrivals and the scoring window: calls that open inside [ramp_s, duration_s - tail_s)
+are scored; earlier ones are the ramp, later ones only keep the server loaded while the scored calls finish.
+Standard library only (PyYAML for --scenario); two streaming passes over the trace (1.4 GB for V1.1-warm).
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import random
+import re
+import sys
+
+SID_RE = re.compile(r'"session_id":\s*"([^"]*)"')
+
+
+def arrivals(rng: random.Random, rate: float, ramp_s: float, horizon_s: float) -> list[float]:
+    """Poisson arrival times (s) in [0, horizon_s) at rate * min(1, t / ramp_s), by thinning."""
+    out: list[float] = []
+    if rate <= 0:
+        return out
+    t = 0.0
+    while True:
+        t += rng.expovariate(rate)
+        if t >= horizon_s:
+            return out
+        ramp = min(1.0, t / ramp_s) if ramp_s > 0 else 1.0
+        if rng.random() < ramp:
+            out.append(t)
+
+
+def scan_rooms(path: str) -> list[str]:
+    rooms: list[str] = []
+    seen: set[str] = set()
+    with open(path) as f:
+        for line in f:
+            m = SID_RE.search(line)
+            if m and m.group(1) not in seen:
+                seen.add(m.group(1))
+                rooms.append(m.group(1))
+    return rooms
+
+
+def mark(row: dict, play: int) -> dict:
+    """Play marker at the start of the first user message: the call's own text goes cold, the system prompt does not."""
+    msgs = [dict(m) for m in row.get("messages") or []]
+    for m in msgs:
+        if m.get("role") == "user" and isinstance(m.get("content"), str):
+            m["content"] = f"play{play} " + m["content"]
+            break
+    return {**row, "messages": msgs}
+
+
+def scenario_duration(path: str) -> float:
+    import yaml
+
+    phases = yaml.safe_load(open(path))["benchmark"]["phases"]
+    return float(next(p for p in phases if p.get("kind", p.get("name")) == "profiling")["duration"])
+
+
+def main(argv: list[str]) -> int:
+    ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
+    ap.add_argument("src")
+    ap.add_argument("dst")
+    ap.add_argument("sidecar")
+    ap.add_argument("--ccu", type=float, required=True, help="target mean concurrent calls")
+    ap.add_argument("--call-duration-s", type=float, default=127.0, help="mean call length the rate is set from")
+    group = ap.add_mutually_exclusive_group(required=True)
+    group.add_argument("--duration-s", type=float, help="whole profiling phase: ramp + scored + tail")
+    group.add_argument("--scenario", help="read the profiling phase duration from this scenario YAML")
+    ap.add_argument("--ramp-s", type=float, default=120.0)
+    ap.add_argument("--tail-s", type=float, default=60.0)
+    ap.add_argument("--seed", type=int, default=42)
+    a = ap.parse_args(argv)
+
+    duration = a.duration_s if a.duration_s is not None else scenario_duration(a.scenario)
+    horizon = duration - a.tail_s
+    if horizon <= a.ramp_s:
+        sys.exit(f"duration {duration:g} s leaves no scored window after ramp {a.ramp_s:g} s and tail {a.tail_s:g} s")
+    rate = a.ccu / a.call_duration_s
+    rng = random.Random(a.seed)
+    rooms = scan_rooms(a.src)
+    if not rooms:
+        sys.exit("no rooms in trace")
+    order = rooms[:]
+    rng.shuffle(order)
+    times = arrivals(rng, rate, a.ramp_s, horizon)
+    # arrival i -> (room, play): the shuffle cycles once every room has been opened
+    plays: dict[str, list[tuple[int, int]]] = {}
+    for i, t in enumerate(times):
+        plays.setdefault(order[i % len(order)], []).append((i // len(order) + 1, int(round(t * 1000))))
+
+    first_seen: set[str] = set()
+    written = 0
+    with open(a.src) as fin, open(a.dst, "w") as fout:
+        for line in fin:
+            m = SID_RE.search(line) if line.strip() else None
+            sid = m.group(1) if m else None
+            if sid is None or sid not in plays:
+                continue
+            row = json.loads(line)
+            opening = sid not in first_seen
+            first_seen.add(sid)
+            for play, t_ms in plays[sid]:
+                out = dict(row) if play == 1 else mark(row, play)
+                if play > 1:
+                    out["session_id"] = f"{sid}~{play}"
+                if opening:
+                    out.pop("delay", None)
+                    out["timestamp"] = t_ms
+                fout.write(json.dumps(out) + "\n")
+                written += 1
+
+    t0 = times[0] if times else 0.0
+    sidecar = {
+        "ccu": a.ccu, "call_duration_s": a.call_duration_s, "calls_per_s": rate, "ramp_s": a.ramp_s,
+        "duration_s": duration, "tail_s": a.tail_s, "seed": a.seed, "arrivals": len(times),
+        "rooms_in_trace": len(rooms), "max_play": max((p for v in plays.values() for p, _ in v), default=0),
+        "requests": written, "first_arrival_s": t0,
+        # aiperf's autoOffset and the scorer both take the first arrival as t = 0
+        "score_window_s": [a.ramp_s - t0, horizon - t0],
+    }
+    with open(a.sidecar, "w") as f:
+        json.dump(sidecar, f, indent=2)
+    print(f"open loop: {len(times)} calls at {rate * 60:.1f}/min (ccu {a.ccu:g} / {a.call_duration_s:g} s), "
+          f"max play {sidecar['max_play']}, {written} requests -> {a.dst}")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main(sys.argv[1:]))