[FEAT]: Durable xdist shard transport primitives (no wiring)

[nina-msft]

Description

First PR in a stack of two that adds durable result transport for pytest-xdist, closing two gaps in the current in-memory workeroutput transport:

• Durability — a worker killed mid-run currently loses all of its finished results (nothing is written until a clean pytest_sessionfinish).
• Granular size cap — a single oversized transcript currently replaces a whole worker's payload with a truncation marker, discarding every other result from that worker.

This PR lands only the primitives for the fix. Every symbol is a pure addition to rampart/pytest_plugin/_xdist.py with no call sites — existing behavior is completely unchanged. It establishes the vocabulary the wiring PR consumes, so it can be reviewed small and green on its own.

What's added

Symbol	Purpose
ShardWriter	Append-only, flush-per-result JSONL writer. A worker's finished results hit disk immediately, so they survive an abrupt worker death (e.g. SIGKILL). An oversized record is written as a marker line naming its nodeid/index instead of discarding the whole shard.
read_worker_shard	Controller-side recovery. Parses a worker shard line by line, tolerates a truncated/partial final line, and surfaces corrupt/oversized records as DroppedRecord entries so the caller can mark the run incomplete. nodeid/index are taken authoritatively from the record envelope, mirroring deserialize_worker_data's existing trust posture (never trusting values embedded in the serialized result body).
shard_eligible	Gates shard transport to runs where every --tx gateway is a local popen worker (shared filesystem). Remote (ssh=/socket=) or proxied (via=) topologies are ineligible and will fall back to workeroutput in PR B.
SHARD_DIR_KEY	Stash key constant for the controller→worker shard-directory channel used in PR B.

The full-fidelity Result round-trip reuses the existing _serialize_result/_deserialize_result pair, so shards carry the same fidelity as the current transport.

🔜 Coming in PR B — wire durable transport (the behavior change)

See draft here: nina-msft#3

PR A is inert on its own. The follow-up (dev/nina-msft/xdist-durable-transport, stacked on this branch) turns the primitives on:

• Controller creates the shard directory and hands it to workers via a pytest_configure_node channel (SHARD_DIR_KEY).
• Workers append each result to their shard as it completes and stamp "transport": "shard".
• The atomic flip: workeroutput is slimmed to a completeness sentinel (result bodies removed) and handle_testnodedown switches to reading results from shards — these two changes must land together, which is exactly why they're held out of this PR.
• Mandatory remote/non-eligible fallback to workeroutput (so single-process and remote runs are unaffected).
• Slow pytester tests proving the wiring: worker-kill recovery, oversized-record isolation, crash-item-absent, and clean-run parity; plus the docs/usage/xdist.md update.

Splitting here keeps the "flip the source of truth" as one reviewable semantic change and keeps this PR a zero-risk, additive foundation.

Breaking changes

None. No existing code path calls any of the new symbols; behavior is identical to main.

Checklist

• pre-commit run --all-files passes
• Tests added or updated for changes
• Documentation updated

[Copilot]

Pull request overview

Adds the foundational, (currently) inert primitives needed to support durable per-worker result transport for pytest-xdist within RAMPART’s pytest plugin, without changing any existing execution path yet.

Changes:

• Introduces shard eligibility detection (shard_eligible) plus a controller→worker stash key (SHARD_DIR_KEY) to gate durable transport to local popen topologies.
• Adds durable JSONL shard writing (ShardWriter) and controller-side recovery (read_worker_shard) with dropped-record accounting.
• Expands unit coverage to validate shard eligibility, writer behavior, and shard recovery scenarios (missing file, truncated last line, oversized/corrupt records, authoritative envelope metadata).

Reviewed changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated 1 comment.

File	Description
rampart/pytest_plugin/_xdist.py	Adds shard transport primitives: eligibility gating, shard writer, shard reader, and dropped-record representation.
tests/unit/pytest_plugin/test_xdist.py	Adds unit tests covering shard eligibility matrix and shard write/read recovery behavior.

[spencrr]

@nina-msft We spoke offline about the filesystem vs in-mem serialized+streamed approaches. What about the following?

sequenceDiagram
    participant T as Test / RAMPART execution
    participant W as xdist worker
    participant E as execnet
    participant C as xdist controller
    participant S as RAMPART sink

    T->>W: execution emits Result(s)
    W->>W: ResultCollectionHandler records in item collector
    W->>W: call-phase pytest_runtest_makereport
    W->>W: serialize Result envelope(s)<br/>onto namespaced TestReport attribute
    W->>E: xdist testreport event
    E->>C: serialized TestReport
    C->>C: reconstruct TestReport
    C->>C: pytest_runtest_logreport
    C->>C: validate + merge Result envelope(s)
    Note over C: repeat incrementally per item

    W-->>C: workeroutput completion sentinel<br/>(expected count + trial specs)
    C->>C: verify streamed count / mark incomplete
    C->>S: emit one unified report

