runtime

Run model turns, keep each conversation inside its context window, and capture what native engines print.

Native output

A local engine such as LiteRT is a C++ library loaded into the Python process. It prints its messages straight to file descriptors 1 and 2, the process’s standard output and standard error, so they never pass through Python’s print or logging. When a turn fails, for example because the prompt was longer than the model’s window, the only explanation is often one of those lines. This section captures the descriptors during a model call and keeps the lines that explain a failure.

interesting picks those lines out of captured text. It drops the engine’s routine startup messages, keeps lines that mention an error, a failure, an exceeded limit or running out of memory, and returns the last limit of them.

interesting

def interesting(
    text, limit:int=4
):

The lines of captured output a person should see: complaints, not chatter.

Engines have no log level to filter on, so interesting matches words. In this log three lines are startup chatter and one says why the call failed:

log = '''Created TensorFlow Lite XNNPACK delegate for CPU.
Loading model from /models/gemma-e4b
I0000 00:00:1730000000.000000 metal delegate ready
ERROR: input tokens 21014 exceeds max_num_tokens 16384
'''
interesting(log)

Repeated lines collapse into one, and only the most recent limit lines are kept. An engine that fails on every step of a tool loop prints the same error many times, and the user needs to see it once.

test_eq(interesting('ERROR: oom\nERROR: oom\nERROR: oom'), ['ERROR: oom'])
interesting('\n'.join(f'error {i}' for i in range(10)), limit=2)

captured is a context manager that records what is written to descriptors 1 and 2 inside its with block. It copies the output rather than hiding it: the bytes still reach the terminal, and the last 8,000 bytes of each descriptor (MAX_KEEP) are kept in cap.text. cap.problems runs interesting over that text.

Only one captured block is active at a time across threads, because two threads redirecting the same descriptor would restore each other’s copies and leave the real descriptor detached. A thread that cannot take the lock within half a second runs its block without capture. Setting RAMABANA_NO_NATIVE_CAPTURE=1 turns capture off.

captured

def captured(
    fds:tuple=(1, 2), enabled:NoneType=None
):

Copy what is written to descriptors 1 and 2 into text, while the output still reaches the terminal.

capture

def capture(
    fn, *a, **kw
):

Call fn and return (result, problems); an exception carries them out too.

capture(fn, *args) is the form to use around a single call. It calls fn inside captured and returns a pair: the result, and the problem lines the engine printed. A call can succeed and still print a warning worth showing, as here, where the engine truncated its input:

def _engine_call():
    os.write(2, b'Created TensorFlow Lite XNNPACK delegate for CPU.\n')
    os.write(2, b'WARNING: input exceeds max_num_tokens 16384, truncating\n')
    return 'a short reply'

capture(_engine_call)

When fn raises, capture attaches the problem lines to the exception as native_output and raises it again. Native failures often arrive as an exception with an empty message, and this keeps the real explanation with it:

def _engine_dies():
    os.write(2, b'ERROR: failed to allocate 4.00 GiB for KV cache\n')
    raise RuntimeError('')
try: capture(_engine_dies)
except RuntimeError as e: err = e
err.native_output

Sizing a conversation

These functions decide when a conversation has grown too big for its model. Each limit is a fraction of the context window rather than a fixed number of tokens, so the same code works for a 4k-token local model and a 200k-token hosted one. A fixed reserve would not: with 16k tokens held back from a 4k window, every turn would be due for compaction, and the agent would compact a two-message conversation forever.

estimate_tokens counts the tokens in a text. threshold gives the token count at which compaction becomes due, and should_compact compares a count with it. halvings returns a budget followed by its halves, for retrying a request whose size was only estimated.

should_compact

def should_compact(
    used, ctx, reserve:NoneType=None
):

Whether used tokens against a ctx window has crossed the line.

threshold

def threshold(
    ctx, reserve:NoneType=None
):

The token count at which a conversation should be compacted, or None when there is no window. reserve is config.toml’s window.reserve, else RESERVE.

halvings

def halvings(
    budget, tries:int=3, floor:int=256
):

A budget and its halvings, to retry a prompt whose fit was only estimated.

estimate_tokens

def estimate_tokens(
    text, count:NoneType=None
):

Tokens in text: exact via count when a tokenizer is at hand, CHARS_PER_TOKEN otherwise.

estimate_tokens uses the model’s own tokenizer when you pass one as count, and otherwise divides the length of the text by CHARS_PER_TOKEN (3.25). The estimate is rough, and close enough for budgeting.

estimate_tokens('the exporter writes one module per notebook'), estimate_tokens('', ), estimate_tokens('x' * 400)

threshold(ctx) holds back a reserve below the window: the full RESERVE of 16,384 tokens for a large window, and at most a quarter of the window for a smaller one. Without the quarter-window cap, a 4k model would be due for compaction before its first turn.

threshold(200_000), threshold(32_768), threshold(4096)

test_eq(threshold(4096), 3072)
test_eq(threshold(0), None)
should_compact(3000, 4096), should_compact(3072, 4096)

