cli

Ramabana in a terminal: a transcript of blocks, a status bar, and one line to type in.

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.

from teleprint.testing import EmuTty as _EmuTty

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)

_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 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 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

def ext_keys_ok(
    tty
):

Whether tty is a real terminal at both ends.

---

source

ext_key

def ext_key(
    code, mods:int=1, event:int=1
):

The Key for a kitty or modifyOtherKeys report, or None.

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 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 picks the active one, and code_theme, hl_theme and code_bg give the matching colours for highlighted code. plan_text draws the plan checklist in the active palette.

---

source

plan_text

def plan_text(
    md
):

The plan checklist as themed Text, reading GRUVBOX per call.

---

source

set_theme

def set_theme(
    name:str='github-dark'
):

Select the active palette; auto means the default.

---

source

code_bg

def code_bg(
    palette:NoneType=None
):

The code block background: the palette’s bg1.

---

source

hl_theme

def hl_theme(
    name:NoneType=None
):

The Lumis theme hl colours code with under the named palette.

---

source

code_theme

def code_theme(
    name:NoneType=None
):

The pygments style to highlight code with under the named palette.

---

source

guide_text

def guide_text(
    text
):

The guide with its headers and commands styled.

---

source

key_card

def key_card(
    text
):

The key list, with its labels and separators styled.

print(HELP)

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, Attachment, 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 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 handles this in two steps. 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 from png_size. Then 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 builds a Picture for a file, or returns None when this terminal or this file cannot be drawn. 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 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

def media_line(
    path
):

The transcript line for a saved picture.

---

source

draw_png

def draw_png(
    path, cols:int=24
):

Send and place path at the cursor, or '', for callers without frames.

---

source

picture

def picture(
    path, cols:int=24, rows:int=12
):

A Picture for path, or None where this terminal or this file cannot be drawn.

---

source

Picture

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

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

def png_size(
    path
):

(width, height) in pixels from a PNG’s IHDR, or None.

---

source

kitty_graphics

def kitty_graphics():

Whether this terminal speaks the kitty graphics protocol.

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''))

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 captures the file’s text when the prompt is submitted, and 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 finds the @path references in a prompt. 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

def file_note(
    atts
):

The grounded text from selected workspace files and folder listings, with their exact paths.

---

source

FileAttachment

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

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

def steer_note(
    atts
):

What a steered line says about its attachments: a tool result carries no picture, so each goes by path.

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 methods that attach files, attach, paste and attach_file, are defined after Ui further down. The tests here check the path parser on its own.

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])

from unittest.mock import patch as mock_patch

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'), [])

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 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 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 checks the typed line against the trigger words in MENUS and returns a title and a list of Options, or None. ChoiceMenu draws the row above the input line and handles the keys that move through it and pick one. Each 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 closes the tmux panes of run_shell_bg commands that have finished; their output stays readable through cmd_output.

---

source

run_turn

async def run_turn(
    ui, prompt
):

One turn, streamed into the transcript.

---

source

close_done_shells

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

def ChoiceMenu(
    title, options
):

The options row: one line per Option, above the input line.

---

source

options_for

def options_for(
    text
):

(title, options) for a prompt worth asking how* about, or None.*

---

source

Option

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 highlights a piece of code in a given language, and diff_rich colours a diff, both through hl in the active palette and its Lumis theme from 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 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: 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, which shows the first DIFF_OPEN lines with a hint until you open the block.

---

source

OpenDiff

def OpenDiff(
    ui, blk, text
):

A result shown DIFF_OPEN lines deep, with a hint, until the reader opens it.

---

source

opens

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

def changed_table(
    rows, cap:int=12
):

The turn’s changed files, largest cap first, then a total over every one.

---

source

diff_rich

def diff_rich(
    s
):

s coloured by hl.diff_rich in the active palette.

---

source

hl_text

def hl_text(
    code, lang
):

code coloured by fastpylight in the active palette’s theme.

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 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 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 for a one-line message, stream for reply text as it arrives, and paint to redraw.

---

source

Ui

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 <cfg>/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 is what that rule matches. Without rules, A behaves like a. 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

def note_memory(
    text
):

#note TEXT: keep a line for later sessions.

---

source

Ui.always_allow

def always_allow():

Always allow the pending request’s pattern and approve it; False if none.

---

source

Ui.ring

def ring():

Ring the bell if on and nobody typed for BELL_IDLE seconds.

---

source

ask_pattern

def ask_pattern(
    ask
):

