pyrepl

A Python prompt you own, and an agent in the layer above it.

Outputs

A kernel reports what a cell produced as a stream of messages: text printed to stdout, a result value, a rich display, an error. A notebook stores the same things as a list of dicts in nbformat, the notebook file format. ExecOutcome holds the result of one request in that shape: whether it succeeded, its outputs, its execution count and its error, if any. The terminal, the session log that log_cell writes, and the tests all read the same list.

output_text flattens such a list to plain text, which is what tests, logs and a plain terminal need. The cell below shows each kind of output.

output_text

def output_text(
    outputs
):

Flatten notebook outputs for tests, logs and plain terminal fallbacks.

ExecOutcome

def ExecOutcome(
    ok:bool=True, outputs:list=<factory>, execution_count:int | None=None, error:str | None=None
)->None:

One kernel request in nbformat’s output shape.

outs = [{'output_type': 'stream', 'text': 'hello\n'},
        {'output_type': 'execute_result', 'data': {'text/plain': '42'}},
        {'output_type': 'error', 'ename': 'ValueError', 'evalue': 'bad', 'traceback': []}]
test_eq(output_text(outs), 'hello\n42\nValueError: bad')
test_eq(output_text([{'output_type': 'stream', 'text': ['a', 'b']}]), 'ab')
test_eq(output_text([{'output_type': 'error', 'ename': 'E', 'evalue': 'v',
                      'traceback': ['line one', 'line two']}]), 'line one\nline two')
test_eq(output_text([]), '')

The kernel

Kernel starts and drives a private ipykernel, the standard Python kernel for Jupyter. start launches it in cwd, waits until it is ready, and runs a bootstrap cell that starts Dhrishti inside it. Dhrishti serves the kernel’s namespace from a background thread. The bootstrap prints the server’s port behind a marker, and start reads the marker to set base, the server’s address. The session is registered under the name <project>-pyrepl-<pid>, so another terminal can find it.

execute runs one cell and returns an ExecOutcome. It passes each output to on_output as it arrives, so the terminal can show it at once. complete asks the kernel for tab completions at a cursor position, interrupt stops the running cell, and shutdown stops the kernel, forcing it when it does not stop within timeout. A Kernel that is not running answers execute with an error and complete with no matches, as the second cell shows.

Kernel

def Kernel(
    cwd:str='.'
):

A private ipykernel with Dhrishti serving its live namespace.

kernel = await Kernel('.').start()
r = await kernel.execute('a = 6 * 7\na')
test_eq((r.ok, output_text(r.outputs)), (True, '42'))
assert kernel.base.startswith('http://127.0.0.1:')

dead = Kernel('.')
r = await dead.execute('1 + 1')
test_eq((r.ok, r.error), (False, 'kernel is not running'))

r = await kernel.execute('1/0')
test_eq(r.ok, False)
assert 'ZeroDivisionError' in r.error and 'ZeroDivisionError' in output_text(r.outputs)

r = await kernel.execute("for i in range(3): print(i)")
test_eq(len([o for o in r.outputs if o['output_type'] == 'stream']), 1)
test_eq(output_text(r.outputs), '0\n1\n2')

await kernel.execute('import json')
matches, start = await kernel.complete('json.du', 7)
test_eq((sorted(matches), start), (['dump', 'dumps'], 5))

test_eq(await dead.complete('json.du', 7), ([], 7))

The host

The agent needs a host whose Python tools reach this kernel. Its file, search and web tools stay the ordinary ones. Dhrishti is the piece that changes: it sends the agent’s run_python, inspect_python and variable listing over HTTP to the agent half of the Dhrishti API. Every call it makes goes to /agent/api/..., which needs no token, and none of these paths reads the owner token that opens /api/exec, /api/set and /api/promote. So no tool call can change the owner’s namespace. The only code here that reads the owner token is promote, which runs when the person types /promote.

log_cell keeps the whole session in one notebook, the session log that Dhrishti keeps. The person’s cells go in as code cells with their real outputs, and each agent turn goes in as markdown cells for the prompt and the answer, in the order they happened.

DhrishtiHost