Reading a conversation

A backend stores its conversation as a list of message dictionaries in urai’s format, which is the same for every model. Each message has a role (user, assistant or tool) and content. An assistant message that calls tools also carries tool_calls, and the result of each call comes back as a tool message. Compaction reads this list directly, using urai’s resp_text for a message’s text and tool_rows for its tool calls.

The examples below use this four-message conversation: a request, a tool call, the tool’s result, and the reply.

hist = [
    {'role': 'user', 'content': 'the exporter duplicates cells, please fix it'},
    {'role': 'assistant', 'content': 'Reading the exporter.',
     'tool_calls': [{'function': {'name': 'view_file', 'arguments': {'path': 'nbdev/maker.py'}}}]},
    {'role': 'tool', 'content': 'def _make_exists(self, cells):\n    with self.fname.open("a") as f: ...'},
    {'role': 'assistant', 'content': 'It appends, so a second export duplicates every named cell.'},
]
[(m.get('role', '?'), resp_text(m)[:40], [name for name, _ in tool_rows(m)]) for m in hist]

serialise writes a list of messages as the text a summarizing model reads. Each message goes in a <message> tag, with its tool calls named in a <tool-calls> tag, and each text is clipped to mx characters (2,000 by default). Tool results make up most of a conversation’s length and little of its meaning, so clipping them costs little.

split_previous

def split_previous(
    msgs
):

Split off a leading summary: (previous_summary_or_None, remaining_msgs).

serialise

def serialise(
    msgs, mx:int=2000
):

Messages as the tagged block the summarizer reads, each text clipped to mx.

print(serialise(hist))

Once a conversation has been compacted, its first message is a user message that starts with SUMMARY_PREFIX and holds the summary. split_previous separates that summary from the messages after it. The next compaction can then update the existing summary with the new messages, instead of summarising the old summary as if it were ordinary conversation.

prev, rest = split_previous([{'role': 'user', 'content': SUMMARY_PREFIX + '## Goal\nfix the exporter'}] + hist)
prev, len(rest)

test_eq(split_previous(hist), (None, hist))
split_previous([])

summarise_prompt builds the request sent to the summarizing model. With no earlier summary it asks for a new one in a fixed format: goal, constraints, progress, key decisions, next steps and critical context. When split_previous finds an earlier summary, it asks for that summary to be updated with the new messages instead. extra adds a line saying what the summary should focus on.

The summary is usually written by a cheaper model with a smaller window, so max_tokens caps the size of the request. When the whole request does not fit, the earlier summary is cut to a quarter of the budget, and the messages after it fill what is left, cut from the end. The instructions come last and are kept whole while the budget can hold them.

fold_summary

def fold_summary(
    msgs, summariser, extra:str='', max_tokens:NoneType=None, count:NoneType=None
):

Summarise msgs in runs that each fit max_tokens, every summary updated with the next run; ’’ when a step answers nothing.

transcript_runs

def transcript_runs(
    msgs, budget, count:NoneType=None
):

msgs in consecutive runs whose transcript fits budget tokens; a message that alone is larger makes a run of its own.

summarise_prompt

def summarise_prompt(
    msgs, extra:str='', max_tokens:NoneType=None, count:NoneType=None
):

The bounded prompt handed to the summarizer, choosing fresh or update instructions.

With no budget the request holds the whole conversation, one <message> per message, followed by the format instructions:

p = summarise_prompt(hist)
test_eq(p.count('<message'), len(hist))
estimate_tokens(p)

With a budget, the request always fits. When the budget is too small to hold the instructions as well, they are cut too, as at 120 and 30 tokens below. A request with cut instructions still gets an answer, while a request over the limit would be refused by the engine.

[estimate_tokens(summarise_prompt(hist, max_tokens=n)) for n in (2000, 400, 120, 30)]

for n in (2000, 400, 120, 30): assert estimate_tokens(summarise_prompt(hist, max_tokens=n)) <= n
summarise_prompt(hist, max_tokens=30)

Summarising needs a second model call, and sometimes no summarizing model is available or its summary cannot be trusted. Surgical compaction needs no model. surgical_history rewrites old messages as short lines, one per request, reply, tool call and tool result, each cut to a token budget for its role. What was asked, what was called, what came back and what was concluded all survive, at a fraction of the size.

truncate_middle does the cutting. It keeps the start and end of a text and replaces the middle with …, because paths, error messages and diffs carry most of their meaning at the ends. Cut to 12 tokens, this long name keeps its file and its last function:

truncate_middle

def truncate_middle(
    text, budget, count:NoneType=None, mark:str=' … '
):

Keep both ends of text inside a token budget.

surgical_history

def surgical_history(
    msgs, policy:NoneType=None, count:NoneType=None
):

Render old history as a compact, readable DSL while preserving tool evidence.

truncate_middle('nbs/01_runtime.ipynb::Compactor.compact::summarise_prompt', 12)

