# The Transport-Driven Pattern in Depth

Authoritative source: `simready_benchmark/engines/nvcf_plugin.py`,
`simready_benchmark/engines/nvcf_session.py`,
`simready_benchmark/runner/work_pool.py`, and
`simready_benchmark/execution/worker_http.py`. This reference walks the NVCF plugin
and session as the canonical transport-driven implementation, end to end.

The worker HTTP contract described here (`POST /run-test`, `GET /health`, 503
plus `Retry-After`) ships as a Technology Preview and can change between
releases. The `EnginePlugin` and `EngineSession` authoring surface is stable.

## Discovering Endpoints

`NvcfPlugin.discover(engines_toml_path=None)` pulls worker endpoints from two
sources:

1. `[nvcf.<name>]` tables in `engines.toml`. Each table becomes one
   `EngineConfig`. The lookup order is an explicit path argument, then the
   `SIMREADY_ENGINES_TOML` environment variable, then `./engines.toml`, then
   `~/.simready-benchmark/engines.toml`.
2. CLI-injected URLs, passed as `NvcfPlugin(extra_urls=[...])`. The `--workers`
   flag wires into this. Each URL becomes a synthetic config named
   `cli_worker_<index>` with `api_key_env="NVCF_API_KEY"`.

For each discovered endpoint, the plugin records the URL and auth metadata in a
private side table, `self._url_by_config`, keyed by config name:

```python
self._url_by_config[config_name] = {
    "url": url,
    "api_key_env": config_data.get("api_key_env"),
    "function_id": config_data.get("function_id"),
    "version_id": config_data.get("version_id"),
}
```

The `EngineConfig` itself carries only identity: `name`, an empty
`executable_path`, a `version` (default `"nvcf-remote"`), and config-level tags
(default `["nvcf", "kit"]`). The transport details never touch the
`EngineConfig`, because the framework logs and serializes it. `start_session`
later looks the URL and auth up by `engine_config.name`.

`validate(engine_config)` reads the same side table, checks that the URL has a
scheme and a netloc through `urllib.parse.urlparse`, and warns (does not fail)
when the named API-key environment variable is unset, so a local unauthenticated
emulator still validates.

## The Worker Thread

`start_session` constructs an `NvcfSession` and awaits `session.start()`.
`start()` loads the plan, builds lookup tables, spawns a daemon
`threading.Thread` targeting `_run`, and awaits an `engine_ready` event with a
30-second deadline (`_ENGINE_READY_TIMEOUT`). If the event does not appear, it
raises `TimeoutError`.

The thread is the worker. It owns the claim loop for the life of the session and
runs a synchronous HTTP client. `is_alive()` returns `thread.is_alive()`
combined with a 300-second inactivity check (`IS_ALIVE_INACTIVITY_TIMEOUT`)
against the timestamp of the last emitted event. Long POSTs still count as
activity because `idle_keepalive` emissions bump the timestamp on every
no-work poll.

`cleanup()` sets the stop event and joins the thread through `run_in_executor`
with a 30-second deadline (`_THREAD_JOIN_TIMEOUT`), so the event loop is never
blocked while the thread winds down.

## The Two-Level Claim Loop

The `WorkPool` (`simready_benchmark/runner/work_pool.py`) is a file-locked
`work_pool.json` shared by every session. Construct it one time before the loop.
Claiming is two levels.

Level 1 is the asset claim. `claim_asset(session_id, engine_tags, plan)`
associates the session with a work item and returns a `WorkItem`, or `None` when
no compatible work remains. The pool filters work items by engine-tag
compatibility, so a session sees only work it can run. The priority order is
assets the session already works on, then fresh unclaimed assets, then assets
claimed by other sessions. Several sessions can work the same asset
concurrently.

Level 2 is the test claim. `claim_test(work_item_id, session_id)` atomically
flips one pending test from `pending` to `claimed` and returns its
`WorkTestEntry`, or `None` when no pending tests remain on that item. When it
returns `None`, the session releases the asset (`release_asset`) and goes back
to level 1.