def DhrishtiHost(
    roots, # the folders the agent is confined to; each must exist
    base, *, ns:NoneType=None, # live namespace; fresh dict when None
    approvals:NoneType=None, # write approval handler
    note:NoneType=None, # status callback
    web:bool=True, # use fossick when installed
    index:bool=True, # start a Kosha sync for every open root
    graph:bool=False, # build Kosha's call graph during sync
    rerank:bool=True, # rerank Kosha hits with flashrank
    rerank_model:NoneType=None, # flashrank model; None uses its default
    memory:NoneType=None, # memory, ask and watch backend
    apis:NoneType=None, # API specification backend
    kernel:NoneType=None, # live kernel; None runs in process
    read_outside:bool=False, # let read-only tools name any path on this machine
    deny:tuple=('*/.ssh/*', '*/.aws/*', '*/.gnupg/*', '*/.config/gcloud/*', '*/.netrc', '*/.git-credentials', '*/.codex/auth.json', '*/.claude/.credentials.json', '*/.env', '*/.env.*', '*/id_rsa*', '*/id_ed25519*', '*.pem', '*.key', '*.p12'), # what `read_outside` still refuses to open
    read_roots:tuple=(), # folders the read tools search and no write may touch
    tmux:NoneType=None, # None detects `$TMUX`; False off; True on; or this host's own pane (target or `Pane`)
    read_guard:NoneType=None, # `(ReadCheck) -> None | str | Refusal`: what the read tools may send
    model_local:NoneType=None, # `() -> (local, model)` for the turn's model
):

A LocalHost that starts on a Dhrishti overlay, for callers that build one directly.

use_kernel

def use_kernel(
    host, base
):

Point host’s Python at a Dhrishti overlay in place, keeping every other tool group.

Dhrishti

def Dhrishti(
    base
):

The agent half of a Dhrishti session, in the shape of a kernel a LocalHost can hold.

Dhrishti is everything a LocalHost needs from a kernel: the calls run, inspect, list_vars, log_cell and describe, and the two facts scopes and kind. run always uses the overlay, whatever scope the model asks for, and inspect accepts only the isolated and overlay scopes. The cells below run code in the overlay and read it back.

from shalya.host import LocalHost

k = Dhrishti(kernel.base)
k.scopes, k.kind, k.run('from_the_kernel = 1')

test_eq(k.scopes, ('isolated', 'overlay'))
test_eq(k.kind, 'ipykernel')
test_eq(k.run('from_the_kernel = 1'), '(ok)')
test_eq(k.inspect('from_the_kernel + 1', 'overlay'), '2')
assert 'from_the_kernel' in k.list_vars()
assert 'only honours' in k.inspect('1+1', scope='nonsense')
test_eq(k.base, kernel.base)

There are two ways to give a host this kernel. use_kernel(host, base) takes the host a session already has and points its Python at the Dhrishti session at base, in place. Every other tool group stays as it was, so a host with a vault keeps its memory tools. The terminal calls it when you switch to Python mode or attach to a session, and so does mk_agent(python=URL). DhrishtiHost(roots, base) builds a new LocalHost that starts on a Dhrishti session, for code that creates its host directly.

The first cell below compares the tool names of a host with memory before and after use_kernel, and nothing is added or lost. After that, a DhrishtiHost shows the overlay at work: the agent’s owner = None leaves the person’s owner as it was. The cells after it check log_cell, a session that has no log, and describe, which keeps falsy values such as 0 and ''.

from shalya.tools import tools_for

class Store:
    def tree(self, document=''): return []
    def search(self, query, limit=8): return []
    def read(self, node_id): return {}
    def topics(self, limit=12): return []
    def forget(self, doc_id): return False
    def remember(self, text, title=None, tags=()): return {'doc_id': 'd1', 'title': 'x'}
    def ask(self, q, ref=None, instruction='', **kw): return {'answer': '', 'cited': []}
    def watch(self, target, **kw): return {'id': 'w1'}
    def watches(self, due_only=False): return []
    def unwatch(self, watch_id): return False
    def poll(self): return {'ran': 0, 'checked': 0, 'results': []}

vaulted = LocalHost(['.'], index=False, web=False, memory=Store())
before = {t.__name__ for t in tools_for(vaulted)}
use_kernel(vaulted, kernel.base)
after = {t.__name__ for t in tools_for(vaulted)}
len(before), sorted(before - after), sorted(after - before)