Each role has its own marker: § for the user, » for the assistant’s text, ▶ for a tool call and > for its result. The lines of a tool result are joined with ¶.

print(surgical_history(hist))

SURGICAL_POLICY sets the budgets: 2,000 tokens for a user message, 150 for the assistant’s text, 60 for a tool call and 35 for a result. The user’s own words are kept nearly whole, and tool results are cut hardest. policy overrides any of them:

print(surgical_history(hist, policy={'result': 8, 'assistant': 20}))

Reorienting the model

After compaction the model sees a summary where its conversation used to be, and it can draw wrong conclusions from that. reorient writes the reminder placed after the summary. It says whether the Python kernel survived, so that the model neither rebuilds variables that still exist nor uses ones that are gone. It tells the model that skill text it read earlier is no longer in context and has to be read again with read_skill, naming the skills when skills is given. It also tells the model not to answer a finished request a second time: if the latest request is complete, the model replies in one line and waits. REORIENT is the reminder with the default arguments.

reorient

def reorient(
    kernel_alive:bool=True, skills:tuple=()
):

The system reminder placed after a compacted or otherwise rewritten conversation.

print(reorient(skills=('nbdev', 'fossick')))

When the kernel restarted, the reminder says the namespace is empty. Getting this wrong would send the model looking for variables that no longer exist, or rebuilding ones that do.

test_eq('do not re-import anything' in reorient(kernel_alive=False), False)
print(reorient(kernel_alive=False)[:180])

assert 'inspect_python' in reorient() and 'list_vars' not in reorient()

Notices on the way in

Before a prompt reaches the model, prompt_notices checks it with a few string tests and returns notices: short instructions that tell the model how to treat the prompt. They follow the UserPromptSubmit hook in Answer.AI’s aai-coding prompts, and each one answers a mistake models often make.

A prompt ending in ? is a question. The model answers it and stops, without starting the work it implies (Q_NOTICE).
A prompt containing “please read”, “read the” or “have a look at” asks the model to read the target in full before answering (READ_NOTICE).
A bare approval such as “go”, “ok” or “do it” approves only what the user explicitly agreed to, not everything the model proposed earlier (APPROVAL_NOTICE).
A prompt starting with “BTW” is a side request. The model answers it and then goes back to the task it was on (BTW_NOTICE).
A prompt starting with a verb such as fix, add, run or rename, or ending in as NAME, asks for action. The model does the work and checks it before replying (ACTION_NOTICE).

Three one-character prompts are shortcuts and get only their own notice: . continues the agreed task, ; asks for the last answer in simpler words, and , asks what the user needs to answer next.

notices_block

def notices_block(
    prompt
):

The notices for prompt as one reminder to append to it, or ''.

prompt_notices

def prompt_notices(
    prompt
):

Notices a submitted prompt earns, from aai-coding’s UserPromptSubmit hook.

A question gets Q_NOTICE and nothing else. A bare “go” gets APPROVAL_NOTICE:

test_eq(prompt_notices('can you fix the exporter?'), [Q_NOTICE])
prompt_notices('go')

A prompt can earn several notices. Here an imperative gets one, a side request that asks for a read gets two, and “thanks” gets none:

[len(prompt_notices(p)) for p in ('fix the exporter', 'BTW please read nbs/01_runtime.ipynb', 'thanks')]

notices_block is the function the agent calls. It wraps the notices in one <system-reminder> block, ready to append to the prompt, or returns '' when there are none.

print(notices_block('ok'))

test_eq(notices_block('thanks'), '')
notices_block('thanks')

Fitting a notebook into the prompt

A prompt can carry the user’s open notebook inside a <notebook> tag, one <cell> per cell. Each cell has a compact attribute that the user sets: keep, auto or discard. compact_notebook_context leaves the prompt alone when it fits. When it does not, it removes discard cells, then auto cells, each in notebook order, until the prompt fits. It never removes a keep cell. A <context-compacted> tag records how many cells were left out.

fits is a function that says whether a prompt fits. The agent passes the backend’s fits, so the decision is made against the real window of the model in use.

compact_notebook_context

def compact_notebook_context(
    prompt, fits
):

Reduce a tagged notebook only when prompt does not fit. A keep cell is never removed.

A notebook of three cells, one with each setting, and a question about it. When everything fits, the prompt comes back unchanged:

nb = '''<notebook path=analysis.ipynb>
<cell id=1 compact="discard">print(load_everything())  # 900 lines of output</cell>
<cell id=2 compact="auto">df = read_csv('sales.csv')</cell>
<cell id=3 compact="keep">model.fit(df)  # the cell the question is about</cell>
</notebook>

why did the fit fail?'''
print(compact_notebook_context(nb, fits=lambda s: True))

With a lower limit the discard cell goes first, and <context-compacted> records it:

out = compact_notebook_context(nb, fits=lambda s: len(s) < 270)
print(out)

A lower limit still removes the auto cell as well. The keep cell stays, even though the result is still over the limit, because the user chose to keep it.

