Add hang-snapshot session index to pytest summary

- `_testing/trace.py`: add `_SNAPSHOT_INDEX` session- scoped list
  populated by `_do_capture_snapshot()` on each successful dump;
  add TODO for future `TRACTOR_TRACE_HOLD=1` pause-on-hang mode
- `_testing/pytest.py`: add `pytest_terminal_summary` hook that
  prints all captured snapshot dirs at end-of-session so paths
  don't get buried in scrollback

(this patch was generated in some part by [`claude-code`][claude-code-gh])
[claude-code-gh]: https://github.com/anthropics/claude-code
subint_forkserver_backend
Gud Boi 2026-05-13 19:00:18 -04:00
parent e329c3108c
commit fb87c36263
2 changed files with 64 additions and 0 deletions

View File

@ -765,3 +765,37 @@ def set_fork_aware_capture(
# request=request, # request=request,
# start_method=start_method, # start_method=start_method,
# ) # )
def pytest_terminal_summary(
terminalreporter,
exitstatus: int,
config: pytest.Config,
) -> None:
'''
End-of-session summary: list all
`fail_after_w_trace`/`afk_alarm_w_trace` snapshot dirs
captured during the run so the human doesn't have to scroll
back through captured-stderr lines to find dump paths.
Reads from `tractor._testing.trace._SNAPSHOT_INDEX` which is
populated by `_do_capture_snapshot()` on each successful
snapshot capture.
No-op when zero snapshots were captured (most sessions).
'''
from .trace import _SNAPSHOT_INDEX
if not _SNAPSHOT_INDEX:
return
tr = terminalreporter
tr.write_sep('=', 'tractor hang-snapshot index')
tr.write_line(
f'{len(_SNAPSHOT_INDEX)} `fail_after_w_trace` / '
f'`afk_alarm_w_trace` snapshot(s) captured this session:'
)
for label, path in _SNAPSHOT_INDEX:
tr.write_line(f' {label}')
tr.write_line(f'{path}')

View File

@ -1000,6 +1000,31 @@ class AFKAlarmTimeout(TimeoutError):
''' '''
# Session-scoped list of snapshot (label, dump_dir) tuples
# captured by `fail_after_w_trace` / `afk_alarm_w_trace` during
# the current process lifetime. Populated by
# `_do_capture_snapshot()` on each successful dump. The
# `pytest_terminal_summary` hook in `tractor._testing.pytest`
# reads this at end-of-session to print an index of all
# snapshot dirs so the human doesn't have to scroll back through
# captured-stderr lines to find paths.
_SNAPSHOT_INDEX: list[tuple[str, Path]] = []
# TODO: follow-up — `TRACTOR_TRACE_HOLD=1` pause-on-hang mode.
# When env-var-enabled, `_do_capture_snapshot` would block on
# `input('press Enter to continue...')` reading from
# `sys.__stdin__` AFTER the dump succeeds, BEFORE re-raising the
# original exception. This lets a human invoke
# `acli.ptree`/`acli.bindspace_scan` from a second terminal
# while the cancel-cascade is frozen mid-flight — currently
# impossible because the per-test reaper fixture sweeps
# orphans within ~0.6s of the timeout firing. See discussion
# 2026-05-13: orphans visible in snapshot's `trace.txt`
# (depth_3 / depth_1 init-adopted procs) but invisible to any
# post-test `acli.*` invocation.
def _do_capture_snapshot( def _do_capture_snapshot(
*, *,
label: str, label: str,
@ -1015,6 +1040,10 @@ def _do_capture_snapshot(
Returns the snapshot `Path` on success, `None` if capture Returns the snapshot `Path` on success, `None` if capture
itself failed (with a banner printed to stderr). itself failed (with a banner printed to stderr).
Appends `(label, dump_dir)` to the session-scoped
`_SNAPSHOT_INDEX` on success so the `pytest_terminal_summary`
hook can render an index at end-of-session.
''' '''
target_pid: int = pid if pid is not None else os.getpid() target_pid: int = pid if pid is not None else os.getpid()
# NOTE: print to `sys.__stderr__` (the ORIGINAL unredirected # NOTE: print to `sys.__stderr__` (the ORIGINAL unredirected
@ -1054,6 +1083,7 @@ def _do_capture_snapshot(
f'pid={target_pid}); snapshot at: {dump_dir}', f'pid={target_pid}); snapshot at: {dump_dir}',
file=sys.__stderr__, file=sys.__stderr__,
) )
_SNAPSHOT_INDEX.append((label, dump_dir))
return dump_dir return dump_dir