`_run` loops over `claim_asset` and, for each claimed asset, calls
`_process_asset`, which loops over `claim_test`. When the pool has no work, the
session emits `idle_keepalive` and waits `NO_WORK_POLL_INTERVAL` (5 seconds) on
the stop event before polling again. The loop exits on the stop event (set by
`cleanup`), on running out of work, or on the unavailability timeout.

## The POST and Result Handling

For each claimed test the session derives the per-test output directory with
`per_test_runtime_dir(output_dir, asset_key, test_name)`, emits `test_start`
**before** the POST (so the runner no-progress watchdog sees activity while the
POST blocks), then dispatches.

`_post_run_test` picks the transport by the worker hostname. Any hostname other
than `api.nvcf.nvidia.com` takes the direct path (`_post_direct`); the managed
NVCF hostname takes the NVCF Functions path (`_post_via_nvcf_functions`). The
request body is identical for both transports:

```python
{
    "test_name": tdef.name,
    "test_version": tdef.version,
    "test_file": tdef.test_file,
    "asset_path": <resolved absolute host path>,
    "asset_key": <directory-safe relative key>,
    "config": <per-test config overrides>,
    "test_feature_names": [...],
    "asset_validated_features": [...] | None,
    "content_root": <SIMREADY_CONTENT_ROOT>,
}
```

The body carries `asset_key` (identity), never `output_dir` (a host path). This
decouples user-machine paths from worker-side paths, so the same contract works
for a shared-filesystem emulator, a bind-mounted Docker worker, and an NVCF
function with no shared filesystem.

### The Direct Path

`_post_direct` POSTs JSON to `<worker_url>/run-test` with a
`Content-Type: application/json` header. Bearer auth is applied opportunistically:
only when `api_key_env` is configured and the named environment variable is set.
The HTTP timeout is `2 * KIT_TEST_TIMEOUT_S` (the per-test timeout defaults to
1800 seconds through `SIMREADY_KIT_TEST_TIMEOUT_S`), set higher than the worker's
own timeout so the worker times out first and returns a structured failure rather
than a transport timeout. The response is unwrapped: if it is a dict with a
`result_json` dict inside, that inner dict is the result; otherwise the parsed
response is the result. A non-object response raises `RuntimeError`, which is an
application failure.

### The NVCF Functions Path

`_post_via_nvcf_functions` POSTs to
`/v2/nvcf/pexec/functions/<function_id>/versions/<version_id>` with a mandatory
bearer token, a `Content-Type` header, and an `NVCF-POLL-SECONDS` header clamped
to `_NVCF_POLL_SECONDS_MAX` (300). On HTTP 200 it unwraps and returns. On HTTP
202 (queued) it reads the `NVCF-REQID` header and polls
`/v2/nvcf/pexec/status/<reqid>` every `_NVCF_POLL_INTERVAL_S` (10 seconds) until
the status is `fulfilled` (return the result), `errored` or `failed` (return a
fail-shaped result), or the total elapsed time exceeds `KIT_TEST_TIMEOUT_S`
(return a fail-shaped timeout result). The 202-poll path is the NVCF-specific
addition over the direct path; the request body is unchanged.

### Completing the Test

After a successful POST the session calls `complete_test(work_item_id, test_id,
{"status": ..., "duration": ...})`, which returns `True` when the asset is fully
done. It then writes the host-side mirror `result.json` and emits
`test_complete`. When `complete_test` reports the asset done, the session calls
`release_asset` and emits `asset_complete`.

## Transport Failure Versus Application Failure

This split is the heart of the pattern, and the part a subprocess plugin does
not need. A local subprocess is either running or not; a network engine has a
"the engine is fine, the network is slow" mode to recover from.

`_is_transport_failure(exc)` returns `True` for exactly these:

- An `HTTPError` with status 503, 425, or 429. 503 is the worker saying it is
  not ready (for example a cold-starting engine), 425 is "Too Early," and 429 is
  "Too Many Requests." `HTTPError` is a subclass of `URLError`, so this check
  comes first.
- Any other `URLError` (an OS or socket-level error with no HTTP response).
- `TimeoutError`, `socket.timeout`, or `ConnectionError`.
- A bare `OSError` (DNS failure, socket reset before any response).

