---
name: embed-in-pipeline
description: Use when integrating the simready-benchmark framework into a larger pipeline, CI job, or service, either by driving the Python API in-process or by invoking the CLI as a subprocess and consuming its report and exit code. Covers building the required discovery and validation-source objects, gating on the run outcome, and streaming progress events.
license: Apache-2.0
compatibility: Requires simready-benchmark (pip install simready-benchmark). In-process embedding uses the public Planner, Runner, and Reporter API. Subprocess embedding shells out to the simready-benchmark CLI. Kit-backed tests also require simready-benchmark-engine-kit and a separately-installed Kit / Isaac Sim runtime.
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`, and treat the installed module docstrings and `simready-benchmark --help` as the authoritative source for the API surface described here. A signature, import path, or argument documented here might have changed in a different version.

## Choose Subprocess or In-Process

Two integration modes, with different trade-offs:

- **Subprocess.** Your pipeline runs `simready-benchmark` as a child process and reads its exit code and JSON report. Simplest, most robust, and fully isolated. Prefer this unless you have a specific reason to embed in-process.
- **In-process.** Your pipeline imports `simready_benchmark` and drives `Planner`, `Runner`, and `Reporter` directly in the same interpreter. Use this when you need the live event stream, programmatic plan edits, or to avoid a second process.

## Remote Assets Require Their Backend In-Process

If any asset path is a Nucleus (`omniverse://`) or S3 (`s3://`) URL, the process that runs the framework needs that scheme's file-access backend, because discovery, the planner's validation reads, and the stamper all resolve the asset through the file-access layer.

- `omniverse://` requires `omni.client`, which ships with Kit and Isaac Sim, not with the standalone `simready-benchmark` wheel. For **in-process** embedding this means your own process must run inside Kit or Isaac Sim (or a Python that has `omni-client` installed) whenever you pass Nucleus assets, including for the planner and the stamper, not just discovery. For **subprocess** embedding, invoke the `simready-benchmark` CLI from that same Kit or Isaac Sim Python.
- `s3://` requires `boto3` (`pip install boto3`).

Without the backend, the framework raises a clear error on first use rather than silently skipping the asset. Local paths need neither dependency, so a standalone Python is enough when every asset is local.

## Subprocess Mode

Invoke the CLI and gate on the exit code. The exit code is the contract: `0` means every test passed, `1` means one or more failed, `2` means invalid arguments, `3` means a required plan or results directory was not found, and `4` means setup is not ready (fix the environment and re-run). Refer to the run-tests skill for the full flag set and exit-code table.

```python
import json
import os
import subprocess


def run_simready_subprocess(project_config: str, output_dir: str) -> int:
    proc = subprocess.run(
        ["simready-benchmark", "--project-config", project_config, "--output-dir", output_dir],
        capture_output=True,
        text=True,
    )
    # 0 = all passed, 1 = test failures. 4 = setup not ready: the environment
    # could not run part of the plan, so the results are not trustworthy. Treat
    # it as an environment problem to fix and re-run, not as a test verdict.
    # Anything else is an operational error.
    if proc.returncode == 4:
        raise RuntimeError(
            "simready-benchmark exited 4 (setup not ready). Check the Engines section "
            "of the report:\n%s" % proc.stderr
        )
    if proc.returncode not in (0, 1):
        raise RuntimeError(
            "simready-benchmark exited %d:\n%s" % (proc.returncode, proc.stderr)
        )

    # Optional: read the machine-readable report for per-asset detail.
    report_path = os.path.join(output_dir, "report", "test_results_index.json")
    with open(report_path, encoding="utf-8") as handle:
        report = json.load(handle)

    return proc.returncode
```

## In-Process Mode

`Planner.plan()` takes two objects that have no zero-argument constructor: a `discovery` and a `validation_source`. Building them is the part an integrator gets wrong. Use the concrete classes below, not the `AssetDiscovery` / `ValidationSource` protocols (those are typing protocols and cannot be instantiated).

