coding_agent_mixed_benchmark.py

gistfile1.txt

#!/usr/bin/env python3
"""
Mixed coding-agent benchmark for long-context LLM serving.

This benchmark is designed to reproduce two production effects that a fixed
batch-concurrency test hides:
  1. decode requests streaming while new 100k-token prefill requests arrive
  2. long-running, multi-repository contexts that can warm, evict, and reload KV

It uses only Python standard library modules.
"""

from __future__ import annotations

import argparse
import json
import math
import random
import re
import statistics
import subprocess
import sys
import threading
import time
import urllib.error
import urllib.parse
import urllib.request
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from typing import Any


ROOT_DIR = Path(__file__).resolve().parent.parent
REPORT_DIR = ROOT_DIR / "stress-test" / "reports"
API_URL = "http://localhost:30001/v1/chat/completions"
MODELS_URL = "http://localhost:30001/v1/models"
METRICS_URL = "http://localhost:30001/metrics"
PROM_URL = "http://localhost:9090"
MODEL = "glm-5.1"

PROFILES = {
    "mtp": {
        "label": "MTP / no HiCache",
        "compose": ROOT_DIR / "compose.yaml",
        "note": "EAGLE speculative decoding enabled, hierarchical cache disabled",
    },
    "hicache": {
        "label": "HiCache / no MTP",
        "compose": ROOT_DIR / "compose-hicache.yaml",
        "note": "hierarchical cache enabled, EAGLE/MTP disabled",
    },
}

SYSTEM_PROMPT = """You are a coding agent embedded in an IDE. You repeatedly read
large repository context, plan patches, write code, and review edge cases. Prefer
concrete code and test snippets over generic advice."""

CODE_MODULE = '''```python
# file: packages/{repo}/src/{name}_{idx}.py
from __future__ import annotations

import asyncio
import hashlib
import json
import logging
import time
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any, Optional

logger = logging.getLogger(__name__)


@dataclass(slots=True)
class {cls}Config:
    tenant_id: str = "{repo}"
    timeout_sec: float = 30.0
    cache_ttl_sec: int = 300
    max_retries: int = 3
    enable_metrics: bool = True
    enable_audit: bool = True
    shard_id: int = 0


@dataclass(slots=True)
class {cls}Record:
    id: str
    payload: dict[str, Any]
    created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
    updated_at: Optional[datetime] = None
    version: int = 1
    deleted: bool = False

    def fingerprint(self) -> str:
        raw = json.dumps(self.payload, sort_keys=True, default=str)
        return hashlib.sha256(raw.encode("utf-8")).hexdigest()[:16]


class {cls}Repository:
    def __init__(self, pool, table: str = "{table}"):
        self._pool = pool
        self._table = table

    async def get(self, record_id: str, tenant_id: str) -> Optional[{cls}Record]:
        query = (
            f"SELECT id, payload, created_at, updated_at, version, deleted "
            f"FROM {{self._table}} WHERE id = $1 AND tenant_id = $2"
        )
        row = await self._pool.fetchrow(query, record_id, tenant_id)
        if row is None:
            return None
        return {cls}Record(
            id=row["id"],
            payload=dict(row["payload"]),
            created_at=row["created_at"],
            updated_at=row["updated_at"],
            version=row["version"],
            deleted=row["deleted"],
        )

    async def upsert(self, record: {cls}Record, tenant_id: str) -> {cls}Record:
        query = f"""
            INSERT INTO {{self._table}}
                (id, tenant_id, payload, created_at, updated_at, version, deleted)
            VALUES ($1, $2, $3, $4, $5, $6, $7)
            ON CONFLICT (id, tenant_id) DO UPDATE SET
                payload = EXCLUDED.payload,
                updated_at = EXCLUDED.updated_at,
                version = {{self._table}}.version + 1,
                deleted = EXCLUDED.deleted
            WHERE {{self._table}}.version = $6
            RETURNING id, payload, created_at, updated_at, version, deleted
        """
        row = await self._pool.fetchrow(
            query,
            record.id,
            tenant_id,
            record.payload,
            record.created_at,
            datetime.now(timezone.utc),
            record.version,
            record.deleted,
        )
        if row is None:
            raise RuntimeError(f"optimistic lock failed for {{record.id}}")
        return {cls}Record(**dict(row))


class {cls}Service:
    def __init__(self, repo: {cls}Repository, config: {cls}Config):
        self._repo = repo
        self._config = config
        self._cache: dict[str, tuple[{cls}Record, float]] = {{}}
        self._locks: dict[str, asyncio.Lock] = {{}}

    def _cache_key(self, record_id: str) -> str:
        return f"{{self._config.tenant_id}}:{{record_id}}"

    async def fetch(self, record_id: str) -> {cls}Record:
        cache_key = self._cache_key(record_id)
        now = time.time()
        cached = self._cache.get(cache_key)
        if cached and now - cached[1] < self._config.cache_ttl_sec:
            return cached[0]
        lock = self._locks.setdefault(cache_key, asyncio.Lock())
        async with lock:
            cached = self._cache.get(cache_key)
            if cached and now - cached[1] < self._config.cache_ttl_sec:
                return cached[0]
            for attempt in range(self._config.max_retries):
                try:
                    record = await asyncio.wait_for(
                        self._repo.get(record_id, self._config.tenant_id),
                        timeout=self._config.timeout_sec,
                    )
                    if record is None or record.deleted:
                        raise KeyError(record_id)
                    self._cache[cache_key] = (record, time.time())
                    return record
                except asyncio.TimeoutError:
                    logger.warning("fetch timeout record=%s attempt=%d", record_id, attempt + 1)
            raise TimeoutError(record_id)

    async def save(self, record: {cls}Record) -> {cls}Record:
        saved = await self._repo.upsert(record, self._config.tenant_id)
        self._cache[self._cache_key(saved.id)] = (saved, time.time())
        return saved

    async def delete(self, record_id: str) -> None:
        record = await self.fetch(record_id)
        record.deleted = True
        await self.save(record)
        self._cache.pop(self._cache_key(record_id), None)
```
'''

