---
name: write-online-runner
description: Use when the engine runs across a network boundary rather than as a local subprocess. Authors a transport-driven engine plugin that dispatches claimed tests to a remote or online worker over a transport such as HTTP, as the NVCF plugin does.
license: Apache-2.0
compatibility: Requires simready-benchmark (pip install simready-benchmark). The worker process on the far end is yours to build and deploy; the framework only drives the host-side plugin and session.
metadata:
  author: NVIDIA
  library: simready_benchmark
  version: "2026.05"
---

## Confirm the Installed Version

This skill ships inside the `simready-benchmark` wheel, so the copy installed with the framework matches the installed code. If you are working from a cached copy or from memory rather than the installed package, confirm the framework version with `simready-benchmark --version`. Treat the installed module docstrings (rendered in the API reference) and `simready-benchmark --list-tests` as the authoritative source for the APIs, entry points, and names described here, because a signature or name documented here might have changed in a different version.

The worker HTTP contract used as the reference here (`POST /run-test`, `GET
/health`, 503 plus `Retry-After`) ships as a Technology Preview and can change
between releases. The authoring surface you implement, the `EnginePlugin` and
`EngineSession` abstract base classes in `simready_benchmark.engines.base`, is
stable.

## When to Write a Transport-Driven Plugin

A transport-driven plugin is the right shape when the engine runs somewhere the
orchestrator host cannot reach through a process handle, such as a managed
cloud function, a container on another host, or any service behind an HTTP
endpoint. The plugin holds a `threading.Thread` per session. That thread runs
the work-claim loop locally and POSTs each claimed test to the remote worker.
The worker runs the actual engine.

If the engine runs locally as a process you spawn and drive with command-line
arguments, you want the subprocess shape instead, modeled by the Kit plugin.
This skill covers only the transport shape, modeled by the NVCF plugin
(`simready_benchmark.engines.nvcf_plugin` and `simready_benchmark.engines.nvcf_session`).

## The EnginePlugin and EngineSession Contract

Two abstract base classes define the whole plugin contract. Both live in
`simready_benchmark.engines.base`.

`EnginePlugin` has six abstract methods, and a subclass must implement all six.

```python
class EnginePlugin(ABC):
    @abstractmethod
    def name(self) -> str: ...

    @abstractmethod
    def tags(self) -> List[str]: ...

    @abstractmethod
    def discover(self) -> List[EngineConfig]: ...

    @abstractmethod
    def validate(self, engine_config: EngineConfig) -> ValidationResult: ...

    @abstractmethod
    async def start_session(
        self,
        engine_config: EngineConfig,
        work_dir: str,
        log_file: str,
        session_env: Optional[Dict[str, str]] = None,
    ) -> "EngineSession": ...

    @abstractmethod
    async def stop_session(self, session: "EngineSession") -> None: ...
```

`EngineSession` has three abstract methods.

```python
class EngineSession(ABC):
    @abstractmethod
    def is_alive(self) -> bool: ...

    @abstractmethod
    async def cleanup(self) -> None: ...

    @abstractmethod
    def get_session_id(self) -> str: ...
```

`start_session`, `stop_session`, and `cleanup` are `async`; the framework
awaits them from a single event loop that also drives every other concurrent
session. A transport session runs its claim loop on a `threading.Thread` and
joins that thread inside `cleanup` through `run_in_executor`, so the join never
blocks the event loop. For the full method-by-method signatures, the
`EngineConfig` fields, and the `ValidationResult` shape, refer to
[references/engine-abcs.md](references/engine-abcs.md).

## A Minimal Worked Example

Here is a compact transport-driven plugin for a fictional engine, `acme_cloud`,
that POSTs each claimed test to a remote URL. It implements all six
`EnginePlugin` methods and all three `EngineSession` methods.

`acme_cloud_plugin/plugin.py`:

```python
import os
from typing import Dict, List, Optional
from urllib.parse import urlparse

from simready_benchmark.engines.base import (
    EngineConfig,
    EnginePlugin,
    ValidationResult,
)


class AcmeCloudPlugin(EnginePlugin):
    """Engine plugin that dispatches tests to a remote acme_cloud worker."""

    def __init__(self) -> None:
        self._session_counter = 0
        # config name -> {"url": str, "api_key_env": Optional[str]}.
        # Endpoint and auth metadata are kept off EngineConfig (which the
        # framework logs and serializes) in this private side table.
        self._url_by_config: Dict[str, Dict[str, Optional[str]]] = {}

    def name(self) -> str:
        return "acme_cloud"

    def tags(self) -> List[str]:
        # Plugin-level tags stay disjoint across installed plugins by
        # convention. The "kit" capability rides on each EngineConfig below.
        return ["acme_cloud"]

    def discover(self) -> List[EngineConfig]:
        url = os.environ.get("ACME_CLOUD_URL", "")
        if not url:
            return []
        self._url_by_config["acme_default"] = {
            "url": url,
            "api_key_env": "ACME_CLOUD_API_KEY",
        }
        return [
            EngineConfig(
                name="acme_default",
                executable_path="",  # transport-driven: no local executable
                version="acme-remote",
                tags=["acme_cloud", "kit"],  # config-level: endpoint runs kit tests
            )
        ]

    def validate(self, engine_config: EngineConfig) -> ValidationResult:
        info = self._url_by_config.get(engine_config.name, {})
        url = info.get("url", "")
        parsed = urlparse(url) if url else None
        if not parsed or not parsed.scheme or not parsed.netloc:
            return ValidationResult(
                valid=False,
                message="acme_cloud config %r has no valid URL" % engine_config.name,
            )
        warnings = []
        key_env = info.get("api_key_env")
        if key_env and not os.environ.get(key_env):
            warnings.append("API key env var %s not set; calling unauthenticated" % key_env)
        return ValidationResult(valid=True, message="OK", warnings=warnings)

    async def start_session(self, engine_config, work_dir, log_file, session_env=None):
        from acme_cloud_plugin.session import AcmeCloudSession

        self._session_counter += 1
        session_id = "acme_cloud_%d_%d" % (os.getpid(), self._session_counter)
        info = self._url_by_config.get(engine_config.name, {})
        if not info.get("url"):
            raise RuntimeError("no URL configured for %r" % engine_config.name)

        session = AcmeCloudSession(
            session_id=session_id,
            cfg=engine_config,
            worker_url=info["url"],
            api_key_env=info.get("api_key_env"),
            session_env=session_env or {},
        )
        await session.start()
        return session

    async def stop_session(self, session) -> None:
        await session.cleanup()
```

`acme_cloud_plugin/session.py`:

```python
import asyncio
import json
import os
import socket
import threading
import time
import urllib.error
import urllib.request
from datetime import datetime, timezone

from simready_benchmark.engines.base import EngineSession

NO_WORK_POLL_INTERVAL = 5.0
PROBE_INTERVAL_S = 15.0
HEALTH_PROBE_TIMEOUT_S = 3.0
UNAVAILABLE_SLEEP_S = 1.0
THREAD_JOIN_TIMEOUT = 30.0
UNAVAILABLE_TIMEOUT_S = float(os.environ.get("ACME_CLOUD_UNAVAILABLE_TIMEOUT_S", "1800"))
IS_ALIVE_INACTIVITY_TIMEOUT = 300.0


class AcmeCloudSession(EngineSession):
    """A worker thread that claims tests and POSTs them to a remote worker."""

    def __init__(self, session_id, cfg, worker_url, api_key_env, session_env):
        self._session_id = session_id
        self._cfg = cfg
        self._worker_url = worker_url.rstrip("/")
        self._api_key_env = api_key_env
        self._session_env = dict(session_env or {})
        self._thread = None
        self._stop_event = threading.Event()
        self._event_list = []
        self._event_lock = threading.Lock()
        self._last_activity_time = time.monotonic()
        self._plan = None
        self._test_by_id = {}
        self._content_root = ""
        self._output_dir = ""
        self._engine_available = True
        self._next_probe_at = 0.0
        self._unavailable_since = None

    # ----- EngineSession contract -----

    def is_alive(self):
        if self._thread is None or not self._thread.is_alive():
            return False
        return (time.monotonic() - self._last_activity_time) < IS_ALIVE_INACTIVITY_TIMEOUT

    async def cleanup(self):
        # cleanup() is async. Joining the worker thread would block the
        # event loop, so hand the join off to an executor.
        self._stop_event.set()
        thread = self._thread
        if thread is not None:
            await asyncio.get_event_loop().run_in_executor(None, thread.join, THREAD_JOIN_TIMEOUT)

    def get_session_id(self):
        return self._session_id

    def drain_events(self):
        with self._event_lock:
            events = list(self._event_list)
            self._event_list.clear()
        return events

    # ----- Lifecycle -----

    async def start(self):
        from simready_benchmark.core.models import load_plan

        plan_path = self._session_env.get("SIMREADY_PLAN_PATH", "")
        if not plan_path:
            raise RuntimeError("AcmeCloudSession requires SIMREADY_PLAN_PATH in session_env")
        self._plan = load_plan(plan_path)
        self._test_by_id = {t.id: t for t in self._plan.tests}
        self._content_root = self._session_env.get("SIMREADY_CONTENT_ROOT", "") or ""
        self._output_dir = self._session_env.get("SIMREADY_OUTPUT_DIR", "") or ""

        self._thread = threading.Thread(
            target=self._run, name="acme-%s" % self._session_id, daemon=True
        )
        self._thread.start()

        deadline = time.monotonic() + 30.0
        while time.monotonic() < deadline:
            with self._event_lock:
                ready = any(e.get("type") == "engine_ready" for e in self._event_list)
            if ready:
                return
            await asyncio.sleep(0.05)
        raise TimeoutError("AcmeCloudSession %s did not become ready" % self._session_id)

    def _emit(self, event_type, **kwargs):
        event = {"type": event_type, "timestamp": datetime.now(timezone.utc).isoformat()}
        event.update(kwargs)
        with self._event_lock:
            self._event_list.append(event)
        self._last_activity_time = time.monotonic()
```

