[FEAT]: Snapshot collector results at call phase

rampart/pytest_plugin/_collection.py

@@ -58,6 +58,16 @@ def deactivate_collector(token: Token[ResultCollector | None]) -> None:
     _active_collector.reset(token)
 
 
+def get_active_collector() -> ResultCollector | None:
+    """Return the collector active for the current test, if any.
+
+    Returns:
+        ResultCollector | None: The active collector, or None when no
+            test collector is installed on the current context.
+    """
+    return _active_collector.get()
+
+
 class ResultCollector:
     """Accumulates Result objects produced during a single test.
 

rampart/pytest_plugin/plugin.py

@@ -41,6 +41,7 @@
     ResultCollector,
     activate_collector,
     deactivate_collector,
+    get_active_collector,
 )
 from rampart.pytest_plugin._session import RampartSession
 from rampart.pytest_plugin._xdist import (
@@ -69,15 +70,20 @@
     "pytest_addoption",
     "pytest_collection_modifyitems",
     "pytest_configure",
+    "pytest_runtest_makereport",
     "pytest_sessionfinish",
     "pytest_terminal_summary",
     "pytest_testnodedown",
     "pytest_unconfigure",
 ]
 
+# Config-scoped stash keys: one entry per pytest session.
 _rampart_key = pytest.StashKey[RampartSession]()
 _session_start_key = pytest.StashKey[float]()
 
+# Item-scoped stash key: a per-test snapshot consumed by xdist streaming.
+_call_results_key = pytest.StashKey[list[Result]]()
+
 # Module-level constants are an acceptable exception in a hook-based
 # plugin module where there is no natural owning class.
 
@@ -454,6 +460,42 @@ def _rampart_collect(  # pytest discovers this via autouse=True
         )
 
 
+@pytest.hookimpl(wrapper=True)
+def pytest_runtest_makereport(
+    item: pytest.Item,
+    call: pytest.CallInfo[None],
+) -> Generator[None, pytest.TestReport, pytest.TestReport]:
+    """Snapshot the active collector's results at the call phase.
+
+    On xdist workers, copies the per-test results onto the item stash
+    while the collector is still active — before the autouse fixture's
+    teardown drains and deactivates it. Downstream xdist streaming reads
+    this snapshot to deliver results per test rather than in a single
+    end-of-worker batch.
+
+    Restricted to worker processes: single-process and controller runs
+    never consume the snapshot, so skipping it there keeps those paths
+    allocation-free and byte-identical to their pre-snapshot behavior.
+
+    Runs as a wrapper so it is never skipped by the firstresult builtin
+    and always returns the report unchanged.
+
+    Args:
+        item (pytest.Item): The test item being reported.
+        call (pytest.CallInfo[None]): The call phase information.
+
+    Returns:
+        Generator[None, pytest.TestReport, pytest.TestReport]: The
+            unchanged report produced by downstream hookimpls.
+    """
+    report = yield
+    if call.when == "call" and is_xdist_worker(config=item.config):
+        collector = get_active_collector()
+        if collector is not None:
+            item.stash[_call_results_key] = collector.results
+    return report
+
+
 def _has_sink_hook_impl(*, config: pytest.Config) -> bool:
     """Return True if any plugin implements ``pytest_rampart_sinks``.
 

tests/unit/pytest_plugin/test_collection.py

@@ -5,6 +5,7 @@
 
 from __future__ import annotations
 
+import asyncio
 from unittest.mock import MagicMock
 
 from rampart.core.execution import ExecutionEvent, ExecutionEventData
@@ -15,6 +16,7 @@
     _active_collector,
     activate_collector,
     deactivate_collector,
+    get_active_collector,
     record_result,
 )
 
@@ -190,3 +192,49 @@ def test_nested_activation(self) -> None:
         assert len(outer.results) == 1
 
         deactivate_collector(outer_token)
+
+
+class TestGetActiveCollector:
+    """get_active_collector exposes the ContextVar-scoped collector."""
+
+    def test_returns_none_when_inactive(self) -> None:
+        token = _active_collector.set(None)
+        try:
+            assert get_active_collector() is None
+        finally:
+            _active_collector.reset(token)
+
+    def test_returns_active_collector(self) -> None:
+        collector = ResultCollector()
+        token = activate_collector(collector)
+        try:
+            assert get_active_collector() is collector
+        finally:
+            deactivate_collector(token)
+
+    def test_returns_none_after_deactivate(self) -> None:
+        baseline = _active_collector.set(None)
+        try:
+            collector = ResultCollector()
+            token = activate_collector(collector)
+            deactivate_collector(token)
+            assert get_active_collector() is None
+        finally:
+            _active_collector.reset(baseline)
+
+    async def test_sees_results_from_child_task_async(self) -> None:
+        collector = ResultCollector()
+        token = activate_collector(collector)
+
+        async def _body() -> None:
+            await asyncio.sleep(0)
+            record_result(result=_make_result(summary="child"))
+
+        try:
+            await asyncio.create_task(_body())
+            active = get_active_collector()
+            assert active is collector
+            assert len(active.results) == 1
+            assert active.results[0].summary == "child"
+        finally:
+            deactivate_collector(token)

