Skip to content

Commit fd7e0c8

Browse files
authored
Enable concurrent Terminal-Bench trials with per-trial sessions (pytorch#23744)
1 parent 328e52a commit fd7e0c8

4 files changed

Lines changed: 135 additions & 8 deletions

File tree

Lines changed: 6 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -1,24 +1,23 @@
11
[server]
22
python = "/path/to/model-env/bin/python"
3-
module = "executorch.examples.models.muse_glimmer.serving.serve"
4-
worker_bin = "/path/to/executorch/cmake-out/examples/models/muse-glimmer/muse_glimmer_worker"
5-
model_path = "/path/to/muse-glimmer-k-quant-17G-128K-text-dflash-metal.pte"
3+
module = "executorch.examples.models.muse_glimmer.serving.serve_batching"
4+
worker_bin = "/path/to/executorch/cmake-out/examples/models/muse-glimmer-batching/muse_glimmer_batching_worker"
5+
# Requires an MLX solo export with --use-offgraph-kv-cache.
6+
model_path = "/path/to/muse-glimmer-offgraph/model.pte"
67
# Keep tokenizer.json, tokenizer_config.json, and chat_template.jinja together.
78
tokenizer_path = "/path/to/tokenizer/tokenizer.json"
89
hf_tokenizer = "/path/to/tokenizer"
910
model_id = "muse-glimmer-30b"
1011
max_context = 131072
11-
max_sessions = 2 # One scratch session and one named session for cross-turn KV reuse.
1212
tool_parser = "atem"
13-
artifact_mode = "dflash"
1413
# host = "0.0.0.0" # Exposes the server on all network interfaces.
1514
# port = 8000
1615

1716
[terminal_bench]
18-
tasks = ["fix-git"]
17+
tasks = ["fix-git", "git-multibranch", "log-summary-date-ranges", "db-wal-recovery"]
1918
max_output_tokens = 4096
2019
temperature = 0.8
2120
step_limit = 100
2221
attempts = 1
23-
session_affinity = "glimmer-terminal-bench"
22+
concurrency = 4
2423
# agent_host = "host.lima.internal" # Override automatic Docker host detection.

‎examples/llm_server/evals/terminal_bench/README.md‎

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -44,6 +44,17 @@ Configuration details:
4444
- Optional `session_affinity` sends an `x-session-affinity` header for KV reuse.
4545
It requires worker support for named sessions; sequential trials share the ID.
4646

47+
For concurrent tasks, set `[terminal_bench].concurrency` (default `1`), remove
48+
`session_affinity`, and select a batching server and compatible export. Server
49+
limits use the launcher's defaults. Keep task concurrency within the server's
50+
session and request capacity; configure `[server]` options as needed.
51+
Each trial/attempt uses Harbor's trial UUID as its KV-session ID. Cleanup runs in
52+
Harbor's END hook before the trial slot is released, retrying transient failures
53+
for up to two minutes. Persistent cleanup errors stop the run to avoid reusing
54+
unreleased capacity. Harbor retains the same mini-SWE-agent loop and per-trial
55+
verification.
56+
The run also checks that evaluation session IDs match the Harbor trial UUIDs.
57+
4758
`--dry-run` writes resolved commands to `run.json`. Use `--task NAME` to override
4859
the task list and `--output DIR` to select a new results directory. Each run owns
4960
its server process.
@@ -60,3 +71,5 @@ its server process.
6071

6172
Model revisions, export/build commands, and dependency versions are not captured;
6273
retain them alongside results.
74+
With concurrency, `metrics.json` sums per-request timings, which overlap;
75+
these sums are not elapsed engine time or aggregate throughput.

‎examples/llm_server/evals/terminal_bench/runner.py‎

Lines changed: 36 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -62,6 +62,7 @@ def load_config(path):
6262
defaults = {
6363
"tasks": ["fix-git"],
6464
"attempts": 1,
65+
"concurrency": 1,
6566
"step_limit": 100,
6667
"max_output_tokens": 512,
6768
"temperature": 0.0,
@@ -83,6 +84,13 @@ def load_config(path):
8384
affinity = options["session_affinity"]
8485
if affinity is not None and (not isinstance(affinity, str) or not affinity.strip()):
8586
raise ValueError("[terminal_bench].session_affinity must be a nonempty string")
87+
concurrency = options["concurrency"]
88+
if type(concurrency) is not int or concurrency < 1:
89+
raise ValueError("[terminal_bench].concurrency must be a positive integer")
90+
if concurrency > 1 and affinity is not None:
91+
raise ValueError(
92+
"Remove [terminal_bench].session_affinity for concurrent trials"
93+
)
8694
return config
8795

8896

@@ -140,7 +148,7 @@ def commands(config, output):
140148
"--ae",
141149
f"no_proxy={agent_host}",
142150
"--n-concurrent",
143-
"1",
151+
str(options["concurrency"]),
144152
"--n-attempts",
145153
str(options["attempts"]),
146154
"--max-retries",
@@ -150,6 +158,18 @@ def commands(config, output):
150158
"--job-name",
151159
"harbor",
152160
]
161+
if options["concurrency"] > 1:
162+
module = "executorch.examples.llm_server.evals.terminal_bench.sessions"
163+
harbor[harbor.index("--agent") + 1] = f"{module}:SessionMiniSweAgent"
164+
host = "127.0.0.1" if server["host"] == "0.0.0.0" else server["host"]
165+
harbor.extend(
166+
[
167+
"--plugin",
168+
f"{module}:SessionCleanup",
169+
"--plugin-kwarg",
170+
f"base_url=http://{host}:{server['port']}/v1",
171+
]
172+
)
153173
for task in options["tasks"]:
154174
harbor.extend(["--include-task-name", task])
155175
probe = [
@@ -246,6 +266,7 @@ def metrics(log, context, reserve):
246266
)
247267
}
248268
result["requests"] = len(turns)
269+
result["session_ids"] = sorted({turn["session_id"] for turn in turns})
249270
result["peak_prompt_tokens"] = max(
250271
(int(turn["prompt_tokens"]) for turn in turns), default=0
251272
)
@@ -342,6 +363,7 @@ def run(config, output, serve, harbor, probe):
342363
print("Generation preflight passed.", flush=True)
343364
job = subprocess.Popen(
344365
harbor,
366+
env=env,
345367
stdout=harbor_log,
346368
stderr=subprocess.STDOUT,
347369
start_new_session=True,
@@ -388,6 +410,19 @@ def run(config, output, serve, harbor, probe):
388410
raise RuntimeError(
389411
"Missing server metrics or context limit exceeded; see metrics.json"
390412
)
413+
if config["terminal_bench"]["concurrency"] > 1:
414+
trial_ids = {
415+
str(json.loads(path.read_text())["id"])
416+
for path in (output / "harbor").glob("*/result.json")
417+
}
418+
session_ids = set(timing["session_ids"])
419+
if session_ids != trial_ids:
420+
raise RuntimeError(
421+
"Concurrent session IDs do not match Harbor trial UUIDs: "
422+
f"unexpected={sorted(session_ids - trial_ids)}, "
423+
f"missing={sorted(trial_ids - session_ids)}; "
424+
"see metrics.json and harbor/*/result.json"
425+
)
391426

392427

393428
def main(argv=None):
Lines changed: 80 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,80 @@
1+
# Copyright (c) Meta Platforms, Inc. and affiliates.
2+
# All rights reserved.
3+
#
4+
# This source code is licensed under the BSD-style license found in the
5+
# LICENSE file in the root directory of this source tree.
6+
7+
"""Per-trial KV sessions for Harbor's unchanged mini-SWE-agent loop."""
8+
9+
import asyncio
10+
import logging
11+
from urllib.parse import quote
12+
13+
import httpx
14+
import yaml
15+
16+
from harbor.agents.installed.mini_swe_agent import MiniSweAgent
17+
18+
RETRYABLE_STATUS_CODES = {408, 409, 429, 500, 502, 503, 504}
19+
20+
21+
class SessionMiniSweAgent(MiniSweAgent):
22+
async def run(self, instruction, environment, context):
23+
if self.context_id is None:
24+
raise ValueError("Harbor must assign the agent's trial context_id")
25+
config = yaml.safe_load(self._config_yaml) if self._config_yaml else {}
26+
headers = (
27+
config.setdefault("model", {})
28+
.setdefault("model_kwargs", {})
29+
.setdefault("extra_headers", {})
30+
)
31+
headers["X-ExecuTorch-Session-ID"] = str(self.context_id)
32+
self._config_yaml = yaml.safe_dump(config, sort_keys=False)
33+
await super().run(instruction, environment, context)
34+
35+
36+
class SessionCleanup:
37+
def __init__(self, base_url: str):
38+
self.base_url = base_url.rstrip("/")
39+
40+
async def on_job_start(self, job):
41+
# Harbor awaits END hooks before releasing the trial's concurrency slot.
42+
job.on_trial_ended(self.on_trial_end)
43+
44+
async def on_trial_end(self, event):
45+
session_id = quote(str(event.trial_id), safe="")
46+
try:
47+
async with (
48+
asyncio.timeout(120),
49+
httpx.AsyncClient(trust_env=False, timeout=20) as client,
50+
):
51+
delay = 1
52+
while True:
53+
try:
54+
response = await client.delete(
55+
f"{self.base_url}/sessions/{session_id}"
56+
)
57+
except httpx.TransportError as error:
58+
reason = str(error)
59+
else:
60+
if response.status_code not in RETRYABLE_STATUS_CODES:
61+
response.raise_for_status()
62+
return
63+
reason = f"HTTP {response.status_code}"
64+
logging.getLogger(__name__).warning(
65+
"Retrying session cleanup for trial %s (%s) in %ss: %s",
66+
event.trial_name,
67+
session_id,
68+
delay,
69+
reason,
70+
)
71+
await asyncio.sleep(delay)
72+
delay = min(delay * 2, 5)
73+
except (httpx.HTTPError, TimeoutError) as error:
74+
raise RuntimeError(
75+
f"Could not release session {session_id} for trial {event.trial_name}; "
76+
"aborting evaluation to avoid silent session leaks"
77+
) from error
78+
79+
async def on_job_end(self, job_result):
80+
pass

0 commit comments

Comments
 (0)