test_eq(before - after, set())
test_eq(after - before, set())
assert 'memory_read' in after and 'run_python' in after
test_eq(vaulted.scopes, ('isolated', 'overlay'))
test_eq(vaulted.kernel_kind, 'ipykernel')
test_eq(vaulted.run_python('after_the_swap = 2'), '(ok)')
assert 'after_the_swap' in vaulted.list_vars()
test_eq(use_kernel(vaulted, kernel.base), vaulted)

host = DhrishtiHost(['.'], kernel.base, web=False, index=False)
await kernel.execute("owner = {'kept': 1}")

test_eq(host.run_python('mine = len(owner)'), '(ok)')
assert 'mine' in host.list_vars() and 'owner' in host.list_vars()

host.run_python('owner = None')
r = await kernel.execute('owner')
test_eq(output_text(r.outputs), "{'kept': 1}")

assert 'Error' in DhrishtiHost(['.'], 'http://127.0.0.1:1', web=False, index=False).run_python('1')

from fastcore.nbio import read_nb

r = await kernel.execute('logged = 1')
host.log_cell('logged = 1', r.outputs)
host.log_cell('**user**\n\nwhat is logged?', cell_type='markdown')
nb = read_nb(host.agent_log)
test_eq([c.cell_type for c in nb.cells[-2:]], ['code', 'markdown'])
test_eq(nb.cells[-2].outputs, r.outputs)

_real_api = _api
_api = lambda base, path, params=None, timeout=60: {'log': None}
try: nolog = DhrishtiHost(['.'], kernel.base, web=False, index=False)
finally: _api = _real_api
dead_host = DhrishtiHost(['.'], 'http://127.0.0.1:1', web=False, index=False)
test_eq((nolog.agent_log, dead_host.agent_log), (None, None))
test_eq([h.log_cell('x = 1', []) for h in (nolog, dead_host)], [None, None])

_real_api = _api
def _fake_rows_api(base, path, params=None, timeout=60):
    return {'groups': [{'nodes': [
        {'name': 'zero_val', 'type': 'int', 'value': 0},
        {'name': 'blank_val', 'type': 'str', 'value': ''},
        {'name': 'missing_val', 'type': 'NoneType', 'value': None},
    ]}]}
_api = _fake_rows_api
try: described = host.describe()
finally: _api = _real_api
test_eq(described['zero_val'], '0')
test_eq(described['blank_val'], '')
test_eq(described['missing_val'], 'NoneType')

Completion, with the namespace described

Tab completion at the Python prompt draws on two sources. The kernel supplies the candidates. Its completer is IPython’s, with jedi behind it, so it knows imports, attribute chains, dictionary keys and file paths. Dhrishti has no completion endpoint, and its view of the namespace covers only top-level names, so it would complete worse.

Dhrishti supplies the descriptions instead. annotate adds the type and shape Dhrishti reports to each candidate that names a live variable, so the menu shows df -> DataFrame [1200×8] rather than df. Reading descriptions needs no token, so they come from the agent half of the API. They are optional: if the lookup is slow or fails, the menu shows the bare names, never an empty list.

annotate

def annotate(
    matches, described
):

Candidate names with their type and shape where the session knows one.

test_eq(annotate(['df', 'dfs'], {'df': 'DataFrame [3×2]'}), ['df -> DataFrame [3×2]', 'dfs'])
test_eq(annotate(['x'], {}), ['x'])
test_eq(annotate([], {'x': 'int'}), [])
test_eq(annotate(['df.copy'], {'df': 'DataFrame [3×2]'}), ['df.copy'])

Attaching to a session someone else owns

Sometimes the kernel already exists. Leela, for example, runs a Python session with its own Dhrishti server, and you want a Ramabana agent beside it rather than a second kernel. ramabana --attach NAME starts no kernel. It finds the running server and points the agent’s host at it. The other program keeps the person’s prompt, so in this case there is no Python mode, and Ramabana is only the agent.

find_session(attach) turns a name into the server’s address. A URL passes straight through. Otherwise it reads Dhrishti’s registry of live sessions and tries, in order, an exact session name, a name of the form <attach>-pyrepl-<pid>, and the name of the folder the session runs in. When two sessions match it raises with both rather than pick one, and when none matches it names the sessions that are running. sessions() lists the live sessions, which is what /kernels and ramabana --kernels print.