TASKS = [
    "Review the architecture and identify a concrete refactor for auth, caching, and observability.",
    "Write the implementation patch with complete code snippets and unit tests.",
    "Audit the patch for race conditions, retry behavior, and multi-tenant isolation bugs.",
    "Prepare a PR summary with migration risks and a verification checklist.",
]


@dataclass
class MetricsSnapshot:
    prompt_tokens_total: float = 0.0
    generation_tokens_total: float = 0.0
    num_requests_total: float = 0.0
    cached_tokens_device: float = 0.0
    cached_tokens_host: float = 0.0
    cached_tokens_storage: float = 0.0
    evicted_tokens_total: float = 0.0
    load_back_tokens_total: float = 0.0
    hicache_host_used_tokens: float = 0.0
    cache_hit_rate: float | None = None


class MetricsSampler:
    def __init__(self, interval_s: float):
        self.interval_s = interval_s
        self.samples: list[dict[str, Any]] = []
        self._stop = threading.Event()
        self._thread: threading.Thread | None = None

    def start(self) -> None:
        self._thread = threading.Thread(target=self._run, daemon=True)
        self._thread.start()

    def stop(self) -> None:
        self._stop.set()
        if self._thread:
            self._thread.join(timeout=5)

    def _run(self) -> None:
        while not self._stop.is_set():
            sample = scrape_metrics_values()
            sample["time"] = time.time()
            self.samples.append(sample)
            self._stop.wait(self.interval_s)


def log(message: str) -> None:
    print(message, flush=True)


def run_cmd(args: list[str], timeout: int = 180) -> subprocess.CompletedProcess[str]:
    return subprocess.run(args, capture_output=True, text=True, timeout=timeout)


def docker_compose(compose_file: Path, action: str) -> None:
    cmd = ["docker", "compose", "-f", str(compose_file), action]
    if action == "up":
        cmd.extend(["-d", "--remove-orphans"])
    result = run_cmd(cmd)
    if result.returncode != 0:
        raise RuntimeError(f"{' '.join(cmd)} failed\n{result.stdout}\n{result.stderr}")


def stop_all_profiles() -> None:
    for profile in PROFILES.values():
        compose = profile["compose"]
        if compose.exists():
            for attempt in range(3):
                result = run_cmd(["docker", "compose", "-f", str(compose), "down"], timeout=600)
                if result.returncode == 0:
                    break
                log(f"warning: failed to stop {compose.name}: {result.stderr.strip()}")
                if attempt < 2:
                    time.sleep(30)


def wait_for_server(timeout_s: int = 1800) -> None:
    deadline = time.time() + timeout_s
    last = ""
    while time.time() < deadline:
        try:
            with urllib.request.urlopen(MODELS_URL, timeout=5) as resp:
                if resp.status == 200:
                    return
        except Exception as exc:  # noqa: BLE001
            last = str(exc)
        time.sleep(5)
    raise TimeoutError(f"server not ready: {last}")


def post_json(url: str, payload: dict[str, Any], timeout_s: int) -> dict[str, Any]:
    req = urllib.request.Request(
        url,
        data=json.dumps(payload).encode("utf-8"),
        headers={"Content-Type": "application/json"},
    )
    with urllib.request.urlopen(req, timeout=timeout_s) as resp:
        return json.loads(resp.read().decode("utf-8"))


def prompt_tokens_via_api(prompt: str, max_tokens: int = 1, timeout_s: int = 600) -> int:
    payload = {
        "model": MODEL,
        "messages": [{"role": "system", "content": SYSTEM_PROMPT}, {"role": "user", "content": prompt}],
        "max_tokens": max_tokens,
        "temperature": 0.0,
        "stream": False,
    }
    body = post_json(API_URL, payload, timeout_s)
    return int((body.get("usage") or {}).get("prompt_tokens") or 0)