What an approval rule would match for this request.

---

source

Ui.nudge

def nudge():

Ring and re-show a question after REASK_EVERY seconds; True if re-shown.

---

source

Ui.reask

def reask(
    body:str=''
):

Show the pending question with body, ring, and restart the nudge wait.

---

source

parse_answer

def parse_answer(
    text
):

(ok, note) for a typed approval answer (y, n, n: reason), or None.

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')

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

def place_pics():

Put every picture back where its block sits, after each frame.

---

source

Ui.on_media

def on_media(
    paths
):

Show pictures a tool wrote, mid-turn, from the model’s thread.

---

source

Ui.show_media

def show_media(
    media, session:str=''
):

Save the pictures the model itself returned and show each one.

---

source

Ui.show_pic

def show_pic(
    path
):

Draw one saved picture where possible, and name its path.

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

def chip():

Plan mode or the approval mode as one short token, or None when nothing is gated.

---

source

Ui.cycle_approve

def cycle_approve():

Cycle ask, edits, auto, then plan mode, then ask again; from off, go to ask.

---

source

Ui.tighten_approve

def tighten_approve():

Make approvals one step stricter, never looser.

---

source

Ui.approve_mode

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.

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

def reply(
    text
):

A model reply as Markdown in this palette, prose compacted.

---

source

compact_md

def compact_md(
    text
):

Markdown with blank rows between plain paragraphs made hard breaks.

---

source

Reply

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

def ThemedCode(
    lexer_name:str, theme:str
)->None:

A fenced block on the palette’s background: a diff by diff_rich, known code by fastpylight, else pygments.

---

source

fence_lang

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

def paste(
    text
):

Insert pasted text, or attach the files when the paste is nothing but paths.

---

source

Ui.on_key

def on_key(
    k
):

Give transcript navigation priority; Ctrl-P/Ctrl-N recall prompts.

---

source

Ui.on_mouse

def on_mouse(
    ev
):

Route a mouse event to the browsing view, or here when /mouse is on.

---

source

Ui.set_mouse

def set_mouse(
    want:str=''
):

Take or release the mouse on the main screen; off by default.

---

source

Ui.leave_transcript

def leave_transcript():

Close the browsing view and return the mouse unless /mouse is on.

---

source

Ui.enter_transcript

def enter_transcript():

Open the browsing view, borrowing the mouse while it is up.

---

source

Ui.submit

def submit():

Submit the line and remember it for Ctrl-P/Ctrl-N recall.

---

source

Ui.remember

def remember(
    line
):

Keep line for Ctrl-P/Ctrl-N recall.

---

source

Ui.recall

def recall(
    step
):

Move through submitted prompts, preserving the draft beyond the newest entry.

---

source

Ui.tail

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

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, 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 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

async def complete_python():

Tab: complete as far as the candidates agree, and leave the menu up to cycle.

---

source

Ui.on_output

def on_output(
    output
):

Print one nbformat output as the block it means.

---

source

Ui.run_code

async def run_code(
    code
):

Run one owner-typed Python cell, streaming its outputs into blocks.

---

source

Ui.attach_session

async def attach_session(
    name
):

Point the agent’s Python at someone else’s dhrishti session.

---

source

Ui.enable_agent_proxy

async def enable_agent_proxy():

Bind this session’s agent at the Python prompt and in the agent’s layer.

---

source

Ui.proxy_label

def proxy_label():

A kernel-unique name for this session’s agent.

---

source

Ui.enter_python

async def enter_python():

Take the line for Python, starting the kernel once and pointing the agent at it.

---

source

Ui.use_host

def use_host(
    host
):

Move the agent onto host, re-briefing the running turn backend with its tools.

---

source

Ui.log_cell

def log_cell(
    source, outputs:NoneType=None, cell_type:str='code'
):

Write one cell to the session notebook, when the host keeps one.

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 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 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

async def run_shell_line(
    command
):

Run one owner-typed ! line in the first open folder, printing its output and exit code.

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 draws a reply. It uses Reply, Rich’s Markdown with code blocks drawn by ThemedCode, and compact_md, which joins plain paragraphs with single line breaks to save rows. 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 decided that “refactor the runtime” earns a menu, ChoiceMenu.choose takes one keystroke at a time and returns (done, option), and Ui decides what the chosen option means for the prompt. Below, down moves the cursor and enter picks “plan”.

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()))

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 whose model replies are scripted. Typing “refactor the runtime” and pressing enter opens the options row on screen.

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())

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.

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.

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.

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.

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.

