From f3530b2f6b6efda45156759b2b7685261a97a26a Mon Sep 17 00:00:00 2001 From: goodboy Date: Sun, 18 Jan 2026 18:18:34 -0500 Subject: [PATCH] Add `pexpect`-based `pdbp`-REPL offline helper Add a new `snippets/claude_debug_helper.py` to provide a programmatic interface to `tractor.pause()` debugger sessions for incremental data inspection matching the interactive UX but able to be run by `claude` "offline" since it can't seem to feed stdin (so it claims) to the `pdb` instance due to lack of ability to allocate a tty internally. The script-wrapper is based on `tractor`'s `tests/devx/` suite's use of `pexpect` patterns for driving `pdbp` prompts and thus enables automated-offline execution of REPL-inspection commands **without** using incremental-realtime output capture (like a human would use it). Features: - `run_pdb_commands()`: batch command execution - `InteractivePdbSession`: context manager for step-by-step REPL interaction - `expect()` wrapper: timeout handling with buffer display - Proper stdin/stdout handling via `pexpect.spawn()` Example usage: ```python from debug_helper import InteractivePdbSession with InteractivePdbSession( cmd='piker store ldshm zecusdt.usdtm.perp.binance' ) as session: session.run('deduped.shape') session.run('step_gaps.shape') ``` (this patch was generated in some part by [`claude-code`][claude-code-gh]) [claude-code-gh]: https://github.com/anthropics/claude-code --- snippets/claude_debug_helper.py | 256 ++++++++++++++++++++++++++++++++ 1 file changed, 256 insertions(+) create mode 100755 snippets/claude_debug_helper.py diff --git a/snippets/claude_debug_helper.py b/snippets/claude_debug_helper.py new file mode 100755 index 00000000..97467d8a --- /dev/null +++ b/snippets/claude_debug_helper.py @@ -0,0 +1,256 @@ +#!/usr/bin/env python +''' +Programmatic debugging helper for `pdbp` REPL human-like +interaction but built to allow `claude` to interact with +crashes and `tractor.pause()` breakpoints along side a human dev. + +Originally written by `clauded` during a backfiller inspection +session with @goodboy trying to resolve duplicate/gappy ohlcv ts +issues discovered while testing the new `nativedb` tsdb. + +Allows `claude` to run `pdb` commands and capture output in an "offline" +manner but generating similar output as if it was iteracting with +the debug REPL. + +The use of `pexpect` is heavily based on tractor's REPL UX test +suite(s), namely various `tests/devx/test_debugger.py` patterns. + +''' +import sys +import os +import time + +import pexpect +from pexpect.exceptions import ( + TIMEOUT, + EOF, +) + + +PROMPT: str = r'\(Pdb\+\)' + + +def expect( + child: pexpect.spawn, + patt: str, + **kwargs, +) -> None: + ''' + Expect wrapper that prints last console data before failing. + + ''' + try: + child.expect( + patt, + **kwargs, + ) + except TIMEOUT: + before: str = ( + str(child.before.decode()) + if isinstance(child.before, bytes) + else str(child.before) + ) + print( + f'TIMEOUT waiting for pattern: {patt}\n' + f'Last seen output:\n{before}' + ) + raise + + +def run_pdb_commands( + commands: list[str], + initial_cmd: str = 'piker store ldshm xmrusdt.usdtm.perp.binance', + timeout: int = 30, + print_output: bool = True, +) -> dict[str, str]: + ''' + Spawn piker process, wait for pdb prompt, execute commands. + + Returns dict mapping command -> output. + + ''' + results: dict[str, str] = {} + + # Disable colored output for easier parsing + os.environ['PYTHON_COLORS'] = '0' + + # Spawn the process + if print_output: + print(f'Spawning: {initial_cmd}') + + child: pexpect.spawn = pexpect.spawn( + initial_cmd, + timeout=timeout, + encoding='utf-8', + echo=False, + ) + + # Wait for pdb prompt + try: + expect(child, PROMPT, timeout=timeout) + if print_output: + print('Reached pdb prompt!') + + # Execute each command + for cmd in commands: + if print_output: + print(f'\n>>> {cmd}') + + child.sendline(cmd) + time.sleep(0.1) + + # Wait for next prompt + expect(child, PROMPT, timeout=timeout) + + # Capture output (everything before the prompt) + output: str = ( + str(child.before.decode()) + if isinstance(child.before, bytes) + else str(child.before) + ) + results[cmd] = output + + if print_output: + print(output) + + # Quit debugger gracefully + child.sendline('quit') + try: + child.expect(EOF, timeout=5) + except (TIMEOUT, EOF): + pass + + except TIMEOUT as e: + print(f'Timeout: {e}') + if child.before: + before: str = ( + str(child.before.decode()) + if isinstance(child.before, bytes) + else str(child.before) + ) + print(f'Buffer:\n{before}') + results['_error'] = str(e) + + finally: + if child.isalive(): + child.close(force=True) + + return results + + +class InteractivePdbSession: + ''' + Interactive pdb session manager for incremental debugging. + + ''' + def __init__( + self, + cmd: str = 'piker store ldshm xmrusdt.usdtm.perp.binance', + timeout: int = 30, + ): + self.cmd: str = cmd + self.timeout: int = timeout + self.child: pexpect.spawn|None = None + self.history: list[tuple[str, str]] = [] + + def start(self) -> None: + ''' + Start the piker process and wait for first prompt. + + ''' + os.environ['PYTHON_COLORS'] = '0' + + print(f'Starting: {self.cmd}') + self.child = pexpect.spawn( + self.cmd, + timeout=self.timeout, + encoding='utf-8', + echo=False, + ) + + # Wait for initial prompt + expect(self.child, PROMPT, timeout=self.timeout) + print('Ready at pdb prompt!') + + def run( + self, + cmd: str, + print_output: bool = True, + ) -> str: + ''' + Execute a single pdb command and return output. + + ''' + if not self.child or not self.child.isalive(): + raise RuntimeError('Session not started or dead') + + if print_output: + print(f'\n>>> {cmd}') + + self.child.sendline(cmd) + time.sleep(0.1) + + # Wait for next prompt + expect(self.child, PROMPT, timeout=self.timeout) + + output: str = ( + str(self.child.before.decode()) + if isinstance(self.child.before, bytes) + else str(self.child.before) + ) + self.history.append((cmd, output)) + + if print_output: + print(output) + + return output + + def quit(self) -> None: + ''' + Exit the debugger and cleanup. + + ''' + if self.child and self.child.isalive(): + self.child.sendline('quit') + try: + self.child.expect(EOF, timeout=5) + except (TIMEOUT, EOF): + pass + self.child.close(force=True) + + def __enter__(self): + self.start() + return self + + def __exit__(self, *args): + self.quit() + + +if __name__ == '__main__': + # Example inspection commands + inspect_cmds: list[str] = [ + 'locals().keys()', + 'type(deduped)', + 'deduped.shape', + ( + 'step_gaps.shape ' + 'if "step_gaps" in locals() ' + 'else "N/A"' + ), + ( + 'venue_gaps.shape ' + 'if "venue_gaps" in locals() ' + 'else "N/A"' + ), + ] + + # Allow commands from CLI args + if len(sys.argv) > 1: + inspect_cmds = sys.argv[1:] + + # Interactive session example + with InteractivePdbSession() as session: + for cmd in inspect_cmds: + session.run(cmd) + + print('\n=== Session Complete ===')