Batch Eval System

The batch-eval system runs one saved graph across many episodes — optionally fanned out over N parallel workers — and aggregates metrics, persists each run, and streams progress. Today it runs in two processes: a POST /api/eval/v2/start only loads the shared model singletons in the long-lived backend and queues the run; the actual BatchEvalRunner executes in a separate subprocess that JobScheduler spawns once GPU admission clears. Its hard promise is that worker_count is a speed dial, never a result dial — 1 worker and 8 workers score bit-identically. For the rationale see ADR-eval-002 + ADR-server-003; for the surrounding runtime objects see backend-overview + component-registry.

One batch run, two processes. The backend (top) loads the heavy model singletons once and queues the run; JobScheduler spawns an eval subprocess (bottom) that borrows those model servers by URL but spawns and owns its own N tagged env servers. The whole class stack — BatchEvalRunner ⊃ EnvWorkerPool ⊃ LoopRunner ⊃ GraphExecutor — lives inside the subprocess.

1. What the system does

Batch eval answers one question: take a graph the user already drew and ran on the canvas, and run it unattended across many episodes — then report aggregate metrics. The solution has three moves, and the whole page is those three moves traced through code:

1. Isolate the run in its own OS process. A crashing episode, a leaked GL context, or a CUDA OOM cannot take the interactive backend down with it, because by default the run executes in a subprocess JobScheduler spawned just for it (§ 2–§ 4).
2. Load each scarce resource once, at the right scope. The heavy model singletons (a VLM, a policy network) stay in the long-lived backend and are borrowed by every run over HTTP; the stateful env simulators are replicated per worker inside the run's own process (§ 4).
3. Drive N episodes concurrently. N worker coroutines each pinned to their own env, results merged under a lock (§ 5).

The promise that survives all three moves: worker_count changes how fast the run finishes, never what it scores. Two properties the code is built to keep — both verifiable, both with a cost named in § 5 / § 7:

• Worker count must not change the result. One execution path parameterized by N; at worker_count=1 the pool, the per-worker queues, and the inference coalescer all degenerate to single-tenant rather than taking a separate code path.
• Partial results still count. A timed-out or crashed episode still returns whatever metrics + step count were harvested up to the failure (eval_batch.py · BatchEvalRunner._run_one_episode); failure is just another way an episode finishes.

The smallest run is a single POST — a graph name and an episode count; everything else defaults. On the default path the response is queued, not running: the backend has only enqueued the run, and nothing executes until the scheduler admits it (§ 3).

POST /api/eval/v2/start
{ "graph_name": "navgpt_mp3d", "episode_count": 10 }
# → { "run_id": "20260616_…", "status": "queued", "via_subprocess": true }
#   default path: the backend only QUEUES the run; nothing executes yet.

2. Two processes, by default

A batch run is split across two OS processes that each load a different half of the world. The request handler start_eval_v2 (api/execution/eval.py · start_eval_v2) branches on one flag:

use_subprocess = req.via_subprocess and os.environ.get("AGENTCANVAS_EVAL_SUBPROCESS", "1") != "0"

The reason for the split is the second move of § 1. A VLM or policy network is many gigabytes and takes minutes to load; running it once per backend and serving every run is the whole point of having a backend. So on the default path the backend only ever loads the shared half: ensure_shared_nodesets_for_graph walks the graph's nodeset prefixes and, for each, continues past anything whose parallelism != "shared" (registry.py · ensure_shared_nodesets_for_graph). The env simulators — stateful, per-call, one-GL-context-per-process — cannot be shared, so they are deliberately not loaded here; the subprocess will spawn its own (§ 4). The backend then computes the URLs of the shared servers it has up and hands them to the subprocess via the spec (§ 3).

3. Admission & spawn — JobScheduler

submit only enqueues; nothing runs until a later tick admits it. On the subprocess path the handler builds a spec — the eval config, a scheduling block (marginal_vram_mb, exclusive_gpu, priority), the full graph JSON, the shared-server URLs, and an optional active-workspace overlay — and calls scheduler.submit(spec) (eval.py · start_eval_v2), which returns immediately with {run_id, status: "queued"}.

submit (job_scheduler.py · submit) generates a second-precision run_id (with a collision suffix), creates outputs/eval_runs/{run_id}/, writes spec.json and shared_urls.json, bumps the shared-singleton refcounts, writes an initial summary.json with status="pending", appends a _QueuedJob, and returns. It does not spawn.