The worker thread body (`_run`, `_process_asset`, the POST, and the
transport-failure handling) is shown in the next two sections.

## The Claim Loop Inside the Worker Thread

The worker thread runs the same two-level claim protocol that the local engine
uses, but it calls the `WorkPool` API from plugin code directly because the
engine runs on the far side of a network call. Construct the `WorkPool` one time
before the loop.

```python
def _run(self):
    from simready_benchmark.runner.work_pool import WorkPool

    self._emit("engine_ready", session_id=self._session_id)
    pool = WorkPool(self._session_env["SIMREADY_WORK_POOL"])
    engine_tags = list(self._cfg.tags)
    engine_version = self._cfg.version

    exit_reason = "no_work"
    while not self._stop_event.is_set():
        # Gate on availability after a transport failure (next section).
        if not self._engine_available and not self._wait_for_recovery():
            exit_reason = "acme_unavailable_timeout"
            break

        item = pool.claim_asset(self._session_id, engine_tags, self._plan)
        if item is None:
            self._emit("idle_keepalive", session_id=self._session_id)
            if self._stop_event.wait(timeout=NO_WORK_POLL_INTERVAL):
                exit_reason = "shutdown_signal"
                break
            continue
        self._emit("asset_start", asset=item.asset, session_id=self._session_id)
        self._process_asset(pool, item)
        if self._stop_event.is_set():
            exit_reason = "shutdown_signal"
            break

    self._emit("session_exit", reason=exit_reason, session_id=self._session_id)
```

`_process_asset` claims one test at a time, POSTs it, writes `result.json`, and
completes the test. Note the order: emit `test_start` before the POST so the
runner no-progress watchdog sees activity even when the POST blocks for minutes.

```python
def _process_asset(self, pool, item):
    from simready_benchmark.core.asset_key import asset_rel_key
    from simready_benchmark.core.output_layout import per_test_runtime_dir

    asset_key = asset_rel_key(item.asset, self._content_root)
    while True:
        if self._stop_event.is_set():
            pool.release_asset(item.id, self._session_id)
            return

        te = pool.claim_test(item.id, self._session_id)
        if te is None:
            pool.release_asset(item.id, self._session_id)
            return

        tdef = self._test_by_id.get(te.test_id)
        test_output_dir = per_test_runtime_dir(self._output_dir, asset_key, tdef.name)
        os.makedirs(test_output_dir, exist_ok=True)

        self._emit("test_start", test=tdef.name, asset=item.asset, session_id=self._session_id)

        try:
            result = self._post_run_test(tdef, item, te, asset_key)
        except Exception as exc:
            if self._is_transport_failure(exc):
                self._handle_transport_failure(pool, item, te, exc)
                return  # test released back to the pool; do NOT write result.json
            result = {"status": "fail", "duration": 0, "message": "transport error: %s" % exc}

        asset_done = pool.complete_test(
            item.id, te.test_id,
            {"status": result.get("status", "unknown"), "duration": result.get("duration", 0)},
        )
        self._write_result_json(test_output_dir, tdef, asset_key, result)
        self._emit(
            "test_complete", test=tdef.name, asset=item.asset,
            status=result.get("status", "unknown"), session_id=self._session_id,
        )
        if asset_done:
            pool.release_asset(item.id, self._session_id)
            self._emit("asset_complete", asset=item.asset, session_id=self._session_id)
            return
```