find_session only reads the registry and never calls Dhrishti’s serve_in_kernel again. Called on a session that is already serving, serve_in_kernel returns the right port but also repoints that session’s logs and resets its agent mode, so a readonly session would come back restricted.

Promotion is not available here. The agent’s work shows in the shared session notebook and over the API that the owning program already watches, and the person promotes from that program, which holds the token.

sessions

def sessions():

Every live Dhrishti session: what to pass to --attach, and where it is.

find_session

def find_session(
    attach
):

Base URL of the live Dhrishti session named attach, or attach if it is a URL.

test_eq(find_session('http://127.0.0.1:9999'), 'http://127.0.0.1:9999')
test_fail(lambda: find_session('no-such-session-anywhere'), contains='no live dhrishti session')

The real registry is shared by every Dhrishti session on this machine, so the next cell replaces dhrishti.serving.active with a fixed list for each check and runs the real resolution code against it. It covers a match by full name and by project name, a refusal when two sessions share a name, a project name that must not match a longer one that starts the same way, and a match by folder name.

from unittest.mock import patch

def _entry(name, port, cwd='/x/proj'):
    return {'name': name, 'port': port, 'pid': port, 'cwd': cwd, 'base': f'http://127.0.0.1:{port}', 'started': 0}

one = [_entry('proj-pyrepl-111', 9001, '/home/proj')]
dupe_name = [_entry('proj-pyrepl-111', 9001, '/home/a'), _entry('proj-pyrepl-111', 9002, '/home/b')]
by_cwd = [_entry('some-other-name-222', 9003, '/home/leela')]

with patch('dhrishti.serving.active', lambda: one):
    test_eq(find_session('proj-pyrepl-111'), 'http://127.0.0.1:9001')
    test_eq(find_session('proj'), 'http://127.0.0.1:9001')

with patch('dhrishti.serving.active', lambda: dupe_name):
    try:
        find_session('proj-pyrepl-111')
        assert False, 'expected a refusal'
    except RuntimeError as e:
        msg = str(e)
        assert 'matches more than one' in msg
        assert '/home/a' in msg and '/home/b' in msg
        assert '9001' in msg and '9002' in msg

cross_project = [_entry('ramabana-pyrepl-1', 9004, '/home/ramabana'),
                 _entry('ramabana-extra-pyrepl-2', 9005, '/home/ramabana-extra')]

with patch('dhrishti.serving.active', lambda: cross_project):
    test_eq(find_session('ramabana'), 'http://127.0.0.1:9004')

with patch('dhrishti.serving.active', lambda: by_cwd):
    test_eq(find_session('leela'), 'http://127.0.0.1:9003')
    test_fail(lambda: find_session('eela'), contains='no live dhrishti session')

test_eq(find_session('http://127.0.0.1:9999'), 'http://127.0.0.1:9999')

Colour

hl colours a line of Python for the terminal with rich’s syntax highlighter, which uses pygments. Both are already installed for the transcript. It returns a rich Text whose .plain is exactly the source, or uncoloured Text when pygments cannot read it. Colour changes only what is drawn. log_cell, the session notebook and /copy keep the plain source, because text with colour codes in it cannot be pasted back as code.

hl

def hl(
    src
):

src as highlighted Text with .plain == src, or plain Text if pygments cannot lex it.

t = hl('x = 1  # note')
test_eq(t.plain, 'x = 1  # note')
assert t.spans
test_eq(hl('').plain, '')

The prompt, against a live kernel

The prompt is the terminal’s cli.Ui, and Python mode is one of its modes. The cells below drive it from here because this notebook has the live kernel. They draw the terminal on an emulated screen, EmuTty, and use fake_agent in place of a model. Agent mode is unchanged: the same blocks, folding, approvals, slash commands and status bar. Python mode changes what a typed line means, from a prompt for the model to code for the kernel.

The cells check switching modes with /agent and /python (or /py), /help and /vars in Python mode, prompt history, how each kind of kernel output is shown, a turn that reaches the model and is written to the session notebook, attaching an image in agent mode, and that an @ inside Python code is never taken for an attachment.