tight = compact_notebook_context(nb, fits=lambda s: len(s) < 150)
test_eq('compact="keep"' in tight, True)
test_eq('load_everything' in tight, False)
print(tight)

The compactor

Compactor decides when a backend’s conversation is due for compaction, and carries it out. It keeps the most recent messages as they are and replaces everything older with a summary, followed by the reorient reminder. The summary comes from a second model call, made through the summariser(prompt, sp) function you pass to compact. With strategy='surgical' it uses surgical_history instead and calls no model.

due(backend) says whether the backend has crossed its threshold. compact returns the summary, or '' when it did nothing, and note says what happened either way. on_compact, when given, is called with each new summary. If the summarizer fails or returns nothing, the conversation is left alone. When you give the summarizer’s window as summary_ctx, no request is larger than it: fold_summary splits a history too large for one request into runs that each fit (transcript_runs), summarises the first, and updates that summary with each run after it, so a history of any length compacts. A failed or empty call is retried at half and then a quarter of that size.

Compaction lives in its own class rather than in a rishi callback because it needs a second model call and then replaces the first model’s history, and a callback does not do both.

Compactor

def Compactor(
    reserve:NoneType=None, # None reads config.toml's `window.reserve`, else `RESERVE`, each time
    keep_recent:NoneType=None, # None reads `window.keep_recent`, else `KEEP_RECENT`
    auto:bool=True, # compact on crossing the threshold
    kernel_alive:bool=True, # what the reorientation note may promise
    on_compact:NoneType=None, # called with each new checkpoint
    strategy:str='summary', # 'summary' or deterministic 'surgical'
):

Decide when to compact, and do it.

budget(ctx) is how many tokens of recent conversation survive compaction: KEEP_RECENT (20,000), capped at half of what the window holds besides the system prompt and tools (overhead). Without the cap, a 4k model would count its whole conversation as recent, find nothing older to summarise, and report everything is recent; nothing to compact until the engine refused the turn. compact also keeps at most about half of the conversation, and the part it keeps always starts at a user message.

c = Compactor()
c.budget(200_000), c.budget(32_768), c.budget(4096)

test_eq(c.budget(4096), 2048)
test_eq(c.note, 'not compacted')
c.strategy

Compacting needs a running backend and a summarizer. testing defines doubles for both and runs a compaction end to end.

Answers, without the thinking

Reasoning models think before they answer. Some runtimes return the thinking in a separate channel, and others leave it in the reply, wrapped in <think>...</think>. The agent’s one-shot jobs read the reply as data, such as a label, a summary or a completion to insert, so the thinking has to come off first. A classifier that returned the model’s reasoning would return no label at all.

answer_only removes the thinking from a finished reply, using urai’s split_think.

answer_only

def answer_only(
    text
):

A one-shot reply with the model’s thinking removed, however the runtime left it.

In the simple case the thinking is wrapped in a pair of tags:

answer_only('<think>7 times 6. Six sevens are 42.</think>\n\n42')

Some chat templates open the <think> block in the prompt itself, so the reply starts inside the thinking and only the closing tag appears. answer_only handles that too. A reply that opens a block and never closes it, because it hit the output cap, has no answer and gives ''. A reply with no tags comes back as it is.

answer_only('Six sevens are 42. They asked for only the number.\n</think>\n\n42')

test_eq(answer_only('<think>cut off at the cap and never closed'), '')
test_eq(answer_only('thinking, and then the cap hit\n'), 'thinking, and then the cap hit')
test_eq(answer_only('42'), '42')
answer_only('<think>a</think>yes')

A model whose template opens the thinking block does so at every step of a turn: after each tool call, the next reply begins inside a new thought. prefills_think tells whether a model’s template does this, by rendering a one-message prompt and looking for an unclosed <think>. For such a model RishiBackend passes the stream through a ThinkFilter, which drops text until it sees </think>, lets the rest through, and starts dropping again after each tool call. Use answer_only on a finished reply and ThinkFilter on a stream. The filter also counts the characters of thinking (thought) and of answer (answer), so the backend can report a turn that spent all its output thinking. One-shot jobs avoid the problem by asking the model not to think.

ThinkFilter

def ThinkFilter():

Drop a template-opened thinking block out of a raw chunk stream. A tool call re-arms it.

prefills_think

def prefills_think(
    chat
):

Whether this model’s chat template opens a <think> block and leaves the model to close it.

The closing tag can be split across chunks, and the filter still finds it. The answer comes through whole:

def chunks(*texts): return [{'content': [{'type': 'text', 'text': t}]} for t in texts]
def texts(chunks): return ''.join(resp_text(c) for c in chunks)
f = ThinkFilter()
texts(f(chunks('Six sevens', ' are 42.\n</th', 'ink>\n\n', '42')))

test_eq(texts(ThinkFilter()(chunks('deliberating', '</think>', 'yes'))), 'yes')

tool = {'content': [{'type': 'tool_call', 'name': 'search_code', 'arguments': {}}]}
out = list(ThinkFilter()(chunks('think 1', '</think>', 'calling ') + [tool] + chunks('think 2', '</think>', 'done')))
test_eq(texts(out), 'calling done')
test_eq(out[1], tool)