Loading

xdist already streams every setup/call/teardown TestReport from workers as tests execute, including for popen, ssh+socket, and proxied gateways. Its worker calls pytest_report_to_serializable() and sends a testreport event. The controller reconstructs the report and re-emits pytest_runtest_logreport.

Pytest’s default serializer copies the complete report.__dict__, and TestReport restores unknown fields through **extra. This is also an established plugin pattern:

• pytest-html attaches structured data as report.extras.
• pytest-rerunfailures attaches report.rerun.
• Core subtests uses explicit report serialization because it introduces a distinct SubtestReport type.
• Session-wide plugins such as pytest-cov use workeroutput, which is appropriate for data that has no natural per-item report.

We can namespace all of this under a private field such as report._rampart_results, rather than user_properties. Like you mentioned, user_properties is intentionally consumed by reporters such as JUnit XML, so putting complete prompts, responses, tool calls, and transcripts there could leak or substantially inflate JUnit output.

The small workeroutput sentinel would retain the valuable completeness properties of the current implementation:

• Expected result count.
• Trial specs.
• Schema/transport version.
• Clean worker completion.

That gives us these failure semantics:

Failure	Streamed design
Worker dies after previous items completed	Previously received item reports remain merged
Worker dies after call report, before teardown/session finish	Current item remains merged
Worker dies during the test body, before its call report	Current item is missing; xdist emits a crash failure and missing sentinel marks the RAMPART run incomplete
Remote worker/channel dies	Previously received reports survive; unlike shards, this works for remote gateways
One result exceeds the cap	Send a dropped-result marker; mark run incomplete without dropping sibling results
Worker exits cleanly	Sentinel verifies streamed count and contributes trial specs
Controller dies	In-memory aggregate is lost; the proposed shards do not currently implement controller restart/recovery either

There is one implementation precondition: today _rampart_collect absorbs results during fixture teardown, after the call report has already been created. A call-report transport needs access to the collector snapshot before that point. I view the collector-lifecycle change as separate work and am not proposing that it be folded into this PR.

Why I prefer evaluating this before committing to shards

The shard approach introduces a second transport with a materially different support envelope:

• It works only for local popen workers; remote workers fall back to the existing worker-session batch and retain its larger failure window.
• It writes attacker-influenced prompts, responses, tool calls, and transcripts to cleartext temporary files.
• It requires directory provisioning, controller→worker path propagation, writer lifetime management, cleanup, retention options, reader recovery, and transport dispatch.
• Controller SIGKILL/OOM can leave sensitive shard files behind indefinitely; in-process cleanup cannot prevent that.
• The configured limit is per record, so total disk use is effectively unbounded across a large run.

The shard primitives are carefully implemented and well tested. My concern is whether the filesystem is necessary at all, rather than the code quality of the primitives.

The filesystem approach does have one important potential advantage: large transcripts do not pass through and queue on the controller’s execnet report channel. Streaming large payloads could increase serialization cost, channel backpressure, controller RSS, and scheduling latency. I think we should measure that rather than assume it dominates.

Could we prototype the report-streamed path and compare it using representative p50/p95/max RAMPART result sizes under:

• -n 4 local workers.
• A remote or socket gateway.
• Worker termination after completed items.
• One oversized result among normal siblings.
• Thousands of small results.
• Representative 1 MiB, 10 MiB, and maximum-size transcripts.

The comparison should measure wall time, controller/worker RSS, bytes transferred, disk footprint, and recovered-result completeness.

If full-result streaming proves too expensive, a compact hybrid may be preferable:

• Stream the safety status, scores, trial identity, and compact metadata in the TestReport.
• Put unusually large transcripts behind an explicit artifact abstraction and stream an artifact reference.
• Keep completeness and gating data on the controller for every topology.

[bashirpartovi]

@spencrr I like that idea but it's definitely not cheap to implement. I think this is worth prototyping before we decide which direction to take. We can mock the agent side and generate synthetic Result payloads, but still run real xdist workers so we're testing the actual report channel.

I'd probably start with the collector timing since results are currently absorbed during teardown. Then we can try a few sample payload sizes and see whether streaming the full results holds up or whether we need the hybrid.

@nina-msft does this sound good to you too?