def build_context(target_tokens: int, repo_id: int, timeout_s: int) -> tuple[str, int]:
    rng = random.Random(9000 + repo_id)
    names = [
        ("UserAuth", "users"),
        ("Billing", "invoices"),
        ("Inventory", "items"),
        ("Notification", "events"),
        ("Analytics", "metrics"),
        ("Workflow", "jobs"),
        ("Payment", "transactions"),
        ("Search", "documents"),
        ("Permissions", "grants"),
        ("Scheduler", "jobs"),
    ]
    parts = [
        f"# Repository Context repo-{repo_id}\n",
        "The following source files are open in the coding agent workspace.\n\n",
    ]
    # Initial ratio comes from previous GLM-5.1 calibration: roughly 5 chars per token
    while len("".join(parts)) < target_tokens * 5:
        idx = len(parts)
        name, table = names[idx % len(names)]
        suffix = "".join(rng.choices("abcdefghijklmnopqrstuvwxyz", k=5))
        parts.append(
            CODE_MODULE.format(
                repo=f"repo_{repo_id}",
                name=f"{name.lower()}_{suffix}",
                cls=name,
                table=table,
                idx=idx,
            )
        )
        parts.append("\n\n")
    text = "".join(parts)
    tokens = prompt_tokens_via_api(f"{text}\n\n---\n\n{TASKS[0]}", timeout_s=timeout_s)
    if abs(tokens - target_tokens) <= target_tokens * 0.05:
        return text, tokens

    # Adjust by truncating on module boundary when needed.
    if tokens > target_tokens:
        ratio = target_tokens / tokens
        cut = int(len(text) * ratio)
        text = text[:cut]
        fence = text.rfind("```")
        if fence > 0:
            text = text[: fence + 3]
    else:
        # Rare: append more modules.
        while tokens < target_tokens * 0.95:
            idx = len(parts)
            name, table = names[idx % len(names)]
            parts.append(CODE_MODULE.format(repo=f"repo_{repo_id}", name=name.lower(), cls=name, table=table, idx=idx))
            parts.append("\n\n")
            text = "".join(parts)
            tokens = prompt_tokens_via_api(f"{text}\n\n---\n\n{TASKS[0]}", timeout_s=timeout_s)
    tokens = prompt_tokens_via_api(f"{text}\n\n---\n\n{TASKS[0]}", timeout_s=timeout_s)
    return text, tokens


def messages_for(context: str, repo_id: int, request_id: int, kind: str) -> list[dict[str, str]]:
    task = TASKS[request_id % len(TASKS)]
    nonce = f"{kind}-{repo_id}-{request_id}-{int(time.time() * 1000)}"
    if kind == "decode":
        instruction = (
            f"Request {nonce}. {task}\n"
            "Give a detailed code review and implementation patch. Include code blocks, tests, and reasoning."
        )
    elif kind == "prefill":
        instruction = (
            f"Request {nonce}. Read the repository context and return only a concise JSON checklist with 3 items."
        )
    else:
        instruction = (
            f"Request {nonce}. Continue the coding-agent workflow for this repository. Reuse prior architecture context, "
            "summarize risks, and provide the next patch step."
        )
    return [
        {"role": "system", "content": SYSTEM_PROMPT},
        {"role": "user", "content": f"{context}\n\n---\n\n{instruction}"},
    ]


def send_stream(
    context: str,
    repo_id: int,
    request_id: int,
    kind: str,
    max_tokens: int,
    timeout_s: int,
) -> dict[str, Any]:
    start = time.time()
    result: dict[str, Any] = {
        "request_id": request_id,
        "repo_id": repo_id,
        "kind": kind,
        "status": "pending",
        "scheduled_at": start,
        "start_time": start,
        "ttft_s": None,
        "latency_s": None,
        "prompt_tokens": None,
        "completion_tokens": None,
        "max_chunk_gap_s": None,
        "avg_chunk_gap_s": None,
        "gap_over_1s": 0,
        "gap_over_2s": 0,
        "error": "",
    }
    payload = {
        "model": MODEL,
        "messages": messages_for(context, repo_id, request_id, kind),
        "max_tokens": max_tokens,
        "temperature": 0.2,
        "stream": True,
        "stream_options": {"include_usage": True},
    }
    req = urllib.request.Request(
        API_URL,
        data=json.dumps(payload).encode("utf-8"),
        headers={"Content-Type": "application/json"},
    )
    token_times: list[float] = []
    try:
        with urllib.request.urlopen(req, timeout=timeout_s) as resp:
            usage: dict[str, Any] = {}
            for raw_line in resp:
                line = raw_line.decode("utf-8", errors="replace").strip()
                if not line.startswith("data: "):
                    continue
                data = line[6:]
                if data == "[DONE]":
                    break
                try:
                    chunk = json.loads(data)
                except json.JSONDecodeError:
                    continue
                if chunk.get("usage"):
                    usage = chunk["usage"]
                for choice in chunk.get("choices") or []:
                    delta = choice.get("delta") or {}
                    content = delta.get("content") or delta.get("reasoning_content") or ""
                    if content:
                        now = time.time()
                        token_times.append(now)
                        if result["ttft_s"] is None:
                            result["ttft_s"] = now - start
            result["status"] = "success"
            result["latency_s"] = time.time() - start
            result["prompt_tokens"] = int(usage.get("prompt_tokens") or 0)
            result["completion_tokens"] = int(usage.get("completion_tokens") or 0)
    except urllib.error.HTTPError as exc:
        result["status"] = f"http_{exc.code}"
        result["latency_s"] = time.time() - start
        result["error"] = exc.read().decode("utf-8", errors="replace")[:500]
    except Exception as exc:  # noqa: BLE001
        result["status"] = "error"
        result["latency_s"] = time.time() - start
        result["error"] = str(exc)[:500]

    gaps = [b - a for a, b in zip(token_times, token_times[1:])]
    if gaps:
        result["max_chunk_gap_s"] = max(gaps)
        result["avg_chunk_gap_s"] = statistics.fmean(gaps)
        result["gap_over_1s"] = sum(1 for g in gaps if g >= 1.0)
        result["gap_over_2s"] = sum(1 for g in gaps if g >= 2.0)
    return result