f = ThinkFilter()
test_eq(texts(f(chunks('cut off at the cap'))), '')
test_eq((f.thinking, f.thought, f.answer), (True, 18, 0))

f = ThinkFilter()
test_eq(texts(f(chunks('deliberating', '</think>\n\n'))), '')
test_eq((f.thinking, f.answer), (False, 0))

Usage

Usage holds the token counts and cost of a turn: tokens in and out, cached tokens, tokens written to the cache, reasoning tokens, the cost in dollars, and the number of turns. Two Usage values add up, so a session’s total is sum of its turns, and subtracting an earlier running total from a later one gives what was used in between. Its repr is the one-line summary shown in the terminal’s status bar.

Usage

def Usage(
    model:str='', input:int=0, output:int=0, total:int=0, cached:int=0, cache_write:int=0, reasoning:int=0,
    cost:float=0.0, turns:int=0
)->None:

Token counts and cost for a turn or a session; values add up.

a = Usage(model='anthropic/claude-sonnet-5', input=12_000, output=800, total=12_800, cached=9_000, cost=0.041, turns=1); a

The sum takes the model name from the right-hand value, so a session that switched models reports the model it uses now:

b = Usage(model='openai/gpt-5.6-terra', input=3_000, output=200, total=3_200, cost=0.008, turns=1)
sum([a, b], None)

test_eq((a + None).total, a.total)
test_eq(sum([a, b], None).turns, 2)
a.dict()['cached']

Backend

Backend is the contract the agent uses to talk to a model. One backend holds one conversation. It starts the model on first use, sends a message with send or streams the reply with stream, asks one-shot questions with oneshot, counts tokens, and replaces or restores the conversation’s history. Subclasses supply the engine-specific methods whose names start with an underscore. RishiBackend, further down, is the one the agent uses.

send, stream and oneshot do not raise when the model fails. A model that cannot start, a stream that breaks and an empty reply all become text: note holds the latest status, problems keeps the last 20 problem lines, and turn_error says why the last turn failed. A caller can always show the user what the backend returned.

The examples below run the base class itself, which has no engine, to show these rules.

fit_content

def fit_content(
    c, room, save, how:str='', count:NoneType=None, tail:int=0
):

Message content c within room tokens: the newest pictures that fit stay and the rest are saved and named, then text past the room becomes an excerpt. The last tail characters of the last text part are always kept.

excerpt

def excerpt(
    text, room, save, how:str='', count:NoneType=None, tail:int=0
):

text cut to its head and its end within room tokens, the last tail characters always kept; the whole goes to save(data, suffix) and the note names the file.

omitted

def omitted(
    n, path, how:str=''
):

The note an excerpt carries where n characters were cut: where the whole is, and how to read it.

trim_hist

def trim_hist(
    hist, need, count:NoneType=None, keep:int=2, save:NoneType=None, how:str=''
):

Free about need tokens from hist in place, the oldest tool results first and then the oldest pictures; the last keep messages stay whole. With save, a trimmed result is kept in a file its note names. Returns what was freed.

msg_tokens

def msg_tokens(
    m, count:NoneType=None
):

One history message’s tokens: its text, its tool calls’ arguments, and IMG_TOKENS a picture.

img_tokens

def img_tokens():

margin

def margin():

Backend

def Backend(
    spec, sp:str='', tools:tuple=(), approve:NoneType=None, tool_max_len:NoneType=None, shared:bool=False, **kw
):

The contract a model engine meets: start, send or stream a turn, count tokens, and replace history.

The base class has no engine, so start fails. It returns None and records the NotImplementedError in note instead of raising it:

b = Backend(resolve('gemma-e4b'))
b.start(), b.note

send on a backend that could not start returns the same note, and the failure is in problems. A caller that never checks for errors still shows the user something true:

test_eq(b.send('hello'), b.note)
b.problems

pending_tokens(msg) estimates what a message will add to the conversation. Each image or audio part counts as img_tokens(), config.toml’s window.image_tokens or IMG_TOKENS (1,600), so a 120 KB image is not counted as the tokens of its bytes:

test_eq(b.pending_tokens([bytes(120_000), 'what is this?']),
        b.pending_tokens('what is this?') + IMG_TOKENS)

problem adds a line to problems. It skips a line identical to the one before it and keeps the latest 20:

for _ in range(30): b.problem('ERROR: oom')
len(b.problems), b.problems[-1]

Before the model has started there is no tokenizer, so count_tokens falls back to estimate_tokens. used_tokens is the size of the conversation so far, and pct_full is that size as a share of the window:

b.count_tokens('a prompt of some length'), b.used_tokens, round(b.pct_full, 3)

fits(msg) says whether the conversation plus msg still fits in the window, keeping back a quarter of the window for the reply, but no less than 128 and no more than 1,024 tokens. This is the fits that compact_notebook_context receives. The gemma-e4b spec used here has a 16,384-token window, and a spec with no window (ctx=0) fits anything:

big = 'x' * 200_000
b.fits('a short prompt'), b.fits(big)

test_eq(Backend(ModelSpec('unbounded', 'remote', 'v/m', ctx=0)).fits(big), True)
b.spec.ctx

fits charges the pending message margin() more than its estimate, config.toml’s window.margin or EST_MARGIN (1.15), because a count from characters runs short on code and escapes. A hosted chat’s token_count is what its last step reported, so after replace_hist or restore_hist, used_tokens gives window_tokens, an estimate of the briefing, the tool schemas and every message, until the chat reports again. Without that, a compaction left the old count in place, fits still failed, and each retry compacted again. A hosted chat that has not reported a step yet counts window_tokens too, and one not started yet counts the conversation it will restore, so the first turn after a resume is measured before it goes out.

trim(need) frees tokens from the oldest tool results first, then the oldest pictures, and leaves the newest two messages whole. With spill set, a trimmed result is written whole to a file first, and its note names the file and says how to read it (how). shrink_recent(need) cuts the newest two messages a person or a tool wrote, largest first, which trim never touches. fit_content(content, room, save) does the cutting: pictures past what fits are saved and named, the newest kept, and text past the room becomes excerpt, its head and its end around a note such as …[52,000 characters omitted; full text at <path>; read it in pieces with view_file(path, start=N)]. Nothing is lost: the whole is in the file:

from urai import mk_msgs
t = trim_hist(h := mk_msgs(['look at this', 'done']) + [{'role': 'tool', 'tool_call_id': 'c', 'content': 'r' * 9000}] + mk_msgs(['and now', 'ok']), 100)
test_eq(h[2]['content'], TRIMMED)
assert t > 100 and h[-1]['content'] == 'ok'

saved = []
def save(data, suffix): saved.append(data); return Path(f'/tmp/full{len(saved)}{suffix}')
cut = fit_content('head ' + 'x' * 20_000 + ' end', 200, save, 'read it in pieces with view_file(path, start=N)')
assert cut.startswith('head') and cut.endswith(' end') and 'full text at /tmp/full1.txt' in cut and saved[0].endswith(b' end')
pics = fit_content([b'\x89PNG\r\n\x1a\n1', b'\x89PNG\r\n\x1a\n2', b'\x89PNG\r\n\x1a\n3', 'which?'], 2 * IMG_TOKENS, save)
test_eq(pics[0], b'\x89PNG\r\n\x1a\n3')   # the newest picture stays
assert 'which?' in pics and '2 of the 3 pictures' in pics[-1]
print(cut[:60] + ' ... ' + cut[-160:])

oneshot(prompt) asks one question in a throwaway conversation and returns the answer with any thinking removed. It cuts the prompt to fit the window, keeping its end, and returns if_busy (by default '') while another call is using the backend. This subclass stands in for a model that leaves its thinking in the reply:

class Thinker(Backend):
    def _start(self): return self
    def _oneshot(self, prompt, sp, max_tokens): return 'They want one label.\n</think>\n\nsuccess'

test_eq(Thinker(resolve('gemma-e4b')).oneshot('the tests pass'), 'success')
Thinker(resolve('gemma-e4b')).oneshot('the tests pass')

The history methods behave differently: replace_hist, restore_hist and revise_last_assistant raise when no model is running, because there is no conversation to change. replace_hist(summary, keep) is what Compactor calls. snapshot_hist and restore_hist save and restore a conversation, for branching. resume_hist hands a saved conversation to a backend that has not started yet, and applies it when the model starts.

test_fail(lambda: b.replace_hist('summary'), contains='not running')
test_fail(lambda: b.restore_hist([]), contains='not running')
test_fail(lambda: b.revise_last_assistant('...'), contains='not running')

RishiBackend

RishiBackend is the only implementation of Backend, because rishi runs every model, local or hosted. It turns a ModelSpec (a model’s name, runtime, id and window, from core) into a rishi Chat, converts rishi’s usage into Usage, and handles what differs between runtimes:

A LiteRT model gets its window as the engine’s token ceiling, max_num_tokens, and a llama.cpp or ollama model gets it as n_ctx.
A Claude model is asked for high reasoning effort (CLAUDE_EFFORT) unless RAMABANA_EFFORT names another level.
On MLX, one-shot questions run in a separate conversation on the same engine, so a completion never sees earlier completions.
A local model’s window is set from the window it was trained for, capped by Ramabana’s limit for that model, and narrowed again if the engine reports a smaller one.
When a hosted service refuses the tool schemas, the backend moves that model to the tags channel, where the schemas travel in the system prompt and calls come back as <tool_call> blocks in the text, and retries once.
When a reply on the tags channel describes a tool call instead of making one, the backend sends one reminder (TAG_REMINDER) asking the model to make the call.

