From 7509e313ff948c513667e49f2dafd753e1f34dc3 Mon Sep 17 00:00:00 2001
From: goodboy <goodboy_foss@protonmail.com>
Date: Wed, 13 May 2026 16:47:17 -0400
Subject: [PATCH] Mv core impl `tractor_diag.xsh` to `_testing.trace`
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Extract all pure-Python diagnostic helpers (`dump_proc_tree`,
`dump_hung_state`, `scan_bindspace`, `dump_all`, `resolve_pids`,
`ensure_sudo_cached`, etc.) from the xonsh xontrib into a new
`tractor/_testing/trace.py` module so the same logic is callable
from both the `acli.*` terminal aliases AND in-test capture-on-hang
fixtures.

Deats,
- `_testing/trace.py`: new module (1171 lines) — proc-tree walker,
  hung-state dumper, bindspace scanner, `dump_all()` snapshot
  archiver, `AFKAlarmTimeout` exc, `fail_after_w_trace()` async CM
  (trio `fail_after` + auto-snapshot on `TooSlowError`),
  `afk_alarm_w_trace()` sync CM (`signal.alarm` + snapshot on
  `SIGALRM`), plus pytest fixture wrappers for both.
- `_testing/pytest.py`: re-export the two fixtures via `from .trace
  import` so pytest plugin-discovery picks them up.
- `tractor_diag.xsh`: thin terminal wrappers that import from
  `_testing.trace` — drops ~627 lines of inline impl. Add
  `acli.dump_all` alias for full snapshot-bundle CLI access.

(this patch was generated in some part by [`claude-code`][claude-code-gh])
[claude-code-gh]: https://github.com/anthropics/claude-code
---
 tractor/_testing/pytest.py |   14 +
 tractor/_testing/trace.py  | 1171 ++++++++++++++++++++++++++++++++++++
 xontrib/tractor_diag.xsh   |  735 ++++------------------
 3 files changed, 1294 insertions(+), 626 deletions(-)
 create mode 100644 tractor/_testing/trace.py

diff --git a/tractor/_testing/pytest.py b/tractor/_testing/pytest.py
index a0a5bfb6..5aca2908 100644
--- a/tractor/_testing/pytest.py
+++ b/tractor/_testing/pytest.py
@@ -38,6 +38,20 @@ import tractor
 from tractor.spawn._spawn import SpawnMethodKey
 import trio
 
+# Re-export `_testing.trace`'s pytest fixtures so they're
+# picked up by pytest's plugin-discovery (this module is
+# loaded via `pytest_plugins` from `pyproject.toml`). The
+# `noqa: F401` annotations make linters tolerate the
+# unused-looking imports — they're load-bearing for pytest
+# discovery. The fixtures share their `name=` kw with the
+# underlying CM functions; the python-level identifiers
+# below carry the `_fixture` suffix to avoid module-scope
+# collision (see `_testing/trace.py` for details).
+from .trace import (  # noqa: F401
+    afk_alarm_w_trace_fixture,
+    fail_after_w_trace_fixture,
+)
+
 # Sub-plugin: zombie-subactor + UDS sock-file + shm
 # reaping fixtures live in `tractor._testing._reap`
 # alongside the underlying detection/cleanup helpers.
