pshnb IPython magic

Provides psh persistent bash magics in Jupyter and IPython 
__file__ = './00_core.ipynb'

Foundations

env = dict(os.environ, TERM='dumb', PS1='$', PS2='$')
eshell = os.environ['SHELL']
sh = pexpect.spawn(eshell, encoding='utf-8', env=env)

This cell creates the initial shell process using pexpect.spawn() with a basic environment configuration. The key setup includes:

  • Environment variables: Sets TERM='dumb' to prevent terminal formatting issues, and standardizes prompts with PS1='$' and PS2='$'
  • Shell spawning: Creates a persistent shell process using the user’s default shell ($SHELL) with UTF-8 encoding
  • Process persistence: The sh object maintains a continuous connection to the shell, allowing commands to be sent and responses captured across multiple interactions

PS1 is the primary shell prompt (what you see when the shell is ready for a command), and PS2 is the secondary prompt (shown when a command spans multiple lines or is incomplete). pexpect is a Python library for controlling interactive command-line programs by automating the sending of inputs and reading of outputs. It’s particularly useful for automating tasks that normally require human interaction. spawn is pexpect’s main function that starts a child process (like a shell) and returns a pexpect object that can communicate with it. Unlike subprocess, spawn maintains an interactive session - you can send commands, wait for responses, and continue the conversation with the same process instance.

This establishes the foundation for persistent shell state - variables, directory changes, and other shell modifications will persist between command executions.

env = dict(os.environ, TERM='dumb', PS1='', PS2='')
sh = pexpect.spawn(eshell, encoding='utf-8', env=env)
sh.sendline('stty -echo')
sh.readline()
echo = os.urandom(8).hex()
echo_re = re.compile(fr'^{echo}\s*$', flags=re.MULTILINE)
sh.sendline(f'export PS1=""')
sh.sendline('set +o vi +o emacs')
sh.sendline('echo '+echo)
sh.expect(echo_re, timeout=2)
0

This cell implements a more sophisticated shell setup that addresses output formatting and command echo issues:

  • Echo suppression: Uses stty -echo to prevent command echoing, which is crucial for clean output capture
  • Empty prompts: Sets PS1="" and PS2="" to eliminate prompt strings that would interfere with output parsing
  • Line editing disabled: Runs set +o vi +o emacs to disable interactive line editing features that can cause issues in programmatic shell interaction
  • Synchronization mechanism: Establishes a unique echo token using os.urandom(8).hex() and corresponding regex pattern for reliable command completion detection

The echo token system is particularly important - it provides a reliable way to detect when a command has finished executing by sending a unique marker and waiting for it to appear in the output.

sh.sendline('ls | head -3')
sh.sendline('echo '+echo)
sh.expect(echo_re, timeout=2)
print(sh.before)
00_core.ipynb
AGENTS.md
CHANGELOG.bak

This cell demonstrates the practical application of the configured shell for command execution and output capture:

  • Command execution: Sends ls | head -3 to list the first three files in the current directory
  • Completion detection: Uses the echo token mechanism (echo [random_hex]) followed by sh.expect(echo_re) to reliably detect when the command has finished
  • Clean output extraction: Captures the command output via sh.before, which contains everything received before the expected echo token

This pattern shows how the kernel can execute arbitrary bash commands while maintaining clean separation between command output and shell control mechanisms. The timeout parameter (2 seconds) provides protection against hanging commands, which is essential for a responsive interactive environment.

source

ShellInterpreter


def ShellInterpreter(
    debug:bool=False, timeout:int=4, shell_path:str='bash', sudo:bool=False, dumb:bool=True, ssh:NoneType=None
):

Initialize self. See help(type(self)) for accurate signature.

sh = ShellInterpreter()
print(sh('ls | head -3'))
00_core.ipynb
AGENTS.md
CHANGELOG.bak

NB: requires passwordless sudo to use this:

sh = ShellInterpreter(sudo=True)

sh('cd')
print(sh('pwd'))
sh('cd ..')
print(sh('pwd'))
print(sh('whoami'))
/var/root
/var
root

source

ShellInterpreter.sync_cwd


def sync_cwd(
    
):

Sync session’s cwd to match pshnb’s cwd

source

shell_replace


def shell_replace(
    s, shell:NoneType=None
):

Replace @{var} refs in s with their variable values, if they exist

b = 1

a = '''asdf
$@{b} @{aa}
fdsa'''

print(shell_replace(a))
asdf
$1 @{aa}
fdsa

source

PshMagic


def PshMagic(
    shell, sudo:bool=False, timeout:int=2, expand:bool=True, o:NoneType=None, ssh:NoneType=None
):

Initialize self. See help(type(self)) for accurate signature.