Everything else is an application failure: an `HTTPError` with any status code
not in {503, 425, 429}, a `RuntimeError` from a malformed response, and anything
else. The worker answered and said no, so the result counts.

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

A 503 must not write a failing `result.json`. The test bounces back to the pool.

## The Release-and-Probe Recovery

On a transport failure, `_handle_transport_failure` releases the in-flight test
and marks the engine unavailable:

- `release_test(work_item_id, test_id, session_id)` flips the test from
  `claimed` back to `pending`, clears its `session_id` and `claimed_at`, and
  removes the session from the work item's `claimed_by` list. It does NOT
  increment the retry count, because the test never ran. (Only a crashed running
  session increments retries, through `reclaim_crashed_tests`.)
- The session sets `_engine_available = False`, schedules the next probe at
  `now + _ENGINE_PROBE_INTERVAL_S` (15 seconds), records `_unavailable_since` if
  not already set, and emits an `engine_warning`.

The main loop then gates on availability before each `claim_asset`. While
unavailable it sleeps in `_UNAVAILABLE_SLEEP_S` (1-second) increments so the stop
event is observed promptly. When `now >= _next_probe_at`, it issues
`_probe_engine_health`, a `GET <worker_url>/health` with a 3-second timeout
(`_HEALTH_PROBE_TIMEOUT_S`) that returns `True` on any 2xx. On the first
successful probe the session flips back to available, clears
`_unavailable_since`, emits `engine_recovered`, and falls through into
`claim_asset` on the same iteration. On a failed probe it reschedules the next
probe 15 seconds out.

While the worker is down, other sessions (including a local Kit session in a
mixed-mode run) keep draining the pool, because the released test is `pending`
again and the engine-unavailable state is local to this one session.

## The Unavailability Timeout

If the engine stays unavailable longer than `_ENGINE_UNAVAILABLE_TIMEOUT_S`
(1800 seconds by default, configurable through the
`SIMREADY_NVCF_UNAVAILABLE_TIMEOUT_S` environment variable), the loop sets
`exit_reason = "nvcf_unavailable_timeout"`, breaks, and emits `session_exit`.
The session exits cleanly rather than spinning on a dead worker forever.

## The Mirror-Layout asset_key Forcing

The reporter aggregates per-test results by the `asset` field across every
plugin in a run. A worker can legitimately return an absolute path on its own
filesystem in its result payload. If that path lands in the host-side
`result.json`, the reporter sees two assets where there is one, and a mixed-mode
run shows the same asset twice under two keys.

`_write_result_json` forces the orchestrator's relative `asset_key` onto the
`asset` field with a direct assignment (not `setdefault`), so the worker's own
value never wins:

```python
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)
data.setdefault("engine", "kit")
data.setdefault("engine_version", self._cfg.version)
```

The file is written atomically so a partial write never leaves a half-baked
`result.json` on disk.

## The Worker HTTP Contract

The worker side (`simready_benchmark/execution/worker_http.py`) is the reference
contract the session POSTs against. It is a Technology Preview and can change.

- `POST /run-test` accepts the JSON body shown above and validates the required
  fields (`asset_path`, `test_name`, `test_file`, `test_version`, `config`,
  `content_root`, `asset_key`, `test_feature_names`, `asset_validated_features`).
  It derives its own worker-local output directory from `asset_key` and
  `test_name`, so it never sees a host path. A malformed body returns 400.
- When the worker engine is not yet ready, `POST /run-test` returns HTTP 503
  with a `Retry-After` header (advisory; the worker uses 15 seconds, which
  roughly matches the host-side probe interval). This is the transport-retryable
  signal: the session releases the test and backs off rather than failing it.
- All other dispatch outcomes (pass, fail, timeout-fail, malformed-result-fail)
  return HTTP 200 with the parsed result dict. The session treats these as
  application-level results.
- `GET /health` returns 200 when the worker engine is ready and its heartbeat is
  fresh, otherwise 503. The session's `_probe_engine_health` polls this during
  recovery.

The single-flight worker serializes concurrent POSTs with an in-process lock, so
each session targets one worker URL. To run several workers concurrently,
declare several `[nvcf.<name>]` configs; each drives its own session against its
own URL.