diff --git a/tractor/_testing/trace.py b/tractor/_testing/trace.py
new file mode 100644
index 00000000..b45883fe
--- /dev/null
+++ b/tractor/_testing/trace.py
@@ -0,0 +1,1171 @@
+# tractor: distributed structured concurrency.
+# Copyright 2018-eternity Tyler Goodlet.
+
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+'''
+Pure-Python diagnostic state-capture for hung
+`pytest`/`tractor` process trees.
+
+This module is the load-bearing core for two consumers:
+
+1. The `xontrib/tractor_diag.xsh::acli.*` xonsh aliases
+   (`acli.ptree`, `acli.hung_dump`, `acli.bindspace_scan`,
+   `acli.dump_all`) — interactive terminal diag tools.
+
+2. In-test "capture-on-hang" helpers like
+   `fail_after_w_trace()` / `afk_alarm_w_trace()` that drop a
+   full diag snapshot to disk when a test exceeds its timeout
+   budget instead of just emitting a context-less
+   `trio.TooSlowError`.
+
+All public dump-* functions RETURN formatted text rather than
+printing, so callers can render to a terminal OR write to a
+file. `dump_all()` does the file-writing for snapshot-archive
+use cases.
+
+Sudo policy:
+  Per-pid kernel `stack` + `py-spy dump` need `CAP_SYS_PTRACE`,
+  invoked via `sudo -n`. Two modes:
+
+  - `allow_sudo_prompt=True` (terminal CLI default):
+    `ensure_sudo_cached()` prompts the user once via `sudo -v`
+    if creds aren't cached, then re-uses them per-pid.
+
+  - `allow_sudo_prompt=False` (pytest / in-test default):
+    silently skip sudo-required diagnostics; emit a banner
+    pointing the human at `sudo -v && acli.hung_dump <pid>`
+    for a follow-up manual capture.
+
+'''
+from __future__ import annotations
+
+import json
+import os
+import re
+import signal
+import subprocess as sp
+from contextlib import (
+    AbstractAsyncContextManager,
+    AbstractContextManager,
+    asynccontextmanager,
+    contextmanager,
+)
+from datetime import datetime
+from io import StringIO
+from pathlib import Path
+from typing import (
+    AsyncIterator,
+    Callable,
+    Iterator,
+    TypeAlias,
+)
+
+
+# Public type aliases for the `fail_after_w_trace` /
+# `afk_alarm_w_trace` fixture-returned CM-factory callables.
+# Test signatures can annotate the fixture param directly::
+#
+#     def test_foo(
+#         fail_after_w_trace: FailAfterWTraceFactory,
+#     ):
+#         async with fail_after_w_trace(5.0):
+#             ...
+#
+# NOTE the fixture name intentionally shadows the underlying
+# `fail_after_w_trace` function at test-fn scope; pytest's
+# param-resolution overrides the module-level import, so the
+# fixture-returned CM-factory wins inside the test body.
+#
+# `Callable[..., ...]` keeps the kwargs surface loose (caller
+# can pass `label=`, `pid=`, `out_dir=`); precise checking of
+# the first-arg `seconds` is left to runtime since most callers
+# pass an `int|float` literal.
+FailAfterWTraceFactory: TypeAlias = Callable[
+    ...,
+    AbstractAsyncContextManager[None],
+]
+AfkAlarmWTraceFactory: TypeAlias = Callable[
+    ...,
+    AbstractContextManager[None],
+]
+
+try:
+    import psutil
+except ImportError:
+    psutil = None
+
+try:
+    import pytest as _pytest
+except ImportError:
+    # `trace.py`'s pure-Python core (proc-tree + bindspace +
+    # dump_*) is intentionally pytest-free so the `xontrib`
+    # CLI can `import` it from any venv. The fixtures at
+    # the bottom of this module require `pytest` and are
+    # only defined when it's importable.
+    _pytest = None
+
+
+# matches tractor's UDS sock naming: `<actor_name>@<pid>.sock`
+_UDS_SOCK_RE = re.compile(
+    r'^(?P<name>.+)@(?P<pid>\d+)\.sock$'
+)
+
+
+# ---------------------------------------------------------------
+# pid + proc-tree resolution
+# ---------------------------------------------------------------
+
+def resolve_pids(arg: str) -> list[int]:
+    '''
+    Resolve a numeric pid OR a `pgrep -f` pattern to a list of
+    pids. Returns `[]` on no match.
+
+    '''
+    if arg.isdigit():
+        return [int(arg)]
+    try:
+        out: str = sp.check_output(
+            ['pgrep', '-f', arg],
+            text=True,
+        )
+    except sp.CalledProcessError:
+        return []
+    return [int(p) for p in out.split() if p]
+
+
+def walk_tree_psutil(pid: int) -> list:
+    '''Flat `[Process, *descendants]` via `psutil` (or `[]`).'''
+    if psutil is None:
+        return []
+    try:
+        p = psutil.Process(pid)
+    except psutil.NoSuchProcess:
+        return []
+    return [p] + p.children(recursive=True)
+
+
+def _walk_tree_with_depth(pid: int) -> Iterator[tuple]:
+    '''Yield `(proc, depth)` pairs walking `pid`'s subtree.'''
+    if psutil is None:
+        return
+    try:
+        root = psutil.Process(pid)
+    except psutil.NoSuchProcess:
+        return
+    yield root, 0
+    stack: list = [(root, 0)]
+    seen: set = {pid}
+    while stack:
+        parent, d = stack.pop()
+        try:
+            kids = parent.children()
+        except psutil.NoSuchProcess:
+            continue
+        for k in kids:
+            if k.pid in seen:
+                continue
+            seen.add(k.pid)
+            yield k, d + 1
+            stack.append((k, d + 1))
+
+
+def _walk_tree_pgrep(pid: int) -> list[int]:
+    '''psutil-less fallback — recursive `pgrep -P`.'''
+    out: list[int] = [pid]
+    try:
+        kids: list = sp.check_output(
+            ['pgrep', '-P', str(pid)],
+            text=True,
+        ).split()
+    except sp.CalledProcessError:
+        return out
+    for k in kids:
+        out.extend(_walk_tree_pgrep(int(k)))
+    return out
+
+
+def _which_cgroup_slice(pid: int) -> str | None:
+    '''
+    Return `'system'` / `'user'` / `None` for `pid`'s top-level
+    systemd cgroup slice. See the full `xontrib` docstring on
+    `_which_cgroup_slice` for the bucket semantics.
+
+    '''
+    try:
+        with open(f'/proc/{pid}/cgroup') as f:
+            cg: str = f.read()
+    except (
+        FileNotFoundError,
+        PermissionError,
+        ProcessLookupError,
+        OSError,
+    ):
+        return None
+    if '/system.slice/' in cg:
+        return 'system'
+    if '/user.slice/' in cg:
+        return 'user'
+    return None
+
+
+def _ppid_from_proc(pid: int) -> int | None:
+    '''
+    Read `ppid` from `/proc/<pid>/stat`. Returns None on race
+    (proc died) / permission / non-linux.
+
+    NB: stat field [1] is `(comm)` which can contain spaces +
+    parens — `rsplit(')', 1)` is the safe way to skip past it.
+
+    '''
+    try:
+        with open(f'/proc/{pid}/stat') as f:
+            stat: str = f.read()
+        after_comm: str = stat.rsplit(')', 1)[1].strip()
+        return int(after_comm.split()[1])  # state(0) ppid(1)
+    except (
+        FileNotFoundError,
+        PermissionError,
+        ProcessLookupError,
+        OSError,
+    ):
+        return None
+
+
+# ---------------------------------------------------------------
+# sudo probe / prompt
+# ---------------------------------------------------------------
+
+def is_sudo_cached() -> bool:
+    '''
+    Quietly probe whether `sudo` creds are cached. Never
+    prompts — safe to call from non-interactive contexts.
+
+    '''
+    try:
+        return sp.run(
+            ['sudo', '-n', 'true'],
+            capture_output=True,
+        ).returncode == 0
+    except FileNotFoundError:
+        return False
+
+
+def ensure_sudo_cached() -> bool:
+    '''
+    Like `is_sudo_cached()` but PROMPTS interactively via
+    `sudo -v` if not yet cached. Suitable for terminal-CLI use
+    only — DO NOT call from inside a pytest run.
+
+    '''
+    if is_sudo_cached():
+        return True
+    print(
+        '[tractor-trace] needs `sudo` for '
+        '/proc/<pid>/stack and `py-spy dump`; caching creds '
+        'via `sudo -v`...'
+    )
+    try:
+        rc: int = sp.run(['sudo', '-v']).returncode
+    except KeyboardInterrupt:
+        print('  cancelled — proceeding without sudo')
+        return False
+    except FileNotFoundError:
+        print('  sudo not on PATH — proceeding without sudo')
+        return False
+    return rc == 0
+
+
+# ---------------------------------------------------------------
+# dump_proc_tree (== acli.ptree)
+# ---------------------------------------------------------------
+
+def dump_proc_tree(
+    roots: list[int],
+    *,
+    flag_tree: bool = False,
+) -> str:
+    '''
+    Severity-classified proc-tree rendering of `roots` and
+    their descendants. Returns formatted text.
+
+    Buckets (severity-ordered):
+      - zombies:       `status in (Z, X)`
+      - orphans:       `ppid==1`, NOT in a systemd cgroup slice
+      - system-slice:  `ppid==1`, under `/system.slice/`
+      - user-slice:    `ppid==1`, under `/user.slice/.../*.scope`
+      - live:          real (`ppid > 1`) parent
+
+    `flag_tree=True` additionally prepends a flat walk-order
+    `## tree` section preserving parent-child shape.
+
+    '''
+    buf = StringIO()
+
+    def echo(line: str = '') -> None:
+        buf.write(line + '\n')
+
+    if psutil is None:
+        echo(
+            'ptree requires `psutil`; '
+            'install via `uv pip install psutil`'
+        )
+        return buf.getvalue()
+
+    # statuses considered "defunct"
+    defunct_statuses: set = {
+        psutil.STATUS_ZOMBIE,
+        getattr(psutil, 'STATUS_DEAD', 'dead'),
+    }
+
+    seen: set = set()
+    walk_order: list = []
+    live: list = []
+    orphans: list = []
+    system_slice: list = []
+    user_slice: list = []
+    zombies: list = []
+    gone: list = []
+    pid_to_bucket: dict = {}
+
+    for r in roots:
+        for (p, depth) in _walk_tree_with_depth(r):
+            if p.pid in seen:
+                continue
+            seen.add(p.pid)
+            try:
+                status: str = p.status()
+                ppid: int = p.ppid()
+            except psutil.NoSuchProcess:
+                gone.append(p.pid)
+                continue
+            entry = (p, depth)
+            if status in defunct_statuses:
+                zombies.append(entry)
+                pid_to_bucket[p.pid] = 'zombies'
+            elif ppid == 1:
+                slice_kind: str | None = _which_cgroup_slice(p.pid)
+                if slice_kind == 'system':
+                    system_slice.append(entry)
+                    pid_to_bucket[p.pid] = 'system-slice'
+                elif slice_kind == 'user':
+                    user_slice.append(entry)
+                    pid_to_bucket[p.pid] = 'user-slice'
+                else:
+                    orphans.append(entry)
+                    pid_to_bucket[p.pid] = 'orphans'
+            else:
+                live.append(entry)
+                pid_to_bucket[p.pid] = 'live'
+            walk_order.append(entry)
+
+    total: int = (
+        len(live)
+        + len(orphans)
+        + len(system_slice)
+        + len(user_slice)
+        + len(zombies)
+    )
+    echo(f'# ptree: {total} procs across roots {roots}')
+
+    hdr: str = (
+        '  ' + 'PID'.rjust(7)
+        + '  ' + 'PPID'.rjust(7)
+        + '  ' + 'STATUS'.ljust(10)
+        + '  CMD'
+    )
+
+    def _row(entry, bucket: str | None = None) -> str:
+        p, depth = entry
+        tree_pfx: str = ('   ' * depth) + ('└─ ' if depth > 0 else '')
+
+        parent_anno: str = ''
+        if (
+            bucket is not None
+            and depth > 0
+        ):
+            try:
+                parent_pid: int = p.ppid()
+            except psutil.NoSuchProcess:
+                parent_pid = 0
+            if parent_pid and parent_pid != 1:
+                parent_bucket: str | None = pid_to_bucket.get(parent_pid)
+                if (
+                    parent_bucket is not None
+                    and parent_bucket != bucket
+                ):
+                    parent_anno = (
+                        f'  [parent: {parent_pid} '
+                        f'(in `{parent_bucket}`)]'
+                    )
+
+        try:
+            cmd: str = (
+                ' '.join(p.cmdline())[:140]
+                or '[' + p.name() + ']'
+            )
+            r: str = '  ' + str(p.pid).rjust(7)
+            r += '  ' + str(p.ppid()).rjust(7)
+            r += '  ' + p.status().ljust(10)
+            r += '  ' + tree_pfx + cmd + parent_anno
+            return r
+        except psutil.ZombieProcess:
+            try:
+                ppid_str: str = str(p.ppid())
+                name: str = p.name()
+            except psutil.NoSuchProcess:
+                ppid_str, name = '?', '?'
+            r = '  ' + str(p.pid).rjust(7)
+            r += '  ' + ppid_str.rjust(7)
+            r += '  ' + 'zombie'.ljust(10)
+            r += (
+                '  ' + tree_pfx
+                + '[' + name + ' <defunct>]'
+                + parent_anno
+            )
+            return r
+        except psutil.NoSuchProcess:
+            return (
+                '  ' + str(p.pid).rjust(7)
+                + '  (gone mid-walk)'
+            )
+
+    def _section(
+        title: str,
+        procs: list,
+        hint: str = '',
+        bucket: str | None = None,
+    ) -> None:
+        echo()
+        echo(
+            f'## {title} ({len(procs)})'
+            + (f'  — {hint}' if hint else '')
+        )
+        if not procs:
+            echo('  (none)')
+            return
+        echo(hdr)
+        for p in procs:
+            echo(_row(p, bucket=bucket))
+
+    if flag_tree:
+        _section(
+            'tree', walk_order,
+            'flat walk-order, parent-child preserved',
+        )
+
+    _section(
+        'zombies', zombies,
+        'status `Z`/`X`, parent has not reaped',
+        bucket='zombies',
+    )
+    _section(
+        'orphans', orphans,
+        '`ppid==1`, NOT in a `system.slice`/`user.slice` cgroup '
+        '(likely leaked / parent gone)',
+        bucket='orphans',
+    )
+    _section(
+        'system-slice', system_slice,
+        '`ppid==1`, rooted under `/system.slice/` '
+        '(real systemd-managed service — daemon, login '
+        'session manager, etc; not a leak)',
+        bucket='system-slice',
+    )
+    _section(
+        'user-slice', user_slice,
+        '`ppid==1`, rooted under `/user.slice/.../*.scope` '
+        '(desktop-launched app wrapped by systemd-user — '
+        'browser, editor, etc; not a leak)',
+        bucket='user-slice',
+    )
+    _section('live', live, bucket='live')
+
+    if gone:
+        echo()
+        echo(f'## gone-during-walk ({len(gone)}): {gone}')
+
+    return buf.getvalue()
+
+
+# ---------------------------------------------------------------
+# dump_hung_state (== acli.hung_dump)
+# ---------------------------------------------------------------
+
+def dump_hung_state(
+    roots: list[int],
+    *,
+    allow_sudo_prompt: bool = False,
+) -> str:
+    '''
+    Per-pid kernel + python state for a hung pytest/tractor
+    process tree. Walks descendants of each root.
+
+    Captures per-pid:
+      - `/proc/<pid>/wchan` (world-readable)
+      - `/proc/<pid>/stack` (CAP_SYS_PTRACE — needs sudo)
+      - `py-spy dump --pid <N> --locals` (needs sudo)
+
+    Sudo policy controlled by `allow_sudo_prompt`:
+
+    - `True`: call `ensure_sudo_cached()` which prompts via
+      `sudo -v` if creds aren't cached. Use from terminal CLI.
+
+    - `False` (default): only probe via `is_sudo_cached()` —
+      never prompts. If not cached, skip stack+py-spy and emit
+      a banner pointing the human at the manual follow-up cmd.
+      Use from inside a pytest run.
+
+    '''
+    buf = StringIO()
+
+    def echo(line: str = '') -> None:
+        buf.write(line + '\n')
+
+    if allow_sudo_prompt:
+        have_sudo: bool = ensure_sudo_cached()
+    else:
+        have_sudo: bool = is_sudo_cached()
+
+    pids: list[int] = []
+    seen: set = set()
+    for r in roots:
+        if psutil is not None:
+            walk: list[int] = [p.pid for p in walk_tree_psutil(r)]
+        else:
+            walk = _walk_tree_pgrep(r)
+        for pid in walk:
+            if pid not in seen:
+                seen.add(pid)
+                pids.append(pid)
+
+    echo(f'# tree: {pids}')
+
+    if not have_sudo:
+        echo()
+        echo(
+            '💡 sudo creds NOT cached — '
+            '`/proc/<pid>/stack` + `py-spy dump` SKIPPED '
+            'for all pids below.'
+        )
+        echo(
+            '   For full kernel-stack + py-spy frames, '
+            're-run manually with sudo cached:'
+        )
+        echo(f'     sudo -v && acli.hung_dump {pids[0] if pids else "<pid>"}')
+
+    echo()
+    echo('## ps forest')
+    if pids:
+        try:
+            ps_out: str = sp.check_output(
+                [
+                    'ps',
+                    '-o', 'pid,ppid,pgid,stat,cmd',
+                    '-p', ','.join(map(str, pids)),
+                ],
+                text=True,
+            )
+            echo(ps_out.rstrip())
+        except (sp.CalledProcessError, FileNotFoundError) as e:
+            echo(f'  (ps failed: {e})')
+
+    for pid in pids:
+        echo()
+        echo(f'## pid {pid}' + (
+            ''
+            if have_sudo
+            else '  (sudo NOT cached — stack/py-spy SKIPPED)'
+        ))
+
+        for f in ('wchan', 'stack'):
+            path = Path(f'/proc/{pid}/{f}')
+            try:
+                txt: str = path.read_text().rstrip()
+                echo(f'-- /proc/{pid}/{f} --')
+                echo(txt)
+            except PermissionError:
+                if not have_sudo:
+                    echo(
+                        f'-- /proc/{pid}/{f}: '
+                        'PermissionError (no sudo) --'
+                    )
+                    continue
+                try:
+                    txt = sp.check_output(
+                        ['sudo', '-n', 'cat', str(path)],
+                        text=True,
+                        stderr=sp.DEVNULL,
+                    ).rstrip()
+                    echo(f'-- /proc/{pid}/{f} (sudo) --')
+                    echo(txt)
+                except sp.CalledProcessError:
+                    echo(
+                        f'-- /proc/{pid}/{f}: '
+                        'sudo cred expired? rerun --'
+                    )
+            except FileNotFoundError:
+                echo(f'-- /proc/{pid}/{f}: proc gone --')
+
+        echo(f'-- py-spy {pid} --')
+        if not have_sudo:
+            echo('  (skipped — no sudo)')
+            continue
+        try:
+            py_spy_out: str = sp.check_output(
+                ['sudo', '-n', 'py-spy', 'dump', '--pid', str(pid), '--locals'],
+                text=True,
+                stderr=sp.STDOUT,
+            )
+            echo(py_spy_out.rstrip())
+        except (sp.CalledProcessError, FileNotFoundError) as e:
+            echo(f'  (py-spy failed: {e})')
+
+    return buf.getvalue()
+
+
+# ---------------------------------------------------------------
+# scan_bindspace (== acli.bindspace_scan)
+# ---------------------------------------------------------------
+
+def scan_bindspace(arg: str | None = None) -> str:
+    '''
+    Scan a tractor UDS bindspace dir for orphan sock files.
+
+    `arg` semantics:
+      - `None`        -> `$XDG_RUNTIME_DIR/tractor`
+      - bare `<name>` -> `$XDG_RUNTIME_DIR/<name>` (e.g. `piker`)
+      - path          -> use as-is
+
+    Output buckets: `live-active`, `orphaned-alive`,
+    `orphaned-dead`, `non-tractor`.
+
+    '''
+    buf = StringIO()
+
+    def echo(line: str = '') -> None:
+        buf.write(line + '\n')
+
+    runtime: str = os.environ.get(
+        'XDG_RUNTIME_DIR',
+        f'/run/user/{os.getuid()}',
+    )
+    if arg:
+        if arg.startswith('/') or '/' in arg:
+            bs_dir = Path(arg)
+        else:
+            bs_dir = Path(runtime) / arg
+    else:
+        bs_dir = Path(runtime) / 'tractor'
+
+    if not bs_dir.exists():
+        echo(f'(no bindspace at {bs_dir})')
+        return buf.getvalue()
+
+    socks: list = sorted(bs_dir.glob('*.sock'))
+    echo(f'## bindspace {bs_dir} ({len(socks)} sock file(s))')
+
+    live_active: list = []
+    live_orphaned: list = []
+    dead_orphans: list = []
+    bogus: list = []
+
+    for s in socks:
+        m = _UDS_SOCK_RE.match(s.name)
+        if not m:
+            bogus.append(s)
+            continue
+        pid = int(m['pid'])
+        name = m['name']
+        try:
+            os.kill(pid, 0)
+        except ProcessLookupError:
+            dead_orphans.append((s, pid, name))
+            continue
+        except PermissionError:
+            live_active.append((s, pid, name, None))
+            continue
+
+        ppid: int | None = _ppid_from_proc(pid)
+        if ppid == 1:
+            live_orphaned.append((s, pid, name, ppid))
+        else:
+            live_active.append((s, pid, name, ppid))
+
+    echo()
+    echo(
+        f'## live-active ({len(live_active)})  '
+        f'— PID alive, parent still own it'
+    )
+    if not live_active:
+        echo('  (none)')
+    for s, pid, name, ppid in live_active:
+        row: str = '  ' + str(pid).rjust(7)
+        row += '  ' + name.ljust(32)
+        row += '  ' + s.name
+        if ppid is not None:
+            row += f'  (ppid={ppid})'
+        echo(row)
+
+    echo()
+    echo(
+        f'## orphaned-alive ({len(live_orphaned)})  '
+        f'— PID alive but `ppid==1`, parent reaped; '
+        f'`acli.reap` candidate'
+    )
+    if not live_orphaned:
+        echo('  (none)')
+    for s, pid, name, ppid in live_orphaned:
+        row = '  ' + str(pid).rjust(7)
+        row += '  ' + name.ljust(32)
+        row += '  ' + s.name + '  (adopted by init)'
+        echo(row)
+
+    echo()
+    echo(
+        f'## orphaned-dead ({len(dead_orphans)})  '
+        f'— PID gone, sock stale'
+    )
+    if not dead_orphans:
+        echo('  (none)')
+    for s, pid, name in dead_orphans:
+        row = '  ' + str(pid).rjust(7)
+        row += '  ' + name.ljust(32)
+        row += '  ' + s.name + '  (no live proc)'
+        echo(row)
+
+    if bogus:
+        echo()
+        echo(
+            f'## non-tractor ({len(bogus)})  '
+            f'— filename lacks `@<pid>` suffix, '
+            f'cannot determine liveness intrinsically'
+        )
+        for s in bogus:
+            echo(f'  {s.name}')
+        echo()
+        echo('to check liveness manually (needs `iproute2`/`ss`):')
+        for s in bogus:
+            echo(f"  ss -lpx 'src = {s}'")
+
+    if dead_orphans or live_orphaned:
+        echo()
+        echo(
+            'to sweep BOTH orphaned-alive subs '
+            '(graceful SIGINT -> SIGKILL) AND dead-orphan '
+            'socks in one shot:'
+        )
+        echo('  acli.reap --uds')
+
+    if dead_orphans:
+        unlink_cmd: str = ' '.join(str(o[0]) for o in dead_orphans)
+        echo()
+        echo(
+            '(or to unlink dead-orphan socks manually, '
+            "skipping `acli.reap`'s graceful-cancel ladder:)"
+        )
+        echo(f'  rm {unlink_cmd}')
+
+    return buf.getvalue()
+
+
+# ---------------------------------------------------------------
+# dump_all — file-writing snapshot capture
+# ---------------------------------------------------------------
+
+def _default_dump_root() -> Path:
+    '''
+    `$XDG_CACHE_HOME/tractor/hung-dumps/` with
+    `~/.cache/tractor/hung-dumps/` fallback.
+
+    '''
+    cache: str = os.environ.get(
+        'XDG_CACHE_HOME',
+        str(Path.home() / '.cache'),
+    )
+    return Path(cache) / 'tractor' / 'hung-dumps'
+
+
+def dump_all(
+    pid: int,
+    out_dir: Path | None = None,
+    *,
+    label: str,
+    allow_sudo_prompt: bool = False,
+) -> Path:
+    '''
+    Capture full diag snapshot for the proc tree rooted at
+    `pid` into a new sub-directory under `out_dir`.
+
+    Layout:
+      `<out_dir>/<label>__<iso-timestamp>/`
+      ├─ trace.txt        # ptree + hung_state merged
+      ├─ bindspace.txt    # bindspace_scan output
+      └─ meta.json        # {pid, label, captured_at, sudo_cached}
+
+    Returns the snapshot directory `Path`.
+
+    `out_dir` defaults to
+    `$XDG_CACHE_HOME/tractor/hung-dumps/` (fallback
+    `~/.cache/tractor/hung-dumps/`).
+
+    See `dump_hung_state()` for `allow_sudo_prompt` semantics
+    — defaults to False (test-safe).
+
+    '''
+    if out_dir is None:
+        out_dir = _default_dump_root()
+    out_dir = Path(out_dir)
+
+    ts: str = datetime.now().strftime('%Y-%m-%dT%H-%M-%S')
+    # sanitize label for filesystem: collapse anything non-word/-./-
+    # into single underscore, strip leading/trailing underscores.
+    safe_label: str = re.sub(r'[^\w.\-]+', '_', label).strip('_')
+    dump_dir: Path = out_dir / f'{safe_label}__{ts}'
+    dump_dir.mkdir(parents=True, exist_ok=True)
+
+    sudo_ok: bool = (
+        ensure_sudo_cached()
+        if allow_sudo_prompt
+        else is_sudo_cached()
+    )
+
+    # combined trace.txt: ptree first (classified buckets),
+    # then hung_state (per-pid wchan/stack/py-spy)
+    trace_txt: str = (
+        '# ===== ptree =====\n'
+        + dump_proc_tree([pid])
+        + '\n# ===== hung_state =====\n'
+        + dump_hung_state(
+            [pid],
+            allow_sudo_prompt=False,  # already prompted above
+        )
+    )
+    (dump_dir / 'trace.txt').write_text(trace_txt)
+
+    (dump_dir / 'bindspace.txt').write_text(scan_bindspace())
+
+    meta: dict = {
+        'pid': pid,
+        'label': label,
+        'captured_at': ts,
+        'sudo_cached': sudo_ok,
+    }
+    (dump_dir / 'meta.json').write_text(
+        json.dumps(meta, indent=2) + '\n'
+    )
+
+    return dump_dir
+
+
+# ---------------------------------------------------------------
+# in-test capture-on-hang helpers
+# ---------------------------------------------------------------
+#
+# Pair of CMs that combine a tight cooperative/hard timeout with
+# a forced `dump_all()` snapshot BEFORE the failure propagates.
+# The goal: when a test hangs, the human (or AI reviewer) gets a
+# fresh ptree + per-pid wchan/stack + bindspace state captured to
+# disk at the exact moment of the timeout — no need to recreate
+# it after the fact (which is often impossible since the procs
+# have moved on / been reaped).
+#
+# Two variants for two failure shapes:
+#
+#  - `fail_after_w_trace` — async CM wrapping `trio.fail_after`.
+#    Cooperative: cancellation is delivered at the next trio
+#    checkpoint. Use when the hang is at the trio/python level
+#    and the runtime is still scheduling normally.
+#
+#  - `afk_alarm_w_trace` — sync CM wrapping `signal.alarm`.
+#    Hard backstop: raises into the python frame at the next
+#    bytecode boundary regardless of trio's state. Use as a wall-
+#    clock cap when something *below* the trio scheduler is
+#    locking up (e.g. forkserver-launchpad in `os.read`, native-
+#    lock held by a C extension, GIL-hostage class hangs).
+#    Must run on the main thread (signal.alarm constraint).
+#
+# Both default to dumping the CURRENT process tree (i.e. the
+# pytest worker + its subactor descendants). Override `pid=` to
+# scope to a specific actor root.
+# ---------------------------------------------------------------
+
+
+class AFKAlarmTimeout(TimeoutError):
+    '''
+    Raised by `afk_alarm_w_trace`'s SIGALRM handler when the
+    alarm fires. Subclass of `TimeoutError` so existing
+    `except TimeoutError:` catches still match.
+
+    '''
+
+
+def _do_capture_snapshot(
+    *,
+    label: str,
+    pid: int | None,
+    out_dir: Path | None,
+    seconds: float,
+    timeout_kind: str,  # 'fail_after' | 'afk_alarm'
+) -> Path | None:
+    '''
+    Run `dump_all()` inside a best-effort try-block — never let
+    capture failure mask the original timeout exception.
+
+    Returns the snapshot `Path` on success, `None` if capture
+    itself failed (with a banner printed to stderr).
+
+    '''
+    target_pid: int = pid if pid is not None else os.getpid()
+    try:
+        dump_dir: Path = dump_all(
+            target_pid,
+            out_dir=out_dir,
+            label=label,
+            # in-test default: never prompt for sudo (would
+            # deadlock pytest); the dump_hung_state banner
+            # points the human at `sudo -v && acli.hung_dump`
+            # for a follow-up manual capture.
+            allow_sudo_prompt=False,
+        )
+    except Exception as e:
+        import sys
+        print(
+            f'[{timeout_kind}_w_trace] '
+            f'⚠️  dump_all() failed: {e!r} '
+            f'(label={label!r}, pid={target_pid})',
+            file=sys.stderr,
+        )
+        return None
+
+    import sys
+    print(
+        f'[{timeout_kind}_w_trace] '
+        f'⏰ timed out after {seconds}s (label={label!r}, '
+        f'pid={target_pid}); snapshot at: {dump_dir}',
+        file=sys.stderr,
+    )
+    return dump_dir
+
+
+@asynccontextmanager
+async def fail_after_w_trace(
+    seconds: float,
+    *,
+    label: str,
+    pid: int | None = None,
+    out_dir: Path | None = None,
+) -> AsyncIterator[None]:
+    '''
+    Async CM: `trio.fail_after(seconds)` + on-timeout
+    `dump_all()` snapshot BEFORE the `trio.TooSlowError`
+    propagates.
+
+    Parameters
+    ----------
+    seconds:
+        timeout budget passed to `trio.fail_after`.
+    label:
+        snapshot dir prefix (e.g. test name).
+    pid:
+        root pid to snapshot. Defaults to current process —
+        which under pytest is the test worker, and its
+        descendants are the spawned subactor tree.
+    out_dir:
+        snapshot parent dir. Defaults to
+        `$XDG_CACHE_HOME/tractor/hung-dumps/`.
+
+    On `trio.TooSlowError`:
+      1. Capture `dump_all()` (best-effort; failure is logged
+         to stderr but doesn't mask the original exception).
+      2. Re-raise so the test fails normally.
+
+    Example
+    -------
+    >>> async with fail_after_w_trace(
+    ...     5.0,
+    ...     label='test_multierror_fast_nursery',
+    ... ):
+    ...     await some_hangy_thing()
+
+    '''
+    # local import — trio is a hard dep of tractor, but the
+    # rest of `trace.py` is trio-free (used from xontrib cli).
+    # Keeping the import scoped here means `trace.py` stays
+    # importable from a plain-python REPL.
+    import trio
+
+    try:
+        with trio.fail_after(seconds):
+            yield
+    except trio.TooSlowError:
+        _do_capture_snapshot(
+            label=label,
+            pid=pid,
+            out_dir=out_dir,
+            seconds=seconds,
+            timeout_kind='fail_after',
+        )
+        raise
+
+
+@contextmanager
+def afk_alarm_w_trace(
+    seconds: int,
+    *,
+    label: str,
+    pid: int | None = None,
+    out_dir: Path | None = None,
+) -> Iterator[None]:
+    '''
+    Sync CM: arm `signal.alarm(seconds)`, on SIGALRM fire
+    `dump_all()` then raise `AFKAlarmTimeout` so the test
+    fails.
+
+    Hard-kill backstop for cases where `trio.fail_after`
+    cannot deliver cancellation — e.g. python-level GIL-
+    hostage hangs, native locks held by C extensions, or a
+    forkserver-launchpad parked in `os.read()`.
+
+    Constraints
+    -----------
+    - Must be invoked from the MAIN thread (`signal.alarm`
+      can only be armed on main thread).
+    - Cannot be nested with other SIGALRM consumers — the
+      previous handler is restored on exit, but two
+      overlapping `afk_alarm` CMs will clobber each other.
+
+    Parameters mirror `fail_after_w_trace`. `seconds` is
+    clamped to integer (signal.alarm granularity).
+
+    Example
+    -------
+    >>> with afk_alarm_w_trace(
+    ...     60, label='test_sigint_closes_lifetime_stack',
+    ... ):
+    ...     trio.run(main)
+
+    '''
+    seconds_int: int = max(1, int(seconds))
+
+    def _handler(signum, frame):
+        raise AFKAlarmTimeout(
+            f'afk_alarm fired after {seconds_int}s '
+            f'(label={label!r})'
+        )
+
+    prev_handler = signal.signal(signal.SIGALRM, _handler)
+    signal.alarm(seconds_int)
+    try:
+        yield
+        signal.alarm(0)  # disarm on clean exit
+    except AFKAlarmTimeout:
+        # alarm already self-cleared; capture diag + re-raise
+        _do_capture_snapshot(
+            label=label,
+            pid=pid,
+            out_dir=out_dir,
+            seconds=seconds_int,
+            timeout_kind='afk_alarm',
+        )
+        raise
+    finally:
+        # belt-and-suspenders: ensure alarm is disarmed even
+        # on non-alarm exception paths (e.g. test failed for a
+        # different reason inside the body).
+        signal.alarm(0)
+        signal.signal(signal.SIGALRM, prev_handler)
+
+
+# ---------------------------------------------------------------
+# pytest fixture wrappers
+# ---------------------------------------------------------------
+# Pre-bind the snapshot `label=` to `request.node.name` so tests
+# don't have to plumb `request: pytest.FixtureRequest` AND
+# `label=request.node.name` through every call site.
+#
+# Re-exported from `tractor._testing.pytest` so they're picked up
+# by pytest's plugin-discovery (per the `pytest_plugins` entry in
+# `pyproject.toml`'s `[tool.pytest.ini_options]`).
+# ---------------------------------------------------------------
+
+if _pytest is not None:
+
+    @_pytest.fixture(name='fail_after_w_trace')
+    def fail_after_w_trace_fixture(
+        request: _pytest.FixtureRequest,
+    ) -> FailAfterWTraceFactory:
+        '''
+        Pre-labeled async-CM factory for
+        `fail_after_w_trace`.
+
+        Auto-injects `label=request.node.name` so tests just
+        do::
+
+            async def test_foo(
+                fail_after_w_trace: FailAfterWTraceFactory,
+            ):
+                async with fail_after_w_trace(5.0):
+                    await some_hangy_thing()
+
+        instead of the more verbose::
+
+            async def test_foo(request):
+                async with fail_after_w_trace(
+                    5.0, label=request.node.name,
+                ):
+                    ...
+
+        Any kwarg can still be overridden by the caller (e.g.
+        a custom `label=` for hand-tuned dedup of snapshot
+        dirs when parametrize ids aren't discriminating
+        enough).
+
+        '''
+        @asynccontextmanager
+        async def _bound(seconds, **kwargs):
+            kwargs.setdefault('label', request.node.name)
+            async with fail_after_w_trace(seconds, **kwargs):
+                yield
+
+        return _bound
+
+    @_pytest.fixture(name='afk_alarm_w_trace')
+    def afk_alarm_w_trace_fixture(
+        request: _pytest.FixtureRequest,
+    ) -> AfkAlarmWTraceFactory:
+        '''
+        Pre-labeled sync-CM factory for `afk_alarm_w_trace`.
+
+        Sync sibling of `fail_after_w_trace` — wraps the
+        SIGALRM-based hard wall-clock backstop with auto-
+        injected `label=request.node.name`::
+
+            def test_foo(
+                afk_alarm_w_trace: AfkAlarmWTraceFactory,
+            ):
+                with afk_alarm_w_trace(10):
+                    trio.run(main)
+
+        See `afk_alarm_w_trace` for constraints (must run on
+        main thread; clobbers other SIGALRM consumers).
+
+        '''
+        @contextmanager
+        def _bound(seconds, **kwargs):
+            kwargs.setdefault('label', request.node.name)
+            with afk_alarm_w_trace(seconds, **kwargs):
+                yield
+
+        return _bound
diff --git a/xontrib/tractor_diag.xsh b/xontrib/tractor_diag.xsh
index d48e40c7..273a30fb 100644
--- a/xontrib/tractor_diag.xsh
+++ b/xontrib/tractor_diag.xsh
@@ -6,7 +6,7 @@ prefix-completion treats them as a sub-cmd group — type
 `acli.<TAB>` to see the full set.
 
 Provides:
-  - `acli.ptree <pid|pgrep-pat>`       psutil-backed proc tree,
+  - `acli.ptree <pid|pgrep-pat>`        psutil-backed proc tree,
                                         live + zombies split.
   - `acli.hung_dump <pid|pat> [...]`    kernel `wchan`/`stack` +
                                         `py-spy dump` (incl `--locals`)
@@ -17,6 +17,10 @@ Provides:
                                         (e.g. `piker`, `tractor`);
                                         path -> use as-is.
                                         default: `$XDG_RUNTIME_DIR/tractor`.
+  - `acli.dump_all <pid> [--out-dir]    full snapshot bundle —
+                          [--label]`    ptree + hung_dump + bindspace
+                                        written to a timestamped dir
+                                        for sharing / AI introspection.
   - `acli.reap [opts]`                  SC-polite zombie-subactor
                                         reaper + optional `/dev/shm/`
                                         + UDS sock-file sweeps.
@@ -29,188 +33,30 @@ Or source directly:
   source ./xontrib/tractor_diag.xsh
 
 Pipe-to-paste idiom (xonsh):
-  hung-dump pytest |t /tmp/hung.log
+  acli.hung_dump pytest |t /tmp/hung.log
+
+The diagnostic core lives in `tractor._testing.trace` so it
+can also be invoked from inside pytest tests (e.g. via
+`fail_after_w_trace` / `afk_alarm_w_trace` capture-on-hang
+helpers) — these aliases are just thin terminal wrappers.
 
 Requires `psutil` for full functionality (`ptree` and the
-`hung-dump` tree-walk). Falls back to `pgrep -P` recursion
-if missing.
+`hung_dump` tree-walk). Falls back to `pgrep -P` recursion if
+missing.
 """
 
-import os
-import re
-import subprocess as sp
 from pathlib import Path
 
-try:
-    import psutil
-except ImportError:
-    psutil = None
-    print(
-        '[tractor-diag] `psutil` missing — '
-        'acli.ptree disabled, acli.hung_dump uses pgrep fallback. '
-        '`uv pip install psutil` for full functionality.'
-    )
-
-
-# matches tractor's UDS sock naming: `<actor_name>@<pid>.sock`
-_UDS_SOCK_RE = re.compile(
-    r'^(?P<name>.+)@(?P<pid>\d+)\.sock$'
+from tractor._testing.trace import (
+    dump_all as _dump_all,
+    dump_hung_state,
+    dump_proc_tree,
+    resolve_pids,
+    scan_bindspace,
 )
 
 
-# --- helpers --------------------------------------------------
-
-def _resolve_pids(arg: str) -> list:
-    '''Resolve a numeric pid OR a `pgrep -f` pattern.'''
-    if arg.isdigit():
-        return [int(arg)]
-    try:
-        out = sp.check_output(
-            ['pgrep', '-f', arg],
-            text=True,
-        )
-    except sp.CalledProcessError:
-        return []
-    return [int(p) for p in out.split() if p]
-
-
-def _walk_tree_psutil(pid: int) -> list:
-    '''Flat list `[Process, *descendants]` via psutil.'''
-    try:
-        p = psutil.Process(pid)
-    except psutil.NoSuchProcess:
-        return []
-    return [p] + p.children(recursive=True)
-
-
-def _walk_tree_with_depth(pid: int):
-    '''
-    Yield `(proc, depth)` pairs walking `pid`'s tree. `depth==0`
-    is the root; `depth==1` are direct children, etc. Used by
-    `ptree` to render parent/child relationships visually.
-    '''
-    try:
-        root = psutil.Process(pid)
-    except psutil.NoSuchProcess:
-        return
-    yield root, 0
-    stack: list = [(root, 0)]
-    seen: set = {pid}
-    while stack:
-        parent, d = stack.pop()
-        try:
-            kids = parent.children()
-        except psutil.NoSuchProcess:
-            continue
-        for k in kids:
-            if k.pid in seen:
-                continue
-            seen.add(k.pid)
-            yield k, d + 1
-            stack.append((k, d + 1))
-
-
-def _which_cgroup_slice(pid: int) -> str|None:
-    '''
-    Return which top-level systemd cgroup slice `pid` is
-    rooted in, or `None` if it's not in either:
-
-      - `'system'`: under `/system.slice/...` — typically
-        `.service` units (long-lived daemons explicitly
-        enabled via `systemctl enable`, e.g.
-        `auto-cpufreq.service`, `dbus.service`,
-        `systemd-journald.service`).
-
-      - `'user'`: under `/user.slice/user-<uid>.slice/...`
-        — typically `.scope` units that systemd auto-wraps
-        around desktop-launched apps + login-session
-        procs (e.g. `app-firefox-<id>.scope`,
-        `session-<id>.scope`).
-
-      - `None`: NOT in either slice — pid 1 is NOT
-        managing this proc via cgroup. Combined with
-        `ppid==1`, this is the genuine "leaked / parent
-        died" orphan signal.
-
-    Both slice categories are by-design `ppid==1` (pid 1
-    is actively managing them) and should NOT be flagged
-    as concerning orphans, but distinguishing them is
-    useful: `system.slice` is "real services on this
-    box", `user.slice` is "stuff in your login session".
-
-    Returns `None` on any read error (proc gone, perm
-    denied, non-Linux, etc.) — callers should treat that
-    as "unknown, classify as plain orphan".
-
-    '''
-    try:
-        with open(f'/proc/{pid}/cgroup') as f:
-            cg: str = f.read()
-    except (
-        FileNotFoundError,
-        PermissionError,
-        ProcessLookupError,
-        OSError,
-    ):
-        return None
-    if '/system.slice/' in cg:
-        return 'system'
-    if '/user.slice/' in cg:
-        return 'user'
-    return None
-
-
-def _walk_tree_pgrep(pid: int) -> list:
-    '''psutil-less fallback — recursive `pgrep -P`.'''
-    out = [pid]
-    try:
-        kids = sp.check_output(
-            ['pgrep', '-P', str(pid)],
-            text=True,
-        ).split()
-    except sp.CalledProcessError:
-        return out
-    for k in kids:
-        out.extend(_walk_tree_pgrep(int(k)))
-    return out
-
-
-def _ensure_sudo_cached() -> bool:
-    '''
-    Ensure `sudo` credentials are cached so subsequent
-    `sudo -n` calls succeed without prompting.
-
-    Returns True if cached (or successfully refreshed),
-    False if user cancelled or sudo is unavailable.
-
-    Tries `sudo -n true` first as a no-op probe; if that
-    fails, runs `sudo -v` which prompts interactively to
-    validate/refresh the credential timestamp.
-    '''
-    # probe — already cached?
-    cached = sp.run(
-        ['sudo', '-n', 'true'],
-        capture_output=True,
-    ).returncode == 0
-    if cached:
-        return True
-
-    print(
-        '[tractor-diag] needs `sudo` for /proc/<pid>/stack '
-        'and `py-spy dump`; caching creds via `sudo -v`...'
-    )
-    try:
-        rc = sp.run(['sudo', '-v']).returncode
-    except KeyboardInterrupt:
-        print('  cancelled — proceeding without sudo')
-        return False
-    except FileNotFoundError:
-        print('  sudo not on PATH — proceeding without sudo')
-        return False
-    return rc == 0
-
-
-# --- ptree ---------------------------------------------------
+# --- ptree ----------------------------------------------------
 
 def _ptree(args):
     '''
@@ -220,36 +66,9 @@ def _ptree(args):
 
     usage: acli.ptree [--tree|-t] <pid|pgrep-pattern> [...]
 
-    classification (per-proc, not per-tree):
+    See `tractor._testing.trace.dump_proc_tree()` for the
+    bucket semantics + classification details.
 
-      - zombies:  `status in (Z, X)` — defunct, parent
-                  hasn't reaped (or kernel-marked dead).
-      - orphans:  `ppid == 1` — original parent exited;
-                  has been reparented to init. Includes
-                  the *root* of an abandoned tree AND
-                  any descendant that ended up reparented
-                  to init mid-flight.
-      - live:     real parent (`ppid > 1`), non-defunct.
-
-    Trees of orphan roots are still walked — their
-    descendants show as `live` if they themselves still
-    have a real (non-init) parent (the orphan root), but
-    the orphan root itself appears in `orphans`.
-
-    Cross-bucket parent annotation (always emitted):
-      when a row's parent (by ppid) lives in a *different*
-      severity bucket, the row is suffixed with
-      `[parent: <pid> (in `<bucket>`)]` so the visual
-      `└─` marker still resolves to a findable parent
-      even when bucketing scatters parent and child into
-      separate sections.
-
-    `--tree` / `-t` flag (opt-in):
-      additionally emit a flat walk-order `## tree`
-      section at the top — a contiguous parent-child
-      tree shape with no severity-grouping. Same procs,
-      no annotations needed because each parent appears
-      directly above its children.
     '''
     flag_tree: bool = False
     pos_args: list = []
@@ -262,224 +81,19 @@ def _ptree(args):
     if not pos_args:
         print('usage: acli.ptree [--tree|-t] <pid|pgrep-pattern> [...]')
         return 1
-    if psutil is None:
-        print('ptree requires psutil; install via `uv pip install psutil`')
-        return 1
 
     roots: list = []
     for a in pos_args:
-        roots.extend(_resolve_pids(a))
+        roots.extend(resolve_pids(a))
     roots = sorted(set(roots))
     if not roots:
         print(f'(no procs match: {pos_args})')
         return 1
 
-    # statuses considered "defunct" — STATUS_ZOMBIE is the
-    # common case (`Z`); STATUS_DEAD (`X`) is rarer but kernel-
-    # reported and equally not-coming-back.
-    defunct_statuses: set = {
-        psutil.STATUS_ZOMBIE,
-        getattr(psutil, 'STATUS_DEAD', 'dead'),
-    }
-
-    seen: set = set()
-    walk_order: list = []  # [(proc, depth)] preserved walk order
-    live: list = []        # [(proc, depth)]
-    orphans: list = []
-    # `ppid==1` AND rooted in `/system.slice/` cgroup —
-    # real systemd-managed services (e.g. `auto-cpufreq`,
-    # `NetworkManager`).
-    system_slice: list = []
-    # `ppid==1` AND rooted in `/user.slice/.../*.scope` —
-    # desktop-launched apps wrapped by systemd-user in
-    # transient `.scope` units (e.g. Firefox, browsers,
-    # editors started from a launcher).
-    user_slice: list = []
-    zombies: list = []
-    gone: list = []
-
-    # parent-bucket lookup populated post-classification so
-    # `_row()` can annotate cross-bucket parent refs.
-    pid_to_bucket: dict = {}
-
-    for r in roots:
-        for (p, depth) in _walk_tree_with_depth(r):
-            if p.pid in seen:
-                continue
-            seen.add(p.pid)
-            try:
-                status: str = p.status()
-                ppid: int = p.ppid()
-            except psutil.NoSuchProcess:
-                gone.append(p.pid)
-                continue
-            entry = (p, depth)
-            # severity order:
-            #   zombie > orphan > system-slice > user-slice > live
-            # `ppid==1` splits into:
-            #   - `system-slice` (rooted in `/system.slice/` —
-            #     real services, by-design `ppid==1`)
-            #   - `user-slice` (rooted in
-            #     `/user.slice/.../*.scope` — desktop apps
-            #     wrapped by systemd-user, by-design `ppid==1`)
-            #   - `orphans` (everything else with `ppid==1` —
-            #     genuinely concerning).
-            if status in defunct_statuses:
-                zombies.append(entry)
-                pid_to_bucket[p.pid] = 'zombies'
-            elif ppid == 1:
-                slice_kind: str|None = _which_cgroup_slice(p.pid)
-                if slice_kind == 'system':
-                    system_slice.append(entry)
-                    pid_to_bucket[p.pid] = 'system-slice'
-                elif slice_kind == 'user':
-                    user_slice.append(entry)
-                    pid_to_bucket[p.pid] = 'user-slice'
-                else:
-                    orphans.append(entry)
-                    pid_to_bucket[p.pid] = 'orphans'
-            else:
-                live.append(entry)
-                pid_to_bucket[p.pid] = 'live'
-            walk_order.append(entry)
-
-    total: int = (
-        len(live)
-        + len(orphans)
-        + len(system_slice)
-        + len(user_slice)
-        + len(zombies)
-    )
-    print(f'# ptree: {total} procs across roots {roots}')
-
-    hdr = '  ' + 'PID'.rjust(7) + '  ' + 'PPID'.rjust(7) + '  '
-    hdr += 'STATUS'.ljust(10) + '  CMD'
-
-    def _row(entry, bucket: str|None = None):
-        '''
-        Render `(proc, depth)` as an aligned row. Tree depth is
-        rendered as a `└─` marker on the CMD column so PID/PPID/
-        STATUS stay column-aligned.
-
-        When `bucket` is given AND the row's parent lives in a
-        *different* bucket, append a `[parent: <pid> (in `<b>`)]`
-        suffix so the `└─` marker can be resolved across the
-        severity-section split.
-        '''
-        p, depth = entry
-        tree_pfx = ('   ' * depth) + ('└─ ' if depth > 0 else '')
-
-        # cross-bucket parent annotation; safe to compute up
-        # front because `p.ppid()` is cheap and rarely
-        # raises (parent pid is read from `/proc/<pid>/stat`,
-        # cached by psutil).
-        parent_anno: str = ''
-        if (
-            bucket is not None
-            and depth > 0
-        ):
-            try:
-                parent_pid: int = p.ppid()
-            except psutil.NoSuchProcess:
-                parent_pid = 0
-            if parent_pid and parent_pid != 1:
-                parent_bucket: str|None = pid_to_bucket.get(parent_pid)
-                if (
-                    parent_bucket is not None
-                    and parent_bucket != bucket
-                ):
-                    parent_anno = (
-                        f'  [parent: {parent_pid} '
-                        f'(in `{parent_bucket}`)]'
-                    )
-
-        # NOTE: `psutil.ZombieProcess` is a *subclass* of
-        # `psutil.NoSuchProcess`, but the proc is NOT gone —
-        # it's a zombie whose `/proc/<pid>/cmdline` is empty/
-        # unreadable. Catch it FIRST so we still render a
-        # row (using fields that DO work on zombies: pid,
-        # ppid, status, name).
-        try:
-            cmd = ' '.join(p.cmdline())[:140] or '[' + p.name() + ']'
-            r = '  ' + str(p.pid).rjust(7)
-            r += '  ' + str(p.ppid()).rjust(7)
-            r += '  ' + p.status().ljust(10)
-            r += '  ' + tree_pfx + cmd + parent_anno
-            return r
-        except psutil.ZombieProcess:
-            try:
-                ppid_str = str(p.ppid())
-                name = p.name()
-            except psutil.NoSuchProcess:
-                ppid_str, name = '?', '?'
-            r = '  ' + str(p.pid).rjust(7)
-            r += '  ' + ppid_str.rjust(7)
-            r += '  ' + 'zombie'.ljust(10)
-            r += '  ' + tree_pfx + '[' + name + ' <defunct>]' + parent_anno
-            return r
-        except psutil.NoSuchProcess:
-            return '  ' + str(p.pid).rjust(7) + '  (gone mid-walk)'
-
-    def _section(
-        title: str,
-        procs: list,
-        hint: str = '',
-        bucket: str|None = None,
-    ):
-        print(f'\n## {title} ({len(procs)})' + (f'  — {hint}' if hint else ''))
-        if not procs:
-            print('  (none)')
-            return
-        print(hdr)
-        for p in procs:
-            print(_row(p, bucket=bucket))
-
-    # `--tree` opt-in: emit a flat walk-order section first
-    # so the parent-child tree shape is contiguous (no
-    # severity-grouping). No `bucket` arg → no cross-bucket
-    # annotation, since each parent appears directly above
-    # its children here.
-    if flag_tree:
-        _section(
-            'tree', walk_order,
-            'flat walk-order, parent-child preserved',
-        )
-
-    # severity-ordered: most concerning first. Each section
-    # passes its own `bucket` name so `_row()` can annotate
-    # rows whose parents live in a different section.
-    _section(
-        'zombies', zombies,
-        'status `Z`/`X`, parent has not reaped',
-        bucket='zombies',
-    )
-    _section(
-        'orphans', orphans,
-        '`ppid==1`, NOT in a `system.slice`/`user.slice` cgroup '
-        '(likely leaked / parent gone)',
-        bucket='orphans',
-    )
-    _section(
-        'system-slice', system_slice,
-        '`ppid==1`, rooted under `/system.slice/` '
-        '(real systemd-managed service — daemon, login '
-        'session manager, etc; not a leak)',
-        bucket='system-slice',
-    )
-    _section(
-        'user-slice', user_slice,
-        '`ppid==1`, rooted under `/user.slice/.../*.scope` '
-        '(desktop-launched app wrapped by systemd-user — '
-        'browser, editor, etc; not a leak)',
-        bucket='user-slice',
-    )
-    _section('live', live, bucket='live')
-
-    if gone:
-        print(f'\n## gone-during-walk ({len(gone)}): {gone}')
+    print(dump_proc_tree(roots, flag_tree=flag_tree), end='')
 
 
-# --- hung-dump ------------------------------------------------
+# --- hung-dump -----------------------------------------------
 
 def _hung_dump(args):
     '''
@@ -489,248 +103,116 @@ def _hung_dump(args):
     usage: acli.hung_dump <pid|pgrep-pattern> [...]
 
     note: `/proc/<pid>/stack` and `py-spy dump` typically
-    require CAP_SYS_PTRACE — invoked via `sudo -n`. run
-    `sudo true` first to cache creds.
+    require CAP_SYS_PTRACE — invoked via `sudo -n`. If sudo
+    isn't cached this alias prompts (via `sudo -v`); for the
+    non-interactive equivalent see
+    `tractor._testing.trace.dump_hung_state(allow_sudo_prompt=False)`.
+
     '''
     if not args:
         print('usage: acli.hung_dump <pid|pgrep-pattern> [...]')
         return 1
 
-    # cache sudo creds upfront so per-pid `sudo -n` calls
-    # for `cat /proc/<pid>/stack` and `py-spy dump` don't
-    # each prompt (or silently fail).
-    have_sudo: bool = _ensure_sudo_cached()
-
     roots: list = []
     for a in args:
-        roots.extend(_resolve_pids(a))
+        roots.extend(resolve_pids(a))
     roots = sorted(set(roots))
     if not roots:
         print(f'(no procs match: {args})')
         return 1
 
-    pids: list = []
-    seen: set = set()
-    for r in roots:
-        if psutil is not None:
-            walk = [p.pid for p in _walk_tree_psutil(r)]
-        else:
-            walk = _walk_tree_pgrep(r)
-        for pid in walk:
-            if pid not in seen:
-                seen.add(pid)
-                pids.append(pid)
-
-    print(f'# tree: {pids}')
-    print('\n## ps forest')
-    $[ps -o pid,ppid,pgid,stat,cmd -p @(','.join(map(str, pids)))]
-
-    for pid in pids:
-        print(f'\n## pid {pid}')
-
-        for f in ('wchan', 'stack'):
-            path = Path(f'/proc/{pid}/{f}')
-            try:
-                txt = path.read_text().rstrip()
-                print(f'-- /proc/{pid}/{f} --\n{txt}')
-            except PermissionError:
-                if not have_sudo:
-                    print(
-                        f'-- /proc/{pid}/{f}: '
-                        'PermissionError (no sudo) --'
-                    )
-                    continue
-                try:
-                    txt = sp.check_output(
-                        ['sudo', '-n', 'cat', str(path)],
-                        text=True,
-                        stderr=sp.DEVNULL,
-                    ).rstrip()
-                    print(f'-- /proc/{pid}/{f} (sudo) --\n{txt}')
-                except sp.CalledProcessError:
-                    print(
-                        f'-- /proc/{pid}/{f}: '
-                        'sudo cred expired? rerun --'
-                    )
-            except FileNotFoundError:
-                print(f'-- /proc/{pid}/{f}: proc gone --')
-
-        print(f'-- py-spy {pid} --')
-        if not have_sudo:
-            print('  (skipped — no sudo)')
-            continue
-        try:
-            $[sudo -n py-spy dump --pid @(pid) --locals]
-        except Exception as e:
-            print(f'  (py-spy failed: {e})')
+    print(
+        dump_hung_state(roots, allow_sudo_prompt=True),
+        end='',
+    )
 
 
-# --- bindspace-scan -------------------------------------------
+# --- bindspace-scan ------------------------------------------
 
 def _bindspace_scan(args):
     '''
-    Scan a tractor UDS bindspace dir for orphan sock files
-    (those whose embedded `<pid>` no longer corresponds to
-    a live process).
+    Scan a tractor UDS bindspace dir for orphan sock files.
 
     usage: acli.bindspace_scan [<name>|<dir>]
 
-    - no arg          -> `$XDG_RUNTIME_DIR/tractor`
-                         (or `/run/user/<uid>/tractor`)
-    - bare `<name>`   -> `$XDG_RUNTIME_DIR/<name>`,
-                         for projects like `piker` that bind
-                         their own sibling sub-dir alongside
-                         tractor's default
-    - path (abs or
-      containing `/`) -> use as-is
+    See `tractor._testing.trace.scan_bindspace()` for full arg
+    semantics + output-bucket details.
 
     '''
-    runtime: str = os.environ.get(
-        'XDG_RUNTIME_DIR',
-        f'/run/user/{os.getuid()}',
-    )
-    if args:
-        arg: str = args[0]
-        if (
-            arg.startswith('/')
-            or
-            '/' in arg
-        ):
-            bs_dir = Path(arg)
-        else:
-            # bare name -> `$XDG_RUNTIME_DIR/<name>` so
-            # callers can say `acli.bindspace_scan piker`
-            bs_dir = Path(runtime) / arg
-    else:
-        bs_dir = Path(runtime) / 'tractor'
+    arg: str | None = args[0] if args else None
+    print(scan_bindspace(arg), end='')
 
-    if not bs_dir.exists():
-        print(f'(no bindspace at {bs_dir})')
+
+# --- dump-all (snapshot bundle) ------------------------------
+
+def _dump_all_alias(args):
+    '''
+    Capture a full diag snapshot bundle for a hung proc-tree
+    into a timestamped directory for offline / AI inspection.
+
+    usage: acli.dump_all <pid|pgrep-pat>
+                        [--label <label>]
+                        [--out-dir <path>]
+
+    Writes:
+      <out_dir>/<label>__<ts>/{trace.txt, bindspace.txt, meta.json}
+
+    Defaults:
+      --label   = `manual`
+      --out-dir = `$XDG_CACHE_HOME/tractor/hung-dumps/`
+                  (fallback `~/.cache/tractor/hung-dumps/`)
+
+    '''
+    import argparse
+    parser = argparse.ArgumentParser(
+        prog='acli.dump_all',
+        description=_dump_all_alias.__doc__,
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+    )
+    parser.add_argument(
+        'target',
+        help='pid or pgrep -f pattern',
+    )
+    parser.add_argument(
+        '--label', '-l',
+        default='manual',
+        help='snapshot dir label prefix (default: `manual`)',
+    )
+    parser.add_argument(
+        '--out-dir', '-o',
+        type=Path,
+        default=None,
+        help='snapshot root dir (default: '
+             '$XDG_CACHE_HOME/tractor/hung-dumps/)',
+    )
+    try:
+        ns = parser.parse_args(args)
+    except SystemExit as se:
+        return int(se.code) if se.code is not None else 0
+
+    pids: list = resolve_pids(ns.target)
+    if not pids:
+        print(f'(no procs match: {ns.target})')
         return 1
 
-    socks = sorted(bs_dir.glob('*.sock'))
-    print(f'## bindspace {bs_dir} ({len(socks)} sock file(s))')
+    # snapshot scoped to ONE root — pick the first matched
+    # pid. Multi-root snapshots can be done by invoking
+    # `acli.dump_all <pid>` per root.
+    root_pid: int = pids[0]
+    if len(pids) > 1:
+        print(
+            f'[acli.dump_all] {len(pids)} pids matched '
+            f'{ns.target!r}; snapshotting tree from {root_pid} '
+            f'(re-run per-pid for others: {pids[1:]})'
+        )
 
-    live_active: list = []  # PID alive AND ppid != 1
-    live_orphaned: list = []  # PID alive AND ppid == 1 (init-adopted)
-    dead_orphans: list = []  # PID gone, sock stale
-    bogus: list = []
-
-    def _ppid(pid: int) -> int | None:
-        '''
-        Read `/proc/<pid>/stat` -> ppid. Returns None on race
-        (proc died between `os.kill(pid, 0)` succeeding and this
-        read), permission errors, or non-linux.
-        '''
-        try:
-            with open(f'/proc/{pid}/stat') as f:
-                # field [3] of `man 5 proc` `/proc/<pid>/stat`
-                # NB: field [1] is `(comm)` which can contain
-                # spaces and parens — split from the *last*
-                # `)` to avoid that bullshit.
-                stat: str = f.read()
-            after_comm: str = stat.rsplit(')', 1)[1].strip()
-            return int(after_comm.split()[1])  # state(0) ppid(1)
-        except (FileNotFoundError, PermissionError, ProcessLookupError, OSError):
-            return None
-
-    for s in socks:
-        m = _UDS_SOCK_RE.match(s.name)
-        if not m:
-            bogus.append(s)
-            continue
-        pid = int(m['pid'])
-        name = m['name']
-        try:
-            os.kill(pid, 0)
-        except ProcessLookupError:
-            dead_orphans.append((s, pid, name))
-            continue
-        except PermissionError:
-            # exists but owned by another user — treat as live-active
-            # (we can't read its /proc/<pid>/stat to check ppid)
-            live_active.append((s, pid, name, None))
-            continue
-
-        # PID is alive in our euid view; classify by ppid
-        ppid: int | None = _ppid(pid)
-        if ppid == 1:
-            # adopted by init -> the original parent reaped
-            # without cleaning up this sub. Same class as
-            # what `acli.reap` detects.
-            live_orphaned.append((s, pid, name, ppid))
-        else:
-            live_active.append((s, pid, name, ppid))
-
-    print(f'\n## live-active ({len(live_active)})  — PID alive, parent still own it')
-    if not live_active:
-        print('  (none)')
-    for s, pid, name, ppid in live_active:
-        row = '  ' + str(pid).rjust(7)
-        row += '  ' + name.ljust(32)
-        row += '  ' + s.name
-        if ppid is not None:
-            row += f'  (ppid={ppid})'
-        print(row)
-
-    print(
-        f'\n## orphaned-alive ({len(live_orphaned)})  '
-        f'— PID alive but `ppid==1`, parent reaped; '
-        f'`acli.reap` candidate'
+    dump_dir = _dump_all(
+        root_pid,
+        out_dir=ns.out_dir,
+        label=ns.label,
+        allow_sudo_prompt=True,  # CLI: ok to prompt
     )
-    if not live_orphaned:
-        print('  (none)')
-    for s, pid, name, ppid in live_orphaned:
-        row = '  ' + str(pid).rjust(7)
-        row += '  ' + name.ljust(32)
-        row += '  ' + s.name + '  (adopted by init)'
-        print(row)
-
-    print(f'\n## orphaned-dead ({len(dead_orphans)})  — PID gone, sock stale')
-    if not dead_orphans:
-        print('  (none)')
-    for s, pid, name in dead_orphans:
-        row = '  ' + str(pid).rjust(7)
-        row += '  ' + name.ljust(32)
-        row += '  ' + s.name + '  (no live proc)'
-        print(row)
-
-    if bogus:
-        print(
-            f'\n## non-tractor ({len(bogus)})  '
-            f'— filename lacks `@<pid>` suffix, '
-            f'cannot determine liveness intrinsically'
-        )
-        for s in bogus:
-            print(f'  {s.name}')
-        # show a copy-pastable `ss` cmd per sock so the
-        # caller can resolve listener-PID externally
-        # (e.g. for piker's `chart.sock` / `pikerd.sock`
-        # style flat names). `ss -lpx 'src = <path>'`
-        # prints `users:(("<proc>",pid=<N>,fd=<M>))` for
-        # the listening side; empty output -> nobody's
-        # listening -> safe to unlink.
-        print(
-            '\nto check liveness manually '
-            '(needs `iproute2`/`ss`):'
-        )
-        for s in bogus:
-            print(f"  ss -lpx 'src = {s}'")
-
-    if dead_orphans or live_orphaned:
-        print(
-            '\nto sweep BOTH orphaned-alive subs (graceful '
-            'SIGINT -> SIGKILL) AND dead-orphan socks in one shot:'
-        )
-        print('  acli.reap --uds')
-
-    if dead_orphans:
-        unlink_cmd = ' '.join(str(o[0]) for o in dead_orphans)
-        print(
-            '\n(or to unlink dead-orphan socks manually, '
-            "skipping `acli.reap`'s graceful-cancel ladder:)"
-        )
-        print(f'  rm {unlink_cmd}')
+    print(f'[acli.dump_all] snapshot written to: {dump_dir}')
 
 
 # --- acli.reap ------------------------------------------------
@@ -916,6 +398,7 @@ _TCLI_ALIASES: dict = {
     'acli.ptree': _ptree,
     'acli.hung_dump': _hung_dump,
     'acli.bindspace_scan': _bindspace_scan,
+    'acli.dump_all': _dump_all_alias,
     'acli.reap': _tractor_reap,
 }