from ramabana.testing import fake_agent
from ramabana.cli import CompletionMenu, Ui, run_turn
from teleprint.compositor import Compositor
from teleprint.keys import Key
from teleprint.testing import EmuTty        # not `pyghostty.EmuTty`; see nbs/05_cli.ipynb

tty = EmuTty(80, 40)   # tall enough for /help: `term.text()` has no scrollback
comp = Compositor(tty)
comp._register_signals = lambda: None    # nbdev runs this async cell on a worker thread
await comp.start()
agent, be = fake_agent(host=host, replies=['`owner` is a dict with one key.'])
ui = Ui(comp, agent)
ui.kernel = kernel          # `enter_python` adopts it rather than start a second
await ui.enter_python()
comp.on_key = ui.on_key
ui.paint()
test_eq(ui.mode, 'python')
assert 'python' in ui.status().plain

ui.buf.insert('/agent'); test_eq(ui.submit(), None)
test_eq(ui.mode, 'agent')
assert ui.prompt().plain.startswith('▌')
ui.buf.insert('/py'); await ui.submit()
test_eq(ui.mode, 'python')

ui.buf.insert('/help'); ui.submit()
assert 'python' in tty.term.text()

ui.buf.insert('/vars'); test_eq(ui.submit(), None)
assert 'owner' in ui.transcript.block_text(list(comp.blocks.values())[-1])

ui.buf.clear()
ui.buf.insert('/detach nothing.png')
ui.complete = CompletionMenu(ui.buf, ['/detach'], start=0, show=8)
test_eq(ui.submit(), None)
test_eq((ui.complete, ui.buf.text), (None, ''))
test_eq(ui.history[-1], '/detach nothing.png')

ui.buf.insert('recalled_py = 1')
await ui.submit()
test_eq(ui.history[-1], 'recalled_py = 1')
ui.recall(-1)
test_eq(ui.buf.text, 'recalled_py = 1')
ui.buf.clear()

ui.on_output({'output_type': 'stream', 'name': 'stdout', 'text': 'printed\n'})
ui.on_output({'output_type': 'execute_result', 'data': {'text/plain': '42'}})
ui.on_output({'output_type': 'error', 'ename': 'ValueError', 'evalue': 'bad', 'traceback': []})
kinds = [b.tag for b in comp.blocks.values()][-3:]
test_eq(kinds, ['note', 'reply', 'error'])

await ui.run_code('from_the_prompt = 99')
r = await kernel.execute('from_the_prompt')
test_eq(output_text(r.outputs), '99')
ui.mode = 'agent'
await run_turn(ui, 'what is in owner?')
assert be.sent, 'the turn reached the backend'
test_eq([c.cell_type for c in read_nb(host.agent_log).cells[-2:]], ['markdown', 'markdown'])
ui.mode = 'python'

_real_host = agent.host
try:
    for h in (nolog, dead_host):
        agent.host = h
        await ui.run_code('logless = 1')
        test_eq(ui.turn, None)
finally: agent.host = _real_host
test_eq(output_text((await kernel.execute('logless')).outputs), '1')

_real_log = host.log_cell
host.log_cell = lambda *a, **kw: 1/0
try: await ui.run_code('wedge_check = 1')
except ZeroDivisionError: pass
finally: host.log_cell = _real_log
test_eq(ui.turn, None)

from dataclasses import replace

pics = Path(tempfile.mkdtemp())
(pics / 'shot.png').write_bytes(b'\x89PNG\r\n\x1a\n' + b'0' * 64)
shot = pics / 'shot.png'

ui.mode = 'agent'
be.spec = replace(be.spec, ctx=128_000)
ui.buf.insert(f'/attach {shot}'); ui.submit()
assert 'shot.png' in ui.attach_row().plain
ui.buf.insert('what is in the picture?')
await ui.submit()
assert isinstance(be.sent[-1], list) and shot.read_bytes() in be.sent[-1]
test_eq(ui.attachments, [])
ui.mode = 'python'

leak = pics / '@leak.png'
leak.write_bytes(b'\x89PNG\r\n\x1a\n' + b'0' * 64)

ui.mode = 'python'
ui.buf.insert("py_matmul = type('M', (), {'__matmul__': lambda self, o: 42})() @ 1")
await ui.submit()
r = await kernel.execute('py_matmul')
test_eq(output_text(r.outputs), '42')
test_eq(ui.attachments, [])

