API Reference

Command line

pidprobe snap <PID> [--pretty] [--timeout SECONDS] [--no-mask]

snap injects the built-in collectors -- plus every installed collector plugin -- into a running CPython 3.14+ process and prints one JSON snapshot. The default output is a single compact line so it can be piped straight into jq; --pretty indents it instead. --timeout (default: 5 seconds) is a hard budget for the whole probe. --no-mask turns off secret masking, which is otherwise on.

pidprobe eval <PID> <EXPR> [--pretty] [--timeout SECONDS] [--no-mask]

eval evaluates one Python expression inside a running process and prints its rendered value, without collecting a snapshot:

$ pidprobe eval 12345 'len(queue)'
{"pid":12345,"expression":"len(queue)","type":"int","result":"12","masking_enabled":true}

The expression is compiled in the target against a copy of its __main__ namespace, and the result is rendered by the same rules as stack locals -- the bounds and the masking below both apply, and result is therefore always a string. type names the result's type and survives masking.

Note

Statements are not expressions: pidprobe eval 12345 'cache = {}' comes back as a SyntaxError rather than rebinding anything. Evaluating a call can still have side effects, because the target runs it — the same call you would make in a debugger.

Anything the expression raises -- a SyntaxError from compiling it, a NameError, or an exception from the expression itself -- comes back as the failure it is, never as a timeout: the target answers with the exception instead of going quiet.

Secret masking

Values bound to a credential-like name -- password, passwd, passphrase, pwd, secret, token, apikey, accesskey, privatekey, credential or authorization, matched with case and separators ignored -- are replaced with "<masked>". Local variables and values under a matching string key in a dictionary are both covered, and stacks.masking_enabled records whether masking was on. eval matches the same patterns against the expression text, so pidprobe eval 12345 api_key is masked and reports "masking_enabled": true.

Warning

Masking happens inside the target process, so a masked value never crosses the return channel. --no-mask removes that guarantee: raw credentials then land in the output, and in whatever you pipe it into.

Every value is rendered within fixed bounds -- 3 levels of nesting, 10 elements per container, 200 characters per repr() and 2000 characters in total -- with what was left out marked as ... or ...<truncated>.

pidprobe diff <PID> --interval SECONDS [--count N] [--pretty] [--timeout SECONDS]

diff samples one process over and over and prints only what moved between two consecutive samples, which is what finds a leak: a growing type is invisible in any single snapshot and tedious to spot across two full ones.

$ pidprobe diff 12345 --interval 5 --count 3 | jq -c '.objects.types[0]'
{"type":"app.models.Session","before":1204,"after":3861,"delta":2657}
{"type":"app.models.Session","before":3861,"after":6498,"delta":2637}

Each delta is printed as one JSON object on its own line -- JSON Lines -- and flushed as soon as it is computed, so the stream can be piped into jq or a log while it is still running. --pretty indents each delta instead, which is for reading: the output is then no longer one delta per line.

--count N takes N snapshots and therefore prints N - 1 deltas, since a delta needs a pair; --count 1 prints nothing. Without --count the command samples until it is interrupted with Ctrl-C, which ends it cleanly with exit code 130 and no traceback. --interval is measured between the starts of two samples, so the time a probe itself costs is taken off the wait rather than added to it.

Note

Only the objects, gc and fds collectors run -- the three sections a delta is defined for. Stacks and plugin sections are not sampled at all, so diff stops the target for less time per sample than snap does, and --no-mask has nothing to apply to: a delta reports counters, never values read out of the target.

pidprobe doctor [PID] [--json] [--pretty]

doctor explains whether attaching would work, without attaching: it never injects anything, so it is safe to point at a production process. Given a PID it also examines that process; without one it reports only on this environment, which is what you want before there is a target to name.

$ pidprobe doctor 12345
pidprobe doctor: checking this environment against pid 12345

  OK      prober_remote_debug   pidprobe runs cpython 3.14.6 with remote debugging enabled
  OK      return_channel        an AF_UNIX return channel binds at /tmp/pidprobe-3f9a1c2e/s.sock
  OK      collector_plugins     5 collectors will run: stacks, objects, gc, fds, sqlalchemy
  FAIL    ptrace_scope          kernel.yama.ptrace_scope is 2
            cause: at scope 2 only a process holding CAP_SYS_PTRACE may attach to anything, ...
            confirm: cat /proc/sys/kernel/yama/ptrace_scope
            fix: run pidprobe as root or with CAP_SYS_PTRACE, or relax the knob with ...
  ...

1 check failed; attaching to pid 12345 will not work

Every check that fails or warns carries all four of the things you need: which check it was, the cause, a confirm command you can run yourself, and the fix. The type rejects a check built without them, so no diagnosis can come back as a bare "Permission denied". --json prints the same report as {"pid", "attachable", "checks"} for tooling.

Troubleshooting has one section per check, under the same names and in the same order the report prints them, so a failing check reads straight across into the page that explains it.

Note

A check that cannot be answered here is reported as SKIPPED rather than guessed at: ptrace_scope on macOS, task_for_pid on Linux, and everything that reads another process' environment or namespace off Linux. Skipped checks still print the command that would answer them.

Warning

Establishing the target's Python version means running the target's own executable with -c -- a process cannot be asked for its version from the outside. That only happens when the binary's name identifies it as an interpreter (python*, pypy*), so pointing doctor at an arbitrary pid never executes an arbitrary program; it reports target_python as a warning instead.

Global options

pidprobe [--timeout SECONDS] [--debug] <command> ...

--timeout before the subcommand sets the hard probe budget for whichever command follows, overriding the built-in default of 5 seconds; the same option after the subcommand overrides it in turn, so the more specific one wins. doctor accepts it and ignores it, because it never attaches and so has nothing to budget.