The POST carries `asset_key` (identity), never `output_dir` (a host path), so
the worker reconstructs its own per-test directory without seeing host paths.

```python
def _post_run_test(self, tdef, item, te, asset_key):
    body = {
        "test_name": tdef.name,
        "test_version": tdef.version,
        "test_file": tdef.test_file,
        "asset_path": item.asset,
        "asset_key": asset_key,
        "config": (item.config_overrides or {}).get(te.test_id, {}),
        "content_root": self._content_root,
    }
    data = json.dumps(body).encode("utf-8")
    req = urllib.request.Request(self._worker_url + "/run-test", data=data, method="POST")
    req.add_header("Content-Type", "application/json")
    if self._api_key_env and os.environ.get(self._api_key_env):
        req.add_header("Authorization", "Bearer %s" % os.environ[self._api_key_env])
    with urllib.request.urlopen(req, timeout=3600) as resp:
        payload = json.loads(resp.read().decode("utf-8"))
    # Unwrap the inner result if the worker wrapped it, else use the payload.
    if isinstance(payload, dict) and isinstance(payload.get("result_json"), dict):
        return payload["result_json"]
    return payload
```

## The Transport-Failure Release-and-Probe Pattern

Draw a deliberate line between a transport failure and an application failure.
A transport failure means the worker did not refuse the work; it just cannot
answer right now. An application failure means the worker answered and said no.

`_is_transport_failure` treats exactly these as transport failures: an
`HTTPError` with status 503, 425, or 429; any other `URLError`; `TimeoutError`,
`socket.timeout`, or `ConnectionError`; and a bare `OSError`. Everything else,
including an `HTTPError` with any other status code and a `RuntimeError` from a
malformed response, is an application failure.

```python
@staticmethod
def _is_transport_failure(exc):
    if isinstance(exc, urllib.error.HTTPError):
        return exc.code in (503, 425, 429)
    if isinstance(exc, urllib.error.URLError):
        return True
    if isinstance(exc, (TimeoutError, socket.timeout, ConnectionError)):
        return True
    return isinstance(exc, OSError)
```

On a transport failure, release the in-flight test back to the pool with
`release_test` (which does NOT increment the retry count), mark the engine
unavailable, and back off. A 503 must not write a failing `result.json`; the
test bounces back to the pool for this or another session to pick up after
recovery.

```python
def _handle_transport_failure(self, pool, item, te, exc):
    pool.release_test(item.id, te.test_id, self._session_id)
    now = time.monotonic()
    self._engine_available = False
    self._next_probe_at = now + PROBE_INTERVAL_S
    if self._unavailable_since is None:
        self._unavailable_since = now
    self._emit(
        "engine_warning", session_id=self._session_id,
        kind="acme_transport_failure", worker_url=self._worker_url,
        message="transport failure against %s: %s -- releasing test and backing off"
        % (self._worker_url, exc),
    )
```

The loop then probes `GET /health` on an interval until the worker returns a
2xx response, at which point it resumes. If the engine stays unavailable longer
than the timeout (`UNAVAILABLE_TIMEOUT_S`, 1800 seconds, configurable through
the `ACME_CLOUD_UNAVAILABLE_TIMEOUT_S` environment variable), the session exits.

```python
def _wait_for_recovery(self):
    """Return True when the engine recovers, False after the timeout expires."""
    if self._unavailable_since is not None and (
        time.monotonic() - self._unavailable_since > UNAVAILABLE_TIMEOUT_S
    ):
        return False
    if time.monotonic() < self._next_probe_at:
        self._stop_event.wait(timeout=UNAVAILABLE_SLEEP_S)
        return True  # still backing off; re-enter the loop without claiming
    if self._probe_engine_health():
        self._engine_available = True
        self._unavailable_since = None
        self._emit("engine_recovered", session_id=self._session_id, worker_url=self._worker_url)
        return True
    self._next_probe_at = time.monotonic() + PROBE_INTERVAL_S
    self._stop_event.wait(timeout=UNAVAILABLE_SLEEP_S)
    return True

def _probe_engine_health(self):
    req = urllib.request.Request(self._worker_url + "/health", method="GET")
    try:
        with urllib.request.urlopen(req, timeout=HEALTH_PROBE_TIMEOUT_S) as resp:
            return 200 <= resp.getcode() < 300
    except Exception:
        return False
```

