from __future__ import annotations
import hashlib
import json
import os
import pwd
import re
import shutil
import socket
import subprocess
import time
import uuid
from dataclasses import dataclass
from datetime import UTC, datetime
from pathlib import Path
from schema import ProcessMemory, Request, Side
from terminal import Terminal
# Fixture shape: "many sessions, many large sessions" plus a deep subagent chain.
# Scale mirrors real long-lived installs: hundreds of saved sessions, one very
# large transcript, and a spawn ledger with enough edges to stress hydration.
MEDIUM_COUNT = 150
MEDIUM_MESSAGES = 120
LARGE_COUNT = 3
LARGE_MESSAGES = 2000
HUGE_MESSAGES = 16000
FANOUT_COUNT = 40
FANOUT_MESSAGES = 20
SUBAGENT_DEPTH = 6
SUBAGENT_MESSAGES = 400
BASE_TIMESTAMP_MS = 946684800000 # 2000-01-01T00:00:00Z; fixed so fixtures are deterministic
TOOL_OUTPUT_LINES = 60
SEARCH_PLACEHOLDER = "Search sessions"
AGENTS_VIEW_FOOTER = "Ctrl+N new"
ROSTER_COUNT = re.compile(r"agents\s+(\d+) running, (\d+) idle, (\d+) inactive")
LEFT_ARROW = "\x1b[D"
RIGHT_ARROW = "\x1b[C"
DOWN_ARROW = "\x1b[B"
EXPAND_ARROW = "\x1b[1;3C"
BACKSPACE = "\x7f"
ENTER = "\r"
FIXTURE_NAMESPACE = uuid.UUID("b0d5f4a1-6d64-4bf8-9d7a-2f4f0a9c1e10")
def session_id(kind: str, index: int) -> str:
return str(uuid.uuid5(FIXTURE_NAMESPACE, f"prime-agent-ui-bench/{kind}/{index}"))
def session_name(kind: str, index: int) -> str:
return f"ui-bench-{kind}-{index:02d}"
def tail_marker(kind: str, index: int) -> str:
return f"ui-bench-tail-{kind}-{index:02d}"
@dataclass(frozen=True)
class FixtureSpec:
"""Session ids the probe addresses; every id is derived deterministically."""
large: list[str]
medium: list[str]
root: str
subagents: list[str]
fanout: list[str]
@property
def resume_id(self) -> str:
"""Cold-resume target; never touched by the warm scenario."""
return self.large[1]
@property
def switch_id(self) -> str:
"""Warm in-session /resume target; the very large transcript."""
return self.large[2]
@property
def open_id(self) -> str:
"""Search-and-open target from the agents view."""
return self.large[0]
def _iso(ms: int) -> str:
return datetime.fromtimestamp(ms / 1000, tz=UTC).isoformat(timespec="milliseconds").replace("+00:00", "Z")
def _session_lines(
name: str,
identifier: str,
count: int,
cwd: Path,
*,
depth: int,
tail: str | None,
) -> list[dict]:
"""One deterministic session transcript: header, name, model, then a realistic message mix."""
timestamp = BASE_TIMESTAMP_MS
parent: str | None = None
lines: list[dict] = []
prefix = identifier.replace("-", "")[:8]
def link(entry_id: str) -> dict:
nonlocal parent, timestamp
timestamp += 1000
entry = {"id": entry_id, "parentId": parent, "timestamp": _iso(timestamp)}
parent = entry_id
return entry
lines.append(
{
"type": "session",
"version": 3,
"id": identifier,
"timestamp": _iso(timestamp),
"cwd": str(cwd),
"rlmDepth": depth,
}
)
entry = link(f"{prefix}n")
entry.update({"type": "session_info", "name": name})
lines.append(entry)
entry = link(f"{prefix}m")
entry.update({"type": "model_change", "provider": "prime-inference", "modelId": "internal/glm-5.3-fast"})
lines.append(entry)
usage = {
"input": 100,
"output": 50,
"cacheRead": 0,
"cacheWrite": 0,
"totalTokens": 150,
"cost": {"input": 0.0001, "output": 0.0001, "cacheRead": 0, "cacheWrite": 0, "total": 0.0002},
}
for index in range(count):
final = index == count - 1
entry = link(f"{prefix}u{index}")
entry.update(
{
"type": "message",
"message": {
"role": "user",
"timestamp": 0,
"content": (
f"Please continue task {name} step {index}: review the module, run the checks, "
"and summarize findings for the migration notes."
),
},
}
)
lines.append(entry)
text = (
f"Working on {name} step {index}. The parser handles nested records correctly; the next edit "
"keeps the schema stable while trimming the duplicated branch."
)
if final and tail:
text = f"Finished all steps. {tail}"
content = [
{
"type": "thinking",
"thinking": (
f"Step {index} for {name}: check the invariants, then update the affected call "
"sites before running the full suite again to confirm no behavior changed."
),
},
{"type": "text", "text": text},
]
if not final:
content.append(
{"type": "toolCall", "id": f"call-{index}", "name": "ipython", "arguments": {"code": "pass"}}
)
entry = link(f"{prefix}a{index}")
entry.update(
{
"type": "message",
"message": {
"role": "assistant",
"timestamp": 0,
"content": content,
"api": "responses",
"provider": "prime-inference",
"model": "internal/glm-5.3-fast",
"usage": usage,
"stopReason": "stop" if final else "toolUse",
},
}
)
lines.append(entry)
if not final:
entry = link(f"{prefix}t{index}")
entry.update(
{
"type": "message",
"message": {
"role": "toolResult",
"toolCallId": f"call-{index}",
"toolName": "ipython",
"content": [{"type": "text", "text": "all checks passed\n" * TOOL_OUTPUT_LINES}],
"isError": False,
"timestamp": 0,
},
}
)
lines.append(entry)
return lines
def spawn_ledger_path(agent_dir: Path, sessions_dir: Path) -> Path:
canonical = Path(os.path.realpath(sessions_dir))
digest = hashlib.sha256(str(canonical).encode()).hexdigest()[:16]
return agent_dir / "rlm-ledger" / f"{digest}.jsonl"
def write_fixtures(agent_dir: Path, workspace: Path, spec: FixtureSpec, *, uid: int | None = None) -> None:
"""Regenerate every session file and the spawn ledger from scratch for one trial.
The harness runs as root; pass the benchmark user's ids so the spawned CLI can
read and append its own sessions.
"""
sessions_dir = agent_dir / "sessions"
for directory in (sessions_dir, agent_dir / "session-artifacts", agent_dir / "session-leases"):
shutil.rmtree(directory, ignore_errors=True)
sessions_dir.mkdir(parents=True, exist_ok=True)
def write(identifier: str, name: str, count: int, *, depth: int = 0, tail: str | None = None) -> Path:
path = sessions_dir / f"{identifier}.jsonl"
path.write_text(
"".join(
json.dumps(line, separators=(",", ":")) + "\n"
for line in _session_lines(name, identifier, count, workspace, depth=depth, tail=tail)
)
)
return path
for index in range(1, LARGE_COUNT + 1):
identifier = spec.large[index - 1]
messages = HUGE_MESSAGES if index == 3 else LARGE_MESSAGES
write(identifier, session_name("large", index), messages, tail=tail_marker("large", index))
for index in range(1, MEDIUM_COUNT + 1):
identifier = spec.medium[index - 1]
write(identifier, session_name("medium", index), MEDIUM_MESSAGES)
parent = write(spec.root, session_name("root", 0), SUBAGENT_MESSAGES, tail=tail_marker("root", 0))
edges = []
for fanout in range(1, FANOUT_COUNT + 1):
identifier = spec.fanout[fanout - 1]
path = write(identifier, session_name("fan", fanout), FANOUT_MESSAGES, depth=1)
edges.append(
{
"childId": identifier,
"parent": str(parent),
"child": str(path),
"depth": 1,
"name": session_name("fan", fanout),
}
)
for depth in range(1, SUBAGENT_DEPTH + 1):
identifier = spec.subagents[depth - 1]
path = write(
identifier,
session_name("sub", depth),
SUBAGENT_MESSAGES,
depth=depth,
tail=tail_marker("sub", depth),
)
edges.append(
{
"childId": identifier,
"parent": str(parent),
"child": str(path),
"depth": depth,
"name": session_name("sub", depth),
}
)
parent = path
ledger = spawn_ledger_path(agent_dir, sessions_dir)
ledger.parent.mkdir(parents=True, exist_ok=True)
records = [
{
"v": 1,
"op": "meta",
"at": _iso(BASE_TIMESTAMP_MS),
"sessionsDir": str(Path(os.path.realpath(sessions_dir))),
}
]
records.extend(
{"v": 1, "op": "spawn", "at": _iso(BASE_TIMESTAMP_MS + edge["depth"]), **edge} for edge in edges
)
ledger.write_text("".join(json.dumps(record, separators=(",", ":")) + "\n" for record in records))
if uid is not None:
gid = pwd.getpwuid(uid).pw_gid
os.chown(agent_dir, uid, gid)
for path in agent_dir.rglob("*"):
os.chown(path, uid, gid)
def fixture_spec() -> FixtureSpec:
return FixtureSpec(
large=[session_id("large", index) for index in range(1, LARGE_COUNT + 1)],
medium=[session_id("medium", index) for index in range(1, MEDIUM_COUNT + 1)],
root=session_id("root", 0),
subagents=[session_id("sub", depth) for depth in range(1, SUBAGENT_DEPTH + 1)],
fanout=[session_id("fan", index) for index in range(1, FANOUT_COUNT + 1)],
)
def process_stats(uid: int) -> list[ProcessMemory]:
"""Snapshot every live process of one user with RSS, PSS, and cumulative CPU time."""
stats: list[ProcessMemory] = []
try:
clock_ticks = os.sysconf("SC_CLK_TCK")
except (ValueError, AttributeError, OSError):
return stats
for status_path in Path("/proc").glob("[0-9]*/status"):
try:
fields = dict(line.split(":", 1) for line in status_path.read_text().splitlines() if ":" in line)
if int(fields["Uid"].split()[0]) != uid or fields["State"].strip().startswith("Z"):
continue
stat_fields = (status_path.parent / "stat").read_text().rsplit(")", 1)[1].split()
cpu_seconds = (int(stat_fields[11]) + int(stat_fields[12])) / clock_ticks
pss = None
try:
for line in (status_path.parent / "smaps_rollup").read_text().splitlines():
if line.startswith("Pss:"):
pss = int(line.split()[1]) * 1024
except (PermissionError, FileNotFoundError, ProcessLookupError):
pass
stats.append(
ProcessMemory(
pid=int(status_path.parent.name),
name=fields["Name"].strip(),
rss=int(fields.get("VmRSS", "0").split()[0]) * 1024,
pss=pss,
cpu=cpu_seconds,
)
)
except (PermissionError, FileNotFoundError, ProcessLookupError, ValueError, IndexError):
continue
return stats
def total_cpu(stats: list[ProcessMemory]) -> float:
return sum(process.cpu or 0.0 for process in stats)
def cpu_delta(start: float, total: float) -> float:
"""CPU seconds consumed since ``start``.
``total`` sums the live process tree, so a child that exits between the two
samples takes its CPU time with it and the raw difference can dip slightly
below zero. That is measurement noise, not negative work, and the report
schema rejects negative observations, so clamp at zero.
"""
return max(0.0, total - start)
def input_ready(terminal: Terminal, marker: str, *, timeout: float = 120) -> float:
"""Return the first successful editor-echo timestamp after the transcript tail appears."""
terminal.until(lambda display: marker in display.text(), timeout)
return terminal.started + terminal.ready(timeout)
def clear_search(terminal: Terminal, *, attempts: int = 5) -> None:
"""Empty the agents-view search box; keystrokes can be eaten by re-entry renders."""
for _ in range(attempts):
terminal.child.send(BACKSPACE * 24)
try:
terminal.until(lambda display: SEARCH_PLACEHOLDER in display.text(), 5)
return
except TimeoutError:
continue
raise TimeoutError("Agents-view search box did not clear")
def open_agents_view(terminal: Terminal, *, clear: bool, timeout: float = 60) -> None:
"""Open a freshly rendered agents view, then optionally clear its preserved search."""
terminal.child.send(LEFT_ARROW)
terminal.until_output(lambda output: AGENTS_VIEW_FOOTER in output, timeout)
if clear:
clear_search(terminal)
def type_query(terminal: Terminal, query: str, *, attempts: int = 3) -> None:
"""Type a search query and verify it echoed before filtering on it."""
for _ in range(attempts):
clear_search(terminal)
terminal.child.send(query)
try:
terminal.until(lambda display: query in display.text(), 8)
return
except TimeoutError:
continue
raise TimeoutError("Agents-view search query did not echo")
def move_down(terminal: Terminal, steps: int) -> None:
"""Move through a deterministic fixture tree without coalescing keypresses."""
for _ in range(steps):
terminal.child.send(DOWN_ARROW)
terminal.settle(0.08)
def expand_subagents(terminal: Terminal, *, attempts: int = 4) -> None:
"""Expand the selected row's subagent list; the toggle retries to survive eaten keystrokes."""
expanded = terminal.display.text().count("▾")
for _ in range(attempts):
terminal.child.send(EXPAND_ARROW)
terminal.settle(1.0)
if terminal.display.text().count("▾") > expanded:
return
raise TimeoutError("Selected agent subagents did not expand")
def roster_inactive(display) -> int | None: # type: ignore[no-untyped-def]
for line in display.text().splitlines():
match = ROSTER_COUNT.search(line)
if match:
return int(match.group(3))
return None
def expected_roster_inactive() -> int:
"""Saved top-level fixture sessions a hydrated roster must list.
Large, medium, and the chain-root fixtures are top-level roster rows, but
the warm scenario holds the switch target live; nested fan-out and
subagent-chain files stay out of the top-level count either way.
"""
top_level = LARGE_COUNT + MEDIUM_COUNT + 1
return top_level - 1
def wait_for_roster(terminal: Terminal, *, minimum: int, settle: float = 3.0, timeout: float = 90) -> float:
"""Wait until the roster lists `minimum` inactive sessions and stops growing.
Saved sessions stream in, but an empty or stalled roster must not count
as settled just because its count stopped moving.
"""
started = time.perf_counter()
last_count: int | None = None
last_change = time.perf_counter()
while time.perf_counter() - started < timeout:
terminal.settle(0.4)
count = roster_inactive(terminal.display)
if (
count is not None
and count == last_count
and count >= minimum
and time.perf_counter() - last_change >= settle
):
return time.perf_counter() - started
if count != last_count:
last_count = count
last_change = time.perf_counter()
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
if not any(s.value is not None and s.trial == 0 for s in side.metrics.get("install", [])):
raise RuntimeError("The first installation must succeed before interactive measurements")
home = homes / user
workspace = home / "workspace"
agent_dir = home / ".prime/agent"
uid = pwd.getpwnam(user).pw_uid
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}
def cpu_total() -> float:
return total_cpu(process_stats(uid))
stop_processes(user)
write_fixtures(agent_dir, workspace, spec, uid=uid)
env = environment(user)
runuser = "/usr/sbin/runuser"
try:
# Cold resume of a large session from the CLI, from process spawn to usable editor.
cpu_start = cpu_total()
terminal = Terminal(
[runuser, "-u", user, "--", "prime-agent", "--resume", spec.resume_id],
workspace,
env,
results / f"ui-resume-{trial}",
)
try:
metric = "resume_large"
bytes_start = terminal.bytes
input_ready(terminal, tail_marker("large", 2))
elapsed = time.perf_counter() - terminal.started
cpu = cpu_delta(cpu_start, cpu_total())
record(side, "resume_large", trial, elapsed) # type: ignore[arg-type]
record(side, "resume_large_cpu", trial, cpu) # type: ignore[arg-type]
note("resume_large", seconds=elapsed, cpu=cpu, pty_bytes=terminal.bytes - bytes_start)
finally:
terminal.close()
stop_processes(user)
# Warm navigation scenario in a single TUI process.
metric = "switch_large"
terminal = Terminal(
[runuser, "-u", user, "--", "prime-agent"], workspace, env, results / f"ui-scenario-{trial}"
)
try:
terminal.ready()
# Warm switch into a different large session from the running editor.
started = time.perf_counter()
cpu_start = cpu_total()
bytes_start = terminal.bytes
terminal.child.send(f"/resume {spec.switch_id}{ENTER}")
input_ready(terminal, tail_marker("large", 3))
elapsed = time.perf_counter() - started
cpu = cpu_delta(cpu_start, cpu_total())
record(side, "switch_large", trial, elapsed) # type: ignore[arg-type]
record(side, "switch_large_cpu", trial, cpu) # type: ignore[arg-type]
note("switch_large", seconds=elapsed, cpu=cpu, pty_bytes=terminal.bytes - bytes_start)
# Session -> agents view: keystroke to rendered roster splash.
metric = "agents_view"
started = time.perf_counter()
cpu_start = cpu_total()
bytes_start = terminal.bytes
open_agents_view(terminal, clear=False)
elapsed = time.perf_counter() - started
cpu = cpu_delta(cpu_start, cpu_total())
record(side, "agents_view", trial, elapsed) # type: ignore[arg-type]
record(side, "agents_view_cpu", trial, cpu) # type: ignore[arg-type]
note("agents_view", seconds=elapsed, cpu=cpu, pty_bytes=terminal.bytes - bytes_start)
# Full roster with many sessions: saved sessions and ledger children stream in.
metric = "agents_roster"
cpu_start = cpu_total()
bytes_start = terminal.bytes
roster_seconds = wait_for_roster(terminal, minimum=expected_roster_inactive())
cpu = cpu_delta(cpu_start, cpu_total())
record(side, "agents_roster", trial, roster_seconds) # type: ignore[arg-type]
record(side, "agents_roster_cpu", trial, cpu) # type: ignore[arg-type]
note("agents_roster", seconds=roster_seconds, cpu=cpu, pty_bytes=terminal.bytes - bytes_start)
# Agents view -> another large session, found by its unique id prefix.
metric = "agents_open"
started = time.perf_counter()
cpu_start = cpu_total()
bytes_start = terminal.bytes
type_query(terminal, spec.open_id[:8])
terminal.until(lambda display: session_name("large", 1) in display.text(), 60)
terminal.settle(0.8)
terminal.child.send(RIGHT_ARROW)
input_ready(terminal, tail_marker("large", 1))
elapsed = time.perf_counter() - started
cpu = cpu_delta(cpu_start, cpu_total())
record(side, "agents_open", trial, elapsed) # type: ignore[arg-type]
record(side, "agents_open_cpu", trial, cpu) # type: ignore[arg-type]
note("agents_open", seconds=elapsed, cpu=cpu, pty_bytes=terminal.bytes - bytes_start)
# Reattach to the worker just opened, excluding search/navigation setup from timing.
metric = "agents_reopen"
open_agents_view(terminal, clear=True)
type_query(terminal, spec.open_id[:8])
terminal.until(lambda display: session_name("large", 1) in display.text(), 60)
terminal.settle(0.8)
started = time.perf_counter()
cpu_start = cpu_total()
bytes_start = terminal.bytes
terminal.child.send(RIGHT_ARROW)
ready_at = input_ready(terminal, tail_marker("large", 1))
elapsed = ready_at - started
cpu = cpu_delta(cpu_start, cpu_total())
record(side, "agents_reopen", trial, elapsed)
record(side, "agents_reopen_cpu", trial, cpu)
note("agents_reopen", seconds=elapsed, cpu=cpu, pty_bytes=terminal.bytes - bytes_start)
# Chain parent: open the root so the deepest subagent opens against a live parent.
metric = "parent_open"
started = time.perf_counter()
cpu_start = cpu_total()
bytes_start = terminal.bytes
open_agents_view(terminal, clear=True)
terminal.settle(1.0)
type_query(terminal, spec.root[:8])
terminal.until(lambda display: session_name("root", 0) in display.text(), 60)
terminal.settle(0.5)
terminal.child.send(RIGHT_ARROW)
input_ready(terminal, tail_marker("root", 0))
elapsed = time.perf_counter() - started
cpu = cpu_delta(cpu_start, cpu_total())
record(side, "parent_open", trial, elapsed) # type: ignore[arg-type]
record(side, "parent_open_cpu", trial, cpu) # type: ignore[arg-type]
note("parent_open", seconds=elapsed, cpu=cpu, pty_bytes=terminal.bytes - bytes_start)
# Deepest subagent session: expand the deterministic fixture tree and
# drill to depth SUBAGENT_DEPTH while its parent session is live.
metric = "subagent_open"
started = time.perf_counter()
cpu_start = cpu_total()
bytes_start = terminal.bytes
open_agents_view(terminal, clear=True)
terminal.settle(1.0)
type_query(terminal, spec.root[:8])
terminal.until(lambda display: session_name("root", 0) in display.text(), 60)
clear_search(terminal)
expand_subagents(terminal)
for depth in range(1, SUBAGENT_DEPTH + 1):
# Filtering keeps the selected parent visible and removes its
# unrelated siblings. Move across its summary row to the child,
# then clear search before the next expansion key.
type_query(terminal, spec.subagents[depth - 1][:8])
terminal.settle(0.5)
move_down(terminal, 2)
clear_search(terminal)
if depth < SUBAGENT_DEPTH:
expand_subagents(terminal)
terminal.child.send(RIGHT_ARROW)
input_ready(terminal, tail_marker("sub", SUBAGENT_DEPTH))
elapsed = time.perf_counter() - started
cpu = cpu_delta(cpu_start, cpu_total())
record(side, "subagent_open", trial, elapsed) # type: ignore[arg-type]
record(side, "subagent_open_cpu", trial, cpu) # type: ignore[arg-type]
note("subagent_open", seconds=elapsed, cpu=cpu, pty_bytes=terminal.bytes - bytes_start)
# Whole-tree memory after the interactions.
metric = "ui_rss"
terminal.settle(1)
processes = process_stats(uid)
if not processes:
raise RuntimeError("No owned processes found for the UI memory measurement")
record(side, "ui_rss", trial, sum(process.rss for process in processes)) # type: ignore[arg-type]
side.processes = processes
(results / f"ui-memory-{trial}.json").write_text(
json.dumps([process.model_dump() for process in processes], indent=2) + "\n"
)
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_delta(cpu_start, cpu_total())
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_delta(cpu_start, cpu_total())
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")