def fetch_metrics_text() -> str:
    try:
        with urllib.request.urlopen(METRICS_URL, timeout=10) as resp:
            return resp.read().decode("utf-8", errors="replace")
    except Exception:
        return ""


def fetch_metrics_snapshot() -> MetricsSnapshot:
    text = fetch_metrics_text()
    snap = MetricsSnapshot()
    for line in text.splitlines():
        if not line or line.startswith("#"):
            continue
        m = re.search(r"\s([-+0-9.eE]+)$", line)
        if not m:
            continue
        value = float(m.group(1))
        if line.startswith("sglang:prompt_tokens_total{"):
            snap.prompt_tokens_total += value
        elif line.startswith("sglang:generation_tokens_total{"):
            snap.generation_tokens_total += value
        elif line.startswith("sglang:num_requests_total{"):
            snap.num_requests_total += value
        elif line.startswith("sglang:cached_tokens_total{"):
            if 'cache_source="device"' in line:
                snap.cached_tokens_device += value
            elif 'cache_source="host"' in line:
                snap.cached_tokens_host += value
            elif 'cache_source="storage"' in line:
                snap.cached_tokens_storage += value
        elif line.startswith("sglang:evicted_tokens_total{"):
            snap.evicted_tokens_total += value
        elif line.startswith("sglang:load_back_tokens_total{"):
            snap.load_back_tokens_total += value
        elif line.startswith("sglang:hicache_host_used_tokens{"):
            snap.hicache_host_used_tokens += value
        elif line.startswith("sglang:cache_hit_rate{"):
            snap.cache_hit_rate = value
    return snap


def scrape_metrics_values() -> dict[str, float]:
    text = fetch_metrics_text()
    values: dict[str, float] = {
        "gen_throughput": 0.0,
        "num_running": 0.0,
        "num_queue": 0.0,
        "cache_hit_rate": 0.0,
        "hicache_host_used_tokens": 0.0,
    }
    for line in text.splitlines():
        if not line or line.startswith("#"):
            continue
        m = re.search(r"\s([-+0-9.eE]+)$", line)
        if not m:
            continue
        value = float(m.group(1))
        if line.startswith("sglang:gen_throughput{"):
            values["gen_throughput"] += value
        elif line.startswith("sglang:num_running_reqs{"):
            values["num_running"] += value
        elif line.startswith("sglang:num_queue_reqs{"):
            values["num_queue"] += value
        elif line.startswith("sglang:cache_hit_rate{"):
            values["cache_hit_rate"] = value
        elif line.startswith("sglang:hicache_host_used_tokens{"):
            values["hicache_host_used_tokens"] += value
    return values


def metrics_delta(before: MetricsSnapshot, after: MetricsSnapshot) -> dict[str, float | None]:
    return {
        "prompt_tokens_total": after.prompt_tokens_total - before.prompt_tokens_total,
        "generation_tokens_total": after.generation_tokens_total - before.generation_tokens_total,
        "num_requests_total": after.num_requests_total - before.num_requests_total,
        "cached_tokens_device": after.cached_tokens_device - before.cached_tokens_device,
        "cached_tokens_host": after.cached_tokens_host - before.cached_tokens_host,
        "cached_tokens_storage": after.cached_tokens_storage - before.cached_tokens_storage,
        "evicted_tokens_total": after.evicted_tokens_total - before.evicted_tokens_total,
        "load_back_tokens_total": after.load_back_tokens_total - before.load_back_tokens_total,
        "hicache_host_used_tokens_after": after.hicache_host_used_tokens,
        "cache_hit_rate_after": after.cache_hit_rate,
    }


def percentile(values: list[float], pct: float) -> float | None:
    if not values:
        return None
    ordered = sorted(values)
    idx = min(len(ordered) - 1, max(0, math.ceil(len(ordered) * pct) - 1))
    return ordered[idx]


def summarize_values(values: list[float]) -> dict[str, float | None]:
    values = [v for v in values if v is not None and math.isfinite(v)]
    return {
        "avg": statistics.fmean(values) if values else None,
        "p05": percentile(values, 0.05),
        "p10": percentile(values, 0.10),
        "p50": percentile(values, 0.50),
        "p90": percentile(values, 0.90),
        "p95": percentile(values, 0.95),
        "max": max(values) if values else None,
        "min": min(values) if values else None,
    }