--debug re-raises an unexpected error instead of summarising it, printing the real traceback. PIDPROBE_DEBUG=1 in the environment does the same, which is the version you want inside a script. It only affects unexpected errors: a diagnosed failure like a timeout is reported the same way either way.

Exit codes

Every command reports its outcome with one of these, so a script can act on what went wrong without parsing the message:

Code	Meaning
0	success
1	probe failed for a reason with no more specific code
2	invalid command line
3	no such process
4	attaching to the target was refused
5	the target did not answer within the timeout
6	the injected code raised inside the target
7	doctor found a check that blocks attaching
70	pidprobe hit an unexpected error (a bug)
130	interrupted with Ctrl-C
141	the reader of stdout closed the pipe

pidprobe --help prints the same table, word for word: both are rendered from one list in the code. The failure itself is explained on stderr as a single pidprobe: ... line for snap, eval and diff; doctor prints its report to stdout whatever it says, because the report is its output, and only the exit code separates a clean environment from a blocked one.

Note

3 and 4 are worth telling apart: a vanished process is nothing to fix and may be worth retrying, while a refused attach needs an operator. 7 likewise means the diagnosis itself succeeded — distinct from 1, which means no diagnosis could be produced.

A snap, eval or diff failure always names pidprobe doctor <PID> in its error, whatever went wrong. A doctor failure does not, since that is where you already are.

Snapshot format

Every snapshot is a JSON object with a schema_version, a meta section and one section per collector, named after that collector:

Key	Contents
meta	Target and prober Python versions, pidprobe version, capture timestamp, measured stop duration inside the target, total elapsed time, and one status report per collector
stacks	Per-thread call stacks: file, line, function and a bounded, credential-masking repr of every local variable, innermost frame first
objects	Counts of GC-tracked objects grouped by type, top 50 by count
gc	Garbage collector state: counts, thresholds and per-generation statistics
fds	Open file descriptors, each with its kind, target and (on Linux) socket addresses
sqlalchemy	Connection pools the target holds, with size, checked-out count and overflow; "available": false when the target never imported SQLAlchemy

Note

A collector that fails costs only its own section: that section becomes null and the reason is reported in meta.collectors. The rest of the snapshot still comes back.

Output schema documents every field of every section, the {status, error, payload} envelope underneath them, and how to validate a snapshot against the JSON Schema the package ships.

Warning

objects only counts containers the garbage collector tracks. Atomic values such as int and str are invisible to it and are not counted.

Delta format

pidprobe diff prints a different document from snap: not a snapshot, but the difference between two of them. Every number that moved is reported as a {"before", "after", "delta"} object, where delta is after - before.

{
  "schema_version": "1.0",
  "meta": {
    "pid": 12345,
    "from": "2026-08-05T14:51:08.863505Z",
    "to": "2026-08-05T14:51:13.867786Z",
    "interval_ms": 5004.281
  },
  "objects": {
    "top_n": 50,
    "total_tracked": { "before": 91204, "after": 93871, "delta": 2667 },
    "distinct_types": { "before": 412, "after": 413, "delta": 1 },
    "types": [
      { "type": "app.models.Session", "before": 1204, "after": 3861, "delta": 2657 },
      { "type": "dict", "before": 30112, "after": 30121, "delta": 9 },
      { "type": "tuple", "before": 18004, "after": 17998, "delta": -6 }
    ]
  },
  "gc": {
    "generations": [
      {
        "generation": 0,
        "collections": { "before": 14, "after": 19, "delta": 5 },
        "collected": { "before": 179, "after": 233, "delta": 54 },
        "uncollectable": { "before": 0, "after": 0, "delta": 0 },
        "count": { "before": 7, "after": 925, "delta": 918 }
      }
    ],
    "garbage_count": { "before": 0, "after": 0, "delta": 0 },
    "freeze_count": { "before": 0, "after": 0, "delta": 0 }
  },
  "fds": { "count": { "before": 31, "after": 31, "delta": 0 } }
}

Key	Contents
meta	The pid, the capture timestamps of the two snapshots (from, to) and the milliseconds actually measured between them
objects.types	Every type whose count moved, ranked by delta from fastest-growing to fastest-shrinking, ties broken by type name
objects.total_tracked, objects.distinct_types	How the totals of the objects section moved
gc.generations	One row per generation both snapshots reported, paired by generation index, with every statistic diffed
gc.garbage_count, gc.freeze_count	How the two gc totals moved
fds.count	How many file descriptors the target gained or lost

A type is left out of objects.types when its count did not move -- a delta reports what changed, and most of the hundreds of types a process holds did not. The gc and fds numbers are reported either way: their shape is fixed and small, and "the collector never ran" is worth telling apart from "nothing happened".

Warning

before or after is null when the type was outside that snapshot's top_n ranking, which is not the same as having no instances. The unknown side is then counted as zero, so delta bounds the change in the direction it moved instead of stating it exactly.

Note

A section is null when either snapshot lacked it, which is what a collector that failed inside the target leaves behind. The schema_version tracks pidprobe's output format as a whole; a delta is not a snapshot and is not described by snapshot.schema.json.

Collector plugins

A section can come from any installed package, not only from pidprobe -- including the sqlalchemy section above, which pidprobe ships as an ordinary plugin. Writing a collector plugin walks through that one as a worked example; the contract is below. Publish an entry point in the pidprobe.collectors group:

[project.entry-points."pidprobe.collectors"]
redis = "my_package.collectors:REDIS"

The entry point resolves either to a collector or to a zero-argument callable returning one. What it resolves to carries name, source and description strings -- a Collector, or anything else matching the CollectorSpec protocol.

source does not run in the prober. It becomes a function body inside the target process and must assign a JSON-serializable value to data, which is published as the top-level section named after name — next to stacks, objects, gc and fds, and reported in meta.collectors like any built-in one. The JSON Schema allows unknown top-level keys for exactly this reason.

