diff --git a/scripts/benchmarks/README.md b/scripts/benchmarks/README.md index 31c648cb1..14681418b 100644 --- a/scripts/benchmarks/README.md +++ b/scripts/benchmarks/README.md @@ -183,6 +183,34 @@ fast sample. Raw results also record PTY bytes per interaction, a proxy for how much the renderer redraws. UI trials default to 3 per revision (`ui_trials` in `config.json`) to bound sandbox cost; the scenario takes roughly 60–90 s per trial. +### Scheduled catalog and cold workers + +After the navigation scenario, the UI trial stops its benchmark-user processes and regenerates +an independent 2,300-session fixture: 2,298 children linked through real ledger edges, their root, +and an unrelated two-message cold target. Each non-target transcript contains 64 assistant usage +entries (147,136 in total), exceeding the 100,000-entry metadata-cache budget. Thirteen children +own scheduled-job artifacts. All schedules are paused, have no next run, and must retain runCount=0. + +The installed daemon runs on a dedicated socket. The probe requires protocol 7 and the negotiated +heartbeat_catalog capability; unsupported revisions fail visibly. It records: + +- **Scheduled catalog, first request** and **repeated request:** complete global heartbeats_list + round trips, with process-tree CPU. The daemon may already have scanned schedules during startup; + the first request does not claim a cold filesystem or empty metadata cache. +- **Cold worker with three catalog scans:** pipeline three global catalog requests, allow 20 ms + for their handlers to enter, then create the unrelated saved session. Time the create request + through a matching ready worker summary. The target must not already be resident. This isolates + worker readiness under scan traffic (#2299), without transcript rendering or PTY delays. + +Every catalog reply must contain exactly the expected 13 jobs, with owner names and first-message +metadata. Competing scans must also complete successfully. Missing jobs, dropped metadata, failed +scans, and wrong/not-ready workers cannot count as improvements. Raw results record how many scan +replies arrived before worker readiness; this is an observed response order, not proof of which +internal scan was executing at each instant. CPU for the cold-worker interval includes scan work. +No schedules execute and no prompts are submitted. Scans remain independent of the ordinary UI +fixture so the added history cannot change the original navigation measurements. Allow additional +fixture-generation and five scan round trips per UI trial; full Linux calibration is still required. + ## Lifecycle and costs The main workflow posts a single marked comment and updates it in place. A new push replaces the diff --git a/scripts/benchmarks/report.py b/scripts/benchmarks/report.py index 4de03c49b..4598dfed5 100644 --- a/scripts/benchmarks/report.py +++ b/scripts/benchmarks/report.py @@ -142,6 +142,27 @@ UI_METRICS = ( PERFORMANCE_NOISE_FLOOR, ), Definition("parent_open_cpu", "CPU, open chain parent", 1000, "ms", 0.05, PERFORMANCE_NOISE_FLOOR), + Definition( + "scheduled_catalog", "Scheduled catalog, first request", 1000, "ms", 0.05, PERFORMANCE_NOISE_FLOOR + ), + Definition("scheduled_catalog_cpu", "CPU, scheduled catalog", 1000, "ms", 0.02, PERFORMANCE_NOISE_FLOOR), + Definition( + "scheduled_catalog_warm", + "Scheduled catalog, repeated request", + 1000, + "ms", + 0.05, + PERFORMANCE_NOISE_FLOOR, + ), + Definition( + "scheduled_catalog_warm_cpu", "CPU, repeated catalog", 1000, "ms", 0.02, PERFORMANCE_NOISE_FLOOR + ), + Definition( + "cold_open_catalog", "Cold worker with three catalog scans", 1000, "ms", 0.05, PERFORMANCE_NOISE_FLOOR + ), + Definition( + "cold_open_catalog_cpu", "CPU, cold worker and scans", 1000, "ms", 0.02, PERFORMANCE_NOISE_FLOOR + ), Definition("ui_rss", "UI memory after interactions", 1e-6, "MB", 10485760, PERFORMANCE_NOISE_FLOOR), ) @@ -377,6 +398,9 @@ def render(report: Report) -> str: "Readiness is the rendered transcript tail plus a confirmed editor echo.", "CPU metrics sum utime+stime across the whole benchmark-user process tree per interaction.", "UI memory sums RSS after the interactions; PTY byte counts are in the raw results.", + "A separate catalog fixture has 2,300 sessions, 2,298 edges, and 13 paused scheduled-job owners.", + "Catalog timings cover first/repeated reads and cold worker creation under three pending scans.", + "All expected jobs and owner metadata are checked; worker readiness excludes TUI rendering.", "Costs estimate full sandbox lifetimes at configured rates, including setup and build.", f"Budget target: ${report.config.budget_usd:g}; not a billing cap. " "Performance changes are informational.", diff --git a/scripts/benchmarks/schema.py b/scripts/benchmarks/schema.py index 18393bdcb..5c8e0f20d 100644 --- a/scripts/benchmarks/schema.py +++ b/scripts/benchmarks/schema.py @@ -42,6 +42,12 @@ Metric = Literal[ "parent_open", "parent_open_cpu", "ui_rss", + "scheduled_catalog", + "scheduled_catalog_cpu", + "scheduled_catalog_warm", + "scheduled_catalog_warm_cpu", + "cold_open_catalog", + "cold_open_catalog_cpu", ] UI_METRIC_KEYS = frozenset( { @@ -60,6 +66,12 @@ UI_METRIC_KEYS = frozenset( "parent_open", "parent_open_cpu", "ui_rss", + "scheduled_catalog", + "scheduled_catalog_cpu", + "scheduled_catalog_warm", + "scheduled_catalog_warm_cpu", + "cold_open_catalog", + "cold_open_catalog_cpu", } ) NonNegative = Annotated[float, Field(ge=0, le=1e15, allow_inf_nan=False)] @@ -235,6 +247,12 @@ PHASE_METRICS: dict[str, tuple[Metric, ...]] = { "parent_open", "parent_open_cpu", "ui_rss", + "scheduled_catalog", + "scheduled_catalog_cpu", + "scheduled_catalog_warm", + "scheduled_catalog_warm_cpu", + "cold_open_catalog", + "cold_open_catalog_cpu", ), } diff --git a/scripts/benchmarks/tests/test_benchmarks.py b/scripts/benchmarks/tests/test_benchmarks.py index ea5c3aab2..b3ad28f18 100644 --- a/scripts/benchmarks/tests/test_benchmarks.py +++ b/scripts/benchmarks/tests/test_benchmarks.py @@ -187,7 +187,8 @@ class ReportTests(unittest.TestCase): report.pr_head.metrics["bundle"] = [] text = render(report) self.assertIn( - "**Overall: 1 regressed · 1 improved · 13 no clear change · 1 incomplete · 16 unavailable.**", + "**Overall: 1 regressed · 1 improved · 13 no clear change · 1 incomplete · " + f"{len(UI_METRICS) + 1} unavailable.**", text, ) tables = text.split("
")[0] diff --git a/scripts/benchmarks/tests/test_catalog.py b/scripts/benchmarks/tests/test_catalog.py new file mode 100644 index 000000000..f6c953082 --- /dev/null +++ b/scripts/benchmarks/tests/test_catalog.py @@ -0,0 +1,227 @@ +from __future__ import annotations + +import io +import json +import tempfile +import time +import unittest +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import MagicMock, Mock, patch + +import ui +from schema import PHASE_METRICS, UI_METRIC_KEYS, Observation, ProcessMemory, Side +from ui import CatalogClient, check_scheduled_jobs, session_id, spawn_ledger_path, write_catalog_fixtures + +HELLO = { + "type": "daemon_hello", + "protocol": {"name": "prime-agent.daemon", "version": 7}, + "serverCapabilities": ["heartbeat_catalog"], +} + + +class CatalogFixtureTests(unittest.TestCase): + def setUp(self): + directory = tempfile.TemporaryDirectory() + self.addCleanup(directory.cleanup) + self.root = Path(directory.name) + self.agent = self.root / "agent" + for name, value in (("CATALOG_COUNT", 8), ("CATALOG_MESSAGES", 2), ("SCHEDULED_OWNERS", 2)): + patcher = patch.object(ui, name, value) + patcher.start() + self.addCleanup(patcher.stop) + + def test_sparse_paused_schedules_use_real_artifact_paths_and_ledger_edges(self): + cold, expected = write_catalog_fixtures(self.agent, self.root) + transcripts = list(self.agent.rglob("*.jsonl")) + ledger = spawn_ledger_path(self.agent, self.agent / "sessions") + self.assertEqual(len(transcripts), ui.CATALOG_COUNT + 1) + records = [json.loads(line) for line in ledger.read_text().splitlines()] + self.assertEqual(len(records) - 1, ui.CATALOG_COUNT - 2) + self.assertNotIn(str(cold), [record.get("child") for record in records]) + entries = [json.loads(line) for line in cold.read_text().splitlines()] + self.assertEqual(sum(entry["type"] == "message" for entry in entries), 2) + self.assertEqual(entries[0]["id"], cold.stem) + artifacts = list(self.agent.rglob("scheduled-jobs.json")) + self.assertEqual(len(artifacts), ui.SCHEDULED_OWNERS) + found = set() + for artifact in artifacts: + job = json.loads(artifact.read_text())["jobs"][0] + found.add(job["id"]) + self.assertEqual(job["status"], "paused") + self.assertNotIn("nextRunAt", job) + self.assertEqual(job["runCount"], 0) + path = Path(job["sessionFile"]) + self.assertEqual( + artifact.parent.resolve(), path.parent.parent / "session-artifacts" / job["sessionId"] + ) + self.assertIn(str(path), [record.get("child") for record in records]) + self.assertEqual(found, expected) + + def test_regeneration_removes_trial_mutations(self): + cold, _ = write_catalog_fixtures(self.agent, self.root) + before = { + str(p.relative_to(self.agent)): p.read_bytes() for p in self.agent.rglob("*") if p.is_file() + } + cold.write_text("mutated") + (self.agent / "sessions/stale.jsonl").write_text("stale") + write_catalog_fixtures(self.agent, self.root) + after = {str(p.relative_to(self.agent)): p.read_bytes() for p in self.agent.rglob("*") if p.is_file()} + self.assertEqual(before, after) + + def test_job_validation_rejects_fast_but_incomplete_or_changed_results(self): + write_catalog_fixtures(self.agent, self.root) + jobs = [json.loads(p.read_text())["jobs"][0] for p in self.agent.rglob("scheduled-jobs.json")] + data = { + "heartbeats": [ + { + "job": job, + "sessionName": ui.session_name("catalog", int(job["id"].split("-")[-1])), + "firstMessage": "task", + } + for job in jobs + ] + } + expected = {job["id"] for job in jobs} + check_scheduled_jobs(data, expected) + for items in ([], [data["heartbeats"][0]] * 2): + with self.assertRaises(RuntimeError): + check_scheduled_jobs({"heartbeats": items}, expected) + data["heartbeats"][0]["sessionName"] = "wrong owner" + with self.assertRaisesRegex(RuntimeError, "metadata"): + check_scheduled_jobs(data, expected) + data["heartbeats"][0]["job"]["status"] = "active" + with self.assertRaisesRegex(RuntimeError, "paused"): + check_scheduled_jobs(data, expected) + + +class CatalogProtocolTests(unittest.TestCase): + def client(self, *messages): + stream = io.BytesIO(b"".join((json.dumps(message) + "\n").encode() for message in messages)) + return CatalogClient(Mock(), stream) + + def test_out_of_order_scan_replies_retain_completion_times(self): + client = self.client( + HELLO, + {"type": "response", "id": "scan", "success": True, "data": {"heartbeats": []}}, + {"type": "response", "id": "open", "success": True, "data": {"sessionId": "cold"}}, + ) + with patch("ui.time.perf_counter", side_effect=range(1, 100)): + opened, completed = client.wait("open") + self.assertEqual(opened["sessionId"], "cold") + self.assertEqual(completed, 6) + data, completed = client.wait("scan") + self.assertEqual(data, {"heartbeats": []}) + self.assertEqual(completed, 3) + + def test_capability_mismatch_and_failed_scan_are_errors(self): + for hello in ( + HELLO | {"serverCapabilities": []}, + HELLO | {"protocol": {"name": "prime-agent.daemon", "version": 8}}, + ): + with self.assertRaisesRegex(RuntimeError, "protocol 7"): + self.client(hello) + client = self.client( + HELLO, {"type": "response", "id": "scan", "success": False, "error": "unreadable"} + ) + with self.assertRaisesRegex(RuntimeError, "unreadable"): + client.wait("scan") + + def test_wire_envelope_and_bounded_eof_failure(self): + client = self.client(HELLO) + client.send("scan-1", {"type": "heartbeats_list"}) + message = json.loads(client.channel.sendall.call_args.args[0]) + self.assertEqual(message["type"], "command") + self.assertEqual(message["protocol"], HELLO["protocol"]) + self.assertEqual(message["id"], message["command"]["id"]) + with self.assertRaisesRegex(RuntimeError, "closed"): + client.wait("scan-1") + with self.assertRaisesRegex(RuntimeError, "limit"): + CatalogClient(Mock(), io.BytesIO(b"x" * 2_000_001)) + + def test_metrics_are_required_for_trial_completion(self): + metrics = { + "scheduled_catalog", + "scheduled_catalog_cpu", + "scheduled_catalog_warm", + "scheduled_catalog_warm_cpu", + "cold_open_catalog", + "cold_open_catalog_cpu", + } + self.assertTrue(metrics <= UI_METRIC_KEYS) + self.assertTrue(metrics <= set(PHASE_METRICS["ui"])) + self.assertNotEqual(session_id("catalog", 0), session_id("catalog", ui.CATALOG_COUNT - 1)) + + +class CatalogTrialTests(unittest.TestCase): + def test_trial_launches_daemon_and_validates_competing_scans_before_recording(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + cold = root / "cold.jsonl" + terminal = Mock(bytes=0, started=time.perf_counter()) + client = Mock(responses={}) + client.send.side_effect = lambda *args: time.perf_counter() + jobs = { + "heartbeats": [ + { + "job": {"id": "catalog-job-1", "status": "paused", "runCount": 0}, + "sessionName": ui.session_name("catalog", 1), + "firstMessage": "task", + } + ] + } + + def reply(identifier): + if identifier == "residents": + data = {"sessions": []} + elif identifier == "cold-open": + data = {"sessionId": cold.stem, "workerState": "ready", "workerPid": 42} + else: + data = jobs + return data, time.perf_counter() + + client.wait.side_effect = reply + with ( + patch("ui.Terminal", return_value=terminal), + patch("ui.input_ready", side_effect=lambda *args: time.perf_counter()), + patch("ui.type_query"), + patch("ui.clear_search"), + patch("ui.expand_subagents"), + patch("ui.wait_for_roster", return_value=0.1), + patch("ui.write_fixtures"), + patch("ui.write_catalog_fixtures", return_value=(cold, {"catalog-job-1"})), + patch("ui.pwd.getpwnam", return_value=SimpleNamespace(pw_uid=123)), + patch("ui.process_stats", return_value=[ProcessMemory(pid=1, name="daemon", rss=100)]), + patch("worker.environment", return_value={}), + patch("worker.stop_processes") as stop, + patch("ui.socket.socket", return_value=MagicMock()), + patch("ui.subprocess.Popen") as launch, + patch("ui.CatalogClient", return_value=client), + ): + for incomplete in (False, True): + with self.subTest(incomplete=incomplete): + jobs["heartbeats"][0]["firstMessage"] = "" if incomplete else "task" + side = Side(sha="a" * 40, metrics={"install": [Observation(trial=0, value=1.0)]}) + ui.ui_measure(Mock(), side, 2, results=root, homes=root, user="benchmark1") + if incomplete: + self.assertIsNotNone(side.metrics["scheduled_catalog"][0].error) + self.assertNotIn("cold_open_catalog", side.metrics) + else: + for metric in PHASE_METRICS["ui"]: + self.assertIsNone(side.metrics[metric][0].error) + self.assertEqual(client.wait.call_args.args, ("scan-2",)) + self.assertEqual( + launch.call_args.args[0][-4:], + [ + "--mode", + "daemon", + "--daemon-socket", + "/tmp/prime-catalog-123-2.sock", + ], + ) + launch.return_value.wait.assert_called_with(timeout=10) + stop.assert_called_with("benchmark1") + + +if __name__ == "__main__": + unittest.main() diff --git a/scripts/benchmarks/ui.py b/scripts/benchmarks/ui.py index 9526e50ee..4a84fec6f 100644 --- a/scripts/benchmarks/ui.py +++ b/scripts/benchmarks/ui.py @@ -6,6 +6,8 @@ import os import pwd import re import shutil +import socket +import subprocess import time import uuid from dataclasses import dataclass @@ -469,6 +471,160 @@ def wait_for_roster(terminal: Terminal, *, minimum: int, settle: float = 3.0, ti raise TimeoutError("Agents-view roster did not settle") +CATALOG_COUNT = 2300 +CATALOG_MESSAGES = 64 +SCHEDULED_OWNERS = 13 + + +def write_catalog_fixtures( + agent_dir: Path, workspace: Path, *, uid: int | None = None +) -> tuple[Path, set[str]]: + """A large ledger with sparse paused schedules and an unrelated two-message cold target.""" + sessions = agent_dir / "sessions" + artifacts = agent_dir / "session-artifacts" + for directory in (sessions, artifacts, agent_dir / "session-leases"): + shutil.rmtree(directory, ignore_errors=True) + sessions.mkdir(parents=True) + parent_id = session_id("catalog", 0) + parent = sessions / f"{parent_id}.jsonl" + children = artifacts / parent_id / "children" + children.mkdir(parents=True) + records = [{"v": 1, "op": "meta", "at": _iso(BASE_TIMESTAMP_MS), "sessionsDir": str(sessions.resolve())}] + jobs = set() + for index in range(CATALOG_COUNT): + cold = index == CATALOG_COUNT - 1 + identifier = session_id("catalog", index) + path = (sessions if index == 0 or cold else children) / f"{identifier}.jsonl" + path.write_text( + "".join( + json.dumps(entry, separators=(",", ":")) + "\n" + for entry in _session_lines( + session_name("catalog", index), + identifier, + 1 if cold else CATALOG_MESSAGES, + workspace, + depth=0 if index == 0 or cold else 1, + tail=None, + ) + ) + ) + if index != 0 and not cold: + records.append( + { + "v": 1, + "op": "spawn", + "at": _iso(BASE_TIMESTAMP_MS), + "childId": identifier, + "parent": str(parent.resolve()), + "child": str(path.resolve()), + "depth": 1, + "name": session_name("catalog", index), + } + ) + if 1 <= index <= SCHEDULED_OWNERS: + job_id = f"catalog-job-{index}" + jobs.add(job_id) + artifact = path.parent.parent / "session-artifacts" / identifier + artifact.mkdir(parents=True) + job = { + "id": job_id, + "status": "paused", + "source": "heartbeat", + "runtimeKind": "subagent", + "activeSessionId": identifier, + "sessionId": identifier, + "sessionFile": str(path.resolve()), + "cwd": str(workspace), + "prompt": "benchmark fixture; must remain paused", + "runCount": 0, + "schedule": {"kind": "interval", "expression": "every 1h", "intervalMs": 3600000}, + "createdAt": _iso(BASE_TIMESTAMP_MS), + "updatedAt": _iso(BASE_TIMESTAMP_MS), + } + (artifact / "scheduled-jobs.json").write_text( + json.dumps({"jobs": [job], "dispatches": []}) + "\n" + ) + ledger = spawn_ledger_path(agent_dir, sessions) + ledger.parent.mkdir(parents=True, exist_ok=True) + ledger.write_text("".join(json.dumps(record) + "\n" for record in records)) + if uid is not None: + gid = pwd.getpwuid(uid).pw_gid + for directory in (sessions, artifacts, ledger.parent): + os.chown(directory, uid, gid) + for path in directory.rglob("*"): + os.chown(path, uid, gid) + return sessions / f"{session_id('catalog', CATALOG_COUNT - 1)}.jsonl", jobs + + +class CatalogClient: + """Small bounded JSONL client; retains out-of-order replies for concurrent scans and create.""" + + def __init__(self, channel: socket.socket, stream): + self.channel = channel + self.stream = stream + self.responses: dict[str, tuple[dict, float]] = {} + hello = self.receive() + if ( + hello.get("type") != "daemon_hello" + or hello.get("protocol", {}).get("name") != "prime-agent.daemon" + or hello["protocol"]["version"] != 7 + or "heartbeat_catalog" not in hello.get("serverCapabilities", []) + ): + raise RuntimeError("Catalog benchmark requires daemon protocol 7 and heartbeat_catalog") + self.protocol = hello["protocol"] + + def receive(self) -> dict: + line = self.stream.readline(2_000_001) + if not line or len(line) > 2_000_000 or not line.endswith(b"\n"): + raise RuntimeError("Daemon closed or exceeded the catalog response limit") + message = json.loads(line) + return message["event"] if message.get("type") == "event" else message + + def send(self, identifier: str, command: dict) -> float: + started = time.perf_counter() + self.channel.sendall( + ( + json.dumps( + { + "type": "command", + "id": identifier, + "protocol": self.protocol, + "clientId": "catalog-benchmark", + "command": {**command, "id": identifier}, + } + ) + + "\n" + ).encode() + ) + return started + + def wait(self, identifier: str) -> tuple[dict, float]: + deadline = time.perf_counter() + 120 + while identifier not in self.responses: + self.channel.settimeout(max(0.001, deadline - time.perf_counter())) + message = self.receive() + if message.get("type") == "response": + self.responses[message["id"]] = (message, time.perf_counter()) + if time.perf_counter() >= deadline: + raise TimeoutError(f"Timed out waiting for {identifier}") + message, finished = self.responses.pop(identifier) + if not message.get("success"): + raise RuntimeError(f"{identifier}: {message.get('error', 'daemon command failed')}") + return message["data"], finished + + +def check_scheduled_jobs(data: dict, expected: set[str]) -> None: + jobs = [item["job"] for item in data["heartbeats"]] + if len(jobs) != len(expected) or {job["id"] for job in jobs} != expected: + raise RuntimeError("Scheduled catalog omitted or duplicated fixture jobs") + if any(job["status"] != "paused" or job["runCount"] != 0 for job in jobs): + raise RuntimeError("Benchmark schedules must stay paused and never execute") + for item in data["heartbeats"]: + index = int(item["job"]["id"].removeprefix("catalog-job-")) + if item.get("sessionName") != session_name("catalog", index) or not item.get("firstMessage"): + raise RuntimeError("Scheduled catalog omitted owner display metadata") + + def ui_measure(request: Request, side: Side, trial: int, *, results: Path, homes: Path, user: str) -> None: """One UI-interaction trial: fresh fixtures, a cold resume, then the warm navigation scenario.""" from worker import clean_error, environment, record, stop_processes @@ -482,6 +638,7 @@ def ui_measure(request: Request, side: Side, trial: int, *, results: Path, homes spec = fixture_spec() details: dict[str, dict] = {} metric = "resume_large" + catalog_daemon = None def note(name: str, *, seconds: float, cpu: float, pty_bytes: int) -> None: details[name] = {"seconds": round(seconds, 4), "cpu": round(cpu, 4), "pty_bytes": pty_bytes} @@ -634,8 +791,97 @@ def ui_measure(request: Request, side: Side, trial: int, *, results: Path, homes ) finally: terminal.close() + # A separate stopped-process workload isolates sparse schedule scans and cold-worker queueing. + metric = "scheduled_catalog" + stop_processes(user) + cold_path, expected_jobs = write_catalog_fixtures(agent_dir, workspace, uid=uid) + socket_path = Path(f"/tmp/prime-catalog-{uid}-{trial}.sock") + with (results / f"catalog-daemon-{trial}.log").open("w") as log: + catalog_daemon = subprocess.Popen( + [ + runuser, + "-u", + user, + "--", + "prime-agent", + "--mode", + "daemon", + "--daemon-socket", + str(socket_path), + ], + cwd=workspace, + env=env, + stdout=log, + stderr=log, + ) + with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as channel: + channel.settimeout(120) + deadline = time.perf_counter() + 120 + while True: + try: + channel.connect(str(socket_path)) + break + except (FileNotFoundError, ConnectionRefusedError): + if catalog_daemon.poll() is not None: + raise RuntimeError("Catalog daemon exited during startup") from None + if time.perf_counter() >= deadline: + raise TimeoutError("Catalog daemon did not start") from None + time.sleep(0.02) + with channel.makefile("rb") as stream: + client = CatalogClient(channel, stream) + for metric in ("scheduled_catalog", "scheduled_catalog_warm"): + cpu_start = cpu_total() + started = client.send(metric, {"type": "heartbeats_list"}) + data, finished = client.wait(metric) + cpu = cpu_total() - cpu_start + check_scheduled_jobs(data, expected_jobs) + record(side, metric, trial, finished - started) + record(side, metric + "_cpu", trial, cpu) + details[metric] = {"seconds": finished - started, "cpu": cpu, "jobs": len(expected_jobs)} + + metric = "cold_open_catalog" + client.send("residents", {"type": "list"}) + residents, _ = client.wait("residents") + cold_id = cold_path.stem + if any(item["sessionId"] == cold_id for item in residents["sessions"]): + raise RuntimeError("Catalog cold target already has a resident worker") + cpu_start = cpu_total() + for index in range(3): + client.send(f"scan-{index}", {"type": "heartbeats_list"}) + # Let read-only handlers enter their scans before registering a new worker. + time.sleep(0.02) + started = client.send( + "cold-open", + { + "type": "create", + "sessionPath": str(cold_path), + "config": {"cwd": str(workspace), "agentDir": str(agent_dir)}, + "launchEnv": env, + }, + ) + summary, finished = client.wait("cold-open") + cpu = cpu_total() - cpu_start + if summary.get("sessionId") != cold_id or summary.get("workerState") != "ready": + raise RuntimeError("Cold worker did not return the expected ready session") + replies_before_open = len(client.responses) + # A failed competing scan is a failed sample, even when the worker opens quickly. + for index in range(3): + data, _ = client.wait(f"scan-{index}") + check_scheduled_jobs(data, expected_jobs) + record(side, metric, trial, finished - started) + record(side, metric + "_cpu", trial, cpu) + details[metric] = { + "seconds": finished - started, + "cpu": cpu, + "scan_replies_before_open": replies_before_open, + "concurrent_scans": 3, + "worker_pid": summary.get("workerPid"), + "sessions": CATALOG_COUNT, + } except Exception as error: record(side, metric, trial, error=clean_error(error)) # type: ignore[arg-type] finally: stop_processes(user) + if catalog_daemon is not None: + catalog_daemon.wait(timeout=10) (results / f"ui-{trial}.json").write_text(json.dumps(details, indent=2, sort_keys=True) + "\n")