def prom_range_summary(query: str, start: float, end: float, step_s: int) -> dict[str, Any]:
    params = {"query": query, "start": start, "end": end, "step": str(step_s)}
    url = f"{PROM_URL}/api/v1/query_range?" + urllib.parse.urlencode(params)
    try:
        with urllib.request.urlopen(url, timeout=30) as resp:
            data = json.loads(resp.read().decode("utf-8"))
    except Exception as exc:  # noqa: BLE001
        return {"error": str(exc)}
    values: list[float] = []
    for series in data.get("data", {}).get("result", []):
        for _, value in series.get("values", []):
            if value in ("NaN", "+Inf", "-Inf"):
                continue
            try:
                f = float(value)
            except ValueError:
                continue
            if math.isfinite(f):
                values.append(f)
    out = summarize_values(values)
    out["samples"] = len(values)
    return out


def schedule_requests(
    contexts: list[tuple[str, int]],
    duration_s: int,
    decode_workers: int,
    long_prefill_interval_s: float,
    revisit_interval_s: float,
    decode_max_tokens: int,
    prefill_max_tokens: int,
    timeout_s: int,
    max_workers: int,
) -> list[dict[str, Any]]:
    start = time.time()
    stop_at = start + duration_s
    request_id = 0
    request_lock = threading.Lock()
    futures = []

    def next_request_id() -> int:
        nonlocal request_id
        with request_lock:
            rid = request_id
            request_id += 1
            return rid

    def decode_loop(worker_id: int) -> list[dict[str, Any]]:
        rows = []
        repo_idx = worker_id % len(contexts)
        while time.time() < stop_at:
            context, _tokens = contexts[repo_idx]
            rows.append(
                send_stream(
                    context,
                    repo_idx,
                    next_request_id(),
                    "decode",
                    decode_max_tokens,
                    timeout_s,
                )
            )
            repo_idx = (repo_idx + 1) % len(contexts)
        return rows

    def timed_loop(kind: str, interval_s: float, max_tokens: int) -> list[dict[str, Any]]:
        rows = []
        if interval_s <= 0:
            return rows
        next_at = start + interval_s
        repo_idx = 0
        while next_at < stop_at:
            delay = next_at - time.time()
            if delay > 0:
                time.sleep(delay)
            context, _tokens = contexts[repo_idx % len(contexts)]
            rows.append(
                send_stream(
                    context,
                    repo_idx % len(contexts),
                    next_request_id(),
                    kind,
                    max_tokens,
                    timeout_s,
                )
            )
            repo_idx += 1
            next_at += interval_s
        return rows

    with ThreadPoolExecutor(max_workers=max_workers) as pool:
        for worker_id in range(decode_workers):
            futures.append(pool.submit(decode_loop, worker_id))
        futures.append(pool.submit(timed_loop, "prefill", long_prefill_interval_s, prefill_max_tokens))
        futures.append(pool.submit(timed_loop, "revisit", revisit_interval_s, decode_max_tokens))
        results = []
        for future in as_completed(futures):
            results.extend(future.result())
    return results


def run_profile(profile_key: str, args: argparse.Namespace) -> dict[str, Any]:
    profile = PROFILES[profile_key]
    if not args.reuse_running:
        log(f"starting profile {profile['label']}")
        stop_all_profiles()
        time.sleep(3)
        docker_compose(profile["compose"], "up")
    wait_for_server()
    time.sleep(args.settle_seconds)

    log(f"building {args.repo_count} repository contexts (~{args.target_input_tokens:,} tokens each)")
    contexts = []
    for repo_id in range(args.repo_count):
        context, tokens = build_context(args.target_input_tokens, repo_id, args.timeout)
        contexts.append((context, tokens))
        log(f"  repo-{repo_id}: {tokens:,} prompt tokens")

    before = fetch_metrics_snapshot()
    sampler = MetricsSampler(interval_s=args.metrics_interval)
    start = time.time()
    sampler.start()
    log(
        f"running mixed workload for {args.duration}s: decode_workers={args.decode_workers}, "
        f"prefill_interval={args.long_prefill_interval}s, revisit_interval={args.revisit_interval}s"
    )
    results = schedule_requests(
        contexts=contexts,
        duration_s=args.duration,
        decode_workers=args.decode_workers,
        long_prefill_interval_s=args.long_prefill_interval,
        revisit_interval_s=args.revisit_interval,
        decode_max_tokens=args.decode_output_tokens,
        prefill_max_tokens=args.prefill_output_tokens,
        timeout_s=args.timeout,
        max_workers=args.max_workers,
    )
    end = time.time()
    sampler.stop()
    after = fetch_metrics_snapshot()

    return {
        "profile_key": profile_key,
        "label": profile["label"],
        "compose": str(profile["compose"]),
        "note": profile["note"] + (" (already running)" if args.reuse_running else ""),
        "context_prompt_tokens": [tokens for _, tokens in contexts],
        "started_at": start,
        "ended_at": end,
        "duration_s": end - start,
        "client_results": results,
        "client_summary": summarize_client_results(results, end - start),
        "sampled_metrics": sampler.samples,
        "sampled_metrics_summary": summarize_sampled_metrics(sampler.samples),
        "prometheus_summary": summarize_prometheus_window(start, end, args.metrics_interval),
        "metrics_delta": metrics_delta(before, after),
    }