Warning

source must not import the library it reports on. The import would run inside the target and load a package that process never asked for, changing what you were trying to observe. Look the module up in sys.modules instead and report "available": false when it is not there. Writing a collector plugin has the worked example and the rest of the rules.

What a snapshot would run is available without probing anything, through available_collectors() for the built-ins plus the plugins and discover_collectors() for the plugins alone.

Note

Discovery never fails a snapshot. A plugin that cannot be imported, hands back something that is not a collector, carries source that does not compile, or claims a name a built-in or an earlier plugin already took is logged on the pidprobe.registry logger and left out; every other section still comes back. A plugin that raises inside the target costs only its own section, exactly like a built-in.

Python API

pidprobe

Public package interface for pidprobe.

COLLECTOR_ENTRY_POINT_GROUP = 'pidprobe.collectors' module-attribute

Entry point group a package publishes its collectors in.

SCHEMA_VERSION = '1.0' module-attribute

Version of the snapshot output format; also embedded in every snapshot.

AttachError

Bases: ProbeError

Raised when pidprobe cannot inject code into the target process.

Wraps the low-level failure from :func:sys.remote_exec and adds a hint about the most likely cause, since the underlying exceptions are indistinguishable without knowing the platform.

Source code in src/pidprobe/_errors.py

class AttachError(ProbeError):
    """Raised when pidprobe cannot inject code into the target process.

    Wraps the low-level failure from :func:`sys.remote_exec` and adds a hint
    about the most likely cause, since the underlying exceptions are
    indistinguishable without knowing the platform.
    """

    def __init__(self, pid: int, hint: str, cause: BaseException | None = None) -> None:
        """Build the error.

        Args:
            pid: Process id pidprobe tried to attach to.
            hint: Human-readable explanation of the likely cause.
            cause: Original exception raised by :func:`sys.remote_exec`.
        """
        detail = f" ({type(cause).__name__}: {cause})" if cause is not None else ""
        super().__init__(f"cannot attach to pid {pid}: {hint}{detail}")
        self.pid = pid
        self.hint = hint

    @classmethod
    def from_cause(cls, pid: int, cause: BaseException) -> AttachError:
        """Create an :class:`AttachError` with a cause-specific hint.

        Args:
            pid: Process id pidprobe tried to attach to.
            cause: Exception raised by :func:`sys.remote_exec`.

        Returns:
            An error whose message names the most likely remedy.
        """
        match cause:
            case ProcessLookupError():
                # A vanished target is its own outcome rather than a refused
                # attach: nothing about the environment needs fixing, so it
                # carries its own type and its own exit code.
                return NoSuchProcessError(pid, _NO_SUCH_PROCESS_HINT, cause)
            case PermissionError():
                hint = _PERMISSION_HINT
            case RuntimeError() | ValueError():
                hint = _REFUSED_HINT
            case _:
                hint = _GENERIC_HINT
        return cls(pid, hint, cause)

__init__(pid, hint, cause=None)

Build the error.

Parameters:

Name	Type	Description	Default
pid	int	Process id pidprobe tried to attach to.	required
hint	str	Human-readable explanation of the likely cause.	required
cause	BaseException | None	Original exception raised by :func:sys.remote_exec.	None

Source code in src/pidprobe/_errors.py

def __init__(self, pid: int, hint: str, cause: BaseException | None = None) -> None:
    """Build the error.

    Args:
        pid: Process id pidprobe tried to attach to.
        hint: Human-readable explanation of the likely cause.
        cause: Original exception raised by :func:`sys.remote_exec`.
    """
    detail = f" ({type(cause).__name__}: {cause})" if cause is not None else ""
    super().__init__(f"cannot attach to pid {pid}: {hint}{detail}")
    self.pid = pid
    self.hint = hint

from_cause(pid, cause) classmethod

Create an :class:AttachError with a cause-specific hint.

Parameters:

Name	Type	Description	Default
pid	int	Process id pidprobe tried to attach to.	required
cause	BaseException	Exception raised by :func:sys.remote_exec.	required

Returns:

Type	Description
AttachError	An error whose message names the most likely remedy.

Source code in src/pidprobe/_errors.py

@classmethod
def from_cause(cls, pid: int, cause: BaseException) -> AttachError:
    """Create an :class:`AttachError` with a cause-specific hint.

    Args:
        pid: Process id pidprobe tried to attach to.
        cause: Exception raised by :func:`sys.remote_exec`.

    Returns:
        An error whose message names the most likely remedy.
    """
    match cause:
        case ProcessLookupError():
            # A vanished target is its own outcome rather than a refused
            # attach: nothing about the environment needs fixing, so it
            # carries its own type and its own exit code.
            return NoSuchProcessError(pid, _NO_SUCH_PROCESS_HINT, cause)
        case PermissionError():
            hint = _PERMISSION_HINT
        case RuntimeError() | ValueError():
            hint = _REFUSED_HINT
        case _:
            hint = _GENERIC_HINT
    return cls(pid, hint, cause)

ChannelError

Bases: ProbeError

Raised when the return channel cannot be set up or returns garbage.

Source code in src/pidprobe/_errors.py

class ChannelError(ProbeError):
    """Raised when the return channel cannot be set up or returns garbage."""

Check dataclass

One diagnostic check and everything needed to act on it.

Attributes:

Name	Type	Description
name	str	Stable identifier of the check, used as the JSON key.
status	CheckStatus	What the check concluded.
summary	str	One line naming what was actually observed.
cause	str	Why that blocks or degrades attaching.
confirm	str	A shell command the user can run to see it independently.
fix	str	The concrete change that resolves it.

Raises:

Type	Description
ValueError	If a warning or failure is built without a cause, a confirmation command and a fix -- the guarantee that no output path is ever a bare "Permission denied".

Source code in src/pidprobe/_diagnosis.py

@dataclass(frozen=True, slots=True)
class Check:
    """One diagnostic check and everything needed to act on it.

    Attributes:
        name: Stable identifier of the check, used as the JSON key.
        status: What the check concluded.
        summary: One line naming what was actually observed.
        cause: Why that blocks or degrades attaching.
        confirm: A shell command the user can run to see it independently.
        fix: The concrete change that resolves it.

    Raises:
        ValueError: If a warning or failure is built without a cause, a
            confirmation command and a fix -- the guarantee that no output
            path is ever a bare "Permission denied".
    """

    name: str
    status: CheckStatus
    summary: str
    cause: str = ""
    confirm: str = ""
    fix: str = ""

    def __post_init__(self) -> None:
        """Reject an unactionable warning or failure."""
        if self.status in _EXPLAINED and not all((self.cause, self.confirm, self.fix)):
            message = (
                f"check {self.name!r} is {self.status.value} and must carry a "
                f"cause, a confirmation command and a fix"
            )
            raise ValueError(message)

    def details(self) -> Iterator[tuple[str, str]]:
        """Yield the populated ``(label, text)`` pairs, in reporting order."""
        for label, text in (
            ("cause", self.cause),
            ("confirm", self.confirm),
            ("fix", self.fix),
        ):
            if text:
                yield label, text

__post_init__()

Reject an unactionable warning or failure.

Source code in src/pidprobe/_diagnosis.py

def __post_init__(self) -> None:
    """Reject an unactionable warning or failure."""
    if self.status in _EXPLAINED and not all((self.cause, self.confirm, self.fix)):
        message = (
            f"check {self.name!r} is {self.status.value} and must carry a "
            f"cause, a confirmation command and a fix"
        )
        raise ValueError(message)

details()

Yield the populated (label, text) pairs, in reporting order.

Source code in src/pidprobe/_diagnosis.py

def details(self) -> Iterator[tuple[str, str]]:
    """Yield the populated ``(label, text)`` pairs, in reporting order."""
    for label, text in (
        ("cause", self.cause),
        ("confirm", self.confirm),
        ("fix", self.fix),
    ):
        if text:
            yield label, text

CheckStatus

Bases: StrEnum

Outcome of one diagnostic check.

Attributes:

Name	Type	Description
OK		The condition attaching needs is satisfied.
WARN		Attaching still works, but something is degraded or unverified.
FAIL		Attaching cannot work until this is fixed.
SKIPPED		The check does not apply here, typically on another platform.

Source code in src/pidprobe/_diagnosis.py

class CheckStatus(StrEnum):
    """Outcome of one diagnostic check.

    Attributes:
        OK: The condition attaching needs is satisfied.
        WARN: Attaching still works, but something is degraded or unverified.
        FAIL: Attaching cannot work until this is fixed.
        SKIPPED: The check does not apply here, typically on another platform.
    """

    OK = "ok"
    WARN = "warn"
    FAIL = "fail"
    SKIPPED = "skipped"

Collector dataclass

One named section of a snapshot, collected inside the target process.

Attributes:

Name	Type	Description
name	str	Snapshot key this collector fills. Must be a Python identifier so it can name the generated function, and must not shadow a reserved snapshot key.
source	str	Python source run as a function body inside the target. It must assign a JSON-serializable value to the name data and may call _pidprobe_target_frames().
description	str	One-line summary used by documentation and --help.

Source code in src/pidprobe/collectors/_base.py

@dataclass(frozen=True, slots=True)
class Collector:
    """One named section of a snapshot, collected inside the target process.

    Attributes:
        name: Snapshot key this collector fills. Must be a Python identifier
            so it can name the generated function, and must not shadow a
            reserved snapshot key.
        source: Python source run as a function body inside the target. It
            must assign a JSON-serializable value to the name ``data`` and may
            call ``_pidprobe_target_frames()``.
        description: One-line summary used by documentation and ``--help``.
    """

    name: str
    source: str
    description: str

    def __post_init__(self) -> None:
        """Reject names that cannot be used as a snapshot key.

        Raises:
            ValueError: If the name is not an identifier or is reserved.
        """
        if not self.name.isidentifier():
            message = f"collector name must be an identifier, got {self.name!r}"
            raise ValueError(message)
        if self.name in RESERVED_SECTION_NAMES:
            message = f"collector name {self.name!r} is reserved by the snapshot format"
            raise ValueError(message)

__post_init__()

Reject names that cannot be used as a snapshot key.

Raises:

Type	Description
ValueError	If the name is not an identifier or is reserved.

Source code in src/pidprobe/collectors/_base.py

def __post_init__(self) -> None:
    """Reject names that cannot be used as a snapshot key.

    Raises:
        ValueError: If the name is not an identifier or is reserved.
    """
    if not self.name.isidentifier():
        message = f"collector name must be an identifier, got {self.name!r}"
        raise ValueError(message)
    if self.name in RESERVED_SECTION_NAMES:
        message = f"collector name {self.name!r} is reserved by the snapshot format"
        raise ValueError(message)

CollectorSpec

Bases: Protocol

The contract a collector fulfils, built-in or third-party.

A collector does not run in the prober: its :attr:source is composed into the script pidprobe injects into the target process, which is why a plugin contributes source text rather than a function to call.

Source code in src/pidprobe/registry.py

