[FEAT]: pytest-xdist support for distributed test execution (#73)

## Description

### Summary

Adds first-class `pytest-xdist` support to RAMPART so attack/probe
sessions can shard across worker processes and still produce a single
coherent report. Worker processes serialize their results through
xdist's `workeroutput` channel; the controller deserializes, merges, and
writes the final report.

### What's included

**New: `_xdist.py`**
- Versioned schema (`rampart.xdist.v1`) for worker → controller payloads
- `_sanitize` JSON-safe coercion at the trust boundary with depth limit,
byte-size cap, and ANSI-escape stripping on deserialize to defend
against terminal injection from worker output
- `SizeLimitError` raised by `finalize_worker` when the serialized
payload exceeds the configured cap; controller logs and continues
- Configurable cap via new pytest CLI option `--rampart-xdist-max-bytes`
/ ini option `rampart_xdist_max_bytes` (default 64 MiB), validated
positive
- Sink resolution re-raises `KeyboardInterrupt`/`SystemExit`, catches
`Exception`

**Trial-group aggregation across workers**
- Trial metadata is captured as data on `RampartSession` at collection
time and shipped through the `rampart.xdist.v1` payload (`trial_specs`),
so the controller aggregates trial groups from merged worker data
instead of `session.items` — which is not reliably populated with trial
clones on the controller at session finish. Without this, trial-group
FAIL/PASS verdicts silently disappeared under xdist even though
per-clone results were present.
- `trial_specs` is back-compatible: payloads without trials emit an
empty list; malformed entries are skipped and non-finite thresholds are
clamped on deserialize.

**Deterministic result ordering & sink resolution (review feedback)**
- Merged results sort by a stable key `(_pytest_nodeid,
_rampart_result_index, _rampart_source_worker)` so aggregated reports
are reproducible regardless of worker completion order. These
bookkeeping keys are namespaced with leading underscores so they cannot
collide with user-supplied result metadata.
`_pytest_nodeid`/`_rampart_result_index` are assigned authoritatively
(absorb-time enumerate; deserialize-time outer-key + enumerate);
`_rampart_source_worker` is tagged on `handle_testnodedown`. The full
node ID (not the trailing `test_name`) is the primary sort key, so
same-named tests in different files, parametrized cases, and trial
clones no longer collide.
- New `pytest_rampart_sinks` hook (`_hooks.py`) is the authoritative
sink source — keyed on hook-impl existence, not truthiness — and
short-circuits the fixture path so projects defining both don't
double-register. The `rampart_sinks` fixture remains the single-process
fallback.
- Controller-side discovery **unwraps a `@pytest.fixture`-wrapped
parameterless `rampart_sinks`** to its underlying function
(`_get_wrapped_function()` with `_fixture_function`/`__wrapped__`
fallbacks, wrapped in try/except) and calls it directly, restoring the
documented session-fixture fallback under `-n`. Fixtures that depend on
other fixtures now warn and point at the hook instead of silently
disappearing.
- `is_xdist_controller` keys off the active distribution (`dist != "no"`
plus `-n`/`--tx` endpoints), not the worker count alone, so `-d`/`--tx`
runs without `-n` and custom schedulers resolve to the controller branch
instead of the empty single-process path.
- Shared `strip_ansi` extracted to `rampart/common/text.py`
(CSI/OSC/DCS/SOS/PM/APC/NF + 7-bit and 8-bit C1, then residual C0/C1
stripping, preserving tab/newline/CR) and reused by both the terminal
summary and the xdist boundary.

**Incomplete runs fail loudly (review feedback)**
- A lost or crashed xdist worker can leave every surviving test green.
For a safety framework "green" must mean "ran and passed", so
`pytest_sessionfinish` forces a non-zero exit status (`OK →
TESTS_FAILED`) whenever the merged session is incomplete. An
already-failing status is left untouched so a stronger failure signal
(e.g. `INTERRUPTED`) is never masked. Workers early-return before
enforcement, so a worker's local incompleteness never decides the
controller's exit code.
- Incomplete-run reasons are also surfaced as a prominent terminal
warning (before the `has_results` guard) when workers crash or report
partial data.

**Updated plugin hooks (`plugin.py`, `_session.py`)**
- `pytest_addoption` registers the new CLI/ini option; `pytest_addhooks`
registers `pytest_rampart_sinks`
- Worker, controller, and non-xdist branches dispatched cleanly; shared
aggregation/gating is hoisted above the branch and only xdist metadata +
sink discovery are guarded by `is_xdist_controller`; the worker branch
wraps `finalize_worker` to surface size-limit truncation as a warning

**Reporting**
- `JsonFileReportSink` now projects `report.metadata`, so xdist run-mode
info (worker_count, dist_mode, incomplete reasons) lands in the emitted
JSON file. The xdist transport serializer and the JSON-report serializer
are deliberately kept separate (full-fidelity round-trip vs. flatter
public shape), with doc-comments on both explaining why they must not be
merged.

**Tests**
- 82 unit tests in `test_xdist.py` (serialization, sanitization, size
limits, ANSI stripping, controller detection, sink resolution/unwrap,
option parsing, ordering determinism incl. reverse-order merge,
trial-spec round-trip/merge)
- 13 unit tests in `test_text.py` for shared `strip_ansi` (CSI, OSC
BEL/ST, DCS, 8-bit CSI, lone C1, residual C0, whitespace preservation,
no over-stripping)
- `test_plugin.py` extended (60 tests) for sink-hook resolution, OSC
sanitizing, the incomplete-warning path, and incomplete-run exit-status
enforcement (`TestIncompleteExitStatus`)
- `test_json_file.py` extended (14 tests) for metadata projection
- Slow-marked unit `test_xdist_aggregation.py` (13 tests) exercising
end-to-end aggregation across real worker subprocesses, including
trial-group verdicts under both `--dist=loadgroup` and `--dist=load`
(run with `-m slow`)

**Docs**
- New `xdist.md` user guide, incl. the `pytest_rampart_sinks` hook, sink
precedence, `--dist=loadgroup` vs default `load` guidance, and a
"Durability limitations" section (workeroutput delivers at worker
shutdown; an over-cap payload is dropped wholesale — durable per-worker
shards are a stacked follow-up)
- Configuration, CI-integration, pytest-integration,
results-and-reporting, and API reference pages updated to describe the
new option (under a dedicated "Pytest Options" section, not under env
vars)
- Architecture page notes the controller/worker split

**Incidental**
- `onedrive.py`: file-level pyright directive suppressing the `Unknown*`
cascade caused by `msgraph-sdk` shipping without type stubs. No behavior
change.
- `_xdist.py`: `# ty: ignore[unresolved-attribute]` on the pytest-xdist
runtime `config.workeroutput` attribute so `ty check` passes under the
repo's pyright → ty migration.

### Security notes

The xdist boundary treats worker output as untrusted:
- All values pass through `_sanitize` → JSON-safe primitives only
- Strings are stripped of the full ECMA-48 escape family on deserialize
(CSI, OSC, DCS/SOS/PM/APC, 7-bit and 8-bit C1, plus residual C0/C1
control bytes), including nested strings inside dicts/lists, to prevent
terminal injection (e.g. OSC hyperlinks/title sets) from a compromised
or misbehaving worker. The sanitizer has no catastrophic-backtracking
exposure.
- Total payload size is capped to prevent controller memory exhaustion;
truncation is recorded with a marker and raised as `SizeLimitError`
- Depth cap (`MAX_METADATA_DEPTH = 6`) prevents pathological nesting

### Validation
- `uv run ruff check rampart tests` → all checks pass; `ruff format
--check` clean
- `uv run ty check` → all checks passed
- `uv run pytest tests/unit -m "not slow"` → 597 passed
- `uv run pytest tests/unit/pytest_plugin/test_xdist_aggregation.py -m
slow` → 13 passed (trial-group verdicts verified under
`--dist=loadgroup` and `--dist=load`)
- `uv run --group docs mkdocs build --strict` → no cross-reference
warnings (remaining output is Windows-only symlink noise from the
font-download plugin; clean on CI/Linux)
- Ran `pytest -n 4 --dist=loadgroup` against the HelpDesk Bot example in
`rampart-examples`: 1 report generated. Running with xdist took ~99.66s
versus ~230.73s serial (yay for speed!)

## Breaking changes

None

## Checklist

- [x] `pre-commit run --all-files` passes
- [x] Tests added or updated for changes
- [x] Documentation updated

---------

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: nina-msft

docs/api/pytest-plugin.md

@@ -14,3 +14,27 @@ RAMPART's pytest integration. Activates automatically when installed.
       members:
         - RampartSession
         - TrialGroupResult
+
+## Parallel Execution Hooks
+
+When `pytest-xdist` is installed, the plugin registers `pytest_testnodedown` (as an optional hook) to merge worker results into the controller session. See [Parallel Execution](../usage/xdist.md) for the data flow and trust boundary.
+
+::: rampart.pytest_plugin._xdist
+    options:
+      members:
+        - SCHEMA_VERSION
+        - WORKEROUTPUT_KEY
+        - SIZE_LIMIT_OPTION
+        - DEFAULT_SIZE_LIMIT_BYTES
+        - WorkerOutputError
+        - SchemaVersionError
+        - SizeLimitError
+        - is_xdist_worker
+        - is_xdist_controller
+        - get_dist_mode
+        - get_worker_count
+        - serialize_worker_data
+        - deserialize_worker_data
+        - finalize_worker
+        - handle_testnodedown
+        - discover_sinks_from_conftest

docs/contributing/architecture.md

@@ -57,6 +57,15 @@ This allows the same evaluator (e.g., `ToolCalled`) to be used in both attack an
 
 Subclasses implement only `_execute_async` and `strategy_name`. They should **not** catch `InfrastructureError` — the base class handles it.
 
+### Pytest Plugin
+
+`pytest_plugin/` integrates RAMPART with pytest:
+
+- `plugin.py` — hook registrations (configure, collection, sessionfinish, terminal summary, optional `pytest_testnodedown`).
+- `_session.py` — session-scoped state container (`RampartSession`), trial-group aggregates, sink registry, idempotency and incomplete-run flags.
+- `_collection.py` — per-test `ResultCollector` and the `ContextVar`-based handler that captures results from executions.
+- `_xdist.py` — pytest-xdist support: detection helpers, JSON-safe serialization of `Result` objects, controller-side merge, and conftest-scanning sink discovery. Workers serialize their results into `config.workeroutput`; the controller deserializes via `pytest_testnodedown` and emits a single unified report. See [Parallel Execution](../usage/xdist.md) for the data flow and trust boundary.
+
 ### PyRIT Bridge
 
 PyRIT is RAMPART's upstream dependency for converters and prompt generation. Its import chain is heavy, so:

docs/index.md

@@ -35,4 +35,5 @@ You provide an **adapter** that connects your agent to the framework. RAMPART pr
 - **Execution strategies** — orchestrate injection, triggering, and evaluation lifecycles
 - **Evaluators** — detect conditions in agent responses (tool calls, text patterns, side effects)
 - **pytest integration** — markers for harm categorization and statistical trials, automatic result collection, terminal summaries
+- **Parallel execution** — run tests across worker processes with `pytest-xdist`; RAMPART produces a single unified report
 - **Reporting** — structured JSON output for CI dashboards

docs/usage/ci-integration.md

@@ -16,6 +16,17 @@ RAMPART tests interact with real or simulated agents and may take longer than un
 pytest tests/ -v --timeout=300
 ```
 
+### Parallel Execution
+
+For faster CI runs, use [`pytest-xdist`](xdist.md):
+
+```bash
+pip install pytest-xdist
+pytest tests/ -n auto
+```
+
+RAMPART aggregates results across worker processes and emits a single unified report under **any** `--dist` mode. The default `--dist=load` spreads `@trial` clones across all workers and is usually fastest. Add `--dist=loadgroup` only when a trial group needs to stay on one worker (e.g. clones share a session fixture or per-group worker state). See [Choosing `loadgroup` vs `load`](xdist.md#choosing-loadgroup-vs-load) for details and security considerations.
+
 ---
 
 ## Trial Markers for Statistical Confidence
@@ -57,11 +68,28 @@ def rampart_sinks() -> list[ReportSink]:
 
 The JSON file contains aggregate statistics and per-result data that CI dashboards can consume.
 
+!!! tip "Running in parallel"
+    Under [`pytest-xdist`](xdist.md), prefer the `pytest_rampart_sinks` hook over the fixture — it is resolved on the controller, so it works the same in single-process and parallel CI runs. See [Registering Sinks](pytest-integration.md#pytest_rampart_sinks-hook).
+
+---
+
+## Pytest Options
+
+RAMPART is configured via pytest options and Python (sinks, adapters, payloads).
+
+### `--rampart-xdist-max-bytes`
+
+Maximum size in bytes of a worker's serialized result payload when running under [`pytest-xdist`](xdist.md). Defaults to `67108864` (64 MB). Workers that exceed the cap log a warning and the controller marks the run as incomplete. Also configurable via the `rampart_xdist_max_bytes` ini option.
+
+```bash
+pytest -n auto --rampart-xdist-max-bytes=134217728   # 128 MB
+```
+
 ---
 
 ## Environment Variables
 
-RAMPART itself does not read environment variables. Your adapter and test configuration typically do. Setting them locally for ad-hoc runs:
+Your adapter and test configuration typically read environment variables. Setting them locally for ad-hoc runs:
 
 === "Linux / macOS"
 

docs/usage/configuration.md

@@ -4,6 +4,16 @@ RAMPART's configurable components: [`LLMConfig`][rampart.core.llm.LLMConfig] for
 
 ---
 
+## Parallel-execution tuning
+
+RAMPART exposes one pytest option for parallel-execution tuning. Other components (LLM endpoints, agent configuration) typically have their own configuration conventions.
+
+| Option | Default | Description |
+|--------|---------|-------------|
+| `--rampart-xdist-max-bytes` (CLI) / `rampart_xdist_max_bytes` (ini) | `67108864` (64 MB) | Maximum size of a worker's serialized result payload when running under [`pytest-xdist`](xdist.md). Workers exceeding the cap are recorded as incomplete in `TestRunReport.metadata`. |
+
+---
+
 ## LLMConfig
 
 Immutable configuration for an LLM endpoint. Used by [`LLMDriver`][rampart.drivers.llm.LLMDriver] and [`Payloads.generate_async()`][rampart.payloads.Payloads.generate_async].

docs/usage/pytest-integration.md

@@ -68,6 +68,9 @@ async def test_with_threshold(adapter):
 - `ERROR` results count against the pass rate (they are not `SAFE`)
 - The trial group aggregate appears in the terminal summary
 
+!!! tip "Running trials in parallel"
+    Under [`pytest-xdist`](xdist.md), aggregation is correct under any `--dist` mode. The default `--dist=load` spreads trial clones across all workers and is usually fastest; use `--dist=loadgroup` only when a trial group must stay on one worker (shared session fixture or per-group worker state). See [Choosing `loadgroup` vs `load`](xdist.md#choosing-loadgroup-vs-load).
+
 ---
 
 ## Fixtures
@@ -98,6 +101,47 @@ def rampart_sinks() -> list[ReportSink]:
     ]
 ```
 
+!!! warning "xdist compatibility"
+    Under [`pytest-xdist`](xdist.md), the controller process discovers fixture-based sinks by calling `rampart_sinks` directly. Fixtures that depend on other fixtures (e.g., `tmp_path_factory`, `request`) cannot be resolved on the controller and are skipped with a warning. Use a parameterless fixture or a module-level list to remain compatible:
+
+    ```python
+    # Resolved on the xdist controller (controller-only — single-process
+    # discovery needs the fixture form above, or the hook below)
+    rampart_sinks = [JsonFileReportSink(output_dir=Path(".report"))]
+    ```
+
+    For sinks that need configuration or dependencies, prefer the
+    `pytest_rampart_sinks` hook below — it is resolved on the controller and works
+    identically in single-process and parallel runs.
+
+---
+
+### `pytest_rampart_sinks` hook
+
+For sinks that need configuration — or to register sinks in a way that behaves
+identically in single-process and `pytest-xdist` runs — implement the
+`pytest_rampart_sinks` hook in your `conftest.py`:
+
+```python
+# conftest.py
+from pathlib import Path
+
+from rampart.reporting import JsonFileReportSink
+
+
+def pytest_rampart_sinks(config):
+    return [JsonFileReportSink(output_dir=Path(".report"))]
+```
+
+The hook receives the active `pytest.Config`, so you can build
+sinks from CLI/ini options or environment variables. Multiple implementations are
+supported; RAMPART emits to the **union** of every returned sink.
+
+**Precedence:** when any `pytest_rampart_sinks` implementation exists, it is
+authoritative and the `rampart_sinks` fixture path is skipped entirely (so a
+project that defines both does not double-register). The fixture remains the
+single-process fallback when no hook implementation is present.
+
 ---
 
 ## Automatic Result Collection

docs/usage/results-and-reporting.md

@@ -91,6 +91,9 @@ class MyDatabaseSink:
 
 Define the `rampart_sinks` fixture in your `conftest.py`. See [pytest Markers & Fixtures](pytest-integration.md#rampart_sinks) for the setup and examples with multiple sinks.
 
+!!! note "Parallel execution"
+    Under [`pytest-xdist`](xdist.md), workers send their results to the controller, which emits sinks **once** with a unified [`TestRunReport`][rampart.reporting.sink.TestRunReport]. For sinks that need configuration, prefer the `pytest_rampart_sinks` hook, which is resolved on the controller and works the same in single-process and parallel runs. The `rampart_sinks` fixture is still supported as a single-process fallback, but on the controller it cannot depend on other fixtures. See [Registering Sinks](xdist.md#registering-sinks-the-pytest_rampart_sinks-hook) for details.
+
 ---
 
 ## TestRunReport