```python
import os

from simready_benchmark import Planner, Runner, Reporter
from simready_benchmark.core.models import save_plan
from simready_benchmark.core.discovery import discover_test_dirs
from simready_benchmark.core.asset_discovery import ExplicitListDiscovery
from simready_benchmark.planner.validation_sources import (
    ChainedSource,
    SiblingJsonSource,
    SidecarJsonSource,
    UsdLayerSource,
)


def run_simready_in_pipeline(
    asset_paths: list[str],
    sr_specs_path: str,
    output_dir: str,
    test_paths: list[str] | None = None,
    feature_overrides: list[str] | None = None,
) -> int:
    # 1. Discovery. ExplicitListDiscovery takes the asset files and directories
    #    directly. For a content tree, use FilesystemWalkDiscovery(content_root)
    #    instead; for simready_search, use AssetSearchDiscovery(project_config).
    discovery = ExplicitListDiscovery(asset_paths)

    # 2. Validation source. This is the default chain the CLI builds: sibling
    #    JSON, then the .simready/validation.json sidecar, then USD layer data.
    validation_source = ChainedSource([
        SiblingJsonSource(),
        SidecarJsonSource(),
        UsdLayerSource(),
    ])

    # 3. Test packs. discover_test_dirs combines entry-point-installed packs
    #    with any extra directories. With no packs installed and no extra
    #    paths, the plan contains zero tests (refer to the zero-test trap below).
    test_dirs = discover_test_dirs(extra_paths=test_paths or [])

    # 4. Build the plan. feature_overrides bypasses the validation gate for the
    #    listed feature prefixes, which is how you test unstamped assets.
    plan = Planner.plan(
        sr_specs_path=str(sr_specs_path),
        test_dirs=test_dirs,
        discovery=discovery,
        validation_source=validation_source,
        feature_overrides=feature_overrides,
    )

    # save_plan is optional; Runner.start takes the Plan object directly. Save
    # it for a record and so --resume / --report-only can find it later.
    save_plan(plan, os.path.join(output_dir, "plan.json"))

    # 5. Run and wait. Runner.start returns a RunHandle immediately and runs in
    #    a background thread. wait() blocks and returns the run summary dict.
    handle = Runner.start(plan, output_dir=output_dir, max_concurrent=2, stamp=True)
    summary = handle.wait()

    # 6. Report. Writes index.html plus report/test_results_index.json.
    Reporter.generate(output_dir=output_dir, fmt="both")

    # 7. Gate. Refer to "Gate on the Right Signal" below.
    if summary is None:
        return 2  # global timeout fired before completion
    if summary["total_work_items"] == 0:
        raise RuntimeError(
            "No tests were planned. Check sr_specs, installed test packs, "
            "and feature_overrides."
        )
    # Setup readiness outranks test failures: a run that could not start part
    # of the plan is not a trustworthy pass or fail.
    if summary["readiness"]["status"] == "not_ready":
        return 4  # fix the environment (refer to the report Engines section)
    return 0 if summary["failed"] == 0 else 1
```

The framework creates `output_dir` and its `state/`, `results/`, and `report/` subdirectories. The original asset directory is never written to.

To find the feature IDs to pass as `feature_overrides`, call `Planner.discover(sr_specs_path, test_dirs)`, which returns the capability tree (`profiles`, `features`, `tests`), or run `simready-benchmark --list-tests`. Passing `feature_overrides=None` (the default) applies no bypass, so unstamped assets plan zero tests. `stamp=True` writes the run outcome into the mirror-copy USDs after a successful run; refer to the stamp-results skill for what that produces.

## Gate on the Right Signal

`handle.wait()` returns a dict with `total_work_items`, `completed`, `failed`, `status`, `readiness`, `started_at`, and `finished_at`. The same dict is written to `<output_dir>/run_summary.json`.

- Check readiness first. `summary["readiness"]["status"] == "not_ready"` (exit code `4` from the CLI) means the environment could not run part of the plan: an engine failed preflight, or no engine covered a required capability. Treat this as "fix setup and re-run", distinct from a test-failure gate. Readiness outranks failures, so `4` outranks `1`: a run that could not start is not a trustworthy pass or fail.
- Gate on `summary["failed"] == 0`. This is exactly what the CLI does to choose its exit code once the run is ready.
- Do NOT gate on `summary["status"]`. `status` is the orchestration state (`"completed"` or `"cancelled"`), not a pass or fail verdict. A run can complete with failures.
- Treat `summary["total_work_items"] == 0` as an error, not a pass. A plan with no tests "passes" trivially, which usually means the test packs are not installed or `feature_overrides` did not match.
- `wait()` returns `None` only if it times out (default global timeout 86400 seconds; override with `Runner.start(..., global_timeout=...)`).

## Stream Progress Events

For live progress, logging, or a CI heartbeat, iterate the event stream instead of blocking on `wait()`. `handle.events()` yields one dict per event until the run reaches a terminal state. Every event has a `type` (a value from `EventType`) and a `timestamp`, plus per-type fields.

```python
handle = Runner.start(plan, output_dir=output_dir)
for event in handle.events():
    if event["type"] == "test_complete":
        logger.info("test done: %s", event)
    elif event["type"] in ("engine_crash", "engine_stuck", "run_error"):
        logger.error("engine problem: %s", event)
summary = handle.wait()  # returns immediately once events() has drained
```

The same events are appended to `<output_dir>/state/events.jsonl`, so a subprocess-mode pipeline can tail that file for the identical stream.

## Publish Results

To copy a completed output directory to a shared location (local, `s3://`, or `omniverse://`), either pass `publish_to` to the reporter or call `publish_results` directly.

```python
from simready_benchmark import publish_results

publish_results(output_dir, "s3://my-bucket/runs/2026-05-29/")
```

## Key Rules

- `discovery` and `validation_source` are required by `Planner.plan()`. Build them from the concrete classes (`ExplicitListDiscovery` / `FilesystemWalkDiscovery` / `AssetSearchDiscovery`, and `ChainedSource([...])`). The `AssetDiscovery` and `ValidationSource` protocols are not instantiable.
- Gate on `summary["failed"]`, never on `summary["status"]`.
- Guard against a zero-test plan (`total_work_items == 0`); it is almost always a configuration miss, not a pass.
- Pass `feature_overrides` to test unstamped assets, otherwise the validation gate skips features with no validation claim.
- `Runner.start` returns immediately; you must call `wait()` or drain `events()` for the run to finish.

For the full signatures of every class and the complete `EventType` list, refer to [references/embedding-api.md](references/embedding-api.md). For the broader Planner, Runner, and Reporter surface, refer to the run-tests skill.
