# pyrepl
## 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`](https://vedicreader.github.io/ramabana/pyrepl.html#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`](https://vedicreader.github.io/ramabana/pyrepl.html#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.
------------------------------------------------------------------------
source
### output_text
``` python
def output_text(
outputs
):
```
*Flatten notebook outputs for tests, logs and plain terminal fallbacks.*
------------------------------------------------------------------------
source
### ExecOutcome
``` python
def ExecOutcome(
ok:bool=True, outputs:list=, execution_count:int | None=None, error:str | None=None
)->None:
```
*One kernel request in nbformat’s output shape.*
``` python
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`](https://vedicreader.github.io/ramabana/pyrepl.html#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
`-pyrepl-`, so another terminal can find it.
`execute` runs one cell and returns an
[`ExecOutcome`](https://vedicreader.github.io/ramabana/pyrepl.html#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`](https://vedicreader.github.io/ramabana/pyrepl.html#kernel)
that is not running answers `execute` with an error and `complete` with
no matches, as the second cell shows.
------------------------------------------------------------------------
source
### Kernel
``` python
def Kernel(
cwd:str='.'
):
```
*A private ipykernel with Dhrishti serving its live namespace.*
``` python
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:')
```
``` python
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`](https://vedicreader.github.io/ramabana/pyrepl.html#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`](https://vedicreader.github.io/ramabana/pyrepl.html#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.
------------------------------------------------------------------------
source
### DhrishtiHost
``` python
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.*
------------------------------------------------------------------------
source
### use_kernel
``` python
def use_kernel(
host, base
):
```
*Point `host`’s Python at a Dhrishti overlay in place, keeping every
other tool group.*
------------------------------------------------------------------------
source
### Dhrishti
``` python
def Dhrishti(
base
):
```
*The agent half of a Dhrishti session, in the shape of a kernel a
`LocalHost` can hold.*
[`Dhrishti`](https://vedicreader.github.io/ramabana/pyrepl.html#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.
``` python
from shalya.host import LocalHost
```
``` python
k = Dhrishti(kernel.base)
k.scopes, k.kind, k.run('from_the_kernel = 1')
```
``` python
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`](https://vedicreader.github.io/ramabana/pyrepl.html#use_kernel),
and nothing is added or lost. After that, a
[`DhrishtiHost`](https://vedicreader.github.io/ramabana/pyrepl.html#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 `''`.
``` python
from shalya.tools import tools_for
```
``` python
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)
```
``` python
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)
```
``` python
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')
```
``` python
from fastcore.nbio import read_nb
```
``` python
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)
```
``` python
_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])
```
``` python
_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`](https://vedicreader.github.io/ramabana/pyrepl.html#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.
------------------------------------------------------------------------
source
### annotate
``` python
def annotate(
matches, described
):
```
*Candidate names with their type and shape where the session knows one.*
``` python
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
`-pyrepl-`, 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()`](https://vedicreader.github.io/ramabana/pyrepl.html#sessions)
lists the live sessions, which is what `/kernels` and
`ramabana --kernels` print.
[`find_session`](https://vedicreader.github.io/ramabana/pyrepl.html#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.
------------------------------------------------------------------------
source
### sessions
``` python
def sessions():
```
*Every live Dhrishti session: what to pass to `--attach`, and where it
is.*
------------------------------------------------------------------------
source
### find_session
``` python
def find_session(
attach
):
```
*Base URL of the live Dhrishti session named `attach`, or `attach` if it
is a URL.*
``` python
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.
``` python
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`](https://vedicreader.github.io/ramabana/pyrepl.html#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.
------------------------------------------------------------------------
source
### hl
``` python
def hl(
src
):
```
*`src` as highlighted `Text` with `.plain == src`, or plain `Text` if
pygments cannot lex it.*
``` python
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`](https://vedicreader.github.io/ramabana/testing.html#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.
``` python
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
```
``` python
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
```
``` python
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()
```
``` python
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'])
```
``` python
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)
```
``` python
from dataclasses import replace
```
``` python
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'
```
``` 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`](https://vedicreader.github.io/ramabana/cli.html#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.
------------------------------------------------------------------------
source
### code_state
``` python
def code_state(
src
):
```
*Whether `src` is a finished statement, an unfinished one, or one that
cannot compile.*
``` python
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')
```
``` python
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')
```
``` python
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'
```
``` 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')
```
``` python
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.
``` python
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`](https://vedicreader.github.io/ramabana/pyrepl.html#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.
``` python
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()
```
``` python
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()
```
``` python
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.
------------------------------------------------------------------------
source
### promote
``` python
def promote(
base, name
):
```
*Move the agent-layer variable `name` into the owner namespace of a
kernel this session started.*
``` python
test_eq(promote('', 'x'), 'no kernel of our own; promote from the surface that started the session')
```
``` python
import ramabana.pyrepl as _pyr
```
``` python
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`](https://vedicreader.github.io/ramabana/pyrepl.html#find_session)
finds it by its registry name, `-pyrepl-`, as
[`Kernel._bootstrap`](https://vedicreader.github.io/ramabana/pyrepl.html#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.
``` python
from dhrishti.serving import active
```
``` python
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()
```
``` python
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)
```
``` python
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`](https://vedicreader.github.io/ramabana/pyrepl.html#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`](https://vedicreader.github.io/ramabana/pyrepl.html#agent_proxy_code)
writes the source of a small `AgentProxy` class that calls the bridge,
and
[`inject_agent_proxy`](https://vedicreader.github.io/ramabana/pyrepl.html#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.
------------------------------------------------------------------------
source
### inject_agent_proxy
``` python
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`](https://vedicreader.github.io/ramabana/pyrepl.html#agent_proxy_code)
through `run_python`.*
------------------------------------------------------------------------
source
### agent_proxy_code
``` python
def agent_proxy_code(
url, token, label:str='local'
):
```
*Source binding `ramabana_agent` (last session) and `ramabana_agents`
(all, by label).*
------------------------------------------------------------------------
source
### AgentBridge
``` python
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.*