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
hist_backfill_fixes
parent
4bfdd388bb
commit
192fe0dc73
|
|
@ -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 ===')
|
||||||
Loading…
Reference in New Issue