comp.on_bytes(b'\r')
await asyncio.sleep(0.4)
print(tty.term.text())

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.

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.

[(b.tag, b.height) for b in comp.blocks.values()]

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.

long = ui.say(Text('\n'.join(f'line {i}' for i in range(40))), 'tool')
long.collapsed, long.height

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.

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.

from ramabana.agent import Act

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. 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.

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')

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')

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 feed. Each of them carries a parent_action_id, which Agent sets from a per-thread stack of the delegate calls in flight, and 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 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 nth of them, which is what alt+1 to alt+9 do.

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)

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.

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)

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')

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))

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)

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()

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, 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:

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.

test_eq(ui2.ask.tool, 'create_file')
tty2.term.text().splitlines()[-1]

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.

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”.

ui2.mode = 'python'
comp2.on_bytes(b'n: put it in docs/ instead')
tty2.term.text().splitlines()[-1]

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.

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.

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.

ui.buf.insert('/tools')
ui.submit()
[l for l in str(comp.blocks[max(comp.blocks)].body[0]).splitlines()[:4]]

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.

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.

test_eq(ui.submit(), None)
ui.buf.text

Running it

mk_agent from ramabana.agent builds what the terminal runs. It makes a WorkspaceHost over the folders named on the command line, an 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 builds the host alone. amain is the terminal loop that runs the Ui, and it stays short because the behaviour lives in 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 starts the turn model and then the code index, the search index behind search_code, so the index never delays the model. 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 <cfg>/kosha.log. warm_up runs start_agent off the loop with off_loop, and prints a note saying why when no model starts.

---

source

Ui.begin

def begin():

Start the model behind the prompt; lines sent meanwhile wait in the queue.

---

source

Ui.warm_up

async def warm_up():

Start the model off the loop; a start that fails says why in a note.

---

source

off_loop

async def off_loop(
    fn
):

fn() on a daemon thread, so quitting never waits for it.

---

source

start_agent

def start_agent(
    agent, closed:function=<lambda>, 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

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 <cfg>/kosha.log.

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, 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 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 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 names the file, inside the session’s own run folder.

For --resume, 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

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

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

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

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

def stderr_to(
    path
):

Append Python’s sys.stderr and faulthandler crash traces to path while the TUI owns the screen.

---

source

AppTty

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:

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))

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 is the entry point behind the ramabana command. It parses the options, builds the agent with mk_agent, and then either runs one turn with ask_once or starts the interactive session with amain. headless_prompt picks the one-turn prompt: the argument, or stdin when the argument is - or empty and stdin is a pipe. 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 explains.

---

source

ask_once

def ask_once(
    agent, prompt, as_json:bool=False
):

Run one turn without a terminal and return the exit code.

---

source

headless_prompt

def headless_prompt(
    prompt
):

The one-shot prompt: the argument, or stdin for - or an empty piped run.

---

source

main

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 <cfg>/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

def host_kw(
    tmux:str='auto'
):

The CLI’s host options: --tmux, and no code index until the model is up.

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 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.

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:

!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 <cfg>/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.

A session over this repository, on a local model, with nothing gated:

ramabana --root . --model qwen-4b --approve none

A session that may read the rest of the machine but writes only here:

ramabana --root . --read-outside

A Python prompt of your own, with the agent reading its namespace through Dhrishti’s overlay:

ramabana --python --root .

One turn for a pipe, with no terminal at all:

ramabana --prompt 'which files import fastllm?'

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 attaches every @path it names.

---

source

run_turn

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

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

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

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

def submit():

Handle /theme, /root and /pane before other commands.

---

source

Ui.open_root

def open_root(
    path:str=''
):

/root lists the open folders; /root add PATH opens another.

---

source

Ui.apply_theme

def apply_theme(
    name:str=''
):

Apply a terminal palette and repaint this UI.

---

source

Ui.stop

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.

import tempfile

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

def on_key(
    k
):

Ctrl-C stops without wiping the composer; alt+up pulls queued text back into it.

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, to <runs_dir>/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, 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 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

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

def close_pane():

Kill the now pane: what was closed, or ’’ when none was open.

---

source

Ui.open_pane

def open_pane(
    arg:str=''
):

/pane opens the now pane in a tmux split on the right; /pane off closes it.

---

source

pane_cmd

def pane_cmd(
    path
):

The shell line that draws the now pane for path.

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()

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

tl_ui.agent.activity.since()[-ACT_TAIL:]

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

_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']}