source

PshMagic.bash


def bash(
    line, cell:NoneType=None
):

::

%bash [-h] [-r [RESET]] [-o] [-x] [-X] [-s [SUDO]] [-S] [-c] [-e ESCAPE [ESCAPE …]] [-t TIMEOUT] [command …]

Run line or cell in persistent shell

positional arguments: command The command to run

options: -h, –help Show this help -r <[RESET]>, –reset <[RESET]> Reset the shell interpreter (optionally run ssh) -o, –obj Return the ShellInterpreter -x, –expand Enable variable expansion -X, –no-expand Disable variable expansion -s <[SUDO]>, –sudo <[SUDO]> Enable sudo -S, –no-sudo Disable sudo -c, –cwd Sync session’s cwd to match pshnb’s cwd -e <ESCAPE [ESCAPE …]>, –escape <ESCAPE [ESCAPE …]> SSH escape code and optional command -t TIMEOUT, –timeout TIMEOUT Set timeout in seconds

source

create_magic


def create_magic(
    shell:NoneType=None
):

# Only required if you don't load the extension
create_magic()
00_core.ipynb
AGENTS.md
CHANGELOG.bak
::

  %bash [-h] [-r [RESET]] [-o] [-x] [-X] [-s [SUDO]] [-S] [-c]
            [-e ESCAPE [ESCAPE ...]] [-t TIMEOUT]
            [command ...]

Run line or cell in persistent shell

positional arguments:
  command               The command to run

options:
  -h, --help            Show this help
  -r <[RESET]>, --reset <[RESET]>
                        Reset the shell interpreter (optionally run ssh)
  -o, --obj             Return the `ShellInterpreter`
  -x, --expand          Enable variable expansion
  -X, --no-expand       Disable variable expansion
  -s <[SUDO]>, --sudo <[SUDO]>
                        Enable sudo
  -S, --no-sudo         Disable sudo
  -c, --cwd             Sync session's cwd to match pshnb's cwd
  -e <ESCAPE [ESCAPE ...]>, --escape <ESCAPE [ESCAPE ...]>
                        SSH escape code and optional command
  -t TIMEOUT, --timeout TIMEOUT
                        Set timeout in seconds
/Users/jhoward/aai-ws/pshnb
/Users/jhoward/aai-ws
!pwd
/Users/jhoward/aai-ws/pshnb

Heredoc example:

cat > tmp.foo << EOF
hi
there
EOF
hi
there
ContextKit
MonsterUI
README.md
os.getcwd()
'/Users/jhoward/aai-ws'
n = 2
2
ContextKit
MonsterUI
echo starting
(sleep 1; echo finished) &
starting
[1] 51853
time.sleep(1.1)
finished

[1]+  Done                       ( sleep 1; echo finished )
::

  %bash [-h] [-r [RESET]] [-o] [-x] [-X] [-s [SUDO]] [-S] [-c]
            [-e ESCAPE [ESCAPE ...]] [-t TIMEOUT]
            [command ...]

Run line or cell in persistent shell

positional arguments:
  command               The command to run

options:
  -h, --help            Show this help
  -r <[RESET]>, --reset <[RESET]>
                        Reset the shell interpreter (optionally run ssh)
  -o, --obj             Return the `ShellInterpreter`
  -x, --expand          Enable variable expansion
  -X, --no-expand       Disable variable expansion
  -s <[SUDO]>, --sudo <[SUDO]>
                        Enable sudo
  -S, --no-sudo         Disable sudo
  -c, --cwd             Sync session's cwd to match pshnb's cwd
  -e <ESCAPE [ESCAPE ...]>, --escape <ESCAPE [ESCAPE ...]>
                        SSH escape code and optional command
  -t TIMEOUT, --timeout TIMEOUT
                        Set timeout in seconds

Reset the interpreter:

/Users/jhoward/aai-ws
/opt/homebrew/bin/bash

sudo:

root

no sudo:

jhoward

timeout:

try: get_ipython().run_line_magic('bash', 'sleep 2')
except TIMEOUT: print("timed out")
timed out

ssh

sh = ShellInterpreter(ssh='ev2')
print(sh('ls | head -3'))
go
Maildir
quarto-1.7.32-linux-amd64.deb
passw = getpass()
sh = ShellInterpreter(ssh='ev2', sudo=passw)
print(sh('pwd'))
print(sh('whoami'))
/root
root
jphmacbook.local
ev2.answer.ai
jhoward
ev2.answer.ai
root

source

load_ipython_extension


def load_ipython_extension(
    ipython
):

Required function for creating magic

source

create_ipython_config


def create_ipython_config(
    
):

Called by pshnb_install to install magic