ui.buf.insert(f'py_leak = 5  # not an attachment: @{leak}')
await ui.submit()
r = await kernel.execute('py_leak')
test_eq(output_text(r.outputs), '5')
test_eq(ui.attachments, [])

Typing more than one line

Ui submits a line when you press enter. That suits a sentence and fails for a block of code, because for i in range(3): is not a program yet. code_state(src) uses the standard library’s codeop.compile_command to say whether the text is complete, incomplete, or invalid, meaning it can never compile. On enter, an incomplete buffer gets a new line under a continuation prompt, a complete one runs, and an invalid one is reported at the prompt without a trip to the kernel. An empty line submits what is pending, as at every Python prompt, and pasted text that compiles runs as one cell.

The first cell below lists the cases. The ones after it type a loop line by line, enter a broken line, and paste two statements.

code_state

def code_state(
    src
):

Whether src is a finished statement, an unfinished one, or one that cannot compile.

test_eq(code_state('x = 1'), 'complete')
test_eq(code_state('for i in range(3):'), 'incomplete')
test_eq(code_state('x = ('), 'incomplete')
test_eq(code_state('x = )'), 'invalid')
test_eq(code_state(''), 'complete')
test_eq(code_state('for i in range(3):\n    print(i)'), 'incomplete')
test_eq(code_state('for i in range(3):\n    print(i)\n'), 'complete')

test_eq(code_state('x = 1\ny = 2\n'), 'complete')
test_eq(code_state('import os\nprint(os.getcwd())'), 'complete')
test_eq(code_state('import os\nfor i in range(2):\n    print(i)'), 'complete')

test_eq(code_state('x = 1\ny = )'), 'invalid')
assert 'unmatched' in _syntax_note('x = 1\ny = )')
assert 'multiple statements' not in _syntax_note('x = 1\ny = )')
test_eq(code_state('x = 1\ny = ('), 'incomplete')

await kernel.execute('multiline_ran = []')
ui.buf.clear()
ui.buf.insert('for i in range(3):')
ui.on_key(Key('enter'))
test_eq(ui.buf.text, 'for i in range(3):\n')
assert ui.turn is None
assert ui.CONT.strip() in tty.term.text()

ui.buf.insert('    multiline_ran.append(i)')
rows, (row, line_no, col) = ui.tail()
test_eq(row, len(rows) - 1)
test_eq(line_no, 1)
test_eq(col, Text(ui.CONT + '    multiline_ran.append(i)').cell_len)
assert ui.CONT in ui.prompt().plain
assert ui.prompt().spans

ui.on_key(Key('enter'))
test_eq(ui.buf.text, 'for i in range(3):\n    multiline_ran.append(i)\n')
out = ui.on_key(Key('enter'))
test_eq(ui.buf.text, '')
await out
r = await kernel.execute('multiline_ran')
test_eq(output_text(r.outputs), '[0, 1, 2]')
test_eq(read_nb(host.agent_log).cells[-1].source, 'for i in range(3):\n    multiline_ran.append(i)\n')

ui.buf.clear()
ui.buf.insert('x = )')
ui.on_key(Key('enter'))
assert 'invalid' in tty.term.text() and ui.turn is None
ui.buf.clear()

ui.mode = 'agent'
ui.buf.insert('for the record:')
out = ui.on_key(Key('enter'))
test_eq(ui.buf.text, '')
if out is not None: await out
ui.mode = 'python'

ui.buf.clear()
ui.paste('pasted_a = 3\npasted_b = pasted_a * 2')
test_eq(code_state(ui.buf.text), 'complete')
out = ui.on_key(Key('enter'))
test_eq(ui.buf.text, '')
if out is not None: await out
r = await kernel.execute('pasted_b')
test_eq(output_text(r.outputs), '6')

ui.buf.clear(); ui.buf.insert('df2 = df.copy()')
assert ui.prompt().spans
blk = ui.say(hl('df2 = df.copy()'), 'user', source='df2 = df.copy()')
test_eq(ui.transcript.block_text(blk), 'df2 = df.copy()')
ui.buf.clear()

Getting back out of Python mode

A line that starts with / is a command. It goes to submit without being compiled, because as Python /agent is invalid and Python mode would have no way out. Tab after /mod completes the command name, as it does in agent mode.