@runtime_checkable
class CollectorSpec(Protocol):
    """The contract a collector fulfils, built-in or third-party.

    A collector does not run in the prober: its :attr:`source` is composed
    into the script pidprobe injects into the target process, which is why a
    plugin contributes source text rather than a function to call.
    """

    @property
    def name(self) -> str:
        """Snapshot key this collector fills; must be a Python identifier."""

    @property
    def source(self) -> str:
        """Python source run as a function body inside the target process.

        It must assign a JSON-serializable value to the name ``data``, and may
        call ``_pidprobe_target_frames()`` to reach the target's stack frames.
        """

    @property
    def description(self) -> str:
        """One-line summary of the section, for documentation and diagnostics."""

description property

One-line summary of the section, for documentation and diagnostics.

name property

Snapshot key this collector fills; must be a Python identifier.

source property

Python source run as a function body inside the target process.

It must assign a JSON-serializable value to the name data, and may call _pidprobe_target_frames() to reach the target's stack frames.

Diagnosis dataclass

Every check pidprobe doctor ran, and what they add up to.

Attributes:

Name	Type	Description
pid	int | None	Target the target-specific checks ran against, or None when only the prober's own environment was examined.
checks	tuple[Check, ...]	The checks, in the order they were run.

Source code in src/pidprobe/_diagnosis.py

@dataclass(frozen=True, slots=True)
class Diagnosis:
    """Every check ``pidprobe doctor`` ran, and what they add up to.

    Attributes:
        pid: Target the target-specific checks ran against, or ``None`` when
            only the prober's own environment was examined.
        checks: The checks, in the order they were run.
    """

    pid: int | None
    checks: tuple[Check, ...]

    @property
    def failures(self) -> tuple[Check, ...]:
        """The checks that block attaching."""
        return self._with_status(CheckStatus.FAIL)

    @property
    def warnings(self) -> tuple[Check, ...]:
        """The checks that flagged something without blocking attaching."""
        return self._with_status(CheckStatus.WARN)

    @property
    def is_attachable(self) -> bool:
        """Whether nothing found would stop pidprobe from attaching."""
        return not self.failures

    def _with_status(self, status: CheckStatus) -> tuple[Check, ...]:
        return tuple(check for check in self.checks if check.status is status)

failures property

The checks that block attaching.

is_attachable property

Whether nothing found would stop pidprobe from attaching.

warnings property

The checks that flagged something without blocking attaching.

Evaluation

Bases: TypedDict

Result of evaluating one expression inside a target process.

Attributes:

Name	Type	Description
pid	int	Process the expression was evaluated in.
expression	str	Expression as it was handed to pidprobe.
type	str	Name of the result's type, which survives masking.
result	str	Safe-repr of the result, or "<masked>" when the expression reads as a credential name and masking is on.
masking_enabled	bool	Whether credential masking was applied.

Source code in src/pidprobe/_eval.py

class Evaluation(TypedDict):
    """Result of evaluating one expression inside a target process.

    Attributes:
        pid: Process the expression was evaluated in.
        expression: Expression as it was handed to pidprobe.
        type: Name of the result's type, which survives masking.
        result: Safe-repr of the result, or ``"<masked>"`` when the
            expression reads as a credential name and masking is on.
        masking_enabled: Whether credential masking was applied.
    """

    pid: int
    expression: str
    type: str
    result: str
    masking_enabled: bool

NoSuchProcessError

Bases: AttachError

Raised when the pid pidprobe was pointed at does not exist.

Separate from a plain :class:AttachError because the remedy is different in kind: nothing about the machine or its policies is wrong, so a script that retries or scans pids wants to tell "gone" apart from "blocked".

Source code in src/pidprobe/_errors.py

class NoSuchProcessError(AttachError):
    """Raised when the pid pidprobe was pointed at does not exist.

    Separate from a plain :class:`AttachError` because the remedy is
    different in kind: nothing about the machine or its policies is wrong, so
    a script that retries or scans pids wants to tell "gone" apart from
    "blocked".
    """

ProbeError

Bases: Exception

Base class for every error raised by pidprobe.

Source code in src/pidprobe/_errors.py

class ProbeError(Exception):
    """Base class for every error raised by pidprobe."""

ProbeTimeoutError

Bases: ProbeError

Raised when the target does not answer within the time budget.

A target only runs injected code at the next bytecode boundary, so a process parked in a long syscall or inside a C extension never answers.

Source code in src/pidprobe/_errors.py

class ProbeTimeoutError(ProbeError):
    """Raised when the target does not answer within the time budget.

    A target only runs injected code at the next bytecode boundary, so a
    process parked in a long syscall or inside a C extension never answers.
    """

    def __init__(self, timeout_seconds: float, pid: int | None = None) -> None:
        """Build the error.

        Args:
            timeout_seconds: Budget that elapsed without an answer.
            pid: Process id of the target, when known.
        """
        super().__init__(
            _TIMEOUT_MESSAGE.format(
                seconds=timeout_seconds,
                pid="<pid>" if pid is None else pid,
            ),
        )
        self.timeout_seconds = timeout_seconds
        self.pid = pid

__init__(timeout_seconds, pid=None)

Build the error.

Parameters:

Name	Type	Description	Default
timeout_seconds	float	Budget that elapsed without an answer.	required
pid	int | None	Process id of the target, when known.	None

Source code in src/pidprobe/_errors.py

def __init__(self, timeout_seconds: float, pid: int | None = None) -> None:
    """Build the error.

    Args:
        timeout_seconds: Budget that elapsed without an answer.
        pid: Process id of the target, when known.
    """
    super().__init__(
        _TIMEOUT_MESSAGE.format(
            seconds=timeout_seconds,
            pid="<pid>" if pid is None else pid,
        ),
    )
    self.timeout_seconds = timeout_seconds
    self.pid = pid

TargetError

Bases: ProbeError

Raised when the injected code failed as a whole inside the target.

A single failing collector is reported per section instead, so for a snapshot this error means the injected script itself could not produce a payload -- for example because the collected data would not serialize. An evaluated expression has no such per-section isolation: whatever it raises inside the target arrives here.