Spawning happens on the scheduler's tick (driven ~1 s by main.py · _scheduler_loop), which runs _reap → _admit → _drain_pending_unloads in that order so VRAM freed by a finished job is available to admission in the same tick. _admit (job_scheduler.py · _admit) walks the queue FIFO and starts the first job that fits:

• Canvas lock. If canvas Play holds ExecutionGuard in the backend, admission is refused outright (the wired-in _canvas_lock_held callback) — the interactive canvas always wins the GPU.
• Resource fit. Per measurable resource (VRAM + RAM): calibrated estimate ≤ measured_free − Σ reservations; a job that doesn't fit is skipped, not blocking. Declared marginal_vram_mb is the fallback when calibration is incomplete; the old static ledger survives only for hosts with no sampler. Details: JobScheduler design doc.
• Exclusive GPU. An exclusive_gpu job can't start while anything else runs, and nothing starts while an exclusive job runs.
• Priority is stored but ignored. The priority field rides on _QueuedJob and surfaces in list_active(), but admission is pure FIFO over queue order (the M1 comment in _admit).

When a job fits, _spawn (job_scheduler.py · _spawn) launches the subprocess. The command line is fixed; the spec never travels on argv or stdin (it's read from spec.json), and start_new_session=True puts the child in its own process group so cancel can SIGTERM the whole tree:

subprocess.Popen(
    [sys.executable, "-m", "app.eval_subprocess_main",
     "--run-dir", str(run_dir), "--backend-url", self._backend_url],
    cwd=self._backend_dir, env=env, stdin=subprocess.DEVNULL,
    start_new_session=True,  # setsid → own pgid for clean cancel
)

_reap polls each child; on exit it reads the child's final summary.json status, or marks the run aborted if the child died without touching _DONE, then releases the shared-singleton refcounts and drops the job from _running (which is what frees its VRAM reservation). cancel (job_scheduler.py · cancel) removes a still-queued job immediately (status cancelled + _DONE); for a running job it SIGTERMs the process group and lets _reap finalize. JobScheduler itself lives once per process on ProcessServices.job_scheduler (state.py), constructed in the backend's lifespan.

4. The eval subprocess cold-starts its own world

The subprocess is a cold Python process — not a fork — so it re-discovers everything from disk before it can run a single node. eval_subprocess_main.main sets PR_SET_PDEATHSIG (so it and its env children die if the backend dies), parses --run-dir, reads spec.json + shared_urls.json, and builds a fresh WorkspaceComponentRegistry (eval_subprocess_main · _build_local_registry_for_run). Its registry shares nothing with the backend's — get_services() returns a per-process ProcessServices (state.py) — so three steps stand between a cold process and a runnable graph:

registry.scan_all()                                    # cold: re-discover nodesets from disk
for name, url in shared_urls.items():
    await registry.register_remote_nodeset(name, url)  # borrow parent's VLM by URL (no 2nd load)
load_result = await registry.ensure_nodesets_for_graph(graph, worker_count=worker_count)

1. scan_all() — re-discover from disk. The backend ran this in its lifespan; the subprocess must redo it or the graph's node types resolve to no handler. This is load-bearing: a missing handler is a silent failure, not an error (§ 7). reload-and-code-freshness covers why each subprocess re-scans (and why that's a feature, not a bug — edits are picked up fresh per launch).
2. register_remote_nodeset(name, url) — borrow, don't load. For each shared server the backend has up, the subprocess fetches its manifest over HTTP, generates proxy node classes pointed at the backend's URL, and installs a no-op server shim (registry.py · register_remote_nodeset). The borrowed process is the backend's — the shim's stop() is intentionally a no-op, so the subprocess exiting never kills the shared VLM. This is how the 14 GB model loads once and serves every run.
3. ensure_nodesets_for_graph(graph, worker_count=N) — spawn the env servers it owns. For each remaining nodeset declared replicated, with worker_count > 1, the registry spawns N tagged auto_host subprocesses (env_habitat#0 … #N-1) via _load_nodeset_as_server — direct python -m app.server.auto_host children so the one-hop PR_SET_PDEATHSIG reaps them with the run.

The subprocess then builds EvalConfig + EvalRun from the spec, wraps run.episodes in a list subclass that atomic-rewrites summary.json on every episode boundary (_install_episode_writer — this is the live-progress channel for the backend's status poll, since the subprocess has no WebSocket clients, § 6.3), runs BatchEvalRunner.execute(), and in a finally calls shutdown_all() to kill its env servers, save_run, and touch_done (the _DONE marker _reap looks for).

5. Inside one run — the class stack

Everything below this point is identical on both the subprocess and legacy paths — it's the class stack from the diagram, BatchEvalRunner ⊃ EnvWorkerPool ⊃ LoopRunner ⊃ GraphExecutor, plus the shared inference tier the proxy calls reach.

5.1 EnvWorkerPool — slots, chunks, pinned leases

The pool turns N tagged env servers into N leasable worker slots, and pins each episode chunk to one slot. BatchEvalRunner.execute opens an EnvWorkerPool(worker_count, env_nodeset) as an async context manager. On enter, if worker_count > 1 and there's an env nodeset, the pool looks up each tagged env panel proxy (get_env_panel("env_habitat#k")) and tagged server URL (get_server_url(env, tag=k)) and packs them into a WorkerHandle's env_panel_overrides + server_url_overrides (env_worker_pool.py · EnvWorkerPool.__aenter__); at worker_count=1 the overrides are empty and lookups fall through to the global registry — bit-identical to canvas Play.

execute then splits the index list into N contiguous chunks (_partition_contiguous, numpy array_split distribution) and gives each worker its own asyncio.Queue; worker k only ever drains queue k, and leases its slot via acquire_specific(worker_id) (a pinned lease that rotates other handles back through the available queue until its own is free). The pinning exists for one reason: a SAPIEN-backed env (SIMPLER) crashes its subprocess after ~15 cross-task set_episode() rebuilds, so each contiguous chunk must stay on one subprocess to bound task switches (env_worker_pool.py · EnvWorkerPool.acquire_specific). N worker coroutines run under one asyncio.gather; they share no step barrier — worker 0 may be on step 8 while worker 1 is on step 2.

5.2 One episode — LoopRunner → GraphExecutor

Each episode is a fresh LoopRunner — the clean-state guarantee is structural, not a reset path. One iteration of a worker's loop (eval_batch.py · BatchEvalRunner._run_one_episode):

1. Set the episode on the worker's tagged env panel: push each cascade field (e.g. dataset → split for Habitat, split → task_id for SIMPLER) via on_field_change, push episode_index, on_action("play") to commit, on_load() to read back episode id / scene / instruction and the env's per-episode step_budget (scene-adaptive envs like HM-EQA publish one).
2. Resolve the step budget: API override → env value → graph value → DEFAULT_STEP_BUDGET; a resolved budget < 1 is refused (it would silently complete a 0-step episode). The wall-clock timeout is resolved_budget × per_step_budget.
3. Build a fresh LoopRunner(env_panel_overrides, server_url_overrides) and asyncio.wait_for(runner.run(graph, principles, step_budget_override=resolved_budget), timeout). The override carries the resolved budget so parallel workers never mutate the shared graph object.
4. Harvest metrics from the executor's graphOut sinks and build the EpisodeResult — in the success branch and the timeout / exception branches (the partial-results property of § 1).
5. Evict + aggregate: drop this episode's key from any nodeset-owned keyed container (_evict_episode_state, worker-safe), then under results_lock append the result, recompute aggregate_metrics + aggregate_by_task, broadcast.

Inside runner.run, LoopRunner builds a GraphExecutor carrying the same overrides and calls executor.run(...) (loop_runner.py · LoopRunner.run). The executor resolves its loop bound (step_budget_override → graph.step_budget → DEFAULT) and fires nodes off a ready queue (graph_executor.py · GraphExecutor.run). Two seams matter for eval:

• Per-worker routing. Each in-graph proxy call (e.g. env_habitat__step_native) asks the executor for a URL override; get_server_url(nodeset) returns this worker's tagged subprocess URL, and server/proxy.py · _make_forward routes the HTTP call there instead of the URL baked into the proxy class (empty override = baked URL = single-worker / canvas).
• Metric harvest. The executor snapshots each graphOut's pending inputs into node.state["_last_inputs"] after the terminal iteration (graph_executor.py); _collect_metrics reads only graph-scope graphOut sinks (_scope_of(node) == GRAPH_SCOPE_ID), treats a portName == "metrics" value as a dict to flatten, and any other scalar port as metrics[portName]. Two guards keep a broken run from masquerading as a result: the executor records every {"error": ...}-shaped node result into its node_errors ledger and finishes such runs with _status="error" — _run_one_episode converts that into episode.status="error" naming the culprit nodes (checked first, since 2026-07-04); and an eval graph that "completed" with no metrics is recorded as an error ("no verdict") so a 0-step drain can't alias to success=0 and poison SR.

5.3 Shared inference — BatchedInferenceServer

The shared model server coalesces the K workers' concurrent calls into one forward pass. When K workers hit the same borrowed server, their requests converge server-side, so the rendezvous lives there, not in the clients (server/batched_inference.py). Each call is keyed by (function_name, config_hash) — two callers with identical config share one _BatchQueue; differing checkpoints stay separate. A queue flushes flush_timeout_ms (default 50 ms) after the last arrival (restart-on-submit timer), so a late caller joins the still-forming batch rather than forcing a partial one out. A server-level _inference_lock serialises handler calls across all keys (single-in-flight), bounding peak device memory to one batch. The contract is pure-functional: no per-caller state, so RNN hidden states must ride explicit hidden_in / hidden_out ports on the wire.

6. Failure & observability

6.1 Termination paths

Six ways an episode ends, all returning a populated EpisodeResult (eval_batch.py · BatchEvalRunner._run_one_episode):

The pool is never blocked by a single stuck episode — each is bounded by asyncio.wait_for at the orchestrator level, and the worker resumes its queue the instant the episode ends. runner is pre-bound to None before the try-block so even a failure during LoopRunner construction is inspectable.

6.2 Persistence

The run dir is laid out so one episode's trajectory can be reconstructed from its own subdir alone (ADR-eval-004):

• outputs/eval_runs/{run_id}/ — run_id = timestamp + collision suffix, e.g. 20260616_143052
  • spec.json, shared_urls.json — input contract written by JobScheduler.submit before spawn
  • summary.json — run summary; rewritten on every episode boundary in subprocess mode (_install_episode_writer) and once more by save_run at exit
  • graph.json — graph frozen at BatchEvalRunner construction (edits to the source graph afterward don't affect the run)
  • stdout.log, stderr.log, _DONE — child output + the terminal marker _reap reads
  • episodes/ep{idx:04d}/ — one self-contained dir per episode
    • log.jsonl — this episode's node-firing log only
    • assets/ — this episode's image artefacts only
    • episode.json — self-describing per-episode record

Active-workspace overlay (ADR-components-009). A run may carry an active_workspace_dir in its spec; JobScheduler._spawn threads it into the subprocess as ACTIVE_WORKSPACE_DIR, so the child's registry overlays workspace/graphs/ + workspace/graph_nodes/ before the frozen base. This is how architect cycles and sibling Claude sessions A/B a graph edit without touching the canonical files. The overlay is resolved at run-start when the graph is frozen; tearing it down afterward doesn't affect the in-flight run.

6.3 Progress: WebSocket vs summary.json

The live-progress channel differs by path, because the WS connection set is per-process. BatchEvalRunner broadcasts eval_progress / eval_episode_done / eval_complete (all tagged source="eval") via state.broadcast, which iterates that process's _ws_connections (state.py · broadcast):

• Legacy in-process path: the runner shares the backend's WS set, so the events reach the UI live.
• Default subprocess path: the subprocess has no WS clients — its _ws_connections is empty — so those broadcasts go nowhere. Live progress instead reaches the UI through summary.json, atomic-rewritten on each episode boundary (§ 4) and served by the backend's /api/eval/v2/status + /queue REST endpoints (pull).

Per-node nav_step / exec_log / nav_status events are suppressed during eval either way: BatchEvalRunner.execute sets ExecutionPrinciples(suppress_nav_events=True), and GraphExecutor gates every such broadcast on session.principles.suppress_nav_events (graph_executor.py), so at N workers × base64 images per step the executor never floods the pipe in the first place.

7. Where the code deviates from the mental model

The opening band hands you a clean model: "replicated ⇒ its own subprocess, shared ⇒ borrowed + coalesced, worker_count is free." The shipped code keeps that model only under the conditions below — these are the honest seams.

node_cls = NODE_HANDLERS.get(node.type)
if node_cls is None:
    log.warning("No handler for node type: %s (id=%s)", node.type, node.id)
    return {}   # missing handler → no-op; the episode silently drains at step 0