def _line(u, s):
    for ch in s: u.on_key(Key(ch, char=ch))
    return u.on_key(Key('enter'))

test_eq(code_state('/agent'), 'invalid')
test_eq(_line(ui, '/agent'), None)
test_eq((ui.mode, ui.buf.text), ('agent', ''))
await _line(ui, '/python')
test_eq((ui.mode, ui.buf.text), ('python', ''))
test_eq(_line(ui, '/help'), None)
test_eq(ui.buf.text, '')
assert 'python' in tty.term.text()
test_eq(_line(ui, '/quit'), 'quit')

ui.buf.clear(); ui.buf.insert('/mod')
test_eq(ui.on_key(Key('tab')), None)
test_eq(ui.buf.text, '/model')
assert ui.complete is not None
ui.complete = None; ui.buf.clear()

_line(ui, 'for i in range(3):')
test_eq(ui.buf.text, 'for i in range(3):\n')
ui.buf.clear()

The main prompt leaves mouse selection to the terminal, so you can select and copy text as usual. enter_transcript turns mouse handling on only while the transcript view is open.

Tab, against the live kernel

complete_python asks the kernel for completions at the cursor. One match completes the buffer in place. Several matches fill in their common prefix and open a menu, each entry described by annotate. The descriptions are fetched off the event loop’s thread, and typing while they are on their way closes the menu without waiting for them, as the last two cells check.

await kernel.execute('completing = {"a": 1}')
ui.buf.clear(); ui.buf.insert('complet')
ui.buf.cursor = len(ui.buf.text)
await ui.complete_python()
test_eq(ui.buf.text, 'completing')
ui.buf.clear()

await kernel.execute('twovals_a = 1\ntwovals_b = 2')
ui.buf.clear(); ui.buf.insert('twovals_'); ui.buf.cursor = len('twovals_')
await ui.complete_python()
assert ui.complete is not None, 'several matches arm the menu'
test_eq(sorted(ui.complete.matches), ['twovals_a', 'twovals_b'])
assert all('->' in d for d in ui.desc), ui.desc

ui.buf.clear(); ui.buf.insert('twoval'); ui.buf.cursor = len('twoval')
await ui.complete_python()
test_eq((ui.buf.text, ui.buf.cursor), ('twovals_', 8))
test_eq(len(ui.complete.matches), 2)
ui.buf.clear()

loop_thread = threading.current_thread()
seen_thread = {}
_real_describe = host.describe
def _watched_describe():
    seen_thread['t'] = threading.current_thread()
    return _real_describe()
host.describe = _watched_describe
try:
    await kernel.execute('threadcheck_a = 1\nthreadcheck_b = 2')
    ui.buf.clear(); ui.buf.insert('threadcheck_'); ui.buf.cursor = len('threadcheck_')
    await ui.complete_python()
finally:
    host.describe = _real_describe
assert seen_thread.get('t') is not None and seen_thread['t'] is not loop_thread
assert all('->' in d for d in ui.desc), ui.desc
ui.buf.clear()

release = threading.Event()
def _blocked_describe():
    release.wait(5)
    return {'twovals_a': 'int', 'twovals_b': 'int'}
_real_describe = host.describe
host.describe = _blocked_describe
ui.complete, ui.desc = None, []
try:
    await kernel.execute('twovals_a = 1\ntwovals_b = 2')
    ui.buf.clear(); ui.buf.insert('twovals_'); ui.buf.cursor = len('twovals_')
    task = asyncio.ensure_future(ui.complete_python())
    for _ in range(500):
        if ui.complete is not None: break
        await asyncio.sleep(0.01)
    assert ui.complete is not None and ui.desc == []
    ui.on_key(Key('x', char='x'))
    test_eq(ui.buf.text, 'twovals_x')
    test_eq(ui.complete, None)
    release.set()
    await task
finally:
    host.describe = _real_describe
    release.set()
test_eq((ui.complete, ui.desc), (None, []))
test_eq(ui.buf.text, 'twovals_x')
ui.buf.clear()

`/promote`