def summarize_client_results(results: list[dict[str, Any]], wall_s: float) -> dict[str, Any]:
    successes = [r for r in results if r["status"] == "success"]
    by_kind: dict[str, list[dict[str, Any]]] = {}
    for r in results:
        by_kind.setdefault(r["kind"], []).append(r)

    def one(rows: list[dict[str, Any]]) -> dict[str, Any]:
        ok = [r for r in rows if r["status"] == "success"]
        completion = [float(r["completion_tokens"]) for r in ok if r.get("completion_tokens")]
        prompt = [float(r["prompt_tokens"]) for r in ok if r.get("prompt_tokens")]
        ttft = [float(r["ttft_s"]) for r in ok if r.get("ttft_s") is not None]
        lat = [float(r["latency_s"]) for r in ok if r.get("latency_s") is not None]
        max_gap = [float(r["max_chunk_gap_s"]) for r in ok if r.get("max_chunk_gap_s") is not None]
        return {
            "requests": len(rows),
            "successful": len(ok),
            "failed": len(rows) - len(ok),
            "prompt_tokens_avg": statistics.fmean(prompt) if prompt else None,
            "completion_tokens_total": sum(completion),
            "completion_tokens_avg": statistics.fmean(completion) if completion else None,
            "output_tps_wall": sum(completion) / wall_s if wall_s > 0 else None,
            "ttft": summarize_values(ttft),
            "latency": summarize_values(lat),
            "max_chunk_gap": summarize_values(max_gap),
            "gap_over_1s_total": sum(int(r.get("gap_over_1s") or 0) for r in ok),
            "gap_over_2s_total": sum(int(r.get("gap_over_2s") or 0) for r in ok),
        }

    summary = {"overall": one(results), "by_kind": {kind: one(rows) for kind, rows in sorted(by_kind.items())}}
    summary["failures"] = [r for r in results if r["status"] != "success"][:5]
    summary["success_count"] = len(successes)
    return summary


def summarize_sampled_metrics(samples: list[dict[str, Any]]) -> dict[str, Any]:
    out = {}
    for key in ["gen_throughput", "num_running", "num_queue", "cache_hit_rate", "hicache_host_used_tokens"]:
        vals = [float(s.get(key, 0.0)) for s in samples]
        out[key] = summarize_values(vals)
    if samples:
        lows = [float(s.get("gen_throughput", 0.0)) for s in samples]
        out["gen_throughput_below_50_ratio"] = sum(1 for v in lows if v < 50) / len(lows)
        out["gen_throughput_below_100_ratio"] = sum(1 for v in lows if v < 100) / len(lows)
    return out


def summarize_prometheus_window(start: float, end: float, step_s: int) -> dict[str, Any]:
    queries = {
        "gen_throughput": "avg(sglang_gen_throughput)",
        "gen_rate_1m": "sum(rate(sglang_generation_tokens_total[1m]))",
        "running": "avg(sglang_num_running_reqs)",
        "queue": "avg(sglang_num_queue_reqs)",
        "ttft_avg_1m": "sum(rate(sglang_time_to_first_token_seconds_sum[1m])) / sum(rate(sglang_time_to_first_token_seconds_count[1m]))",
        "latency_avg_1m": "sum(rate(sglang_e2e_request_latency_seconds_sum[1m])) / sum(rate(sglang_e2e_request_latency_seconds_count[1m]))",
    }
    return {name: prom_range_summary(query, start, end, step_s) for name, query in queries.items()}


def fmt(value: Any, digits: int = 2) -> str:
    if value is None:
        return "N/A"
    if isinstance(value, (int, float)):
        if math.isnan(value) or math.isinf(value):
            return "N/A"
        return f"{value:,.{digits}f}"
    return str(value)


def metric_cells(summary: dict[str, Any], key: str) -> str:
    s = summary.get(key, {})
    return f"{fmt(s.get('avg'))} | {fmt(s.get('p05'))} | {fmt(s.get('p50'))} | {fmt(s.get('p95'))} | {fmt(s.get('max'))}"