tests/unit/pytest_plugin/test_plugin.py

@@ -5,16 +5,24 @@
 
 from __future__ import annotations
 
+from types import SimpleNamespace
 from typing import TYPE_CHECKING, Any, cast
 from unittest.mock import AsyncMock, MagicMock
 
 import pytest
 
 from rampart.core.result import Result, SafetyStatus
 from rampart.core.types import ObservabilityLevel
-from rampart.pytest_plugin._collection import ResultCollectionHandler, ResultCollector
+from rampart.pytest_plugin._collection import (
+    ResultCollectionHandler,
+    ResultCollector,
+    _active_collector,
+    activate_collector,
+    deactivate_collector,
+)
 from rampart.pytest_plugin._session import RampartSession
 from rampart.pytest_plugin.plugin import (
+    _call_results_key,
     _emit_sinks,
     _enforce_incomplete_exit_status,
     _evaluate_gates,
@@ -26,6 +34,7 @@
     _write_trial_group_lines,
     pytest_collection_modifyitems,
     pytest_configure,
+    pytest_runtest_makereport,
     pytest_sessionfinish,
     pytest_terminal_summary,
     pytest_unconfigure,
@@ -969,3 +978,111 @@ def test_incomplete_run_does_not_mask_existing_failure(self) -> None:
             rampart_session=rampart_session,
         )
         assert session.exitstatus == pytest.ExitCode.INTERRUPTED
+
+
+def _make_result(*, summary: str = "result") -> Result:
+    """Build a minimal Result for makereport tests."""
+    return Result(status=SafetyStatus.SAFE, summary=summary)
+
+
+def _make_reporting_item(*, worker: bool = True) -> Any:
+    """Build a mock pytest.Item backed by a real Stash.
+
+    Defaults to a worker-like config so the call-phase snapshot fires;
+    pass worker=False for a single-process or controller config.
+    """
+    item = MagicMock()
+    item.stash = pytest.Stash()
+    if worker:
+        item.config = SimpleNamespace(workerinput={"workerid": "gw0"})
+    else:
+        item.config = SimpleNamespace()
+    return item
+
+
+def _drive_makereport(*, item: Any, when: str, report: Any = None) -> Any:
+    """Drive the makereport wrapper generator and return its result."""
+    call = MagicMock()
+    call.when = when
+    sent = report if report is not None else MagicMock()
+    gen = pytest_runtest_makereport(
+        item=cast("pytest.Item", item),
+        call=cast("pytest.CallInfo[None]", call),
+    )
+    next(gen)
+    try:
+        gen.send(sent)
+    except StopIteration as stop:
+        return stop.value
+    raise AssertionError("makereport wrapper did not return")
+
+
+class TestPytestRuntestMakereport:
+    """The makereport hook snapshots collector results at the call phase."""
+
+    def test_snapshots_results_at_call_phase(self) -> None:
+        item = _make_reporting_item()
+        collector = ResultCollector()
+        collector.record(result=_make_result(summary="captured"))
+        token = activate_collector(collector)
+        try:
+            _drive_makereport(item=item, when="call")
+        finally:
+            deactivate_collector(token)
+
+        snapshot = item.stash[_call_results_key]
+        assert len(snapshot) == 1
+        assert snapshot[0].summary == "captured"
+
+    def test_snapshots_empty_list_when_no_results(self) -> None:
+        item = _make_reporting_item()
+        collector = ResultCollector()
+        token = activate_collector(collector)
+        try:
+            _drive_makereport(item=item, when="call")
+        finally:
+            deactivate_collector(token)
+
+        assert item.stash[_call_results_key] == []
+
+    def test_no_snapshot_at_setup_phase(self) -> None:
+        item = _make_reporting_item()
+        collector = ResultCollector()
+        collector.record(result=_make_result())
+        token = activate_collector(collector)
+        try:
+            _drive_makereport(item=item, when="setup")
+        finally:
+            deactivate_collector(token)
+
+        assert _call_results_key not in item.stash
+
+    def test_no_snapshot_when_no_collector_active(self) -> None:
+        item = _make_reporting_item()
+        token = _active_collector.set(None)
+        try:
+            _drive_makereport(item=item, when="call")
+        finally:
+            _active_collector.reset(token)
+
+        assert _call_results_key not in item.stash
+
+    def test_no_snapshot_when_not_xdist_worker(self) -> None:
+        item = _make_reporting_item(worker=False)
+        collector = ResultCollector()
+        collector.record(result=_make_result())
+        token = activate_collector(collector)
+        try:
+            _drive_makereport(item=item, when="call")
+        finally:
+            deactivate_collector(token)
+
+        assert _call_results_key not in item.stash
+
+    def test_returns_report_unchanged(self) -> None:
+        item = _make_reporting_item()
+        report = object()
+
+        returned = _drive_makereport(item=item, when="call", report=report)
+
+        assert returned is report