For the full pattern, including the NVCF Functions 202-poll path and inline
media transport, refer to
[references/transport-pattern.md](references/transport-pattern.md).

## Forcing the Canonical asset_key Onto result.json

The reporter aggregates results by the `asset` field across every plugin in a
run. A worker can legitimately return an absolute path on its own filesystem in
that field. If that path lands in `result.json`, the reporter sees two assets
where there is one, and a mixed-mode run shows the same asset twice. Force the
orchestrator's relative `asset_key` onto the field before writing the file.

```python
def _write_result_json(self, test_output_dir, tdef, asset_key, result):
    data = dict(result) if result else {}
    data.setdefault("test_name", tdef.name)
    data["asset"] = asset_key  # overwrite, not setdefault
    data.setdefault("session_id", self._session_id)
    out_path = os.path.join(test_output_dir, "result.json")
    with open(out_path, "w", encoding="utf-8") as f:
        json.dump(data, f, indent=2)
```

## Register the Plugin through an Entry Point

Plugins register through the `simready_benchmark.engines` entry-point group. The
value is the dotted import path to the `EnginePlugin` subclass with the
`:ClassName` suffix.

```toml
[project]
name = "acme-cloud-plugin"
version = "0.1.0"
dependencies = ["simready-benchmark"]

[project.entry-points."simready_benchmark.engines"]
acme_cloud = "acme_cloud_plugin.plugin:AcmeCloudPlugin"

[build-system]
requires = ["setuptools>=68"]
build-backend = "setuptools.build_meta"
```

The key on the left is an arbitrary label. The value on the right must carry the
`:ClassName` suffix; this is the one place the engine entry point differs from a
test pack, whose value is a bare module path. The registry instantiates each
loaded class with no arguments and skips any class that is not an
`EnginePlugin` subclass with a warning.

## Verify

After installing the package in the same environment as `simready-benchmark`:

```bash
py -c "from importlib.metadata import entry_points; print([ep.name for ep in entry_points(group='simready_benchmark.engines')])"
```

The label `acme_cloud` appears in the list. Then confirm the framework loads
and discovers it:

```bash
simready-benchmark --show-config
```

The plugin name appears under the discovered engines alongside `kit` and any
other installed plugins. To check that the session POSTs correctly, point the
plugin at a local worker that answers `POST /run-test` and `GET /health`, then
run a small plan and inspect the per-test `result.json` files for the correct
`asset` field.

## Key Rules

- Implement all six abstract `EnginePlugin` methods and all three abstract
  `EngineSession` methods.
- `start_session`, `stop_session`, and `cleanup` are `async`. Never call
  blocking code in them directly. Join the worker thread inside `cleanup`
  through `run_in_executor`.
- Run the claim loop on a `threading.Thread`. Construct the `WorkPool` one time
  before the loop, then call `claim_asset`, `claim_test`, `complete_test`, and
  `release_asset` from plugin code.
- The request body carries `asset_key`, never `output_dir`.
- A transport failure (503, 425, 429, a refused connection, a socket reset, a
  timeout, or a bare OS error) calls `release_test` and backs off. It must not
  write a failing `result.json`.
- Every other outcome is an application failure: call `complete_test` with
  `status="fail"` and no retry.
- After a transport failure, probe `GET /health` on an interval and resume on a
  2xx. Exit the session after the unavailability timeout expires.
- Force the canonical `asset_key` onto `result.json["asset"]` before writing.
- Keep plugin-level `tags()` disjoint across installed plugins; carry shared
  capability tags such as `kit` on each `EngineConfig` instead.
- Register through the `simready_benchmark.engines` group with a `module:ClassName`
  value.

For the full base-class contract, refer to
[references/engine-abcs.md](references/engine-abcs.md). For the NVCF plugin and
session worked pattern in depth, refer to
[references/transport-pattern.md](references/transport-pattern.md).
