# cli
Two small fixes to teleprint are applied when this module loads. Each
checks whether the installed teleprint already has the fix and does
nothing if so; `FRAME_PATCHED` and `INK_PATCHED` record whether the
patch was needed.
The first concerns rows already written to the scrollback. Teleprint
keeps a mark, `_ws`, of how many rows of the document it has committed
there, and each frame writes only the rows past the mark. Those rows are
history in the terminal and cannot be taken back. When a tool result
folds, the document gets shorter, and teleprint used to lower the mark
to match, so the next frame wrote earlier rows a second time. A session
with many folded calls ended up with many copies of its startup banner.
The patch keeps the mark from ever going down.
The second adds a hook, `Compositor.on_frame`, which runs after each
paint. Pictures need it: a picture is placed on the screen after the
text is drawn, as the section on pictures below explains. The hook is
installed before the mark fix, so `_frame.__wrapped__` is the frame
without the mark fix, and the test below uses it to show the duplicated
rows.
``` python
from teleprint.testing import EmuTty as _EmuTty
```
``` python
async def _marks_over_a_fold(frame):
"The ink mark after growth, after a fold, and after growing again, under `frame`."
tty = _EmuTty(40, 8)
comp = Compositor(tty); comp._register_signals = lambda: None
await comp.start()
comp.print_block(Text('BANNER'), tag='note', collapse_at=None, source='BANNER')
tall = comp.print_block(Text('\n'.join(f'line {i}' for i in range(20))), tag='tool',
collapse_at=None, source='x')
out = []
frame(comp); out.append(comp._ws)
tall.collapsed, tall.collapse_at = True, 1
comp._dirty(tall); frame(comp); out.append(comp._ws)
comp.print_block(Text('\n'.join(f'again {i}' for i in range(20))), tag='tool',
collapse_at=None, source='y')
frame(comp); out.append(comp._ws)
tty.close()
return out
_marks = await _marks_over_a_fold(Compositor._frame)
assert _marks[0] > 0, 'the document has to outgrow the window for anything to be inked at all'
test_eq(_marks, sorted(_marks))
if INK_PATCHED:
_bare = await _marks_over_a_fold(Compositor._frame.__wrapped__)
test_eq(_bare[1], 0)
assert _bare != sorted(_bare)
```
``` python
_h_tty = _EmuTty(30, 6)
_h_comp = Compositor(_h_tty); _h_comp._register_signals = lambda: None
await _h_comp.start()
_fired = []
_h_comp.on_frame = lambda: _fired.append(_h_comp._cursor)
_h_comp.print_block(Text('one'), tag='note')
test_eq(len(_fired), 1)
_h_comp.paused = True
_h_comp.print_block(Text('two'), tag='note')
test_eq(len(_fired), 1)
_h_tty.close()
```
Ramabana gives `shift+enter` its own meaning: during a turn it queues
the line as the next turn, where plain `enter` steers the running one.
By default a terminal sends the same byte for both keys, so the program
has to ask the terminal for richer key reports. There are two protocols
for this:
- Kitty’s keyboard protocol, with flag 1 (“disambiguate”), reports a
modified key as `CSI code;mods u`. Ghostty and kitty support it.
- Xterm’s modifyOtherKeys, at level 1, reports it as
`CSI 27;mods;code ~`. Tmux sends this form when its `extended-keys`
option is on, and the kitty form when `extended-keys-format` is
`csi-u`.
The compositor asks for both, and each terminal answers with the one it
supports.
[`ext_key`](https://vedicreader.github.io/ramabana/cli.html#ext_key)
turns either report into the same `Key` the old byte parser would have
produced. Once kitty’s flag is on, `esc`, control letters and alt
letters also arrive in the new form, so the decoder has to cover every
key, and it follows these rules:
- A key release is dropped.
- Keypad keys decode to the same characters and names the old bytes
gave.
- `ctrl+i`, `ctrl+j` and `ctrl+m` stay `tab` and `enter`.
- Kitty’s other private-use keys, such as bare modifiers and media keys,
are dropped rather than typed.
- Caps lock and num lock are ignored, since they are states rather than
key chords.
- Chords with super, hyper or meta have no names here and are dropped.
[`ext_keys_ok`](https://vedicreader.github.io/ramabana/cli.html#ext_keys_ok)
decides whether to ask at all: only when both stdin and stdout are a
real terminal, so tests and pipes never see the escape codes. The
compositor turns the modes off wherever it hands the terminal back: at
`stop`, at `release` when another program borrows the terminal, before a
fatal signal, and while `ctrl+z` has the process suspended. A `ctrl+z`
from the transcript view also leaves the alternate screen while
suspended and comes back on `fg`. Kitty keeps a separate flag stack for
the alternate screen, so the transcript view sets the flag again there
and clears it before leaving.
------------------------------------------------------------------------
source
### ext_keys_ok
``` python
def ext_keys_ok(
tty
):
```
*Whether `tty` is a real terminal at both ends.*
------------------------------------------------------------------------
source
### ext_key
``` python
def ext_key(
code, mods:int=1, event:int=1
):
```
*The `Key` for a kitty or modifyOtherKeys report, or None.*
``` python
def _ks(*chunks):
p = Parser()
return [e.name for c in chunks for e in p.feed(c)]
test_eq(_ks('\x1b[13;2u', '\x1b[27;2;13~', '\x1b\r', '\r'), ['shift+enter', 'shift+enter', 'alt+enter', 'enter'])
test_eq(_ks('\x1b[9;2u', '\x1b[27;2;9~', '\x1b[Z'), ['shift+tab'] * 3)
test_eq(_ks('\x1b[27u', '\x1b[99;5u', '\x03', '\x1b[49;3u', '\x1b1'), ['escape', 'ctrl+c', 'ctrl+c', 'alt+1', 'alt+1'])
test_eq(_ks('\x1b[13;2:3u', '\x1b[57428u'), [])
test_eq(_ks('\x1b[13;2', ':1u'), ['shift+enter'])
test_eq(Parser().feed('\x1b[97u'), [Key('a', 'a')])
test_eq(_ks('\x1b[57414u', '\x1b[106;5u', '\x1b[127;2u', '\x1b[57441u'), ['enter', 'enter', 'backspace'])
test_eq(ext_keys_ok(_EmuTty(10, 4)), False)
```
## Blocks and keys
Every block has a kind, and each kind has its own gutter, the mark in
the left margin. `GUTTERS` maps the nine kinds to their marks: `user`
for your prompts (`▌`), `step` for the agent’s narration (`┆`), `tool`
for a tool call (`│`), `reply` for the answer, `ask` for an approval
question (`◆`), `plan` for the plan checklist (`▸`),
[`note`](https://vedicreader.github.io/ramabana/agent.html#note) for a
message from Ramabana itself (`·`), `error` (`×`), and `media` for a
picture.
A turn is one prompt and everything the agent does to answer it. In the
transcript it reads as a timeline in the order the work happened: a
`step` for each stretch of narration, a `tool` block for each call that
narration introduced, and one `reply` with the answer at the end.
Results fold to one row and narration stays visible, so the answer
always sits below the narration that led to it.
The cell below also holds the palettes. `DARK`, `GRUVBOX` and the other
dicts name the colours of each theme,
[`set_theme`](https://vedicreader.github.io/ramabana/cli.html#set_theme)
picks the active one, and
[`code_theme`](https://vedicreader.github.io/ramabana/cli.html#code_theme),
[`hl_theme`](https://vedicreader.github.io/ramabana/cli.html#hl_theme)
and [`code_bg`](https://vedicreader.github.io/ramabana/cli.html#code_bg)
give the matching colours for highlighted code.
[`plan_text`](https://vedicreader.github.io/ramabana/cli.html#plan_text)
draws the plan checklist in the active palette.
------------------------------------------------------------------------
source
### plan_text
``` python
def plan_text(
md
):
```
*The plan checklist as themed `Text`, reading `GRUVBOX` per call.*
------------------------------------------------------------------------
source
### set_theme
``` python
def set_theme(
name:str='github-dark'
):
```
*Select the active palette; `auto` means the default.*
------------------------------------------------------------------------
source
### code_bg
``` python
def code_bg(
palette:NoneType=None
):
```
*The code block background: the palette’s `bg1`.*
------------------------------------------------------------------------
source
### hl_theme
``` python
def hl_theme(
name:NoneType=None
):
```
*The Lumis theme
[`hl`](https://vedicreader.github.io/ramabana/pyrepl.html#hl) colours
code with under the named palette.*
------------------------------------------------------------------------
source
### code_theme
``` python
def code_theme(
name:NoneType=None
):
```
*The pygments style to highlight code with under the named palette.*
------------------------------------------------------------------------
source
### guide_text
``` python
def guide_text(
text
):
```
*The guide with its headers and commands styled.*
------------------------------------------------------------------------
source
### key_card
``` python
def key_card(
text
):
```
*The key list, with its labels and separators styled.*
``` python
print(HELP)
```
``` python
print(guide_text(GUIDE))
```
## Attachments
A prompt can carry files with it: pictures, sound, or text. In the
terminal you attach one in three ways. You can drop the file on the
window, paste its path, or write `@path` in the prompt. All three end up
as the same attachment. The path parser accepts quoted paths, paths in
brackets, `file://` URIs and backslash-escaped spaces, and a plain
sentence that merely mentions a path is left alone, as the tests below
check.
A picture goes to the model as an image content part, the form a model
API takes images in. Every attachment also carries its absolute path,
and the path still works when the model cannot take the content part. A
model without audio support, for example, can still open a `.wav`
through the file tools.
The attachment helpers themselves,
[`media_paths`](https://vedicreader.github.io/ramabana/tools.html#media_paths),
[`Attachment`](https://vedicreader.github.io/ramabana/tools.html#attachment),
[`media_parts`](https://vedicreader.github.io/ramabana/tools.html#media_parts)
and the rest, live in `ramabana.tools` and are imported here so the
terminal and the other frontends share one parser.
### Pictures, drawn where the terminal can draw
A picture can reach the screen in two ways. The model can return one in
`Resp.media`, or a tool can write an image file during the turn and
report it through `on_media`. On a terminal without kitty graphics, the
picture is saved and its path printed.
[`kitty_graphics`](https://vedicreader.github.io/ramabana/cli.html#kitty_graphics)
decides which kind of terminal this is, from the environment variables
and `$TERM` names in `KITTY_ENV` and `KITTY_TERM`.
On a terminal with kitty graphics the picture is drawn in place. Kitty
positions an image by screen cell, separately from the text. The
compositor redraws its document with absolute cursor moves, so an image
cannot ride along in the text the way prose does. A
[`Picture`](https://vedicreader.github.io/ramabana/cli.html#picture)
handles this in two steps.
[`Picture.send`](https://vedicreader.github.io/ramabana/cli.html#picture.send)
transfers the image bytes to the terminal once. The block itself holds
only the blank rows the picture will cover, sized by
[`img_cells`](https://vedicreader.github.io/ramabana/cli.html#img_cells)
from
[`png_size`](https://vedicreader.github.io/ramabana/cli.html#png_size).
Then
[`Picture.place`](https://vedicreader.github.io/ramabana/cli.html#picture.place)
writes a short placement command at the end of every frame, at the
current position of those blank rows. Each placement reuses the
picture’s image id, which replaces the previous placement, so repaints
never stack copies. If the terminal clears the cells under an image, the
next frame places it again.
[`picture`](https://vedicreader.github.io/ramabana/cli.html#picture)
builds a
[`Picture`](https://vedicreader.github.io/ramabana/cli.html#picture) for
a file, or returns None when this terminal or this file cannot be drawn.
[`draw_png`](https://vedicreader.github.io/ramabana/cli.html#draw_png)
is for code that has no frames to hook into: it sends and places a
picture at the cursor in one go.
[`media_line`](https://vedicreader.github.io/ramabana/cli.html#media_line)
is the transcript line for a saved picture.
Kitty also offers Unicode placeholders, where the image is anchored to
special characters in the text. They are not used, because Kaku and at
least one other terminal ignore the `U=1` flag, draw at the cursor, and
print `U+10EEEE` as a missing glyph, which leaves a block of boxes
behind the picture.
------------------------------------------------------------------------
source
### media_line
``` python
def media_line(
path
):
```
*The transcript line for a saved picture.*
------------------------------------------------------------------------
source
### draw_png
``` python
def draw_png(
path, cols:int=24
):
```
*Send and place `path` at the cursor, or `''`, for callers without
frames.*
------------------------------------------------------------------------
source
### picture
``` python
def picture(
path, cols:int=24, rows:int=12
):
```
*A [`Picture`](https://vedicreader.github.io/ramabana/cli.html#picture)
for `path`, or None where this terminal or this file cannot be drawn.*
------------------------------------------------------------------------
source
### Picture
``` python
def Picture(
path, cols:int=24, rows:int=12
):
```
*One picture on screen: bytes sent once, then re-placed at the end of
every frame.*
------------------------------------------------------------------------
source
### img_cells
``` python
def img_cells(
path, cols, rows:int=12
):
```
*Cell width and height for `path` in a `cols` by `rows` box, keeping
aspect.*
------------------------------------------------------------------------
source
### png_size
``` python
def png_size(
path
):
```
*`(width, height)` in pixels from a PNG’s IHDR, or `None`.*
------------------------------------------------------------------------
source
### kitty_graphics
``` python
def kitty_graphics():
```
*Whether this terminal speaks the kitty graphics protocol.*
``` python
import os as _osr, struct as _struct, zlib as _zlib
def _a_png(w, h):
"A real noise PNG, big enough to need several APC chunks."
def chunk(tag, data):
return (_struct.pack('>I', len(data)) + tag + data
+ _struct.pack('>I', _zlib.crc32(tag + data) & 0xffffffff))
raw = b''.join(b'\x00' + _osr.urandom(3 * w) for _ in range(h))
return (b'\x89PNG\r\n\x1a\n' + chunk(b'IHDR', _struct.pack('>IIBBBBB', w, h, 8, 2, 0, 0, 0))
+ chunk(b'IDAT', _zlib.compress(raw)) + chunk(b'IEND', b''))
```
``` python
import os as _os
_seen = {k: _os.environ.pop(k, None) for k in (*KITTY_ENV, 'TERM', 'TERM_PROGRAM')}
try:
test_eq(kitty_graphics(), False)
test_eq(picture('/tmp/nope.png'), None)
test_eq(draw_png('/tmp/nope.png'), '')
assert media_line('/tmp/a.png').startswith('saved ')
import tempfile
with tempfile.TemporaryDirectory() as d:
png = b'\x89PNG\r\n\x1a\n' + b'x'
p = save_media({'mime': 'image/png', 'data': png}, d)
test_eq(p.name, 'image-1.png')
test_eq(p.read_bytes(), png)
test_eq(p.parent.name, 'media')
test_eq(save_media({'mime': 'image/png', 'data': png}, d).name, 'image-2.png')
test_eq(save_media({'mime': 'image/avif', 'data': png}, d).suffix, '.avif')
_os.environ['RAMABANA_KITTY'] = '1'
with tempfile.TemporaryDirectory() as d:
p = Path(d)/'a.png'
p.write_bytes(_a_png(1200, 600))
pic = picture(p, 24)
test_eq((pic.cols, pic.rows), (24, 6))
assert len(pic.send()) > 2 * APC_CHUNK
test_eq(pic.send().count('m=0'), 1)
assert 'a=t' in pic.send() and 'a=p' not in pic.send()
test_eq(pic.place(), f'\x1b_Ga=p,i={pic.id},p=1,c=24,r=6,C=1,q=2\x1b\\')
test_eq(len(pic.gap().plain.split('\n')), 6)
test_eq(pic.gap().plain.split('\n')[0], 'a.png')
test_eq(media_line(p), str(p))
finally:
_os.environ.pop('RAMABANA_KITTY', None)
for k, v in _seen.items():
if v is not None: _os.environ[k] = v
```
Text files can be attached too, up to `MAX_FILE_ATTACH` characters each.
A
[`FileAttachment`](https://vedicreader.github.io/ramabana/cli.html#fileattachment)
captures the file’s text when the prompt is submitted, and
[`file_note`](https://vedicreader.github.io/ramabana/cli.html#file_note)
adds the text of every attached file to the prompt, each tagged with its
exact path, so the model reads the file as it was at that moment.
[`file_refs`](https://vedicreader.github.io/ramabana/cli.html#file_refs)
finds the `@path` references in a prompt.
[`steer_note`](https://vedicreader.github.io/ramabana/cli.html#steer_note)
covers a line sent during a turn, which reaches the model inside a tool
result; a tool result cannot carry a picture, so the note lists each
attachment by path for the model to open with `view_file`.
The cell after this one makes a few files to attach.
------------------------------------------------------------------------
source
### file_note
``` python
def file_note(
atts
):
```
*The grounded text from selected workspace files and folder listings,
with their exact paths.*
------------------------------------------------------------------------
source
### FileAttachment
``` python
def FileAttachment(
path, text, folder:bool=False
):
```
*One text or notebook source attachment, or a folder’s listing, captured
when the prompt is submitted.*
------------------------------------------------------------------------
source
### file_refs
``` python
def file_refs(
text
):
```
*Paths named as `@path` or `@"path with spaces"` in a prompt, sentence
punctuation removed from a bare one.*
------------------------------------------------------------------------
source
### steer_note
``` python
def steer_note(
atts
):
```
*What a steered line says about its attachments: a tool result carries
no picture, so each goes by path.*
``` python
media_dir = Path(tempfile.mkdtemp())
(media_dir / 'shot.png').write_bytes(b'\x89PNG\r\n\x1a\n' + b'0' * 64)
(media_dir / 'My Photo.png').write_bytes(b'\x89PNG\r\n\x1a\n')
(media_dir / 'note.wav').write_bytes(b'RIFF' + b'0' * 40)
shot, spaced, snd_path = media_dir / 'shot.png', media_dir / 'My Photo.png', media_dir / 'note.wav'
sorted(p.name for p in media_dir.iterdir())
```
The [`Ui`](https://vedicreader.github.io/ramabana/cli.html#ui) methods
that attach files, `attach`, `paste` and `attach_file`, are defined
after [`Ui`](https://vedicreader.github.io/ramabana/cli.html#ui) further
down. The tests here check the path parser on its own.
``` python
test_eq(media_paths(str(shot)), [shot])
test_eq(media_paths(f'"{shot}"'), [shot])
test_eq(media_paths(f"'{shot}'"), [shot])
test_eq(media_paths(f'[{shot}]'), [shot])
test_eq(media_paths(f'file://{shot}'), [shot])
test_eq(media_paths(f' {shot} '), [shot])
test_eq(media_paths(str(spaced)), [spaced])
test_eq(media_paths(str(spaced).replace(' ', '\\ ')), [spaced])
test_eq(media_paths(f'{shot} {snd_path}'), [shot, snd_path])
```
``` python
from unittest.mock import patch as mock_patch
```
``` python
test_eq(media_paths('where is the threshold?'), [])
test_eq(media_paths(f'have a look at {shot} when you get a chance'), [])
test_eq(media_paths(str(media_dir / 'gone.png')), [])
_real_expanduser = Path.expanduser
def _expand_without_home(path):
if str(path).startswith('~'): raise RuntimeError('Could not determine home directory.')
return _real_expanduser(path)
with mock_patch.object(Path, 'expanduser', _expand_without_home):
test_eq(media_paths('keep ~literal text'), [])
test_eq(attach_refs(f'compare @{shot} against the spec'), [shot])
test_eq(attach_refs(f'what is in @{shot}?'), [shot])
test_eq(attach_refs('email me @ the address on file'), [])
```
``` python
img, snd = Attachment(shot), Attachment(snd_path)
test_eq((img.kind, img.mime), ('image', 'image/png'))
test_eq((snd.kind, snd.mime), ('audio', 'audio/wav'))
test_eq(media_note([]), '')
hears, deaf = resolve('gemini/gemini-2.5-flash'), resolve('anthropic/claude-opus-4-5')
test_eq(media_parts([img, snd], hears), [img.data, snd.data])
test_eq(media_parts([img, snd], deaf), [img.data])
assert 'The audio above is attached' in media_note([img, snd], hears)
assert 'does not accept audio input' in media_note([img, snd], deaf)
test_eq(media_parts([img, snd]), [img.data, snd.data])
note = media_note([img, snd])
assert str(img.path) in note and str(snd.path) in note
print(note)
```
## The turn
A turn starts when you press enter on a prompt.
[`run_turn`](https://vedicreader.github.io/ramabana/cli.html#run_turn)
runs one turn and streams it into the transcript. The agent’s
`agent.stream` is a blocking generator that runs on a separate model
thread, so
[`run_turn`](https://vedicreader.github.io/ramabana/cli.html#run_turn)
reads its chunks from a queue. That keeps the event loop free to handle
keystrokes, approval questions and activity updates while the model
works.
Before the turn starts, some prompts get a row of choices. A request
such as “refactor the runtime” can mean “make the change”, “propose a
plan”, or “say what is in scope”, and learning which one the model
picked is costly.
[`options_for`](https://vedicreader.github.io/ramabana/cli.html#options_for)
checks the typed line against the trigger words in `MENUS` and returns a
title and a list of
[`Option`](https://vedicreader.github.io/ramabana/cli.html#option)s, or
None.
[`ChoiceMenu`](https://vedicreader.github.io/ramabana/cli.html#choicemenu)
draws the row above the input line and handles the keys that move
through it and pick one. Each
[`Option`](https://vedicreader.github.io/ramabana/cli.html#option)
carries the instruction it appends to the prompt (its `suffix`), so the
choice reaches the model as part of what you asked for. The `REFACTOR`
options are the only menu at present. The examples of the menu in use
come later, under the surface.
[`close_done_shells`](https://vedicreader.github.io/ramabana/cli.html#close_done_shells)
closes the tmux panes of `run_shell_bg` commands that have finished;
their output stays readable through `cmd_output`.
------------------------------------------------------------------------
source
### run_turn
``` python
async def run_turn(
ui, prompt
):
```
*One turn, streamed into the transcript.*
------------------------------------------------------------------------
source
### close_done_shells
``` python
def close_done_shells(
host
):
```
*Close the tmux panes of `run_shell_bg` commands that have exited;
`cmd_output` keeps what they printed.*
------------------------------------------------------------------------
source
### ChoiceMenu
``` python
def ChoiceMenu(
title, options
):
```
*The options row: one line per
[`Option`](https://vedicreader.github.io/ramabana/cli.html#option),
above the input line.*
------------------------------------------------------------------------
source
### options_for
``` python
def options_for(
text
):
```
*`(title, options)` for a prompt worth asking* how\* about, or None.\*
------------------------------------------------------------------------
source
### Option
``` python
def Option(
key:str, label:str, note:str='', suffix:str=''
)->None:
```
*One choice on the options row: key, label, note and prompt suffix.*
Tool results and the files a turn changed are drawn in colour.
[`hl_text`](https://vedicreader.github.io/ramabana/cli.html#hl_text)
highlights a piece of code in a given language, and
[`diff_rich`](https://vedicreader.github.io/ramabana/cli.html#diff_rich)
colours a diff, both through [hl](04b_hl.ipynb) in the active palette
and its Lumis theme from
[`hl_theme`](https://vedicreader.github.io/ramabana/cli.html#hl_theme).
In a unified diff, each line gets the colour of an addition or a
removal, and the code inside is highlighted for the file’s language. A
diffstat gets green and red bars, and any other text stays gray.
[`changed_table`](https://vedicreader.github.io/ramabana/cli.html#changed_table)
lists the files a turn wrote, largest change first, with a total row at
the end.
Whether a finished call stays open is decided by
[`opens`](https://vedicreader.github.io/ramabana/cli.html#opens): a file
write stays open, and so does any unified diff, while a shell or Python
run folds. An open diff is wrapped in
[`OpenDiff`](https://vedicreader.github.io/ramabana/cli.html#opendiff),
which shows the first `DIFF_OPEN` lines with a hint until you open the
block.
------------------------------------------------------------------------
source
### OpenDiff
``` python
def OpenDiff(
ui, blk, text
):
```
*A result shown `DIFF_OPEN` lines deep, with a hint, until the reader
opens it.*
------------------------------------------------------------------------
source
### opens
``` python
def opens(
act
):
```
*Whether a finished call stays open: a write other than a shell or
python run, or any unified diff.*
------------------------------------------------------------------------
source
### changed_table
``` python
def changed_table(
rows, cap:int=12
):
```
*The turn’s changed files, largest `cap` first, then a total over every
one.*
------------------------------------------------------------------------
source
### diff_rich
``` python
def diff_rich(
s
):
```
*`s` coloured by `hl.diff_rich` in the active palette.*
------------------------------------------------------------------------
source
### hl_text
``` python
def hl_text(
code, lang
):
```
*`code` coloured by fastpylight in the active palette’s theme.*
``` python
d = diff_rich('--- a/x.py\n+++ b/x.py\n@@ -1 +1 @@\n-x = 1\n+x = 2')
test_eq(d.plain.splitlines()[-1], '+x = 2')
assert any(GRUVBOX['green'] in str(s.style) for s in d.spans)
test_eq(diff_rich('3 passed').spans, [])
t = hl_text('a\r\nb = 1', 'python')
test_eq(t.plain, 'a\nb = 1')
assert any(t.plain[s.start:s.end] == '1' for s in t.spans), 'colours stay put after a CRLF'
from rich.console import Console
_con = Console(width=60)
with _con.capture() as cap: _con.print(changed_table([('app/[id]/page.tsx', 3, 1, True)]))
assert 'app/[id]/page.tsx' in cap.get() and '1 file' in cap.get()
```
## The surface
[`Ui`](https://vedicreader.github.io/ramabana/cli.html#ui) is the whole
terminal surface, and the class the rest of this page builds on. It has
three jobs. It paints the tail, the status bar and input line at the
bottom of the screen. It turns tool calls and approval requests into
blocks. And it decides what each keystroke means: `on_key` returns a
coroutine for the caller to start (a turn, for example), the string
`'quit'`, or None when the key was handled.
Every method on
[`Ui`](https://vedicreader.github.io/ramabana/cli.html#ui) is
synchronous and does no terminal I/O of its own, so the examples below
drive it against an emulated terminal with no mocks. The one care it
takes is with threads. Activity and approval callbacks arrive on the
model’s worker thread, and only the event loop thread may touch the
compositor, so those callbacks are handed to the loop through `_post`.
The class is long, and later cells add to it with fastcore’s `@patch`.
The methods a caller is most likely to use are `submit` for the typed
line, `say` to print a block,
[`note`](https://vedicreader.github.io/ramabana/agent.html#note) for a
one-line message, `stream` for reply text as it arrives, and `paint` to
redraw.
------------------------------------------------------------------------
source
### Ui
``` python
def Ui(
comp, agent, loop:NoneType=None
):
```
*The terminal surface: a transcript of blocks, a status bar, and one
line to type in.*
A line starting with `/name` that is not one of the agent’s slash
commands is looked up as a skill or a command file. Command files are
markdown files in `/commands/*.md`, plus a root’s
`.agents/commands/` once the `project_extensions` option is turned on.
So `/exhash fix the loop` sends the prompt “fix the loop” with the
`exhash` skill attached, instead of reporting an unknown command. In a
command file, `$ARGUMENTS` is replaced by everything after the name,
`$1` to `$n` by its separate words, and each `@path` by that file when
it lies inside the open folders. A command file with `subtask: true` in
its frontmatter is sent as a task for `delegate_async`, which runs it in
a sub-agent.
The approval keys come next. While an approval question waits, `y`
approves, `n` refuses, and lowercase `a` approves this request and
everything else for the rest of the session. Capital `A` approves and
also adds a rule that always allows requests like this one, through
`always_allow`, when the approval gate can hold rules;
[`ask_pattern`](https://vedicreader.github.io/ramabana/cli.html#ask_pattern)
is what that rule matches. Without rules, `A` behaves like `a`.
[`parse_answer`](https://vedicreader.github.io/ramabana/cli.html#parse_answer)
reads a typed answer such as `y`, `no` or `n: put it in docs/`.
A waiting question rings the terminal bell with `ring`, but only once
nobody has typed for `BELL_IDLE` seconds. `nudge` shows the question
again every `REASK_EVERY` seconds while it waits, and `reask` does the
showing. `#note TEXT` at the prompt calls `note_memory`, which keeps the
line for the next session’s memory.
------------------------------------------------------------------------
source
### Ui.note_memory
``` python
def note_memory(
text
):
```
*`#note TEXT`: keep a line for later sessions.*
------------------------------------------------------------------------
source
### Ui.always_allow
``` python
def always_allow():
```
*Always allow the pending request’s pattern and approve it; False if
none.*
------------------------------------------------------------------------
source
### Ui.ring
``` python
def ring():
```
*Ring the bell if on and nobody typed for `BELL_IDLE` seconds.*
------------------------------------------------------------------------
source
### ask_pattern
``` python
def ask_pattern(
ask
):
```
*What an approval rule would match for this request.*
------------------------------------------------------------------------
source
### Ui.nudge
``` python
def nudge():
```
*Ring and re-show a question after `REASK_EVERY` seconds; True if
re-shown.*
------------------------------------------------------------------------
source
### Ui.reask
``` python
def reask(
body:str=''
):
```
*Show the pending question with `body`, ring, and restart the `nudge`
wait.*
------------------------------------------------------------------------
source
### parse_answer
``` python
def parse_answer(
text
):
```
*`(ok, note)` for a typed approval answer (`y`, `n`, `n: reason`), or
None.*
``` python
from types import SimpleNamespace
class _QueueWork:
def __init__(self): self.closed = False
def close(self): self.closed = True
class _QueueTask:
def add_done_callback(self, callback): self.callback = callback
def done(self): return False
class _QueueComp:
def __init__(self): self.blocks, self.started = {}, None
def remove_block(self, block): self.blocks.pop(block.id)
def spawn(self, work, name):
self.started = work
return _QueueTask()
class _QueueUi:
agent = SimpleNamespace(leftovers=lambda: [])
_echo, _retract, _replay = Ui._echo, Ui._retract, Ui._replay
_turn_over, start_turn, _next_queued, take_back = Ui._turn_over, Ui.start_turn, Ui._next_queued, Ui.take_back
_queued_prompt = Ui._queued_prompt
flash, mark_read = Ui.flash, Ui.mark_read
def __init__(self, turn=None):
self.comp, self.turn = _QueueComp(), turn
self._queued, self._queued_echo, self._echoed = None, [], []
self._steered, self._queued_typed = [], ''
self._queued_text, self._typed, self._prompt, self._flash = [], '', None, None
self.buf = SimpleNamespace(text='', cursor=0)
self.notes, self.painted, self.attachments = [], False, []
def say(self, body, kind='reply', **kw):
block = type('Block', (), {'id': len(self.comp.blocks) + 1, 'body': str(body), 'tag': kind})()
self.comp.blocks[block.id] = block
return block
def note(self, text, kind='note'):
self.notes.append((text, kind))
return self.say(text, kind)
def paint(self): self.painted = True
u = _QueueUi()
u._echo('now')
now = _QueueWork()
assert u.start_turn(now)
test_eq([(b.body, b.tag) for b in u.comp.blocks.values()], [('now', 'user')])
u = _QueueUi('running')
u._echo('second')
second = _QueueWork()
assert not u.start_turn(second)
test_eq([(b.body, b.tag) for b in u.comp.blocks.values()],
[('queued · it runs when this turn ends · alt+up takes it back · ctrl+c stops this turn', 'note')])
u._echo('third')
third = _QueueWork()
assert not u.start_turn(third)
assert third.closed
assert all(b.body != 'third' for b in u.comp.blocks.values())
assert u._queued is second
u.turn = None
u._next_queued()
assert u.comp.started is second
assert u._queued is None
assert list(u.comp.blocks.values())[-1].body == 'second'
u = _QueueUi('running')
u._echo('/clear me')
u._typed = '/clear me' # what Enter took from the composer
clear = _QueueWork()
u.start_turn(clear)
assert u.take_back()
assert clear.closed and u._queued is None and u.buf.text == '/clear me', 'a queued command goes back as it was typed'
assert all(b.body != '/clear me' for b in u.comp.blocks.values())
u = _QueueUi('running')
u._echo('retry me')
retry = _QueueWork()
u.start_turn(retry)
u.turn = None
def fail_spawn(work, name): raise RuntimeError('busy')
u.comp.spawn = fail_spawn
u._next_queued()
assert u._queued is retry
assert all(b.body != 'retry me' for b in u.comp.blocks.values())
assert u.notes[-1] == ('the waiting message did not start: RuntimeError: busy', 'error')
```
``` python
class _SteerAgent:
def __init__(self, listening=True): self.listening, self.told, self.left = listening, [], []
def steer(self, text):
if self.listening: self.told.append(text)
return 'told' if self.listening else ''
def leftovers(self): out, self.left = self.left, []; return out
_QueueUi.steer = Ui.steer
u = _QueueUi('running'); u.agent = _SteerAgent()
u._echo('look at b')
assert u.steer('look at b')
test_eq(u.agent.told, ['look at b'])
test_eq([b.body for b in u.comp.blocks.values()], ['↪ look at b'])
test_eq(u._flash[0], 'sent · read after the current call')
assert not _QueueUi(None).steer('no turn of ours')
u = _QueueUi('running'); u.agent = _SteerAgent(listening=False)
assert not u.steer('nobody listening') and u.agent.told == []
u = _QueueUi('running'); u.agent = _SteerAgent()
held = _QueueWork()
u.start_turn(held, prompt='queued line')
u.agent.left, u.turn = ['unread steer'], None
u._next_queued()
assert held.closed and u._queued is None
test_eq(u.comp.started.cr_frame.f_locals['prompt'], 'unread steer\n\nqueued line')
u.comp.started.close()
```
------------------------------------------------------------------------
source
### Ui.place_pics
``` python
def place_pics():
```
*Put every picture back where its block sits, after each frame.*
------------------------------------------------------------------------
source
### Ui.on_media
``` python
def on_media(
paths
):
```
*Show pictures a tool wrote, mid-turn, from the model’s thread.*
------------------------------------------------------------------------
source
### Ui.show_media
``` python
def show_media(
media, session:str=''
):
```
*Save the pictures the model itself returned and show each one.*
------------------------------------------------------------------------
source
### Ui.show_pic
``` python
def show_pic(
path
):
```
*Draw one saved picture where possible, and name its path.*
``` python
class _Comp:
cols, rows, paused, _cursor, _screen, _spans = 80, 24, False, (0, 0), [], {}
blocks = {}
class tty:
wrote = []
@classmethod
def write(cls, s): cls.wrote.append(s)
class _Say:
def __init__(self): self.said, self.comp, self.agent, self.pics, self.drawn = [], _Comp(), None, {}, 0
def say(self, body, kind='reply', **kw):
self.said.append((str(body), kind))
return type('B', (), {'id': len(self.said)})
def paint(self): pass
show_pic, place_pics = Ui.show_pic, Ui.place_pics
u = _Say(); Ui.show_media(u, [])
test_eq(u.said, [])
import tempfile
with tempfile.TemporaryDirectory() as d:
u = _Say()
Ui.show_media(u, [{'mime': 'image/png', 'data': b'\x89PNG\r\n\x1a\nx'}], session=d)
assert u.said and (Path(d)/'media'/'image-1.png').exists()
test_eq([k for _, k in u.said], ['note'])
test_eq(u.pics, {})
from ramabana.testing import FullHost
h, away, cwd = FullHost(), Path(tempfile.mkdtemp()), os.getcwd()
u = _Say(); u.agent = type('A', (), {'host': h})()
os.chdir(away) # no session: the agent's host folder, never the working directory
try: Ui.show_media(u, [{'mime': 'image/png', 'data': b'\x89PNG\r\n\x1a\nx'}])
finally: os.chdir(cwd)
assert (h.root/'media'/'image-1.png').exists() and not list(away.iterdir())
```
There are two ways to change the approval mode, one for each direction.
`tighten_approve`, bound to `ctrl+g`, makes approvals one step stricter,
from `auto` to `edits` to `ask` to `off`, and never looser, so pressing
it by accident is harmless. Loosening takes a typed command:
`approve_mode` handles `/approve MODE`, which moves in either direction,
and `/approve` alone prints the current mode.
------------------------------------------------------------------------
source
### Ui.chip
``` python
def chip():
```
*Plan mode or the approval mode as one short token, or None when nothing
is gated.*
------------------------------------------------------------------------
source
### Ui.cycle_approve
``` python
def cycle_approve():
```
*Cycle ask, edits, auto, then plan mode, then ask again; from off, go to
ask.*
------------------------------------------------------------------------
source
### Ui.tighten_approve
``` python
def tighten_approve():
```
*Make approvals one step stricter, never looser.*
------------------------------------------------------------------------
source
### Ui.approve_mode
``` python
def approve_mode(
want:str=''
):
```
*`/approve [off|ask|edits|auto]`, or the current mode when `want` is
empty.*
`cycle_approve`, bound to `shift+tab`, goes round `ask`, `edits` and
`auto`, then into plan mode. In plan mode every tool that acts on the
world is withheld and the model proposes a plan instead. The next press
leaves plan mode and goes back to `ask`. `off` is not on the cycle; from
`off`, the next press goes to `ask`.
`chip` is the short label for the current mode, which always shows at
the left of the row under the status bar. Every one of these controls
also answers a question that was already waiting, so changing the mode
to `auto` approves the request in front of you.
``` python
from ramabana.agent import Approvals
from ramabana.tools import WRITE_TOOLS
class _Planner:
"Enough of an `Agent` for the approval controls: approvals, and plan mode."
plan_mode = False
def __init__(self, approvals): self.approvals = approvals
def set_plan_mode(self, on): self.plan_mode = on; return on
class _Holder:
"Enough of a `Ui` for the approval controls."
def __init__(self, approvals): self.agent = _Planner(approvals)
tighten_approve, approve_mode, cycle_approve, chip = Ui.tighten_approve, Ui.approve_mode, Ui.cycle_approve, Ui.chip
u = _Holder(Approvals(tools=WRITE_TOOLS, mode='auto'))
test_eq(u.tighten_approve(), 'approvals: auto -> edits')
test_eq(u.tighten_approve(), 'approvals: edits -> ask')
test_eq(u.tighten_approve(), 'approvals: ask -> off')
assert 'off' in u.tighten_approve()
test_eq(u.agent.approvals.mode, 'off')
test_eq(u.approve_mode(), 'approvals: off')
test_eq(u.approve_mode('ask'), 'approvals: off -> ask')
test_eq(u.approve_mode('yolo'), 'usage: /approve [off|ask|edits|auto]')
test_eq(u.agent.approvals.mode, 'ask')
test_eq(_Holder(None).tighten_approve(), 'this session runs without approvals')
said = [u.cycle_approve() for _ in range(3)]
test_eq(said[:2], ['approvals: ask -> edits', 'approvals: edits -> auto'])
assert said[2].startswith('plan mode') and u.agent.plan_mode and u.chip().plain == '⏸ plan'
test_eq(u.cycle_approve(), 'approvals: auto -> ask') # leaving plan mode re-enters at ask
assert not u.agent.plan_mode
u.agent.approvals.mode = 'off'
test_eq(u.chip().plain, '⊘ off')
test_eq(u.cycle_approve(), 'approvals: off -> ask')
test_eq(_Holder(None).chip(), None)
```
------------------------------------------------------------------------
source
### Ui.reply
``` python
def reply(
text
):
```
*A model reply as Markdown in this palette, prose compacted.*
------------------------------------------------------------------------
source
### compact_md
``` python
def compact_md(
text
):
```
*Markdown with blank rows between plain paragraphs made hard breaks.*
------------------------------------------------------------------------
source
### Reply
``` python
def Reply(
markup:str, code_theme:str='monokai', justify:JustifyMethod | None=None, style:str | Style='none',
hyperlinks:bool=True, inline_code_lexer:str | None=None, inline_code_theme:str | None=None
)->None:
```
*Rich’s `Markdown` with our code block.*
------------------------------------------------------------------------
source
### ThemedCode
``` python
def ThemedCode(
lexer_name:str, theme:str
)->None:
```
*A fenced block on the palette’s background: a diff by
[`diff_rich`](https://vedicreader.github.io/ramabana/cli.html#diff_rich),
known code by fastpylight, else pygments.*
------------------------------------------------------------------------
source
### fence_lang
``` python
def fence_lang(
name
):
```
*fastpylight’s name for a fence’s language: ‘diff’ for a patch, None
when it has no grammar.*
------------------------------------------------------------------------
source
### Ui.paste
``` python
def paste(
text
):
```
*Insert pasted text, or attach the files when the paste is nothing but
paths.*
------------------------------------------------------------------------
source
### Ui.on_key
``` python
def on_key(
k
):
```
*Give transcript navigation priority; Ctrl-P/Ctrl-N recall prompts.*
------------------------------------------------------------------------
source
### Ui.on_mouse
``` python
def on_mouse(
ev
):
```
*Route a mouse event to the browsing view, or here when `/mouse` is on.*
------------------------------------------------------------------------
source
### Ui.set_mouse
``` python
def set_mouse(
want:str=''
):
```
*Take or release the mouse on the main screen; off by default.*
------------------------------------------------------------------------
source
### Ui.leave_transcript
``` python
def leave_transcript():
```
*Close the browsing view and return the mouse unless `/mouse` is on.*
------------------------------------------------------------------------
source
### Ui.enter_transcript
``` python
def enter_transcript():
```
*Open the browsing view, borrowing the mouse while it is up.*
------------------------------------------------------------------------
source
### Ui.submit
``` python
def submit():
```
*Submit the line and remember it for Ctrl-P/Ctrl-N recall.*
------------------------------------------------------------------------
source
### Ui.remember
``` python
def remember(
line
):
```
*Keep `line` for Ctrl-P/Ctrl-N recall.*
------------------------------------------------------------------------
source
### Ui.recall
``` python
def recall(
step
):
```
*Move through submitted prompts, preserving the draft beyond the newest
entry.*
------------------------------------------------------------------------
source
### Ui.tail
``` python
def tail():
```
*The live-tail description shared with Teleprint’s transcript view. The
plan checklist heads it rather than covering the transcript.*
------------------------------------------------------------------------
source
### Ui.model_row
``` python
def model_row():
```
*The row under the bar: the next turn’s model, from the routing table.*
### Python mode
Python mode gives the same prompt a second meaning. `/python` starts a
Jupyter kernel that belongs to you, and from then on a line you type
runs as Python in that kernel. `/agent` hands the line back to the model
and leaves the kernel running; only `/quit` or the end of the process
shuts it down. Everything else stays the same: the transcript, folding,
approvals, attachments and slash commands. Only what `enter` does with
the line changes.
The agent follows you into the kernel. `enter_python` points the agent’s
Python tools at a Dhrishti overlay on the new kernel. An overlay is a
layer over your namespace: the agent can read every name you defined and
writes its own names into its layer, and it cannot rebind a name you
made. The host stays the same object. `enter_python` then calls
[`Agent.refresh`](https://vedicreader.github.io/ramabana/agent.html#agent.refresh),
which drops the cached skills, registry and tools and gives the running
model backend a fresh briefing with the new tools.
Related methods: `run_code` runs one cell you typed and prints its
outputs as blocks, `complete_python` handles tab completion from the
kernel, `attach_session` points the agent at someone else’s live
Dhrishti session for `--attach`, and `enable_agent_proxy` makes this
session’s agent callable from the Python prompt for `--agent-proxy`.
`/promote NAME` copies one of the agent’s variables into your namespace.
[pyrepl](11_pyrepl.ipynb) holds the parts that judge Python: the kernel,
the overlay host, whether a line compiles, and how to colour it. It is
imported only when Python mode starts, so a plain `ramabana` session
needs no kernel. The tests that need a live kernel are on that page.
------------------------------------------------------------------------
source
### Ui.complete_python
``` python
async def complete_python():
```
*Tab: complete as far as the candidates agree, and leave the menu up to
cycle.*
------------------------------------------------------------------------
source
### Ui.on_output
``` python
def on_output(
output
):
```
*Print one nbformat output as the block it means.*
------------------------------------------------------------------------
source
### Ui.run_code
``` python
async def run_code(
code
):
```
*Run one owner-typed Python cell, streaming its outputs into blocks.*
------------------------------------------------------------------------
source
### Ui.attach_session
``` python
async def attach_session(
name
):
```
*Point the agent’s Python at someone else’s dhrishti session.*
------------------------------------------------------------------------
source
### Ui.enable_agent_proxy
``` python
async def enable_agent_proxy():
```
*Bind this session’s agent at the Python prompt and in the agent’s
layer.*
------------------------------------------------------------------------
source
### Ui.proxy_label
``` python
def proxy_label():
```
*A kernel-unique name for this session’s agent.*
------------------------------------------------------------------------
source
### Ui.enter_python
``` python
async def enter_python():
```
*Take the line for Python, starting the kernel once and pointing the
agent at it.*
------------------------------------------------------------------------
source
### Ui.use_host
``` python
def use_host(
host
):
```
*Move the agent onto `host`, re-briefing the running turn backend with
its tools.*
------------------------------------------------------------------------
source
### Ui.log_cell
``` python
def log_cell(
source, outputs:NoneType=None, cell_type:str='code'
):
```
*Write one cell to the session notebook, when the host keeps one.*
``` python
import ramabana.pyrepl as _pyrepl
from ramabana.testing import MemHost
from ramabana.tools import SessionHost, implemented
@implemented
class _Mem(MemHost, SessionHost):
"A host recording what the agent runs, with the session tool group."
def __init__(self, *a, **kw):
super().__init__()
self.ran, self.web, self.read_outside, self.kernel = [], False, False, None
def run_python(self, code): self.ran.append(code); return '(ok)'
def inspect_python(self, code, scope='isolated'): return '(ok)'
def list_vars(self): return 'mem_var: int = 1'
def restart_kernel(self): return 'restarted'
def terminal_text(self, n=1): return ''
class _FakeKernel:
base = 'http://127.0.0.1:65000'
def __init__(self, cwd='.'): self.cwd = cwd
async def start(self): return self
def _run_python(a): return next(t for t in a.tools if t.__name__ == 'run_python')
def _said(u, c): return u.transcript.block_text(list(c.blocks.values())[-1])
swap_tty = EmuTty(72, 14)
swap_comp = Compositor(swap_tty)
swap_comp._register_signals = lambda: None
await swap_comp.start()
swap_agent, _ = fake_agent(host=_Mem(), approvals=Approvals(tools=WRITE_TOOLS, mode='ask'))
swap_ui = Ui(swap_comp, swap_agent)
_run_python(swap_agent)('before = 1')
test_eq(swap_agent.host.ran, ['before = 1'])
host_before, tools_before = swap_agent.host, swap_agent.tools
_realk, PYREPL_MODULES = _pyrepl.Kernel, ()
_pyrepl.Kernel = _FakeKernel
try: await swap_ui.enter_python()
finally: _pyrepl.Kernel = _realk
test_eq((swap_ui.mode, swap_ui.kernel.base), ('python', _FakeKernel.base))
assert swap_agent.host is host_before
test_eq(swap_agent.host.kernel.base, _FakeKernel.base)
assert swap_agent.approvals.host is swap_agent.host
assert swap_agent.tools is not tools_before
test_eq({t.__name__ for t in tools_before} - {t.__name__ for t in swap_agent.tools}, set())
swap_ui.kernel, swap_ui.mode, swap_agent.host.kernel = None, 'agent', None
PYREPL_MODULES = ('no_such_module',)
try: await swap_ui.enter_python()
finally: PYREPL_MODULES = ()
test_eq((swap_ui.mode, swap_ui.kernel), ('agent', None))
assert 'no_such_module' in _said(swap_ui, swap_comp) and 'ramabana[python]' in _said(swap_ui, swap_comp)
_realf = _pyrepl.find_session
_pyrepl.find_session = lambda name: f'http://127.0.0.1:9{name}'
try: test_eq(await swap_ui.attach_session('001'), 'http://127.0.0.1:9001')
finally: _pyrepl.find_session = _realf
test_eq((swap_ui.kernel, swap_ui.mode, swap_ui.attached), (None, 'agent', 'http://127.0.0.1:9001'))
assert swap_agent.host is host_before
test_eq(swap_agent.host.kernel.base, 'http://127.0.0.1:9001')
_pyrepl.Kernel = lambda root: _FakeKernel()
try: await swap_ui.enter_python()
finally: _pyrepl.Kernel = _realk
test_eq((swap_ui.attached, swap_ui.mode), ('', 'python'))
assert swap_ui.kernel is not None
assert any('left the attached session' in swap_ui.transcript.block_text(b)
for b in swap_comp.blocks.values())
swap_comp.stop()
```
### Shell lines
A line that starts with `!` runs as a shell command in the first open
folder, with no approval question, because you typed it yourself.
`run_shell_line` runs it. Its output prints as one block, followed by
the exit code, and a non-zero exit code prints as an error. The model
never sees the command or its output.
So that the model knows this exists,
[`main`](https://vedicreader.github.io/ramabana/cli.html#main) adds
`SHELL_NOTE` to an interactive session’s `extra_sp`, the extra text
appended to the agent’s briefing, before a resumed session builds its
briefing. The note tells the model it can ask you to run a command with
`!`. A one-turn run from the command line never gets the note. `!` lines
work only in agent mode; in Python mode the line is Python, and
[`code_state`](https://vedicreader.github.io/ramabana/pyrepl.html#code_state)
rejects it because it does not compile.
The command runs in its own process group with stdin from `/dev/null`,
so a command that tries to read from the terminal gets end of file
instead of hanging. `Ui.shell_proc` holds the process while it runs,
which is how `ctrl+c` reaches it (see `stop` further down).
------------------------------------------------------------------------
source
### Ui.run_shell_line
``` python
async def run_shell_line(
command
):
```
*Run one owner-typed `!` line in the first open folder, printing its
output and exit code.*
``` python
sh_tty = EmuTty(72, 14)
sh_comp = Compositor(sh_tty)
sh_comp._register_signals = lambda: None
await sh_comp.start()
sh_ui = Ui(sh_comp, fake_agent(host=FullHost())[0])
sh_ui.buf.insert('!echo made > out.txt; cat out.txt; exit 3')
sh_ui.start_turn(sh_ui.submit())
await asyncio.wait_for(sh_ui.turn, 5)
sh_said = [sh_ui.transcript.block_text(b) for b in sh_comp.blocks.values()]
test_eq(sh_ui.turn, None)
assert (Path(sh_ui.agent.host.roots[0])/'out.txt').exists()
assert any('made' in s for s in sh_said) and any('exit 3' in s for s in sh_said)
sh_said
```
Replies are written in Markdown and drawn with Rich. `MARKDOWN_THEME`
gives headings, inline code, links and lists their own colours in the
active palette, instead of drawing everything in the terminal’s default
colour. The text the model wrote is kept as plain Markdown; only the
drawing adds colour. Rich markup inside a reply, such as `[red]`, is
shown as literal text, so a reply cannot inject terminal styling.
[`Ui.reply`](https://vedicreader.github.io/ramabana/cli.html#ui.reply)
draws a reply. It uses
[`Reply`](https://vedicreader.github.io/ramabana/cli.html#reply), Rich’s
`Markdown` with code blocks drawn by
[`ThemedCode`](https://vedicreader.github.io/ramabana/cli.html#themedcode),
and
[`compact_md`](https://vedicreader.github.io/ramabana/cli.html#compact_md),
which joins plain paragraphs with single line breaks to save rows.
[`fence_lang`](https://vedicreader.github.io/ramabana/cli.html#fence_lang)
maps a code fence’s language to the highlighter’s name for it.
Here is the options row from the turn section in use.
[`options_for`](https://vedicreader.github.io/ramabana/cli.html#options_for)
decided that “refactor the runtime” earns a menu,
[`ChoiceMenu.choose`](https://vedicreader.github.io/ramabana/cli.html#choicemenu.choose)
takes one keystroke at a time and returns `(done, option)`, and
[`Ui`](https://vedicreader.github.io/ramabana/cli.html#ui) decides what
the chosen option means for the prompt. Below, `down` moves the cursor
and `enter` picks “plan”.
``` python
menu = ChoiceMenu(*options_for('refactor the runtime'))
test_eq(menu.choose('down'), (False, None))
test_eq(menu.choose('enter')[1].label, 'plan')
test_eq(ChoiceMenu('x', REFACTOR).choose('s')[1].label, 'scope')
test_eq(ChoiceMenu('x', REFACTOR).choose('escape'), (True, None))
test_eq(options_for('what does the runtime do?'), None)
print('\n'.join(r.plain for r in menu.render()))
```
``` python
from rich.console import Console
color_console = Console(width=72, theme=MARKDOWN_THEME, color_system='truecolor')
styled = Markdown('# Heading\n\nUse `search_code` and **verify** it.\n\n```python\nanswer = 42\n```',
code_theme='github-dark')
lines = color_console.render_lines(styled, pad=False)
colors = {seg.style.color.triplet for line in lines for seg in line
if seg.style and seg.style.color and seg.style.color.triplet}
inline = next(seg for line in lines for seg in line if seg.text == 'search_code')
test_eq(inline.style.bgcolor, None)
test_eq(len(colors) >= 3, True)
```
The rest of the examples on this page build a complete surface in the
notebook: an emulated terminal, a real compositor, and an agent from
[`fake_agent`](https://vedicreader.github.io/ramabana/testing.html#fake_agent)
whose model replies are scripted. Typing “refactor the runtime” and
pressing enter opens the options row on screen.
``` python
tty = EmuTty(72, 14)
comp = Compositor(tty)
comp._register_signals = lambda: None # nbdev executes this async cell on a worker thread
await comp.start()
agent, be = fake_agent(replies=['`threshold` is in `ramabana/runtime.py`, and it caps the reserve.'], local_multimodal=True)
ui = Ui(comp, agent)
comp.on_key = ui.on_key
ui.paint()
print(tty.term.text())
```
``` python
ui.buf.insert('refactor the runtime')
test_eq(ui.submit(), None)
ui.paint()
assert 'a apply' in tty.term.text() and 'p plan' in tty.term.text()
test_eq(ui.menu.choose('escape'), (True, None))
ui.menu, ui.menu_prompt = None, ''
ui.buf.clear()
ui.paint()
```
Cancelling the menu, with `escape` or the `cancel` option, puts the line
back in the composer to edit instead of throwing it away.
``` python
ui.buf.insert('restructure the compactor')
ui.submit()
ui.on_key(Key('escape'))
test_eq(ui.buf.text, 'restructure the compactor')
ui.buf.clear()
ui.paint()
```
This is the resting state. The status bar shows the version, the model
and how full the context is, and the cursor sits on the input line.
Everything above the last two rows is the transcript.
``` python
test_eq(tty.term.text().splitlines()[-1], '▌')
[l for l in tty.term.text().splitlines() if 'tools' in l]
```
Typing goes through the real key parser: the example feeds the bytes a
terminal would send, not ready-made `Key` objects. Writing
`/search_code threshold` inside a prompt names a tool, which runs before
the turn starts.
``` python
comp.on_bytes(b'where is the threshold? /search_code threshold') # a named tool runs ahead of the turn
tty.term.text().splitlines()[-1]
```
While a turn runs, a working marker in the status bar moves on a timer
of its own, so it keeps moving during a silent wait on the network. With
no turn running it shows a steady ready dot.
``` python
resting_status = ui.status().plain
assert VERSION in resting_status
ui.turn = object()
first = ui.status().plain
ui.frame += 1
second = ui.status().plain
ui.turn = None
test_eq('● ' in resting_status and 'WORKING' not in resting_status, True)
test_eq('⠋ working' in first, True)
test_eq('⠙ working' in second, True)
```
Pressing enter prints the prompt as a `user` block and starts the turn.
The short `await` gives the model thread time to run. In the real app
the event loop does that waiting, so nothing blocks the screen.
``` python
comp.on_bytes(b'\r')
await asyncio.sleep(0.4)
print(tty.term.text())
```
``` python
ui.buf.text, ui.buf.cursor = 'x' * 80, 80
ui.paint()
test_eq(comp._cursor[1], len(ui.buf.text) % tty.size[0])
test_eq(0 <= comp._cursor[0] < comp.rows, True)
ui.buf.text, ui.buf.cursor = 'top\nbottom', len('top\nbottom')
ui.paint()
test_eq(comp._cursor[1], len('bottom'))
test_eq(0 <= comp._cursor[0] < comp.rows, True)
ui.buf.clear()
ui.paint()
```
`ctrl+p` and `ctrl+n` walk back and forward through the prompts you have
sent, and keep whatever you had half typed. `remember` stores each
submitted line and `recall` moves through them. With an empty composer,
`up` or `down` opens teleprint’s transcript view instead.
``` python
ui.buf.insert('first prompt')
ui.submit()
ui.buf.insert('second prompt')
ui.submit()
ui.buf.insert('unfinished draft')
ui.on_key(Key('ctrl+p'))
test_eq(ui.buf.text, 'second prompt')
ui.on_key(Key('ctrl+p'))
test_eq(ui.buf.text, 'first prompt')
ui.on_key(Key('ctrl+n'))
ui.on_key(Key('ctrl+n'))
test_eq(ui.buf.text, 'unfinished draft')
ui.buf.clear()
ui.on_key(Key('up'))
test_eq(ui.transcript.active, True)
ui.on_key(Key('g'))
test_eq(ui.transcript.top, 0)
ui.on_key(Key('escape'))
test_eq(ui.transcript.active, False)
```
The transcript now holds the prompt, the tool calls the agent made, and
the reply, each as a block with its own tag and height.
``` python
[(b.tag, b.height) for b in comp.blocks.values()]
```
``` python
test_eq(agent.calls[0][0], 'search_code')
test_eq(any('threshold' in str(b.body) for b in comp.blocks.values()), True)
be.sent and len(be.sent)
```
## Folding
A block taller than `FOLD` (12 rows) starts folded, so a
four-hundred-line result takes one row until you open it. `ctrl+o` opens
all the working of the latest turn, every step and call, and a second
press folds it again. `alt+o` does the same for every turn, and `alt+1`
to `alt+9` open a single entry, newest first. The main screen leaves the
mouse to the terminal, so selecting and copying work as in any
scrollback. In the transcript view, a block opens and closes under the
block cursor or with a click.
``` python
long = ui.say(Text('\n'.join(f'line {i}' for i in range(40))), 'tool')
long.collapsed, long.height
```
``` python
comp.toggle(long)
test_eq(long.collapsed, False)
comp.toggle(long)
long.collapsed
```
A one-line block has nothing to fold, so toggling it does nothing and
raises no error.
``` python
short = ui.say(Text('one line'), 'tool')
comp.toggle(short), short.collapsed
```
A tool call folds to its own summary line rather than at `FOLD` rows,
because the summary is what you want to read. A turn of thirty calls
reads as thirty lines, and any one result is a keystroke away.
The fold is decided when the result arrives. A tool block is printed
when the call starts, one hourglass row tall, and the result is put into
the same block when the call returns. Teleprint’s `print_block` decides
folding once, from the height at print time, which would be 1, so on its
own it would leave every result open. `_act` therefore decides the fold
again at the moment the result lands.
``` python
from ramabana.agent import Act
```
``` python
step = Act(tool='view_file', args={'path': 'runtime.py'}, summary='view runtime.py')
ui.on_act(step)
test_eq((ui.acts[step.id].height, ui.acts[step.id].collapsed), (1, False))
ui.on_act(step.finish('\n'.join(f'line {i}' for i in range(400))))
folded = ui.acts[step.id]
test_eq((folded.height, folded.collapsed), (401, True))
test_eq(len(comp._block_rows(folded)), 1)
assert step.summary in ui.transcript.block_text(folded)
comp.toggle(folded)
test_eq((folded.collapsed, len(comp._block_rows(folded))), (False, 401))
comp.toggle(folded)
```
The agent’s activity feed records each tool call as an
[`Act`](https://vedicreader.github.io/ramabana/agent.html#act). Text the
model streams before a call is narration, so when a call starts,
`_close_seg` closes that text as a `step` block above the call. The
finished timeline stays on screen, and the reply block holds only the
text after the last call.
While the reply is still arriving its block is tagged `reply`. `/copy`
copies the final answer; the tool calls are the foldable trace above it.
``` python
tl_tty = EmuTty(72, 20)
tl_comp = Compositor(tl_tty)
tl_comp._register_signals = lambda: None
await tl_comp.start()
tl_agent, _ = fake_agent()
tl_ui = Ui(tl_comp, tl_agent)
tl_acts = tl_agent.activity
seg = tl_ui.stream(None, 'Looking for it.\n\nTwo places, probably.\n')
call = tl_acts.start('search_code', {'query': 'threshold'})
tl_acts.finish(call, 'runtime.py:88\nagent.py:12')
seg = tl_ui.stream(seg, '## Answer\n\nBoth of them.\n')
tl_ui.flush_stream()
tl_ui._seg_blk = None
timeline = list(tl_comp.blocks.values())
test_eq([b.tag for b in timeline], ['step', 'tool', 'reply'])
test_eq([b.collapsed for b in timeline], [False, True, False])
test_eq(tl_ui.transcript.block_text(timeline[0]), 'Looking for it.\n\nTwo places, probably.')
test_eq(len(tl_comp._block_rows(timeline[1])), 1)
test_eq(tl_ui.transcript.block_text(timeline[-1]), '''## Answer
Both of them.
''')
status_tty = EmuTty(72, 10)
status_comp = Compositor(status_tty)
status_comp._register_signals = lambda: None
await status_comp.start()
status_agent, _ = fake_agent()
status_ui = Ui(status_comp, status_agent)
status_seg = status_ui.stream(None, 'Waiting for approval.\n')
status_ui.note('approved')
status_ui.stream(status_seg, 'Done.\n')
status_blocks = list(status_comp.blocks.values())
test_eq([b.tag for b in status_blocks], ['note', 'reply'])
test_eq(status_ui.transcript.block_text(status_blocks[-1]), 'Waiting for approval.\nDone.\n')
```
``` python
answer = timeline[-1]
test_eq(tl_ui.transcript.block_text(answer), '## Answer\n\nBoth of them.\n')
assert 'copied' in tl_ui.copy_last('reply')
test_eq(tl_ui.fold_work(), False)
test_eq([b.collapsed for b in timeline], [False, False, False])
test_eq(tl_ui.fold_work(), True)
test_eq([b.collapsed for b in timeline], [False, True, False]) # narration is not part of the working
class _FakePlan:
def __init__(self): self.n = 0
def md(self):
rows = [f"{'[x]' if i < self.n else '[ ]'} step {i+1}" for i in range(3)]
return f"**Repair** \u00b7 {self.n}/3 done\n" + '\n'.join(rows)
plan_tty = EmuTty(72, 20)
plan_comp = Compositor(plan_tty)
plan_comp._register_signals = lambda: None
await plan_comp.start()
plan_agent, _ = fake_agent()
plan_ui = Ui(plan_comp, plan_agent)
plan_ui._plan_blk = None
_p, _blk = _FakePlan(), None
for _i in range(3):
_p.n = _i
plan_ui.on_plan(_p)
_blk = plan_ui.stream(_blk, f'step {_i+1}\n')
plan_ui.flush_stream()
_plans = [b for b in plan_comp.blocks.values() if b.tag == 'plan']
test_eq(len(_plans), 1)
assert '2/3 done' in plan_ui.transcript.block_text(_plans[0])
_md = ('**Repair** \u00b7 1/2 done\n'
'[x] `aaa` did the thing -- with a note\n'
'[ ] `bbb` still to do')
_t = plan_text(_md)
test_eq(_t.plain.splitlines()[1], '[x] aaa did the thing -- with a note')
_styles = {_md_line: None for _md_line in ()}
_mark = [sp for sp in _t.spans if _t.plain[sp.start:sp.end] in ('[x]', '[ ]')]
test_eq(len(_mark), 2)
test_eq(len({sp.style for sp in _mark}), 1)
_step = next(sp for sp in _t.spans if _t.plain[sp.start:sp.end].strip() == 'did the thing')
assert _step.style != _mark[0].style
set_theme('dark'); _d = plan_text('[x] `a` x').spans[0].style
set_theme('light'); _l = plan_text('[x] `a` x').spans[0].style
assert _d != _l
set_theme('dark')
```
``` python
tl_ui._turn_at, tl_ui.turn = time.monotonic(), 'a turn'
footer = [r.plain for r in tl_ui.working()]
test_eq(footer[0], f'1 {call.line()}')
test_eq(tl_ui.drillable(), [tl_ui.acts[call.id]])
assert footer[-1].startswith(' step 1 ·'), footer
tl_ui.turn = None
test_eq(tl_ui.working(), [])
```
A delegation, where the agent hands a question to a sub-agent, shows as
one entry with the sub-agent’s calls grouped under it. A sub-agent is a
second agent the main one starts for a piece of work, here through
`delegate_search`. It runs on the same recorded tool wrappers as its
caller, so its calls arrive in the same
[`Activity`](https://vedicreader.github.io/ramabana/agent.html#activity)
feed. Each of them carries a `parent_action_id`, which
[`Agent`](https://vedicreader.github.io/ramabana/agent.html#agent) sets
from a per-thread stack of the delegate calls in flight, and
[`Ui`](https://vedicreader.github.io/ramabana/cli.html#ui) uses it to
put the call under its parent. Without the grouping, the searches of
three sub-agents would appear among the caller’s own calls.
A finished call folds to its summary, unless it is a write or a diff
(see [`opens`](https://vedicreader.github.io/ramabana/cli.html#opens)
above). A failed call never folds, and nor does a running one, so
folding never hides an error or progress. `drillable` lists the turn’s
foldable calls, newest first, and `drill(n)` opens or folds the `n`th of
them, which is what `alt+1` to `alt+9` do.
``` python
parent = tl_acts.start('delegate_search', {'question': 'who reads the threshold?'})
for q in ('threshold callers', 'reserve usage'):
tl_acts.finish(tl_acts.start('search_code', {'query': q}, parent_action_id=parent.id), 'hits')
tl_acts.finish(parent, 'Three callers, all in runtime.py.')
group = tl_ui.acts[parent.id]
test_eq(len(tl_comp.blocks), 4)
test_eq((group.collapsed, len(tl_comp._block_rows(group))), (True, 1))
assert '2 calls' in tl_comp._ansi(tl_comp._block_rows(group)[0][1])
assert all(k.line() in tl_ui.transcript.block_text(group) for k in tl_ui.kids[parent.id])
failed = tl_acts.start('run_shell', {'command': 'pytest'})
tl_acts.finish(failed, 'FAILED test_reserve\nassert 0.9 <= 0.8', ok=False)
test_eq(tl_ui.acts[failed.id].collapsed, False)
```
``` python
test_eq([b.tag for b in tl_ui.drillable()], ['tool', 'tool', 'tool'])
test_eq(tl_ui.drill(2), True)
test_eq(group.collapsed, False)
test_eq(tl_ui.drill(2), True)
test_eq((group.collapsed, tl_ui.drill(99)), (True, False))
assert 'copied' in tl_ui.copy_last('turn')
test_eq(tl_ui._reply, '## Answer\n\nBoth of them.\n')
test_eq(tl_ui.mouse, False)
assert 'mouse on' in tl_ui.set_mouse('on') and tl_ui.mouse
tl_ui.leave_transcript()
test_eq(tl_ui.mouse, True)
assert 'mouse off' in tl_ui.set_mouse('off')
```
## Copying and browsing
There are two ways to copy out of the transcript, and both use the
system clipboard. On the main screen the mouse belongs to the terminal,
so you select and copy as in any scrollback. The transcript view, opened
with `ctrl+r` or `up`, takes the mouse only while it is open
(`enter_transcript` and `leave_transcript`). `/copy` and the view’s `y`
key send the text over OSC 52, a terminal escape code for setting the
clipboard, which works through ssh and tmux.
They copy a block’s source text rather than what is drawn: the Markdown
the model wrote, or the arguments a tool got. Text taken off the screen
would come back wrapped at the terminal’s width, indented by the gutter,
and missing the code fences, so every block records its source when it
is printed. `transcript.block_text` returns it, as the examples show.
The last three examples attach files: `attach` adds a picture once and
refuses a duplicate, `paste` turns a pasted path into an attachment, and
`submit` sends the attachments with the prompt and clears them.
``` python
md = 'Here is the fix.\n\n```python\nthreshold = 42\n```\n\nCall it from `runtime.py`.'
rblk = ui.stream(None, md[:20])
rblk = ui.stream(rblk, md[20:])
test_eq(ui.transcript.block_text(rblk), md)
rendered = '\n'.join(''.join(s.text for s in l) for l in comp._content_lines(rblk))
assert '```' not in rendered
assert '```python' in ui.transcript.block_text(rblk)
```
``` python
u = ui.say(Text('where is the threshold?'), 'user')
test_eq(ui.transcript.block_text(u), 'where is the threshold?')
wrapped = ui.say('x' * 200, 'note')
test_eq(ui.transcript.block_text(wrapped), 'x' * 200)
test_eq(wrapped.height > 1, True)
ui.copy_last('reply')
```
``` python
ui.enter_transcript()
test_eq((ui.transcript.active, ui.transcript.follow), (True, True))
ui.say(Text('search_code("threshold") landed while browsing'), 'tool')
assert 'landed while browsing' in tty.term.text()
sblk = ui.stream(None, 'and a reply streamed ')
sblk = ui.stream(sblk, 'in behind it')
ui.touch(now=True)
assert 'in behind it' in tty.term.text()
ui.transcript.move(-1)
test_eq(ui.transcript.follow, False)
ui.leave_transcript()
test_eq((ui.transcript.active, comp.paused), (False, False))
```
``` python
test_eq(ui.attach(shot), 'attached image shot.png (72B)')
test_eq(ui.attach(shot), 'shot.png is already attached')
ui.attach(snd_path)
test_eq([a.name for a in ui.attachments], ['shot.png', 'note.wav'])
assert 'shot.png' in ui.attach_row().plain and '/detach' in ui.attach_row().plain
test_eq(ui.detach('2'), 'dropped note.wav')
test_eq(ui.detach(), 'dropped 1 attachment')
test_eq(ui.detach(), 'nothing is attached')
test_eq(ui.attach(media_dir / 'gone.png'), 'cannot attach gone.png: no such file')
test_eq(ui.attach_row(), None)
```
``` python
ui.paste(str(shot))
test_eq([a.name for a in ui.attachments], ['shot.png'])
test_eq(ui.buf.text, '')
ui.paste('and this is ordinary text')
test_eq(ui.buf.text, 'and this is ordinary text')
long_text = 'copied text ' * 100 + 'image.png'
ui.paste(long_text)
test_eq(ui.buf.text, 'and this is ordinary text' + long_text)
ui.buf.clear()
```
``` python
ui.buf.insert('what is in this picture?')
await ui.submit()
test_eq(ui.attachments, [])
```
## Approvals in a terminal
An approval is the question Ramabana asks before a tool with an effect
runs: a file write, a shell command, a git commit and so on. The
approval gate,
[`Approvals`](https://vedicreader.github.io/ramabana/agent.html#approvals),
holds the model’s worker thread until a person answers. The request
arrives on that thread and the answer comes from the event loop thread.
While a question waits, the input line is where you answer it: `y`, `n`,
`a` for the rest of the session, or `n:` followed by a reason.
Here a write is requested from another thread, the way a tool call does
it:
``` python
tty2 = EmuTty(72, 12)
comp2 = Compositor(tty2)
comp2._register_signals = lambda: None
await comp2.start()
gated, _ = fake_agent(replies=['done'])
gated.approvals = Approvals(tools=WRITE_TOOLS, host=gated.host)
ui2 = Ui(comp2, gated)
comp2.on_key = ui2.on_key
ui2.paint()
threading.Thread(target=lambda: gated.approvals.request(
'create_file', {'path': '/proj/notes.md', 'text': '# notes\n'}), daemon=True).start()
await asyncio.sleep(0.2)
print(tty2.term.text())
```
The question block shows what you need to decide: the path, whether the
write overwrites a file, and the start of what would be written. The
input line now holds the approval question.
``` python
test_eq(ui2.ask.tool, 'create_file')
tty2.term.text().splitlines()[-1]
```
[`parse_answer`](https://vedicreader.github.io/ramabana/cli.html#parse_answer)
reads a typed answer. A line that is not an answer, such as a prompt
typed at the wrong moment, does not count as a refusal. The question is
shown again with a hint and keeps waiting. A waiting question also rings
and shows itself again every `REASK_EVERY` seconds.
``` python
test_eq([parse_answer(t) for t in ('y', 'YES', 'n', 'no: too broad', 'n:', 'continue.', 'y: and add a test', 'approve', 'deny', 'ok')],
[(True, ''), (True, ''), (False, ''), (False, 'too broad'), (False, ''), None, (True, 'and add a test'), (True, ''), (False, ''), None])
ui2.buf.insert('continue.')
ui2.on_key(Key('enter'))
assert ui2.ask is not None and ui2.ask.pending
assert 'not an answer' in tty2.term.text()
test_eq(ui2.buf.text, '')
rang = []
_write = tty2.write
tty2.write = lambda s: rang.append(s) if s == '\a' else _write(s)
ui2._asked_at -= REASK_EVERY; ui2._key_at -= BELL_IDLE + 1
test_eq(ui2.nudge(), True)
test_eq(rang, ['\a'])
test_eq(ui2.nudge(), False)
ui2.buf.insert('n: beca')
ui2._asked_at -= REASK_EVERY; ui2._key_at = time.monotonic()
test_eq(ui2.nudge(), False)
test_eq((ui2.buf.text, rang), ('n: beca', ['\a']))
ui2._asked_at -= REASK_EVERY; ui2._key_at -= BELL_IDLE + 1
test_eq(ui2.nudge(), False)
test_eq((ui2.buf.text, rang), ('n: beca', ['\a', '\a']))
ui2.buf.clear()
tty2.write = _write
```
The question takes the input line in any mode, including Python mode,
where it is asked below. A reason is a sentence, and running it as
Python would leave the gate unanswered.
`y` and `a` answer at once when nothing has been typed. `n` waits for
enter, so that a reason can follow it: `n` and enter refuses, and
`n: put it in docs/ instead` refuses with that reason, which gives the
model something to act on instead of the word “denied”.
``` python
ui2.mode = 'python'
comp2.on_bytes(b'n: put it in docs/ instead')
tty2.term.text().splitlines()[-1]
```
``` python
comp2.on_bytes(b'\r')
await asyncio.sleep(0.2)
ui2.mode = 'agent'; ui2.paint()
ui2.ask, gated.approvals.history[-1].reply()
```
There are two histories, kept apart. The composer keeps your past
prompts, and teleprint’s blocks are the history of the transcript. `up`
and `down` recall a past prompt only when the cursor cannot move further
inside a prompt of several lines. `ctrl+r` opens teleprint’s transcript
view, where you can search old blocks, open them and copy them.
``` python
test_eq(bool(gated.approvals.history[-1]), False)
test_eq(tty2.term.text().splitlines()[-1], '▌')
[b.tag for b in comp2.blocks.values()][-2:]
```
`a` approves the current request and sets the approval mode to `auto`
for the rest of the session.
``` python
threading.Thread(target=lambda: gated.approvals.request('create_file', {'path': '/proj/b.md'}), daemon=True).start()
await asyncio.sleep(0.2)
comp2.on_bytes(b'a')
await asyncio.sleep(0.2)
gated.approvals.mode, gated.approvals.history[-1].note
```
## Commands
Slash commands such as `/model`, `/tools` and `/cost` are answered by
the agent, not by the terminal, so each one has a single implementation
that every frontend shares. `/help` is answered here, because the keys
it describes belong to the terminal. An unknown command prints an error
block.
``` python
ui.buf.insert('/tools')
ui.submit()
[l for l in str(comp.blocks[max(comp.blocks)].body[0]).splitlines()[:4]]
```
``` python
ui.buf.insert('/nope')
test_eq(ui.submit(), None)
comp.blocks[max(comp.blocks)].tag
```
Commands are recognised before the options row, so `/tools refactor`
runs the command and opens no menu.
``` python
ui.buf.insert('/tools refactor')
test_eq(ui.submit(), None)
test_eq(ui.menu, None)
comp.blocks[max(comp.blocks)].tag
```
An empty line does nothing; no empty turn goes to the model.
``` python
test_eq(ui.submit(), None)
ui.buf.text
```
## Running it
[`mk_agent`](https://vedicreader.github.io/ramabana/agent.html#mk_agent)
from `ramabana.agent` builds what the terminal runs. It makes a
[`WorkspaceHost`](https://vedicreader.github.io/ramabana/vault.html#workspacehost)
over the folders named on the command line, an
[`Agent`](https://vedicreader.github.io/ramabana/agent.html#agent) over
that host, and an approval gate connected to both. A host is the object
that supplies the agent’s tools and decides what they may touch: which
folders, whether the network is reachable, and which writes need
approval.
[`mk_host`](https://vedicreader.github.io/ramabana/agent.html#mk_host)
builds the host alone.
[`amain`](https://vedicreader.github.io/ramabana/cli.html#amain) is the
terminal loop that runs the
[`Ui`](https://vedicreader.github.io/ramabana/cli.html#ui), and it stays
short because the behaviour lives in
[`Ui`](https://vedicreader.github.io/ramabana/cli.html#ui).
The prompt appears before the model is ready. `begin` builds the system
prompt and tools and starts the model on a thread, and the status bar
says `starting model…` meanwhile. A line sent before the model is up
waits in the queue, like a line sent during a turn, and runs once the
model is ready.
[`start_agent`](https://vedicreader.github.io/ramabana/cli.html#start_agent)
starts the turn model and then the code index, the search index behind
`search_code`, so the index never delays the model.
[`sync_index`](https://vedicreader.github.io/ramabana/cli.html#sync_index)
decides whether the index needs rebuilding: it rebuilds when a package
or an open folder changed since the last good build, otherwise reopens
the saved index, and logs which it did to `/kosha.log`. `warm_up`
runs
[`start_agent`](https://vedicreader.github.io/ramabana/cli.html#start_agent)
off the loop with
[`off_loop`](https://vedicreader.github.io/ramabana/cli.html#off_loop),
and prints a note saying why when no model starts.
------------------------------------------------------------------------
source
### Ui.begin
``` python
def begin():
```
*Start the model behind the prompt; lines sent meanwhile wait in the
queue.*
------------------------------------------------------------------------
source
### Ui.warm_up
``` python
async def warm_up():
```
*Start the model off the loop; a start that fails says why in a note.*
------------------------------------------------------------------------
source
### off_loop
``` python
async def off_loop(
fn
):
```
*`fn()` on a daemon thread, so quitting never waits for it.*
------------------------------------------------------------------------
source
### start_agent
``` python
def start_agent(
agent, closed:function=, note:NoneType=None
):
```
*Try the turn model, then start the code index search reads, whether the
model came up or not; the backend, or None with `agent.note` saying
why.*
------------------------------------------------------------------------
source
### sync_index
``` python
def sync_index(
host, cfg:NoneType=None, dirs:NoneType=None
):
```
*Sync the code index when a package or root changed since the last good
sync, else reopen it; `run` or `skip`, logged to `/kosha.log`.*
[`amain`](https://vedicreader.github.io/ramabana/cli.html#amain) is the
interactive loop: one terminal, one event loop, and one owner of the
keyboard. It starts the compositor, connects keys, pastes, resizes and
mouse events to the
[`Ui`](https://vedicreader.github.io/ramabana/cli.html#ui), starts the
model with `begin`, opens Python mode when asked, and on exit restores
the terminal and shuts down the agent and any kernel.
Three helpers keep a broken terminal from breaking the session.
[`AppTty`](https://vedicreader.github.io/ramabana/cli.html#apptty) is
teleprint’s `RealTty` with one change: when the terminal hangs up,
reading stops and writes are dropped, so the session quits instead of
spinning.
[`stderr_to`](https://vedicreader.github.io/ramabana/cli.html#stderr_to)
sends Python’s stderr to a file while the screen belongs to the
compositor, because a warning written straight to the terminal would
scroll the transcript under the compositor’s frames; it also points
`faulthandler` at that file so a hard crash leaves its stack trace
there.
[`stderr_log`](https://vedicreader.github.io/ramabana/cli.html#stderr_log)
names the file, inside the session’s own run folder.
For `--resume`,
[`pick_session`](https://vedicreader.github.io/ramabana/cli.html#pick_session)
lists the newest saved sessions from these folders and asks which to
open, and `show_resumed` draws the last `RESUME_SHOWN` turns of the
reopened session before the first prompt.
------------------------------------------------------------------------
source
### amain
``` python
async def amain(
agent, hint:str='', python:bool=False, attach:str='', agent_proxy:bool=False, bell:bool=True, pane:str='auto',
resumed:bool=False, # draw the resumed session's turns before the first prompt
):
```
*The tty loop: one terminal, one event loop, one keyboard owner.*
------------------------------------------------------------------------
source
### Ui.show_resumed
``` python
def show_resumed(
n:int=10
):
```
*Draw the resumed session’s last `n` turns: each request, what was sent
during it, and its reply.*
------------------------------------------------------------------------
source
### pick_session
``` python
def pick_session(
agent, all_folders:bool=False, # list every folder's sessions, not only those that ran in these folders
ask:function=_no_stdin, out:builtin_function_or_method=print, n:int=20
):
```
*The id of the saved session the person picks from the newest `n`, or ’’
for a new session.*
------------------------------------------------------------------------
source
### stderr_log
``` python
def stderr_log(
agent
):
```
*Where a session’s stderr goes: its own run folder, so the next start
cannot overwrite a crash’s trace.*
------------------------------------------------------------------------
source
### stderr_to
``` python
def stderr_to(
path
):
```
*Append Python’s `sys.stderr` and `faulthandler` crash traces to `path`
while the TUI owns the screen.*
------------------------------------------------------------------------
source
### AppTty
``` python
def AppTty():
```
*`RealTty` that stops reading and drops writes once the terminal hangs
up.*
The agent the terminal gets is a real one, over real folders, with
writes behind the approval gate. `approve='none'` builds it with no gate
at all:
``` python
tempdir = tempfile.mkdtemp()
cli_agent, cli_host = mk_agent([tempdir], approve='ask', web=False)
cli_host.roots, sorted(WRITE_TOOLS & set(t.__name__ for t in cli_agent.tools))
```
``` python
test_eq(cli_agent.approvals.mode, 'ask')
test_eq(cli_agent.approvals.timeout, None)
test_eq(mk_agent([tempdir], approve='none')[0].approvals, None)
cli_agent.routing.name_for()
```
Without a terminal, `ramabana` runs one turn, which suits a pipe or a
script. [`main`](https://vedicreader.github.io/ramabana/cli.html#main)
is the entry point behind the `ramabana` command. It parses the options,
builds the agent with
[`mk_agent`](https://vedicreader.github.io/ramabana/agent.html#mk_agent),
and then either runs one turn with
[`ask_once`](https://vedicreader.github.io/ramabana/cli.html#ask_once)
or starts the interactive session with
[`amain`](https://vedicreader.github.io/ramabana/cli.html#amain).
[`headless_prompt`](https://vedicreader.github.io/ramabana/cli.html#headless_prompt)
picks the one-turn prompt: the argument, or stdin when the argument is
`-` or empty and stdin is a pipe.
[`host_kw`](https://vedicreader.github.io/ramabana/cli.html#host_kw)
holds the host options the terminal adds. Outside tmux, an interactive
session first starts itself again inside a tmux server of Ramabana’s
own, as [setup](19_setup.ipynb) explains.
------------------------------------------------------------------------
source
### ask_once
``` python
def ask_once(
agent, prompt, as_json:bool=False
):
```
*Run one turn without a terminal and return the exit code.*
------------------------------------------------------------------------
source
### headless_prompt
``` python
def headless_prompt(
prompt
):
```
*The one-shot prompt: the argument, or stdin for `-` or an empty piped
run.*
------------------------------------------------------------------------
source
### main
``` python
def main(
prompt:str='', # one turn and exit; omit for interactive
root:str='.', # comma-separated folders: .,~/notes,/srv/app
model:str=None, # turn model; omit for the routing default
approve:str='ask', # ask | edits | auto | off | none (no gate)
web:bool=True, # --no-web takes the web tools offline
read_outside:bool=False, # reads reach any path; writes need --root
subagent_writes:bool=True, # --no-subagent-writes restricts delegates to read-only tools
vault:bool=False, # vishalakshi vault for reads; not in python mode
pii:str='off', # off | redact | refuse PII from the vault
pii_ner:bool=False, # --pii also gates titled names
spec:bool=False, # enable OpenAPI, Azure and Google Discovery tools
theme:str='auto', # auto | dark | light terminal palette
max_tool_calls:str='auto', # auto | 20..400 tool calls per turn
max_steps:str='auto', # auto | 8..80 model/tool-loop steps per turn
cfg:str='~/.config/ramabana', # config dir: skills, extensions, resumable history
resume:str='', # saved session id/prefix, or 'latest'; bare --resume lists this folder's sessions to pick from
all_folders:bool=False, # with a bare --resume, list sessions from every folder
python:bool=False, # start in python mode, on a kernel of your own
attach:str='', # join a live session; --kernels lists them
agent_proxy:bool=False, # expose a restricted agent proxy in PyREPL
kernels:bool=False, # list live sessions and exit
json:bool=False, # with a prompt: print the full result as JSON
bell:bool=True, # --no-bell silences turn-end and approval bells
tmux:str='auto', # auto | on | off: read and run in tmux panes
pane:str='auto', # auto | on | off: the `now` pane; auto in tmux
doctor:bool=False, # check tmux and the now pane, offer to install tmux, and exit
warm:bool=False, # seed a fresh session with dhrona's example rounds
no_warm:bool=False, # start without them, which is the default
optin:str='', # opt-in tool groups: exhash,research,author
profile:str='auto', # auto | small | full: small suits ≤32k local models
learned:bool=True, # --no-learned starts without /learned, the lessons Dhrona keeps
):
```
*Run Ramabana as a terminal agent or Python prompt. Name every folder it
may work on: –root .,~/notes*
------------------------------------------------------------------------
source
### host_kw
``` python
def host_kw(
tmux:str='auto'
):
```
*The CLI’s host options: `--tmux`, and no code index until the model is
up.*
``` python
from contextlib import redirect_stderr
from io import StringIO
from fastcore.script import anno_parser
_p = anno_parser(main.__wrapped__, pos=['prompt'])
test_eq(_p.parse_args(['one question', '--model', 'gpt']).prompt, 'one question')
test_eq(_p.parse_args([]).prompt, '')
test_eq(_p.parse_args(['--python']).python, True)
test_eq(_p.parse_args([]).python, False)
test_eq(_p.parse_args(['--attach', 'proj']).attach, 'proj')
_err = StringIO()
with redirect_stderr(_err): test_eq(main(python=True, vault=True), 2)
assert '--vault' in _err.getvalue()
import ramabana.pyrepl as _pyr
def _refuse(name): raise RuntimeError(f'no live dhrishti session matching {name!r}')
_real_find, _pyr.find_session = _pyr.find_session, _refuse
_err = StringIO()
try:
with redirect_stderr(_err): test_eq(main(attach='nope'), 2)
finally: _pyr.find_session = _real_find
assert 'no live dhrishti session' in _err.getvalue()
```
[`ask_once`](https://vedicreader.github.io/ramabana/cli.html#ask_once)
runs the turn, prints the answer on stdout and any problems on stderr,
and returns the exit code: 0, or 1 when the turn model was not
available. `ramabana --prompt 'what changed?' | less` therefore behaves
like any Unix program. With `as_json=True`, the `--json` option, it
prints the reply, usage, changed files, activity, problems and session
id as one JSON object.
``` python
piped, _ = fake_agent(replies=['nothing changed since the last commit.'])
test_eq(ask_once(piped, 'what changed?'), 0)
piped.note
```
The command line, as `--help` prints it:
``` python
!ramabana --help
```
usage: ramabana [-h] [--prompt (str)] [--root (str)] [--model (str)] [--approve (str)] [--no-web] [--read-outside]
[--no-subagent-writes] [--vault] [--pii (str)] [--pii-ner] [--spec] [--theme (str)]
[--max-tool-calls (str)] [--max-steps (str)] [--cfg (str)] [--resume (str)] [--all-folders] [--python]
[--attach (str)] [--agent-proxy] [--kernels] [--json] [--no-bell] [--tmux (str)] [--pane (str)]
[--doctor] [--warm] [--no-warm] [--optin (str)] [--profile (str)] [--no-learned]
Run Ramabana as a terminal agent or Python prompt. Name every folder it may work on: --root .,~/notes
options:
-h, --help show this help message and exit
--prompt (str) one turn and exit; omit for interactive (default: '')
--root (str) comma-separated folders: .,~/notes,/srv/app (default: .)
--model (str) turn model; omit for the routing default
--approve (str) ask | edits | auto | off | none (no gate) (default: ask)
--no-web --no-web takes the web tools offline (default: True)
--read-outside reads reach any path; writes need --root (default: False)
--no-subagent-writes --no-subagent-writes restricts delegates to read-only tools (default: True)
--vault vishalakshi vault for reads; not in python mode (default: False)
--pii (str) off | redact | refuse PII from the vault (default: off)
--pii-ner --pii also gates titled names (default: False)
--spec enable OpenAPI, Azure and Google Discovery tools (default: False)
--theme (str) auto | dark | light terminal palette (default: auto)
--max-tool-calls (str) auto | 20..400 tool calls per turn (default: auto)
--max-steps (str) auto | 8..80 model/tool-loop steps per turn (default: auto)
--cfg (str) config dir: skills, extensions, resumable history (default: ~/.config/ramabana)
--resume (str) saved session id/prefix, or 'latest'; bare --resume lists this folder's sessions to pick from
(default: '')
--all-folders with a bare --resume, list sessions from every folder (default: False)
--python start in python mode, on a kernel of your own (default: False)
--attach (str) join a live session; --kernels lists them (default: '')
--agent-proxy expose a restricted agent proxy in PyREPL (default: False)
--kernels list live sessions and exit (default: False)
--json with a prompt: print the full result as JSON (default: False)
--no-bell --no-bell silences turn-end and approval bells (default: True)
--tmux (str) auto | on | off: read and run in tmux panes (default: auto)
--pane (str) auto | on | off: the `now` pane; auto in tmux (default: auto)
--doctor check tmux and the now pane, offer to install tmux, and exit (default: False)
--warm seed a fresh session with dhrona's example rounds (default: False)
--no-warm start without them, which is the default (default: False)
--optin (str) opt-in tool groups: exhash,research,author (default: '')
--profile (str) auto | small | full: small suits ≤32k local models (default: auto)
--no-learned --no-learned starts without /learned, the lessons Dhrona keeps (default: True)
The prompt can also be given as the first positional argument, so
`ramabana 'which files import fastllm?'` is the same as `--prompt`. A
bare `--resume` lists this folder’s saved sessions and asks which to
reopen, through
[`pick_session`](https://vedicreader.github.io/ramabana/cli.html#pick_session).
A session over this repository, on a local model, with nothing gated:
``` bash
ramabana --root . --model qwen-4b --approve none
```
A session that may read the rest of the machine but writes only here:
``` bash
ramabana --root . --read-outside
```
A Python prompt of your own, with the agent reading its namespace
through Dhrishti’s overlay:
``` bash
ramabana --python --root .
```
One turn for a pipe, with no terminal at all:
``` bash
ramabana --prompt 'which files import fastllm?'
```
``` python
ui.buf.insert('/guide'); test_eq(ui.submit(), None)
test_eq(ui.transcript.block_text(list(comp.blocks.values())[-1]), GUIDE)
test_eq(guide_text(GUIDE).plain.rstrip(), GUIDE.rstrip())
assert guide_text(GUIDE).spans
assert max(len(l) for l in GUIDE.splitlines()) < 78
ui.buf.insert('/help'); test_eq(ui.submit(), None)
test_eq(ui.transcript.block_text(list(comp.blocks.values())[-1]), HELP)
assert key_card(HELP).spans
for q in ('/quit', '/exit', '/q'):
ui.buf.insert(q); test_eq(ui.submit(), 'quit')
```
`attach_file` attaches one text or notebook file. The file must be
readable under the host’s file policy and inside the open folders; a
picture or sound file goes to `attach` instead. Typing `@` in the prompt
offers file completion from the open folders, in the same tab menu that
completes slash commands, and when the prompt is sent,
[`run_turn`](https://vedicreader.github.io/ramabana/cli.html#run_turn)
attaches every `@path` it names.
------------------------------------------------------------------------
source
### run_turn
``` python
async def run_turn(
ui, prompt
):
```
*Run a turn after attaching its `@` files and folders, saying what was
attached or refused; a word that names nothing is let be.*
------------------------------------------------------------------------
source
### Ui.on_key
``` python
def on_key(
k
):
```
*Offer `@` completion with the same Tab menu used for slash commands; a
folder completed whole opens onto its contents.*
------------------------------------------------------------------------
source
### ref_text
``` python
def ref_text(
s
):
```
*`s` as typed after `@`: quoted when it holds whitespace, a folder’s
quote left open so typing goes on inside it.*
------------------------------------------------------------------------
source
### Ui.attach_file
``` python
def attach_file(
path
):
```
*Attach one readable file, or a folder as its listing, under the host’s
read policy.*
Three commands are handled by the surface itself before anything else:
`/theme NAME` switches palette and repaints with `apply_theme`, `/root`
lists the open folders and `/root add PATH` opens another with
`open_root`, and `/pane` opens or closes the `now` pane, described at
the end of this page.
------------------------------------------------------------------------
source
### Ui.submit
``` python
def submit():
```
*Handle `/theme`, `/root` and `/pane` before other commands.*
------------------------------------------------------------------------
source
### Ui.open_root
``` python
def open_root(
path:str=''
):
```
*`/root` lists the open folders; `/root add PATH` opens another.*
------------------------------------------------------------------------
source
### Ui.apply_theme
``` python
def apply_theme(
name:str=''
):
```
*Apply a terminal palette and repaint this UI.*
------------------------------------------------------------------------
source
### Ui.stop
``` python
def stop():
```
*Escalate repeated Ctrl-C from cancellation to termination to exit.*
`stop` is what `ctrl+c` does, and repeated presses escalate. During a
turn, the first press cancels the turn’s run, the second terminates it,
and the third quits; presses count as repeats only while they come
within the agent’s `cancel_grace` seconds of each other. Each press also
takes queued lines back into the composer, so nothing you typed is lost.
For a `!` command, `stop` signals the command’s process group. The first
`ctrl+c` sends `SIGINT`, and the next sends `SIGKILL`, however long
after the first it comes. A command ended by a signal prints `stopped`,
and the prompt is free once the process exits. Quitting kills a command
that is still running.
`on_key` is patched once more so that `ctrl+c` stops a turn without
clearing the composer, and `alt+up` pulls queued text back into the
composer for editing.
``` python
import tempfile
```
``` python
armed = Path(tempfile.mkdtemp())/'armed'
sh_ui.buf.insert(f"!trap '' INT; touch {armed}; sleep 30")
sh_ui.start_turn(sh_ui.submit())
sh_running = sh_ui.turn
for _ in range(100): # SIGINT before the trap is set would end the shell, so wait until it is
if armed.exists(): break
await asyncio.sleep(0.05)
assert armed.exists()
sh_ui.stop()
await asyncio.sleep(0.6) # longer than `cancel_grace`, and the command ignores SIGINT
test_eq(sh_running.done(), False)
sh_ui.stop()
await asyncio.wait_for(sh_running, 5)
sh_comp.stop()
test_eq((sh_ui.turn, sh_ui.shell_proc), (None, None))
sh_last = sh_ui.transcript.block_text(list(sh_comp.blocks.values())[-1])
assert 'stopped' in sh_last, [sh_ui.transcript.block_text(b) for b in list(sh_comp.blocks.values())[-4:]]
sh_last
```
------------------------------------------------------------------------
source
### Ui.on_key
``` python
def on_key(
k
):
```
*Ctrl-C stops without wiping the composer; alt+up pulls queued text back
into it.*
``` python
class _StopAgent:
cancel_grace = 1
def __init__(self):
self.r = type('R', (), {'id': 'r1', 'terminal': False})()
def run(self, run_id=''): return self.r
def cancel(self, run_id): self.r.terminal = True; return {'state': 'detached'}
def terminate(self, run_id): return {'state': 'terminated'}
def leftovers(self): return []
class _StopUi:
mark_read = Ui.mark_read
def __init__(self):
self.agent, self._stop_at, self._stop_count, self._stop_run = _StopAgent(), 0., 0, ''
self.notes, self.turn, self.starting, self.shell_proc = [], None, False, None
self._queued = None
self._steered = []
def note(self, text): self.notes.append(text)
def take_back(self): return Ui.take_back(self)
def paint(self): return None
def _post(self, fn, *a): return fn(*a)
u = _StopUi()
Ui.stop(u); time.sleep(.02)
Ui.stop(u); time.sleep(.02)
test_eq(u.notes[:4], ['cancelling', 'detached', 'terminating', 'terminated'])
test_eq(Ui.stop(u), 'quit')
```
## The `now` pane
The `now` pane is a small live view, in a tmux split beside the chat, of
what the main agent and each sub-agent are doing at this moment: the
turn, the call it is on, and each sub-agent run with its own calls. The
surface writes a snapshot,
[`now_snapshot`](https://vedicreader.github.io/ramabana/pane.html#now_snapshot),
to `/now.json` from its own loop: at most every `PANE_EVERY`
seconds while calls change or a turn runs, and once more when the turn
ends. `ramabana-pane`, from [pane](18_pane.ipynb), reads that file and
draws it.
- `/pane` opens it with `open_pane` and `/pane off` closes it with
`close_pane`.
- `--pane auto|on|off` decides whether a session starts with it, through
`start_pane`; `auto` opens it only inside tmux.
- [`pane_cmd`](https://vedicreader.github.io/ramabana/cli.html#pane_cmd)
is the shell line that runs the viewer for a given file.
When the viewer crashes rather than quits, the surface reopens it, a
limited number of times.
------------------------------------------------------------------------
source
### Ui.start_pane
``` python
def start_pane(
mode:str='auto'
):
```
*Open the first pane: `on` always tries, `auto` only in tmux with a runs
dir.*
------------------------------------------------------------------------
source
### Ui.close_pane
``` python
def close_pane():
```
*Kill the `now` pane: what was closed, or ’’ when none was open.*
------------------------------------------------------------------------
source
### Ui.open_pane
``` python
def open_pane(
arg:str=''
):
```
*`/pane` opens the `now` pane in a tmux split on the right; `/pane off`
closes it.*
------------------------------------------------------------------------
source
### pane_cmd
``` python
def pane_cmd(
path
):
```
*The shell line that draws the `now` pane for `path`.*
``` python
import tempfile
from ramabana.pane import read_snapshot
class _Pane:
def __init__(self, id): self.id, self.killed, self.dead = id, False, False
def kill(self): self.killed = True
def refresh(self): return self
class _Me:
def __init__(self): self.splits = []
def split(self, where, cmd, size=None): self.splits.append((where, cmd, size)); return _Pane('%9')
pane_tty = EmuTty(72, 14)
pane_comp = Compositor(pane_tty)
pane_comp._register_signals = lambda: None
await pane_comp.start()
pane_ui = Ui(pane_comp, fake_agent(cfg=Path(tempfile.mkdtemp()))[0])
now = pane_ui.agent.runs_dir/'now.json'
act = pane_ui.agent.activity.start('read_file', {'path': 'a.py'})
at = pane_ui._now_at
pane_ui.agent.activity.finish(act, 'ok')
test_eq(pane_ui._now_at, at)
assert read_snapshot(now)['root']['current'].startswith('read_file'), 'the write the call started made'
pane_ui.write_now(force=True)
assert pane_ui._now_at > at and read_snapshot(now)['root']['current'] == ''
me = pane_ui.agent.host.tmux_pane = _Me()
pane_ui.open_pane()
test_eq(me.splits, [('right', pane_cmd(now), '35%')])
pane_ui.open_pane()
test_eq(len(me.splits), 1)
test_eq(pane_ui.close_pane(), 'closed the pane %9')
assert pane_ui.pane is None
pane_tty.close()
```
``` python
from inspect import signature
assert 'graph' not in signature(mk_host).parameters
assert 'graph' not in signature(mk_agent).parameters
assert 'graph' not in signature(main).parameters
```
``` python
tl_ui.agent.activity.since()[-ACT_TAIL:]
```
``` python
tl_ui.turn = 'a turn'
footer = [r.plain for r in tl_ui.working()]
assert all(row[:8].count(':') != 2 for row in footer), footer
assert not any('─' * 18 in row for row in footer), footer
```
``` python
_art = (*Ui.BOW_FRAMES, Ui.COMPACT_BOW)
assert all(cell_len(f) == cell_len(Ui.BOW_FRAMES[0]) for f in _art)
assert not any('\n' in f or ' ' in f for f in _art), 'the track fills every cell'
ui.frame, ui.turn = 0, 'working'
ui._compact_until = 0.0
test_eq(ui.bow(False), Ui.BOW_FRAMES[0])
ui._compact_until = time.monotonic() + 1
test_eq(ui.bow(True), ui.COMPACT_BOW)
ui._compact_until = 0.0
assert set(Ui.BOW_CYCLE) == set(range(len(Ui.BOW_FRAMES))), 'every frame is reached'
_cycle = [Ui.BOW_FRAMES[i] for i in Ui.BOW_CYCLE]
_flight = [f.index('▸') for f in _cycle]
_gaps = [b - a for a, b in zip(_flight, _flight[1:]) if b != a]
test_eq(_gaps, sorted(_gaps))
assert len(set(_gaps)) > 1, 'a constant gap is the metronome this exists to avoid'
assert _cycle.count(Ui.BOW_FRAMES[0]) > _cycle.count(Ui.BOW_FRAMES[-1]), 'slow to leave, quick to go'
assert '◂' in Ui.COMPACT_BOW and '◂' not in ''.join(Ui.BOW_FRAMES), 'compacting closes inward'
_t = ui.bow_text(True, GRUVBOX['green'])
test_eq(_t.plain, ui.bow(True))
assert {str(sp.style) for sp in _t.spans} == {GRUVBOX['gray'], GRUVBOX['green']}
```