Values the agent makes live in its overlay. /promote NAME moves one into the owner’s namespace, where it becomes an ordinary variable at your prompt. Promotion is the owner’s act: it is a slash command the person types and never a tool, so the model can ask for it in words but cannot do it. promote(base, name) reads the session’s owner token and calls Dhrishti’s /api/promote. It works only on a kernel this session started. An attached session’s token belongs to the program that started it, and the terminal says so. The test checks that the command runs off the event loop’s thread and that /promote alone prints its usage.

promote

def promote(
    base, name
):

Move the agent-layer variable name into the owner namespace of a kernel this session started.

test_eq(promote('', 'x'), 'no kernel of our own; promote from the surface that started the session')

import ramabana.pyrepl as _pyr

seen = {}
def _fake_promote(base, name):
    seen.update(base=base, name=name, thread=threading.current_thread())
    return f'promoted {name}'
_real_promote, _pyr.promote = _pyr.promote, _fake_promote
try:
    ui.buf.insert('/promote answer_42')
    await ui.submit()
finally: _pyr.promote = _real_promote
test_eq((seen['base'], seen['name']), (kernel.base, 'answer_42'))
assert seen['thread'] is not threading.current_thread()
assert 'promoted answer_42' in tty.term.text()
test_eq(ui.buf.text, '')

ui.buf.insert('/promote')
test_eq(ui.submit(), None)
assert 'usage: /promote' in tty.term.text()

Attaching to the live session

The kernel this notebook started is itself a live Dhrishti session, so attaching can be tested against a real server. find_session finds it by its registry name, <project>-pyrepl-<pid>, as Kernel._bootstrap sets it. The process id keeps the name apart from any session left over from an earlier run. The cell after it checks that shutdown forces a kernel that does not stop when asked, and the last one shuts this notebook’s kernel down.

from dhrishti.serving import active

own = next((e for e in active() if e.get('base') == kernel.base), None)
assert own, 'the kernel above is serving'
test_eq(find_session(own['name']), kernel.base)

attached = DhrishtiHost(['.'], find_session(own['name']), web=False, index=False)
test_eq(attached.kernel_kind, 'ipykernel')
assert 'owner' in attached.list_vars()

class _StuckKm:
    has_kernel = True
    def __init__(self): self.forced = None
    async def shutdown_kernel(self, now=False):
        self.forced = now
        if not now: await asyncio.sleep(30)

stuck, km = Kernel('.'), _StuckKm()
stuck.km, stuck.kc = km, None
await stuck.shutdown(timeout=0.05)
test_eq(km.forced, True)
test_eq(stuck.km, None)
assert stuck.base is None

class _GoodKm(_StuckKm):
    async def shutdown_kernel(self, now=False): self.forced = now
good = _GoodKm(); ok = Kernel('.'); ok.km, ok.kc = good, None
await ok.shutdown(timeout=5)
test_eq(good.forced, False)

await kernel.shutdown()
test_eq(kernel.alive, False)

The agent at the Python prompt

--agent-proxy, or /agent_proxy inside a session, makes the session’s agent reachable from Python as ramabana_agent, so code at the prompt can read the agent’s token use or attach a chat callback. ramabana_agents holds every agent bound in that namespace, by label.

AgentBridge runs in the Ramabana process. It serves one HTTP path, /agent, on the loopback address, and answers only requests that carry its random token. It offers a fixed set of operations. usage, turn_usage and chat_usage return token counts and cost for the session, the last turn and the turn’s chat. callbacks lists the chat callbacks that are known and attached, and attach_callback(name) attaches one, on the session’s own thread through dispatch. Anything else is refused. agent_proxy_code writes the source of a small AgentProxy class that calls the bridge, and inject_agent_proxy runs that source in the agent’s overlay with run_python. The terminal also runs it in the kernel itself, so the person’s code sees the proxy too.

inject_agent_proxy

def inject_agent_proxy(
    host, url, token, label:str='local'
):

Bind the agent proxy in the agent’s Dhrishti overlay by running agent_proxy_code through run_python.

agent_proxy_code

def agent_proxy_code(
    url, token, label:str='local'
):

Source binding ramabana_agent (last session) and ramabana_agents (all, by label).

AgentBridge

def AgentBridge(
    agent, dispatch:NoneType=None
):

A loopback HTTP endpoint, guarded by a random token, through which kernel code reads this agent’s usage and attaches chat callbacks.