Source code in src/pidprobe/_errors.py

class TargetError(ProbeError):
    """Raised when the injected code failed as a whole inside the target.

    A single failing collector is reported per section instead, so for a
    snapshot this error means the injected script itself could not produce a
    payload -- for example because the collected data would not serialize.
    An evaluated expression has no such per-section isolation: whatever it
    raises inside the target arrives here.
    """

    def __init__(self, pid: int, error: ErrorInfo, *, action: str = "snapshot") -> None:
        """Build the error.

        Args:
            pid: Process id the injected code ran in.
            error: Failure details reported by the target.
            action: What the injected code was asked to do, named first in
                the message so ``snap`` and ``eval`` failures read correctly.
        """
        super().__init__(
            f"{action} failed inside pid {pid}: {error['type']}: {error['message']}",
        )
        self.pid = pid
        self.error = error
        self.action = action

__init__(pid, error, *, action='snapshot')

Build the error.

Parameters:

Name	Type	Description	Default
pid	int	Process id the injected code ran in.	required
error	ErrorInfo	Failure details reported by the target.	required
action	str	What the injected code was asked to do, named first in the message so snap and eval failures read correctly.	'snapshot'

Source code in src/pidprobe/_errors.py

def __init__(self, pid: int, error: ErrorInfo, *, action: str = "snapshot") -> None:
    """Build the error.

    Args:
        pid: Process id the injected code ran in.
        error: Failure details reported by the target.
        action: What the injected code was asked to do, named first in
            the message so ``snap`` and ``eval`` failures read correctly.
    """
    super().__init__(
        f"{action} failed inside pid {pid}: {error['type']}: {error['message']}",
    )
    self.pid = pid
    self.error = error
    self.action = action

add(a, b)

Return the sum of two integers.

Parameters:

Name	Type	Description	Default
a	int	First operand.	required
b	int	Second operand.	required

Returns:

Type	Description
int	The sum of a and b.

Source code in src/pidprobe/core.py

def add(a: int, b: int) -> int:
    """Return the sum of two integers.

    Args:
        a: First operand.
        b: Second operand.

    Returns:
        The sum of *a* and *b*.
    """
    return a + b

available_collectors(*, group=COLLECTOR_ENTRY_POINT_GROUP)

Return the collectors a snapshot runs when it is given none.

Parameters:

Name	Type	Description	Default
group	str	Entry point group to read; defaults to :data:COLLECTOR_ENTRY_POINT_GROUP.	COLLECTOR_ENTRY_POINT_GROUP

Returns:

Type	Description
Collector	data:~pidprobe.collectors.BUILTIN_COLLECTORS in their documented
...	order, followed by the discovered plugins. A plugin cannot take over a
tuple[Collector, ...]	built-in section: one that claims a built-in name is logged and
tuple[Collector, ...]	skipped, so stacks always means what this package documents.

Source code in src/pidprobe/registry.py

def available_collectors(
    *,
    group: str = COLLECTOR_ENTRY_POINT_GROUP,
) -> tuple[Collector, ...]:
    """Return the collectors a snapshot runs when it is given none.

    Args:
        group: Entry point group to read; defaults to
            :data:`COLLECTOR_ENTRY_POINT_GROUP`.

    Returns:
        :data:`~pidprobe.collectors.BUILTIN_COLLECTORS` in their documented
        order, followed by the discovered plugins. A plugin cannot take over a
        built-in section: one that claims a built-in name is logged and
        skipped, so ``stacks`` always means what this package documents.
    """
    builtin = tuple(BUILTIN_COLLECTORS)
    taken = {collector.name for collector in builtin}
    return builtin + tuple(_accepted(discover_collectors(group=group), taken))

diagnose(pid=None)

Run the attach preflight checks and report what they found.

Parameters:

Name	Type	Description	Default
pid	int | None	Target to examine. Omit it to run only the checks that describe the prober's own environment, which is the useful thing to do before there is a process to point at.	None

Returns:

Name	Type	Description
The	Diagnosis	class:~pidprobe._diagnosis.Diagnosis; it is is_attachable
	Diagnosis	when nothing found would stop an injection.

Source code in src/pidprobe/_doctor.py

def diagnose(pid: int | None = None) -> Diagnosis:
    """Run the attach preflight checks and report what they found.

    Args:
        pid: Target to examine. Omit it to run only the checks that describe
            the prober's own environment, which is the useful thing to do
            before there is a process to point at.

    Returns:
        The :class:`~pidprobe._diagnosis.Diagnosis`; it is ``is_attachable``
        when nothing found would stop an injection.
    """
    checks = [
        _check_prober_remote_debug(),
        _check_return_channel(),
        _check_collector_plugins(),
        check_ptrace_scope(),
        check_task_for_pid(),
    ]
    if pid is not None:
        checks += [
            check_target_process(pid),
            check_target_owner(pid),
            check_pid_namespace(pid),
            *check_target_python(pid),
            check_target_remote_debug(pid),
        ]
    return Diagnosis(pid=pid, checks=tuple(checks))

diff_snapshots(before, after)

Report what changed between two snapshots of the same process.

Parameters:

Name	Type	Description	Default
before	Mapping[str, Any]	The earlier snapshot.	required
after	Mapping[str, Any]	A later snapshot of the same process.	required

Returns:

Type	Description
SnapshotDelta	The delta document: meta plus an objects, gc and fds
SnapshotDelta	section. A section is None when either snapshot lacks it, which is
SnapshotDelta	what a collector that failed in the target looks like.

Raises:

Type	Description
ValueError	If the snapshots describe different processes, which would make every number in the delta meaningless.

Source code in src/pidprobe/_diff.py