rishi adds two callbacks to a chat by default: a reminder on every message, and a sliding window that forgets old messages without summarising them. RishiBackend removes both (ROUTINE_CBS), because Compactor does that job. LiteRT runs the tool loop inside the engine, so the text a model writes before a tool call never reaches Ramabana. QUIET_RUNTIMES lists such runtimes, and narrates is False for them.

LiteRT models run on the GPU by default. rishi asks for the GPU, and falls back to the CPU with a warning when the build has no GPU support. RAMABANA_LITERT_BACKEND=cpu pins a model to the CPU, and gpu asks for the GPU explicitly. A backend= argument, or eng_kw={'backend': ...}, takes precedence over the variable. RISHI_LITERT_GPU=0 stops rishi asking for the GPU at all.

make_backend

def make_backend(
    spec, **kw
):

getattr

def __getattr__(
    name
):

RishiBackend

def RishiBackend(
    *a, max_steps:int=40, **kw
):

The Backend that runs every model, local or hosted, through rishi’s Chat.

use_chat

def use_chat(
    f
):

Build model conversations with f for the duration, instead of rishi’s Chat.

use_chat(f) makes backends build their conversations with f instead of rishi’s Chat, for the duration of a with block. testing uses it to put in urai’s CachedChat, which replays a real model’s recorded answers with no weights and no network. Here f is a stub chat that answers from a script, so nothing is downloaded:

from ramabana.core import ModelSpec

built = []
class StubChat:
    def __init__(self, model, **kw): built.append((model, kw)); self.hist = []
    def oneshot(self, prompt, sp='', think=None, max_tokens=None): return 'replayed'
    def close(self): pass

with use_chat(StubChat):
    be = make_backend(ModelSpec('stub', 'remote', 'openai/gpt-x', 8000))
    assert be.start() is not None
    test_eq(built[0][0], 'openai/gpt-x')
    test_eq(be.oneshot('a cheap job'), 'replayed')
    test_eq(be._usage().total, 0)
test_eq(_MK_CHAT, None)

Ramabana does not set LiteRT’s constrained decoding, since rishi turns it on itself when there are tools. A conv_kw you pass goes through unchanged:

def _a_tool(query: str) -> str:
    "A tool."
    return ''
kw = make_backend(resolve('gemma-e4b'), tools=[_a_tool])._runtime_kw()
assert 'enable_constrained_decoding' not in kw.get('conv_kw', {})
assert 'constrained_decoding_config' not in kw.get('conv_kw', {})
mine = dict(constrained_decoding_config='mine')
test_eq(make_backend(resolve('gemma-e4b'), tools=[_a_tool], conv_kw=mine)._runtime_kw()['conv_kw'], mine)

make_backend(spec) is the constructor to call, and it returns a RishiBackend. The old names LlamaBackend and FastllmBackend still work as aliases of RishiBackend, with a DeprecationWarning:

from litert_lm import Backend as LB
r = make_backend(resolve('gemma-e4b'), eng_kw=dict(backend=LB.GPU()))
import warnings
from ramabana import runtime
with warnings.catch_warnings(record=True) as w:
    warnings.simplefilter('always')
    assert runtime.LlamaBackend is runtime.FastllmBackend is runtime.RishiBackend
    assert [x.category for x in w] == [DeprecationWarning, DeprecationWarning]
r.kind, type(r).__name__

('rishi', 'RishiBackend')

For a LiteRT spec the window goes to the engine as max_num_tokens. The cells below also check that RAMABANA_LITERT_BACKEND is applied, and that an explicit backend takes precedence over it:

r._runtime_kw()

r2 = make_backend(resolve('gemma-e4b'), tools=[interesting])
test_eq(r2._runtime_kw().get('conv_kw'), None)
r2._runtime_kw()['eng_kw']

import os
os.environ['RAMABANA_LITERT_BACKEND'] = 'gpu'
try:
    k = make_backend(resolve('gemma-e4b'))._runtime_kw()
    test_eq((type(k['backend']), 'backend' in k['eng_kw']), (type(LB.GPU()), False))
    k2 = make_backend(resolve('gemma-e4b'), eng_kw=dict(backend=LB.CPU()))._runtime_kw()
    test_eq((type(k2['eng_kw']['backend']), 'backend' in k2), (type(LB.CPU()), False))
finally: del os.environ['RAMABANA_LITERT_BACKEND']

None of this has started an engine. A backend starts its model on the first send, stream or oneshot, so building one costs nothing:

r.ready, r.busy, r.note

Runs

A run is one piece of model work that can be cancelled: the turn the user started (kind='root'), or a sub-agent that the turn delegated to (kind='child', made with child). Runs form a tree. Cancelling a run cancels everything under it, and the tool calls of the whole tree count against one budget, through draw.

A run starts pending, becomes running with start, and ends completed, failed, cancelled, detached or terminated. cancel asks the backend to stop and waits up to grace seconds. A run whose backend does not stop in time is detached: the caller treats it as over and ignores anything the model sends later. terminate closes the backend outright.

