---
name: write-engine-plugin
description: Use when you need the framework to drive a local simulator it does not already support. Authors a subprocess-driven engine plugin, a local engine process per session like Kit, registered through the simready_benchmark.engines entry-point group.
license: Apache-2.0
compatibility: Requires simready-benchmark (pip install simready-benchmark). The Kit reference plugin lives in simready-benchmark-engine-kit, which also requires a Kit or Isaac Sim runtime installed separately.
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 Plugin Contract

An engine plugin is the seam that lets the framework drive a backend it does
not ship support for. Everything else, including planning, work distribution,
report rendering, and asset stamping, stays inside the framework. You implement
two abstract base classes from `simready_benchmark.engines.base` and register the
plugin through an entry point. After installation, the same `simready-benchmark`
invocation that drives Kit drives your engine too.

`EnginePlugin` has six abstract methods. 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 shared by every concurrent session, so never call
`time.sleep`, a blocking `process.wait()`, or a blocking network call directly
inside them. Refer to [references/engine-abcs.md](references/engine-abcs.md) for
the full method-by-method contract, the `EngineConfig` fields, and the
`ValidationResult` shape.

## Minimal Worked Example

A complete subprocess-driven plugin for a fictional `acme_engine`. It satisfies
all six `EnginePlugin` methods and all three `EngineSession` methods.

`acme_engine_plugin/plugin.py`:

```python
import asyncio
import os
import subprocess
from pathlib import Path
from typing import Dict, List, Optional

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


class AcmeSession(EngineSession):
    """Subprocess-backed session for the acme_engine binary."""

    def __init__(self, session_id: str, process: subprocess.Popen) -> None:
        self._session_id = session_id
        self._process = process

    def is_alive(self) -> bool:
        return self._process.poll() is None

    async def cleanup(self) -> None:
        # cleanup() is async. Hand the blocking wait off to an executor so
        # the framework event loop stays responsive during shutdown.
        if self._process.poll() is None:
            self._process.terminate()
            loop = asyncio.get_event_loop()
            try:
                await loop.run_in_executor(None, self._process.wait, 10)
            except subprocess.TimeoutExpired:
                self._process.kill()
                await loop.run_in_executor(None, self._process.wait)

    def get_session_id(self) -> str:
        return self._session_id


class AcmeEnginePlugin(EnginePlugin):
    """Engine plugin that drives the fictional acme_engine binary."""

    def __init__(self) -> None:
        self._session_counter = 0

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

    def tags(self) -> List[str]:
        return ["acme"]

    def discover(self) -> List[EngineConfig]:
        exe = os.environ.get("ACME_ENGINE_PATH", "")
        if not exe or not Path(exe).is_file():
            return []
        return [
            EngineConfig(
                name="acme_default",
                executable_path=exe,
                version="1.0.0",
                tags=["acme"],
            )
        ]

    def validate(self, engine_config: EngineConfig) -> ValidationResult:
        if not Path(engine_config.executable_path).is_file():
            return ValidationResult(
                valid=False,
                message="acme_engine executable not found: %s" % engine_config.executable_path,
            )
        return ValidationResult(valid=True, message="OK")

    async def start_session(
        self,
        engine_config: EngineConfig,
        work_dir: str,
        log_file: str,
        session_env: Optional[Dict[str, str]] = None,
    ) -> EngineSession:
        self._session_counter += 1
        session_id = "acme_engine_%d_%d" % (os.getpid(), self._session_counter)

        env = os.environ.copy()
        if session_env:
            env.update(session_env)
        env["SIMREADY_SESSION_ID"] = session_id
        env["SIMREADY_ENGINE_TAGS"] = ",".join(engine_config.tags)
        env["SIMREADY_ENGINE_VERSION"] = engine_config.version

        command = [
            engine_config.executable_path,
            "--autonomous",
            "--work-dir", work_dir,
        ]
        with open(log_file, "w", encoding="utf-8") as log_fp:
            process = subprocess.Popen(
                command,
                env=env,
                stdout=log_fp,
                stderr=subprocess.STDOUT,
                cwd=work_dir,
            )

        # Yield control so the loop stays responsive while the process warms
        # up. Substitute a real readiness check in production: poll a file
        # the engine writes, parse a ready event from stdout, or open a socket.
        await asyncio.sleep(0)
        return AcmeSession(session_id=session_id, process=process)

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

The `engine_config.tags` echoed into `SIMREADY_ENGINE_TAGS` let the engine
process filter the work pool to tests it can run. The engine inside the
subprocess claims those tests and writes per-test `result.json` files itself.

## The Subprocess Lifecycle

The bundled `KitPlugin` and `KitSession` are the production reference for this
pattern. The lifecycle inside `start_session` runs in four phases.

1. **Spawn.** Build the command line, copy `os.environ`, overlay `session_env`
   plus the session-specific keys (`SIMREADY_SESSION_ID`, `SIMREADY_ENGINE_TAGS`,
   `SIMREADY_ENGINE_VERSION`), and launch the process with stdout piped so the
   plugin can read structured events. `KitPlugin` launches through a
   `SubprocessEngine.launch` helper that pipes stdout, folds stderr into stdout,
   and registers the process for atexit cleanup.
2. **Wait for ready.** Drain the process stdout in a background thread, parse
   structured events out of the engine noise, and poll for the readiness event
   with `await asyncio.sleep` against a bounded timeout. `KitPlugin` waits up to
   120 seconds for an `engine_ready` event. If the process exits before the
   event arrives, it raises `RuntimeError` with the log path. If the timeout
   elapses, it kills the process and raises `TimeoutError`.
3. **Drain events.** The background log-capture thread separates structured
   event lines (Kit uses an `@@EVENT@@` prefix) from engine log noise. Events go
   to a shared list the runner drains; noise goes to the log file. The same
   thread records a `last_output_time` timestamp on every line for liveness.
4. **Cleanup.** `stop_session` stops the capture thread and force-kills the
   process tree, including any tracked child PIDs (the case where the engine
   runs under `cmd.exe` on Windows). `cleanup` stops the capture thread and
   releases file handles.

`KitSession.is_alive` adds an output-activity check on top of the bare
`process.poll()`. It treats the session as dead when no stdout has appeared for
`OUTPUT_INACTIVITY_TIMEOUT` (180 seconds), which catches the "process alive but
hung" case a plain poll misses. Refer to
[references/subprocess-pattern.md](references/subprocess-pattern.md) for the
worked Kit pattern in depth, including the command builder and the
log-capture thread.

## Discovering Engines from engines.toml

The bundled plugins read configuration from `engines.toml`. The runner calls
`plugin.discover()` with no arguments. `KitPlugin.discover(self, engines_toml_path=None)`
extends the signature with an optional argument for tests, and any extra keyword
argument must default such that the zero-argument call behaves the same.

The lookup order both bundled plugins follow:

1. An explicit path passed to `discover()`.
2. The `SIMREADY_ENGINES_TOML` environment variable. If it is set but points at
   a missing file, raise `FileNotFoundError`; explicit configuration must never
   silently fall through.
3. `./engines.toml` in the current working directory.
4. `~/.simready-benchmark/engines.toml`.

`KitPlugin` reads a top-level `[kit]` table and treats each `[kit.<instance>]`
subtable as one installation:

```toml
[kit.local]
executable_path = "C:/isaac-sim/isaac-sim.bat"
version = "4.5.0"
tags = ["kit", "isaac"]
custom_args = []
```

Each subtable becomes one `EngineConfig`. A relative `executable_path` is
resolved against the `engines.toml` location. A third-party plugin is not
required to use `engines.toml`; it can inspect the filesystem, query a registry,
or accept configuration any other way. The framework only requires that
`discover()` returns a `List[EngineConfig]`.

## Entry-Point Registration

Plugins register through the `simready_benchmark.engines` entry-point group. Unlike
test packs, which register a bare module path, an engine plugin value is
`module:ClassName` with the class suffix, because the registry instantiates the
class directly.

```toml
[project.entry-points."simready_benchmark.engines"]
acme_engine = "acme_engine_plugin.plugin:AcmeEnginePlugin"
```

The key on the left is an arbitrary label. The value on the right is the dotted
import path to the `EnginePlugin` subclass. The bundled Kit plugin registers
`kit = simready_benchmark_engine_kit:KitPlugin`, where `KitPlugin` is re-exported from
the package `__init__`, so the value resolves through the package rather than a
submodule.

The registry calls `importlib.metadata.entry_points(group="simready_benchmark.engines")`,
loads each class, instantiates it with no arguments, and rejects any result that
is not an `EnginePlugin` instance with a logged warning.

## How to Verify

1. Install the package into the same environment as `simready-benchmark`. From the
   package root, `pip install -e .` works.
2. Confirm the entry point loads:

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

3. Run `simready-benchmark --show-config`. Your plugin and its discovered
   `EngineConfig` entries appear alongside `kit` and any other installed plugin.
4. Run `simready-benchmark --list-tests`. Tests whose engine tags match your config
   tags appear in the listing.
5. Run against an asset. The framework plans, spawns your session, drains
   events, and aggregates the `result.json` files into the HTML report.

## Key Rules

- Implement all six `EnginePlugin` methods and all three `EngineSession`
  methods.
- `start_session`, `stop_session`, and `cleanup` are `async`. Use
  `await asyncio.sleep(...)` for polling, and hand any blocking wait or join to
  `loop.run_in_executor`. A blocking call freezes every concurrent session.
- The entry-point value is `module:ClassName`. The class suffix is required for
  engine plugins.
- Keep `tags()` disjoint across installed plugins by convention. The runner
  binds each config to the plugin that discovered it, so overlap does not break
  dispatch, but it does break human reasoning about which plugin owns a
  capability. Put overlapping capability tags on individual `EngineConfig`
  entries instead.
- `discover()` returns only configurations this plugin can drive. An empty list
  is a valid result, not an error.
- Keep `validate()` cheap and structural. It runs before every `start_session`,
  often many times per invocation. Refuse broken configs with `valid=False`;
  surface soft issues as `warnings`. Save network probes and launches for
  `start_session`.
- Every session must write `result.json` for each completed test and must put
  the orchestrator's canonical `asset_key` in the `"asset"` field. The reporter
  aggregates by that field; an absolute or engine-local path splits one asset
  into two report entries.

For the full ABC contract, refer to [references/engine-abcs.md](references/engine-abcs.md).
For the Kit subprocess pattern in depth, refer to [references/subprocess-pattern.md](references/subprocess-pattern.md).
