refactor(benchmarks): exclude fork DELETE from the timed span

The fork journey deleted each fork inline inside measure, folding the DELETE into the timed op. Collect fork ids in the journey context and delete them in teardown instead, so only the fork POST is measured. Co-authored-by: Isaac
feat(benchmarks): add fork, comment, and runner-file-read journeys
2026-07-09 16:27:36 +08:00 · 2026-07-09 16:20:21 +08:00
4 changed files with 200 additions and 7 deletions
@@ -57,6 +57,8 @@ latency run), `--requests N` / `--concurrency N` (throughput), `--runs N`,
 | `get_session` | `GET /v1/sessions/{id}` — single-session snapshot | (O(1)) |
 | `load_conversation_history` | `GET /v1/sessions/{id}/items` — history read | items/session |
 | `search_sessions` | `GET /v1/sessions?search_query=` — unindexed `LIKE` | total item count |
+| `fork_session` | `POST /v1/sessions/{id}/fork` — fork (deep-copy items); forks deleted in teardown, untimed | items/session |
+| `add_comment` | `POST /v1/sessions/{id}/comments` — create a review comment | write path |

 Read journeys target a **pre-seeded** session when the DB has a corpus; against
 an empty DB they self-seed a small fallback session over HTTP (the
@@ -83,6 +85,12 @@ drift negligible (~2 ms/turn).
 | `warm_turn` | Drive a turn on an already-warm session — steady-state dispatch overhead |
 | `time_to_first_token` | Post a turn; time to the first streamed `output_text` delta |
 | `interrupt` | Interrupt a running (gated) turn; time to cancellation |
+| `read_runner_file` | `GET .../environments/default/filesystem/{path}` — server → runner filesystem read proxy |
+
+`read_runner_file` needs a runner but does **not** drive a turn or call the LLM:
+its setup plants a file via `PUT`, and the timed op is the proxied read (a
+localhost round-trip). Being far cheaper than a turn, it uses a higher iteration
+cap (50) than the full-turn journeys.

 **Only measure what we control.** Full-turn journeys always use the
 **`openai-agents`** SDK harness, which runs **in-process** (a call into the
@@ -212,6 +220,14 @@ seeding.

 ## Follow-ups

+- **Subagent spawn.** A planned full-turn journey (`needs_runner=True`): the
+  parent agent emits a `sys_session_send` tool call, the runner dispatches a
+  child session, and the parent auto-wakes with the collected result. It's
+  fully mockable with the zero-latency mock LLM (no real model) — script the
+  parent's queue to emit the tool call and the child's queue to return a short
+  reply, then poll for the child's marker. It needs the parent bundle to declare
+  a sub-agent under `tools:` (extend `_agent_bundle`); the pattern is in
+  `tests/e2e/test_coder_subagent.py`.
 - **Excluded journeys** (agent-behaviour-dependent, deliberately not measured):
  multi-turn and tool-calling turns (dominated by the agent's own choices) and
  large-history turns (the O(N) `history_to_input_items` conversion is real app
@@ -265,6 +265,11 @@ class BenchEnvironment:
    def _spawn_runner(
        self, base_env: dict[str, str], binding_token: str
    ) -> subprocess.Popen[bytes]:
+        # Point the runner's filesystem workspace at the temp dir so file
+        # writes (e.g. read_runner_file's setup) land there and are cleaned up
+        # on teardown, rather than in the launch cwd (its default).
+        workspace = self._tmp / "workspace"
+        workspace.mkdir(exist_ok=True)
        runner_env = apply_runner_env(
            {
                **base_env,
@@ -272,6 +277,7 @@ class BenchEnvironment:
                "OMNIGENT_RUNNER_TUNNEL_BINDING_TOKEN": binding_token,
                "OMNIGENT_RUNNER_PARENT_PID": str(os.getpid()),
                "RUNNER_SERVER_URL": self.base_url,
+                "OMNIGENT_RUNNER_WORKSPACE": str(workspace),
            }
        )
        return subprocess.Popen(
@@ -360,6 +366,12 @@ class BenchEnvironment:
            "model": self.model,
            "config": {"harness": self.harness},
        }
+        config: dict[str, object] = {
+            "spec_version": 1,
+            "name": name,
+            "prompt": "You are a helpful assistant used for performance benchmarking.",
+            "executor": executor,
+        }
        if self.with_runner:
            executor["auth"] = {
                "type": "api_key",
@@ -367,12 +379,15 @@ class BenchEnvironment:
                "base_url": f"{self.mock_url}/v1",
            }
            executor["connection"] = {"base_url": f"{self.mock_url}/v1", "api_key": "mock-key"}
-        config: dict[str, object] = {
-            "spec_version": 1,
-            "name": name,
-            "prompt": "You are a helpful assistant used for performance benchmarking.",
-            "executor": executor,
-        }
+            # A filesystem env so the runner can serve the resource endpoints
+            # (read_runner_file). Without os_env the runner has no primary
+            # environment to materialize and the filesystem proxy 404s.
+            # sandbox.type=none avoids needing a bwrap binary on the host.
+            config["os_env"] = {
+                "type": "caller_process",
+                "cwd": ".",
+                "sandbox": {"type": "none"},
+            }
        buf = io.BytesIO()
        with tarfile.open(fileobj=buf, mode="w:gz") as tar:
            payload = yaml.safe_dump(config).encode()
@@ -452,6 +467,40 @@ class BenchEnvironment:
        bound.raise_for_status()
        return session_id

+    async def write_runner_file(self, session_id: str, relative_path: str, content: str) -> None:
+        """Write a file into the runner's default environment over HTTP.
+
+        The server proxies the ``PUT`` to the bound runner, which writes to its
+        sandboxed filesystem — so this needs a runner. Used to plant a file the
+        read journey can then fetch back.
+
+        :raises RuntimeError: If not in runner mode.
+        """
+        assert self.client is not None
+        if not self.with_runner:
+            raise RuntimeError("write_runner_file requires with_runner=True")
+        resp = await self.client.put(
+            f"/v1/sessions/{session_id}/resources/environments/default/filesystem/{relative_path}",
+            json={"content": content, "encoding": "utf-8"},
+        )
+        resp.raise_for_status()
+
+    async def read_runner_file(self, session_id: str, relative_path: str) -> None:
+        """Read a file from the runner's default environment over HTTP.
+
+        Times the server → runner filesystem proxy (a localhost round-trip); no
+        LLM is involved. Requires a runner — the server returns 502 without one.
+
+        :raises RuntimeError: If not in runner mode.
+        """
+        assert self.client is not None
+        if not self.with_runner:
+            raise RuntimeError("read_runner_file requires with_runner=True")
+        resp = await self.client.get(
+            f"/v1/sessions/{session_id}/resources/environments/default/filesystem/{relative_path}",
+        )
+        resp.raise_for_status()
+
    async def drive_turn(
        self, session_id: str, text: str, *, timeout: float = _TURN_TIMEOUT_S
    ) -> None:
@@ -13,6 +13,11 @@ v1 journeys are pure HTTP/API (server + DB, no runner, no LLM):
 - ``get_session`` — single-session snapshot load.
 - ``load_conversation_history`` — history read, seeded runner-free via
  ``external_conversation_item`` (see :meth:`BenchEnvironment.seed_items`).
+- ``fork_session`` — fork a session (deep-copy its items), then DELETE.
+- ``add_comment`` — create a review comment on a file (DB write).
+
+``read_runner_file`` needs a runner but no LLM turn: it plants a file in the
+runner environment (setup) and times the server → runner filesystem read proxy.

 The framework (``Journey`` + the two runners) is harness-agnostic and reused
 verbatim by phase-2 full-turn journeys.
@@ -255,6 +260,67 @@ async def _measure_load_history(env: BenchEnvironment, ctx: JourneyContext) -> N
    resp.raise_for_status()


+@dataclass
+class _ForkContext:
+    """Fork-journey context: the session to fork + the forks to clean up.
+
+    ``measure`` records each fork's id here instead of deleting it inline, so
+    the DELETE stays out of the timed span; ``teardown`` removes them after.
+    """
+
+    source_id: str
+    fork_ids: list[str]
+
+
+async def _setup_fork_session(env: BenchEnvironment) -> _ForkContext:
+    """Resolve a session to fork; start an empty fork-id collector."""
+    source_id = await _setup_target_session(env)
+    return _ForkContext(source_id=source_id, fork_ids=[])
+
+
+async def _measure_fork_session(env: BenchEnvironment, ctx: JourneyContext) -> None:
+    assert env.client is not None
+    fork_ctx = cast(_ForkContext, ctx)  # _setup_fork_session
+    forked = await env.client.post(f"/v1/sessions/{fork_ctx.source_id}/fork", json={})
+    forked.raise_for_status()
+    # Record the fork for teardown; deleting it here would fold the DELETE into
+    # the timed span. The fork POST (a deep-copy of the source's items) is the
+    # operation of interest.
+    fork_ctx.fork_ids.append(forked.json()["id"])
+
+
+async def _teardown_fork_session(env: BenchEnvironment, ctx: JourneyContext) -> None:
+    """Delete every fork created during the run (best effort, untimed)."""
+    assert env.client is not None
+    fork_ctx = cast(_ForkContext, ctx)
+    for fork_id in fork_ctx.fork_ids:
+        with contextlib.suppress(httpx.HTTPError):
+            await env.client.delete(f"/v1/sessions/{fork_id}")
+
+
+# Anchor snapshot for the comment journey; the offsets below span it.
+_COMMENT_ANCHOR = "benchmark"
+
+
+async def _measure_add_comment(env: BenchEnvironment, ctx: JourneyContext) -> None:
+    assert env.client is not None
+    session_id = cast(str, ctx)  # _setup_target_session
+    # Each POST creates an independent comment row. Unlike sessions, an
+    # accumulating comment skews no measured read path, so there's no cleanup.
+    # The file need not exist — the handler stores the path + offsets + body.
+    resp = await env.client.post(
+        f"/v1/sessions/{session_id}/comments",
+        json={
+            "path": "bench_target.py",
+            "body": "benchmark review comment",
+            "start_index": 0,
+            "end_index": len(_COMMENT_ANCHOR),
+            "anchor_content": _COMMENT_ANCHOR,
+        },
+    )
+    resp.raise_for_status()
+
+
 # ── runner (full-turn) journeys ──────────────────────────────
 #
 # These drive a real agent turn through the runner + mock LLM (with_runner=True,
@@ -272,6 +338,16 @@ _TURN_PROMPT = "Say hello."
 # drift negligible.
 _RUNNER_MAX_ITERATIONS = 5

+# Iteration cap for the runner filesystem read. It's a proxied localhost read,
+# not a full turn, so it's far cheaper than the drive-a-turn journeys — a higher
+# cap gives a usable p50/p99 while staying well within the CI time budget.
+_RUNNER_FS_MAX_ITERATIONS = 50
+
+# File planted by the read-runner-file setup and fetched by its measure op.
+# ~1 KB — a modest, representative source file, not a stress case.
+_RUNNER_FILE_PATH = "bench_read_target.txt"
+_RUNNER_FILE_CONTENT = "benchmark file content line\n" * 40
+

 async def _setup_turn_agent(env: BenchEnvironment, *, stream: bool = False) -> str:
    """Register the agent + a reset-surviving reply; return the agent id.
@@ -340,6 +416,24 @@ async def _measure_interrupt(env: BenchEnvironment, ctx: JourneyContext) -> None
    await env.drive_and_interrupt(session_id)


+async def _setup_runner_file_session(env: BenchEnvironment) -> str:
+    """Bind a session to the runner and plant a file to read; return its id.
+
+    No turn is driven and no mock reply is configured — the measured op is a
+    filesystem read proxied to the runner, which never calls the LLM.
+    """
+    name = await env.ensure_agent()
+    agent_id = await env.agent_id(name)
+    session_id = await env.create_bound_session(agent_id)
+    await env.write_runner_file(session_id, _RUNNER_FILE_PATH, _RUNNER_FILE_CONTENT)
+    return session_id
+
+
+async def _measure_read_runner_file(env: BenchEnvironment, ctx: JourneyContext) -> None:
+    session_id = cast(str, ctx)  # _setup_runner_file_session
+    await env.read_runner_file(session_id, _RUNNER_FILE_PATH)
+
+
 # ── registry ─────────────────────────────────────────────────

 ALL_JOURNEYS: dict[str, Journey] = {
@@ -383,6 +477,23 @@ ALL_JOURNEYS: dict[str, Journey] = {
            concurrency_safe=True,
            description="GET /v1/sessions?search_query= — unindexed LIKE over titles + items.",
        ),
+        Journey(
+            name="fork_session",
+            kind="latency",
+            measure=_measure_fork_session,
+            setup=_setup_fork_session,
+            teardown=_teardown_fork_session,
+            concurrency_safe=True,
+            description="POST /v1/sessions/{id}/fork — session fork (deep-copy); DELETE untimed.",
+        ),
+        Journey(
+            name="add_comment",
+            kind="latency",
+            measure=_measure_add_comment,
+            setup=_setup_target_session,
+            concurrency_safe=True,
+            description="POST /v1/sessions/{id}/comments — create a review comment.",
+        ),
        # Runner (full-turn) journeys — with_runner=True, openai-agents, mock LLM.
        Journey(
            name="session_cold_start",
@@ -420,6 +531,15 @@ ALL_JOURNEYS: dict[str, Journey] = {
            max_iterations=_RUNNER_MAX_ITERATIONS,
            description="Interrupt a running (gated) turn; time to cancellation.",
        ),
+        Journey(
+            name="read_runner_file",
+            kind="latency",
+            measure=_measure_read_runner_file,
+            setup=_setup_runner_file_session,
+            needs_runner=True,
+            max_iterations=_RUNNER_FS_MAX_ITERATIONS,
+            description="GET .../environments/default/filesystem/{path} — runner file read proxy.",
+        ),
    )
 }

@@ -27,6 +27,8 @@ _SMOKE_JOURNEYS = [
    "get_session",
    "load_conversation_history",
    "search_sessions",
+    "fork_session",
+    "add_comment",
 ]


@@ -177,7 +179,13 @@ async def test_benchmark_smoke_threshold_failure_exits_nonzero() -> None:

 # ── runner (full-turn) journeys ──────────────────────────────

-_RUNNER_JOURNEYS = ["session_cold_start", "warm_turn", "time_to_first_token", "interrupt"]
+_RUNNER_JOURNEYS = [
+    "session_cold_start",
+    "warm_turn",
+    "time_to_first_token",
+    "interrupt",
+    "read_runner_file",
+]


@pytest.mark.timeout(300)