# Embedding API Reference

Exact signatures for the objects an in-process integration needs. The run-tests
skill's `references/python-api.md` covers the broader `Planner`, `Runner`, and
`Reporter` surface; this file focuses on the parts an integrator must construct.

## Asset Discovery

All three concrete classes are importable from `simready_benchmark.core.asset_discovery`
and satisfy the `AssetDiscovery` typing protocol (which is itself not instantiable).
Each exposes `discover() -> list[str]` returning absolute, slash-normalized, sorted
interface-USD paths.

```python
from simready_benchmark.core.asset_discovery import (
    ExplicitListDiscovery,   # __init__(self, paths: list[str])
    FilesystemWalkDiscovery, # __init__(self, content_root: str)
    AssetSearchDiscovery,    # __init__(self, project_config_path: str)
)
```

- `ExplicitListDiscovery(paths)`: takes files and directories. Files ending in
  `.usd` / `.usda` / `.usdc` are taken verbatim; directories are walked with the
  same rules as `FilesystemWalkDiscovery`.
- `FilesystemWalkDiscovery(content_root)`: walks a content root, pruning internal
  directories (`dcc_source/`, `Payload/`, `payloads/`, `configuration/`, dot
  directories). Raises `AssetDiscoveryUnavailable` if `content_root` is missing.
- `AssetSearchDiscovery(project_config_path)`: wraps `simready.search.AssetLibrary`.
  Requires the search dependency; raises `AssetDiscoveryUnavailable` if the import
  fails.

Remote schemes route through their `FileAccess` backend, imported only when a
remote path is used: `omniverse://` needs `omni.client` (which ships with Kit and
Isaac Sim, not the standalone wheel, so embed in-process from inside Kit when you
pass `omniverse://` assets), and `s3://` needs `boto3`. The backend raises a clear
error on first use when its dependency is absent.

## Validation Sources

All importable from `simready_benchmark.planner.validation_sources`. Each exposes
`read(asset_path: str) -> AssetValidation | None`. The default chain mirrors the
CLI's `_build_validation_chain()`.

```python
from simready_benchmark.planner.validation_sources import (
    ChainedSource,                  # __init__(self, sources: list[ValidationSource])
    SiblingJsonSource,              # __init__(self)
    SidecarJsonSource,              # __init__(self)
    UsdLayerSource,                 # __init__(self)
    WorkspaceCacheValidationSource, # __init__(self, project_config_path: str)
)

validation_source = ChainedSource([
    SiblingJsonSource(),   # <asset>.json sibling
    SidecarJsonSource(),   # .simready/validation.json sidecar
    UsdLayerSource(),      # USD customLayerData.SimReady_Metadata.validation
])
```

`ChainedSource.read()` prefers a record with per-feature data (`validated_features
is not None`) over a profile-only record, regardless of chain order, and falls back
to a profile-only record only when no source has per-feature data. It swallows
per-source errors (logged at DEBUG). `WorkspaceCacheValidationSource` needs a project config
and raises `ValidationSourceUnavailable` if the workspace library cannot load; the
CLI appends it only under `--use-asset-search`.

### The Validation Gate

`Planner.plan()` only schedules a feature against an asset when the validation
source reports that feature as validated for the asset. For unstamped assets (no
validation metadata), pass `feature_overrides=["FET003", ...]` to bypass the gate
for those feature-ID prefixes. This is the Python equivalent of the CLI `--features`
flag.

## Test Directory Discovery

```python
from simready_benchmark.core.discovery import discover_test_dirs

test_dirs = discover_test_dirs(extra_paths=None)  # -> list[str]
```

Combines directories from packages registered under the `simready_benchmark.tests`
entry-point group with any `extra_paths` you pass. Extra paths combine with, never
replace, the entry-point packs. A non-directory extra path is skipped with a
warning.

## Planner

```python
from simready_benchmark import Planner

Planner.discover(sr_specs_path: str, test_dirs: list[str]) -> dict
Planner.plan(
    sr_specs_path: str,
    test_dirs: list[str],
    discovery: AssetDiscovery,
    validation_source: ValidationSource,
    feature_overrides: list[str] | None = None,
) -> Plan
```

`discover()` returns the capability tree (`profiles`, `features`, `tests`,
`warnings`). `plan()` returns a `Plan` dataclass. Both are static methods.

## Plan I/O

```python
from simready_benchmark.core.models import load_plan, save_plan

plan = load_plan("path/to/plan.json")   # -> Plan
save_plan(plan, "path/to/plan.json")
```

`Runner.start()` accepts the `Plan` object directly, so `save_plan` is optional. Save
it if you want `--resume` or `--report-only` to find the plan later.