A run also keeps what the user should see while it works. status is the last line the model wrote before its latest tool call. notes keeps up to 100 such messages for the now pane, the live view beside the terminal (pane), and write appends a line to the run’s transcript file when it has one. tell queues a message from the user for a running sub-agent, and drain_inbox hands the queued messages over. air returns a reminder to tell the user where the work stands once 8 tool calls in a row have gone by with less than 100 characters said before them. repeated counts identical calls with identical results in a row, so the agent can tell a model that it is looping.

Run

def Run(
    id:str, kind:str='root', question:str='', model:str='', parent:object=None, grace:float=0.25,
    state:str='pending', started:float=0.0, ended:float=0.0, backend:object=None, log:object=None, status:str='',
    answer:str=''
)->None:

A foreground or delegated model call with bounded cancellation.

said_before_call

def said_before_call(
    hist
):

The text of the message that made the latest tool call in hist, or ’’.

status_line

def status_line(
    text, n:int=80
):

The last line of text worth showing as a status, without bullet or bold, clipped to n chars.

A run’s status comes from the text streamed since the last tool call (hear), or, when nothing was streamed, from the message in the backend’s history that made the call. status_line takes the last line worth showing, without list markers or bold, cut to 80 characters:

class _Said:
    hist = [{'role': 'user', 'content': 'go'},
            {'role': 'assistant', 'content': 'Thought about it.\nReading the loader.', 'tool_calls': [{'id': '1'}]}]

r = Run('run_status')
r.start(_Said())
test_eq(r.on_call().status, 'Reading the loader.')
r.hear('Checking the '); r.hear('tests.\n> 🧠 hmm')
test_eq(r.on_call().status, 'Checking the tests.')
test_eq(status_line('x ' * 60), ('x ' * 40).strip())
test_eq(status_line('- **Reading** it.\n```'), 'Reading it.')

A run whose backend stops when asked ends cancelled. A run whose backend cannot stop ends detached once its grace period runs out:

import threading, time

class _Blocking:
    def __init__(self, stoppable=True): self.stoppable, self.released = stoppable, threading.Event()
    def cancel(self):
        if self.stoppable: self.released.set()

r = Run('run_cancelled', grace=.05)
r.start(_Blocking())
threading.Thread(target=lambda: (r.backend.released.wait(), r.finish()), daemon=True).start()
test_eq(r.cancel().state, 'cancelled')

r = Run('run_detached', grace=.01)
r.start(_Blocking(False))
test_eq(r.cancel().state, 'detached')

current_run() returns the run that the current thread or async task is working for, and run_context(run) sets it for a with block. The agent sets it around each turn and each sub-agent, so code deep inside a tool call can find its run without having it passed down.

run_context

def run_context(
    run
):

Make run the current run inside a with block.

current_run

def current_run():

Once its run is cancelled, a backend’s stream stops yielding, drops what the model sends later, and does not retry:

class _CancelStream(Backend):
    def __init__(self):
        from ramabana.core import ModelSpec
        super().__init__(ModelSpec('cancel-test', 'fake', 'fake/cancel', ctx=1000))
        self.first, self.release, self.calls, self.recoveries = threading.Event(), threading.Event(), 0, 0
    def _start(self): return self
    def _stream(self, msg, **kw):
        self.calls += 1
        yield 'first'
        self.first.set(); self.release.wait()
        yield 'late'
    def cancel(self): self.release.set(); return True
    def _usage(self): return Usage()
    def _recover(self, e): self.recoveries += 1; return True

be, run, chunks = _CancelStream(), Run('run_late', grace=.05), []
run.start(be)
t = threading.Thread(target=lambda: chunks.extend(be.stream('x', run=run)), daemon=True)
t.start(); assert be.first.wait(.2)
run.cancel(); t.join(.2)
test_eq(chunks, ['first'])
test_eq((be.calls, be.recoveries), (1, 0))

terminate ends a run even after it was detached, and closes its backend:

class _Closable:
    def __init__(self): self.closed = threading.Event()
    def close(self): self.closed.set()

r = Run('run_terminated', grace=0)
r.start(_Closable()); r.detach(); r.terminate()
test_eq(r.state, 'terminated')
assert r.backend.closed.wait(.2)

add_cb registers a rishi callback class on a backend. It is applied to the current chat and to any chat that replaces it later. takes_pictures says whether an image returned by a tool can reach the model: the model has to accept images and its transport has to carry them. A model that cannot see a picture gets urai’s text note in its place. start is redefined here so that it applies both when the model starts.

Backend.start

def start():

Backend.takes_pictures

def takes_pictures():

Whether a picture a tool returns can reach this model: it accepts images and its transport carries them.

Backend.add_cb

def add_cb(
    cb
):

Register one Rishi callback class, for this chat and for any that replaces it.

TokenLogger is a rishi callback that prints the usage the provider reports after each response. Set TokenLogger.sink to send the line somewhere other than print. CHAT_CALLBACKS maps the names that Agent.add_chat_callback accepts to their classes.

TokenLogger

def TokenLogger(
    *args, **kwargs
):

Report Rishi’s provider-reported usage after each response.