def generate_report(config: dict[str, Any], results: list[dict[str, Any]], chart_name: str) -> str:
    lines: list[str] = []
    lines.append("# GLM-5.1 Mixed Coding-Agent 压测")
    lines.append("")
    lines.append(f"- **测试时间**: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    lines.append(f"- **目标 input tokens**: {config['target_input_tokens']:,}")
    lines.append(f"- **decode 输出上限**: {config['decode_output_tokens']:,}")
    lines.append(f"- **prefill spike 输出上限**: {config['prefill_output_tokens']:,}")
    lines.append(f"- **持续时间**: {config['duration']}s")
    lines.append(f"- **decode workers**: {config['decode_workers']}")
    lines.append(f"- **long prefill interval**: {config['long_prefill_interval']}s")
    lines.append(f"- **repo_count**: {config['repo_count']}")
    lines.append("")
    lines.append("## 口径")
    lines.append("")
    lines.append("- 这是 open-loop/closed-loop 混合测试：长输出 decode 持续进行，同时周期性插入 100k input 的短输出 prefill 请求。")
    lines.append("- 重点观察 prefill 挤占 decode 时的 `max_chunk_gap`、Grafana 同口径 `sglang_gen_throughput` 低位数，以及 HiCache evict/load_back。")
    lines.append("- 参考方向：SGLang benchmark/profiling、HiCache best practices、SGLang PD disaggregation，以及 Sarathi-Serve 对 prefill/decode interference 的分析。")
    lines.append("")
    for result in results:
        lines.append(f"## {result['label']}")
        lines.append("")
        lines.append(f"- Compose: `{Path(result['compose']).name}`")
        lines.append(f"- Context tokens: {', '.join(fmt(v, 0) for v in result['context_prompt_tokens'])}")
        cs = result["client_summary"]
        overall = cs["overall"]
        lines.append("")
        lines.append("### Client Summary")
        lines.append("")
        lines.append("| Scope | Requests | Success | Output tok/s(wall) | Prompt tok/req avg | Output tok/req avg | TTFT P50/P95 | Latency P50/P95 | Max chunk gap P50/P95 | gaps >=1s | gaps >=2s |")
        lines.append("|---|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|")

        def row(name: str, s: dict[str, Any]) -> str:
            return (
                f"| {name} | {s['requests']} | {s['successful']} "
                f"| {fmt(s.get('output_tps_wall'))} "
                f"| {fmt(s.get('prompt_tokens_avg'), 0)} "
                f"| {fmt(s.get('completion_tokens_avg'), 0)} "
                f"| {fmt(s['ttft'].get('p50'))}/{fmt(s['ttft'].get('p95'))} "
                f"| {fmt(s['latency'].get('p50'))}/{fmt(s['latency'].get('p95'))} "
                f"| {fmt(s['max_chunk_gap'].get('p50'))}/{fmt(s['max_chunk_gap'].get('p95'))} "
                f"| {s.get('gap_over_1s_total', 0)} "
                f"| {s.get('gap_over_2s_total', 0)} |"
            )

        lines.append(row("overall", overall))
        for kind, summary in cs["by_kind"].items():
            lines.append(row(kind, summary))
        lines.append("")
        lines.append("### Metrics Summary")
        lines.append("")
        lines.append("| Metric | Avg | P05 | P50 | P95 | Max |")
        lines.append("|---|---:|---:|---:|---:|---:|")
        sm = result["sampled_metrics_summary"]
        for key in ["gen_throughput", "num_running", "num_queue", "cache_hit_rate", "hicache_host_used_tokens"]:
            lines.append(f"| sampled {key} | {metric_cells(sm, key)} |")
        ps = result["prometheus_summary"]
        for key in ["gen_throughput", "gen_rate_1m", "running", "queue", "ttft_avg_1m", "latency_avg_1m"]:
            lines.append(f"| prometheus {key} | {metric_cells(ps, key)} |")
        lines.append("")
        lines.append("### SGLang Counter Delta")
        lines.append("")
        m = result["metrics_delta"]
        lines.append("| prompt tokens | generation tokens | requests | cached device | cached host | evicted | load_back | host used after |")
        lines.append("|---:|---:|---:|---:|---:|---:|---:|---:|")
        lines.append(
            f"| {fmt(m.get('prompt_tokens_total'),0)} | {fmt(m.get('generation_tokens_total'),0)} | {fmt(m.get('num_requests_total'),0)} "
            f"| {fmt(m.get('cached_tokens_device'),0)} | {fmt(m.get('cached_tokens_host'),0)} "
            f"| {fmt(m.get('evicted_tokens_total'),0)} | {fmt(m.get('load_back_tokens_total'),0)} "
            f"| {fmt(m.get('hicache_host_used_tokens_after'),0)} |"
        )
        lines.append("")
    lines.append("## 图表")
    lines.append("")
    lines.append(f"![Mixed benchmark]({chart_name})")
    lines.append("")
    return "\n".join(lines)


def svg_line(points: list[tuple[float, float]], color: str) -> str:
    if not points:
        return ""
    data = " ".join(f"{x:.1f},{y:.1f}" for x, y in points)
    return f'<polyline points="{data}" fill="none" stroke="{color}" stroke-width="2" />'


def generate_svg(results: list[dict[str, Any]], path: Path) -> None:
    width, height = 1100, 560
    pad_l, pad_r, pad_t, pad_b = 70, 30, 50, 60
    plot_w = width - pad_l - pad_r
    plot_h = height - pad_t - pad_b
    all_samples = []
    for r in results:
        all_samples.extend(r["sampled_metrics"])
    if not all_samples:
        path.write_text("<svg />", encoding="utf-8")
        return
    t0 = min(s["time"] for s in all_samples)
    t1 = max(s["time"] for s in all_samples)
    max_thr = max([float(s.get("gen_throughput", 0)) for s in all_samples] + [1])
    colors = {"mtp": "#2563eb", "hicache": "#dc2626"}

    def x(t: float) -> float:
        return pad_l + (t - t0) / max(1, t1 - t0) * plot_w

    def y(v: float) -> float:
        return pad_t + plot_h - v / (max_thr * 1.1) * plot_h

    lines = [
        f'<svg xmlns="http://www.w3.org/2000/svg" width="{width}" height="{height}" viewBox="0 0 {width} {height}">',
        '<rect width="100%" height="100%" fill="white"/>',
        '<style>text{font-family:Arial,Helvetica,sans-serif;fill:#111827}.grid{stroke:#e5e7eb}.axis{stroke:#9ca3af}.title{font-size:18px;font-weight:700}.label{font-size:12px}</style>',
        f'<text x="{width/2}" y="28" text-anchor="middle" class="title">Mixed Workload: SGLang gen_throughput</text>',
    ]
    for i in range(6):
        yy = pad_t + plot_h - plot_h * i / 5
        val = max_thr * 1.1 * i / 5
        lines.append(f'<line x1="{pad_l}" y1="{yy:.1f}" x2="{pad_l+plot_w}" y2="{yy:.1f}" class="grid"/>')
        lines.append(f'<text x="{pad_l-8}" y="{yy+4:.1f}" text-anchor="end" class="label">{val:.0f}</text>')
    lines.append(f'<line x1="{pad_l}" y1="{pad_t}" x2="{pad_l}" y2="{pad_t+plot_h}" class="axis"/>')
    lines.append(f'<line x1="{pad_l}" y1="{pad_t+plot_h}" x2="{pad_l+plot_w}" y2="{pad_t+plot_h}" class="axis"/>')
    lines.append(f'<text x="{pad_l-48}" y="{pad_t+plot_h/2}" transform="rotate(-90 {pad_l-48},{pad_t+plot_h/2})" text-anchor="middle" class="label">tokens/s</text>')
    for idx, r in enumerate(results):
        color = colors.get(r["profile_key"], "#111827")
        pts = [(x(s["time"]), y(float(s.get("gen_throughput", 0)))) for s in r["sampled_metrics"]]
        lines.append(svg_line(pts, color))
        ly = 52 + idx * 24
        lines.append(f'<line x1="{width-260}" y1="{ly}" x2="{width-232}" y2="{ly}" stroke="{color}" stroke-width="3"/>')
        lines.append(f'<text x="{width-224}" y="{ly+5}" class="label">{r["label"]}</text>')
    lines.append("</svg>")
    path.write_text("\n".join(lines), encoding="utf-8")


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Mixed coding-agent serving benchmark")
    parser.add_argument("--profiles", default="hicache", help="comma-separated: mtp,hicache")
    parser.add_argument("--reuse-running", action="store_true")
    parser.add_argument("--keep-running", action="store_true")
    parser.add_argument("--duration", type=int, default=600)
    parser.add_argument("--repo-count", type=int, default=4)
    parser.add_argument("--target-input-tokens", type=int, default=100_000)
    parser.add_argument("--decode-workers", type=int, default=2)
    parser.add_argument("--decode-output-tokens", type=int, default=2048)
    parser.add_argument("--prefill-output-tokens", type=int, default=64)
    parser.add_argument("--long-prefill-interval", type=float, default=30.0)
    parser.add_argument("--revisit-interval", type=float, default=45.0)
    parser.add_argument("--max-workers", type=int, default=32)
    parser.add_argument("--timeout", type=int, default=2400)
    parser.add_argument("--metrics-interval", type=int, default=5)
    parser.add_argument("--settle-seconds", type=int, default=5)
    parser.add_argument("--report", default="")
    parser.add_argument("--json-output", default="")
    parser.add_argument("--chart", default="")
    return parser.parse_args()


def main() -> int:
    args = parse_args()
    selected = [p.strip() for p in args.profiles.split(",") if p.strip()]
    for key in selected:
        if key not in PROFILES:
            raise SystemExit(f"unknown profile {key}")
    REPORT_DIR.mkdir(parents=True, exist_ok=True)
    stamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    report_path = Path(args.report) if args.report else REPORT_DIR / f"coding_agent_mixed_benchmark_{stamp}.md"
    json_path = Path(args.json_output) if args.json_output else REPORT_DIR / f"coding_agent_mixed_benchmark_{stamp}.json"
    chart_path = Path(args.chart) if args.chart else REPORT_DIR / f"coding_agent_mixed_benchmark_{stamp}.svg"

    log("=" * 80)
    log("Mixed coding-agent benchmark")
    log("=" * 80)
    log(f"profiles={selected}, duration={args.duration}s, decode_workers={args.decode_workers}")
    results = []
    try:
        for idx, key in enumerate(selected):
            # --reuse-running only makes sense for the first selected profile.
            if idx > 0:
                args.reuse_running = False
            results.append(run_profile(key, args))
    finally:
        if not args.keep_running and not args.reuse_running:
            stop_all_profiles()

    config = {
        "profiles": selected,
        "duration": args.duration,
        "repo_count": args.repo_count,
        "target_input_tokens": args.target_input_tokens,
        "decode_workers": args.decode_workers,
        "decode_output_tokens": args.decode_output_tokens,
        "prefill_output_tokens": args.prefill_output_tokens,
        "long_prefill_interval": args.long_prefill_interval,
        "revisit_interval": args.revisit_interval,
    }
    output = {"config": config, "results": results}
    json_path.write_text(json.dumps(output, indent=2, ensure_ascii=False), encoding="utf-8")
    generate_svg(results, chart_path)
    report_path.write_text(generate_report(config, results, chart_path.name), encoding="utf-8")
    log(f"JSON:   {json_path}")
    log(f"Report: {report_path}")
    log(f"Chart:  {chart_path}")
    return 0


if __name__ == "__main__":
    sys.exit(main())