## Runner

```python
from simready_benchmark import Runner

Runner.start(
    plan: Plan,
    output_dir: str,
    engine_plugins: dict[str, EnginePlugin] | None = None,
    max_concurrent: int = 2,
    stamp: bool = False,
    session_mode: str | None = None,   # "batch" | "per_asset"
    **kwargs,
) -> RunHandle
```

Returns immediately; execution runs in a background daemon thread. When
`engine_plugins` is `None`, the runner discovers every plugin registered under the
`simready_benchmark.engines` entry-point group. Recognized `**kwargs`:

| Kwarg | Purpose |
|---|---|
| `runtime=["isaac_sim", ...]` | Restrict to named engines from `engines.toml`. Unknown names hard-fail. |
| `global_timeout=86400` | Watchdog timeout in seconds. |
| `engines_toml=<path>` | Override the 4-tier `engines.toml` lookup. |
| `engine_path=<path>` | Direct engine-binary override (debug). |

## RunHandle

```python
from simready_benchmark.runner.run_handle import RunState  # str Enum

handle.wait(timeout: float | None = None) -> dict | None  # run summary, or None on timeout
handle.events() -> Iterator[dict]                         # blocking until terminal state
handle.is_done() -> bool
handle.cancel() -> None
handle.state      # RunState: RUNNING | COMPLETED | CANCELLED | ERROR
handle.cancelled  # bool
```

### Run Summary Dict

Returned by `wait()` and written to `<output_dir>/run_summary.json`:

| Key | Type | Meaning |
|---|---|---|
| `total_work_items` | int | Total scheduled (asset, test) items. `0` means nothing was planned. |
| `completed` | int | Items that finished. |
| `failed` | int | Items that failed. Gate on this once the run is ready. |
| `status` | str | Orchestration state: `"completed"` or `"cancelled"`. NOT a pass/fail verdict. |
| `readiness` | dict | `{"status": "ready" \| "degraded" \| "not_ready"}`. `not_ready` maps to CLI exit `4` and outranks failures: the environment could not run part of the plan. Check this before gating on `failed`. |
| `started_at` | str | ISO 8601 start time. |
| `finished_at` | str | ISO 8601 finish time. |

## Event Stream

`handle.events()` yields one dict per event; the same events are appended to
`<output_dir>/state/events.jsonl`. Every event has at least:

```python
{"type": "<EventType value>", "timestamp": "<ISO 8601>", ...}
```

Per-type fields vary; for example `run_complete` carries a `summary` key. The full
`EventType` enum (`simready_benchmark.core.protocol.EventType`):

| Group | Values |
|---|---|
| Run | `run_start`, `run_complete`, `run_error`, `run_cancelled` |
| Engine | `engine_starting`, `engine_loading`, `engine_ready`, `engine_warning`, `engine_error`, `engine_crash`, `engine_stuck`, `engine_stop`, `engine_readiness` |
| Driver | `driver_warning` |
| Asset | `asset_start`, `asset_loading`, `asset_warning`, `asset_loaded`, `asset_complete`, `asset_stamp` |
| Test | `test_start`, `test_complete`, `test_fail`, `test_incomplete` |
| Progress | `step`, `requirement_complete`, `feature_complete`, `profile_complete`, `heartbeat` |

## Reporter

```python
from simready_benchmark import Reporter

Reporter.generate(
    output_dir: str | None = None,
    fmt: str = "both",          # "html" | "json" | "both"
    publish_to: str | None = None,
    **kwargs,
) -> dict
```

Static method. Writes `index.html` (plus `page-N.html`) and
`report/test_results_index.json` under `output_dir`, and returns a dict with
`metadata`, `assets`, and `engines` keys. Passing `publish_to` also publishes the directory.

## Publishing

```python
from simready_benchmark import publish_results

publish_results(output_dir: str, destination: str) -> None
```

Walks `output_dir` and copies every file to `destination`, preserving relative
layout. `destination` can be a local path, `s3://...`, or `omniverse://...`. Raises
`FileNotFoundError` if `output_dir` does not exist, and `RuntimeError` if a remote
destination needs an unavailable dependency (`boto3` for S3, `omni.client` for
Omniverse).

## Version and Protocol Guards

```python
from simready_benchmark import get_version, PROTOCOL_VERSION

get_version()      # installed wheel version, for example "2026.5.1"; falls back to __version__
PROTOCOL_VERSION   # on-disk plan / result.json schema version, for example "1.1.0"
```

Use `PROTOCOL_VERSION` to refuse a plan or result file written by an incompatible
framework version.