def diff_snapshots(
    before: Mapping[str, Any],
    after: Mapping[str, Any],
) -> SnapshotDelta:
    """Report what changed between two snapshots of the same process.

    Args:
        before: The earlier snapshot.
        after: A later snapshot of the same process.

    Returns:
        The delta document: ``meta`` plus an ``objects``, ``gc`` and ``fds``
        section. A section is ``None`` when either snapshot lacks it, which is
        what a collector that failed in the target looks like.

    Raises:
        ValueError: If the snapshots describe different processes, which would
            make every number in the delta meaningless.
    """
    before_meta = _meta_of(before)
    after_meta = _meta_of(after)
    pid = after_meta.get("pid")
    if before_meta.get("pid") != pid:
        message = (
            "cannot diff snapshots of different processes: "
            f"{before_meta.get('pid')!r} and {pid!r}"
        )
        raise ValueError(message)
    captured_from = before_meta.get("captured_at")
    captured_to = after_meta.get("captured_at")
    return {
        "schema_version": SCHEMA_VERSION,
        "meta": {
            "pid": pid,
            "from": captured_from,
            "to": captured_to,
            "interval_ms": _interval_ms(captured_from, captured_to),
        },
        "objects": _objects_delta(before.get("objects"), after.get("objects")),
        "gc": _gc_delta(before.get("gc"), after.get("gc")),
        "fds": _fds_delta(before.get("fds"), after.get("fds")),
    }

discover_collectors(*, group=COLLECTOR_ENTRY_POINT_GROUP)

Load every collector published in an entry point group.

Parameters:

Name	Type	Description	Default
group	str	Entry point group to read; defaults to :data:COLLECTOR_ENTRY_POINT_GROUP.	COLLECTOR_ENTRY_POINT_GROUP

Returns:

Type	Description
Collector	The collectors that loaded cleanly, ordered by entry point name so the
...	result does not depend on installation order. Anything that failed to
tuple[Collector, ...]	load or repeated a name already claimed is logged and left out.

Source code in src/pidprobe/registry.py

def discover_collectors(
    *,
    group: str = COLLECTOR_ENTRY_POINT_GROUP,
) -> tuple[Collector, ...]:
    """Load every collector published in an entry point group.

    Args:
        group: Entry point group to read; defaults to
            :data:`COLLECTOR_ENTRY_POINT_GROUP`.

    Returns:
        The collectors that loaded cleanly, ordered by entry point name so the
        result does not depend on installation order. Anything that failed to
        load or repeated a name already claimed is logged and left out.
    """
    published = sorted(entry_points(group=group), key=lambda point: point.name)
    loaded = [
        collector for point in published if (collector := _load(point)) is not None
    ]
    return tuple(_accepted(loaded, set()))

evaluate_in_target(pid, expression, *, timeout_seconds=DEFAULT_TIMEOUT_SECONDS, is_masked=True, allow_socket=True)

Evaluate one expression inside a running CPython 3.14+ process.

Parameters:

Name	Type	Description	Default
pid	int	Target process id.	required
expression	str	Python expression evaluated against a copy of the target's __main__ namespace. Statements are rejected by the target's own compiler, so an evaluation cannot rebind a name the target holds.	required
timeout_seconds	float	Hard budget covering injection and read-back.	DEFAULT_TIMEOUT_SECONDS
is_masked	bool	Set to False to receive credential-like results as they are; masking is otherwise applied inside the target.	True
allow_socket	bool	Set to False to force the tempfile return channel.	True

Returns:

Type	Description
Evaluation	The evaluation document, ready to be serialized to JSON.

Raises:

Type	Description
AttachError	If the target refuses the injection.
ProbeTimeoutError	If the target never reaches a safe evaluation point within the budget.
ChannelError	If the answer does not match the envelope contract.
TargetError	If compiling or evaluating the expression raised inside the target.

Source code in src/pidprobe/_eval.py

def evaluate_in_target(
    pid: int,
    expression: str,
    *,
    timeout_seconds: float = DEFAULT_TIMEOUT_SECONDS,
    is_masked: bool = True,
    allow_socket: bool = True,
) -> Evaluation:
    """Evaluate one expression inside a running CPython 3.14+ process.

    Args:
        pid: Target process id.
        expression: Python expression evaluated against a copy of the
            target's ``__main__`` namespace. Statements are rejected by the
            target's own compiler, so an evaluation cannot rebind a name the
            target holds.
        timeout_seconds: Hard budget covering injection and read-back.
        is_masked: Set to ``False`` to receive credential-like results as
            they are; masking is otherwise applied inside the target.
        allow_socket: Set to ``False`` to force the tempfile return channel.

    Returns:
        The evaluation document, ready to be serialized to JSON.

    Raises:
        AttachError: If the target refuses the injection.
        ProbeTimeoutError: If the target never reaches a safe evaluation
            point within the budget.
        ChannelError: If the answer does not match the envelope contract.
        TargetError: If compiling or evaluating the expression raised inside
            the target.
    """
    envelope = execute_in_target(
        pid,
        build_eval_source(expression, is_masked=is_masked),
        timeout_seconds=timeout_seconds,
        allow_socket=allow_socket,
    )
    payload = payload_of(pid, envelope, action=_ACTION)
    return Evaluation(
        pid=pid,
        expression=expression,
        # Only the two rendered strings come from the target; everything else
        # is what the prober asked for and needs no trusting back.
        type=str(payload.get("type", "")),
        result=str(payload.get("result", "")),
        masking_enabled=is_masked,
    )

iter_snapshot_deltas(pid, *, interval_seconds, count=None, timeout_seconds=DEFAULT_TIMEOUT_SECONDS)

Sample a running process repeatedly, yielding each delta as it is ready.

Parameters:

Name	Type	Description	Default
pid	int	Target process id.	required
interval_seconds	float	Spacing between the starts of two samples. The time a sample itself costs is subtracted from the wait, so a slow probe does not push the whole series later and later.	required
count	int | None	How many snapshots to take. N snapshots produce N - 1 deltas, so count=1 yields nothing at all and count=0 does not touch the target. None keeps sampling until the caller stops consuming the iterator.	None
timeout_seconds	float	Hard budget for each individual probe.	DEFAULT_TIMEOUT_SECONDS

Yields:

Type	Description
SnapshotDelta	One delta per pair of consecutive snapshots, as soon as the later
SnapshotDelta	snapshot of that pair has been taken.

Raises:

Type	Description
ProbeError	Whatever :func:~pidprobe.take_snapshot raises for a sample. A failed sample ends the series instead of being skipped: a target that stopped answering will not start again on its own, and silently widening the interval would misreport every delta that followed.

Source code in src/pidprobe/_diff.py

def iter_snapshot_deltas(
    pid: int,
    *,
    interval_seconds: float,
    count: int | None = None,
    timeout_seconds: float = DEFAULT_TIMEOUT_SECONDS,
) -> Iterator[SnapshotDelta]:
    """Sample a running process repeatedly, yielding each delta as it is ready.

    Args:
        pid: Target process id.
        interval_seconds: Spacing between the *starts* of two samples. The
            time a sample itself costs is subtracted from the wait, so a slow
            probe does not push the whole series later and later.
        count: How many snapshots to take. ``N`` snapshots produce ``N - 1``
            deltas, so ``count=1`` yields nothing at all and ``count=0`` does
            not touch the target. ``None`` keeps sampling until the caller
            stops consuming the iterator.
        timeout_seconds: Hard budget for each individual probe.

    Yields:
        One delta per pair of consecutive snapshots, as soon as the later
        snapshot of that pair has been taken.

    Raises:
        ProbeError: Whatever :func:`~pidprobe.take_snapshot` raises for a
            sample. A failed sample ends the series instead of being skipped:
            a target that stopped answering will not start again on its own,
            and silently widening the interval would misreport every delta
            that followed.
    """
    started = monotonic()
    previous: Mapping[str, Any] | None = None
    taken = 0
    while count is None or taken < count:
        # Scheduled from one origin rather than from the previous sample, so
        # the probes themselves cannot make the series drift.
        _wait_until(started + taken * interval_seconds)
        current = _sample(pid, timeout_seconds)
        taken += 1
        if previous is not None:
            yield diff_snapshots(previous, current)
        previous = current

snapshot_schema()

Return the JSON Schema for snapshot documents.

Returns:

Type	Description
dict[str, Any]	A freshly parsed schema, so callers may modify it (to bundle it into
dict[str, Any]	a larger schema, for instance) without affecting anyone else.

Source code in src/pidprobe/_schema.py

def snapshot_schema() -> dict[str, Any]:
    """Return the JSON Schema for snapshot documents.

    Returns:
        A freshly parsed schema, so callers may modify it (to bundle it into
        a larger schema, for instance) without affecting anyone else.
    """
    text = resources.files(__package__).joinpath(SCHEMA_FILENAME).read_text("utf-8")
    # Any: a JSON Schema is an arbitrarily nested JSON document.
    schema: dict[str, Any] = json.loads(text)
    return schema

take_snapshot(pid, *, timeout_seconds=DEFAULT_TIMEOUT_SECONDS, collectors=None, allow_socket=True)

Collect one snapshot from a running CPython 3.14+ process.

Parameters:

Name	Type	Description	Default
pid	int	Target process id.	required
timeout_seconds	float	Hard budget covering injection and read-back.	DEFAULT_TIMEOUT_SECONDS
collectors	Iterable[Collector] | None	Collectors to run; defaults to :func:~pidprobe.registry.available_collectors, the built-in collectors plus every plugin published in the pidprobe.collectors entry point group.	None
allow_socket	bool	Set to False to force the tempfile return channel.	True

Returns:

Type	Description
Snapshot	The snapshot document, ready to be serialized to JSON.

Raises:

Type	Description
AttachError	If the target refuses the injection.
ProbeTimeoutError	If the target never reaches a safe evaluation point within the budget.
ChannelError	If the answer does not match the envelope contract.
TargetError	If the injected code failed as a whole inside the target.

Source code in src/pidprobe/_snapshot.py

def take_snapshot(
    pid: int,
    *,
    timeout_seconds: float = DEFAULT_TIMEOUT_SECONDS,
    collectors: Iterable[Collector] | None = None,
    allow_socket: bool = True,
) -> Snapshot:
    """Collect one snapshot from a running CPython 3.14+ process.

    Args:
        pid: Target process id.
        timeout_seconds: Hard budget covering injection and read-back.
        collectors: Collectors to run; defaults to
            :func:`~pidprobe.registry.available_collectors`, the built-in
            collectors plus every plugin published in the
            ``pidprobe.collectors`` entry point group.
        allow_socket: Set to ``False`` to force the tempfile return channel.

    Returns:
        The snapshot document, ready to be serialized to JSON.

    Raises:
        AttachError: If the target refuses the injection.
        ProbeTimeoutError: If the target never reaches a safe evaluation
            point within the budget.
        ChannelError: If the answer does not match the envelope contract.
        TargetError: If the injected code failed as a whole inside the target.
    """
    chosen = available_collectors() if collectors is None else tuple(collectors)
    source = compose_collector_source(chosen)
    started = time.perf_counter()
    envelope = execute_in_target(
        pid,
        source,
        timeout_seconds=timeout_seconds,
        allow_socket=allow_socket,
    )
    elapsed_ms = (time.perf_counter() - started) * _MILLISECONDS
    return _document(pid, payload_of(pid, envelope), elapsed_ms)