testing

In-memory hosts and scripted backends for tests and documentation.

Specs

A ModelSpec describes one model: its name, the runtime that serves it, the model id and the context window in tokens (ctx). The two specs here differ mainly in that window. SPEC has 1,000 tokens, small enough that compaction arithmetic can be checked by hand. Compaction is how the agent keeps a long conversation inside the window: it replaces older messages with a summary. SCRIPTED has 8,000 tokens, so compaction never fires in the middle of a recorded demo.

SPEC, SCRIPTED.ctx

A host in memory

MemHost keeps the project’s files in a dict keyed by absolute path, and the file and search tools read and write that dict instead of the disk.

A host says what it can do by the classes it inherits. Each capability class, such as CodeHost for code search or ShellHost for running commands, names a capability group, and tools_for(host) builds the tools of every group the host declares. MemHost starts from NullHost, which declares only the file group and offers a lone view_file, and adds CodeHost and ShellHost. Every other group, the live Python session among them, contributes no tools. Checking that a missing group really removes its tools is one of the main uses of this class.

commands scripts the shell. It maps a command string to (exit_code, output). A command not in the dict exits 0 with no output, and every command that runs is appended to host.cmds. The cell below builds a host with two files.

---

source

MemHost

def MemHost(
    files:NoneType=None, root:str='/proj', commands:NoneType=None
):

A host whose folders live in a dict, so the file tools never touch disk.

host = MemHost({'/proj/a.py': 'def a(): return 1\n', '/proj/b.py': 'def b(): return a() + 1\n'})
host.roots, host.walk()

search is a plain substring scan, but it returns Hit objects, the same type a real code index returns, so code that formats search results sees the usual shape.

host.search('return a'), host.search_note

write stores text in the dict and text_at reads it back, with an empty string for a file that does not exist. That is all Agent.changes() needs to report which files a turn edited, so it can be tested against this host.

host.write('/proj/c.py', 'def c(): pass\n')
host.text_at('/proj/c.py'), host.text_at('/proj/never.py')

run_python records the code instead of running it, so a test can check what the agent tried to run without starting a kernel. A test has to call the method directly. MemHost does not declare the session group, so tools_for gives the model no run_python tool for it, as the second cell checks. host.provides lists the groups the host does declare.

host.run_python('df2 = df.dropna()'), host.ran

test_eq(host.can('session'), False)
test_eq(names(tools_for(host)) & {'run_python', 'list_vars', 'inspect_python'}, set())
sorted(host.provides)

A host with everything

LocalHost is the host the terminal session runs on: real folders on disk, a real shell and a real Python kernel. Some of its groups are switched off because they need something it lacks. Memory and watches need a vault, the web tools need the network, and its inspect_python has only the isolated scope. Tests for those tools find nothing to test on a LocalHost.

FullHost is a LocalHost that fills each gap with a small working implementation over in-process state. Its web is a dict of pages you pass in. Its memory really stores and searches the pages it has read. Its terminal is a transcript you supply. Nothing reaches the network. It works in a fresh temporary folder unless you pass root, so a test or a docs page can write files freely. Three groups stay off because they need a model, an API specification or a vault: ask, api and watch.

---

source

FullHost

def FullHost(
    files:NoneType=None, pages:NoneType=None, root:NoneType=None, terminal:str='', **kw
):

Every host capability in process over a temporary folder, never the network.

Here it is seeded with one file, one web page and a terminal that has already printed a test run. The cell lists every tool tools_for builds for it, and the next cell checks that this is a strict superset of what NullHost gets.

host = FullHost(
    files={'pkg/sizes.py': 'RESERVE = 16_384\n\ndef threshold(ctx):\n    return max(1, ctx - min(RESERVE, ctx // 4))\n'},
    pages={'https://nbdev.fast.ai/api/export.html': (
        '# Exporting a notebook to a library\n\n'
        '## nb_export\n\n'
        'Named exports append to a module another notebook owns, so a second run duplicates them.\n\n'
        '## default_exp\n\n'
        'One module per notebook, named by the default_exp directive.\n')},
    terminal='$ pytest -q\n3 failed, 66 passed')
[t.__name__ for t in tools_for(host)]

names = {t.__name__ for t in tools_for(host)}
assert 'run_shell' in names
test_eq({t.__name__ for t in tools_for(NullHost())} < names, True)
host.kernel_kind, host.concurrent, host.scopes

Underneath it is a real LocalHost. The sandbox that keeps every path inside the open folders, the file tools and the Python namespace are the production code. The next cells view a file through the real view_file tool, show a path outside the folder being refused, and run Python in the live namespace.

view_file, replace_text, create_file, add_root = file_tools(host)
print(view_file('pkg/sizes.py'))

test_fail(lambda: host.check('/etc/passwd'), contains='outside the open folders')
host.run_python('total = threshold_calls = 0\ntotal'), host.list_vars().splitlines()

The web is the dict of pages passed in. web_search matches the query against each page’s url and text, and read_url returns a page. Reading a page also stores it in memory, which gives the memory tools below something to find.

web_search, read_url = web_tools(host)
print(web_search('nbdev'))
print(read_url('https://nbdev.fast.ai/api/export.html')[:80])

Memory is where the agent keeps pages and notes beyond the current conversation. A memory search returns sections of a stored document, each with a breadcrumb naming the document and heading it came from, rather than loose snippets. FullHost makes those sections by splitting a page at its headings, and memory_tree lists the headings of each stored document.

memory_search, memory_read, remember, memory_forget = memory_tools(host)
host.memory_search('duplicates')

test_eq(len(host.remembered), 1)
host.memory_tree()[0]['headings']

memory_forget deletes a stored document and its sections. The same search then finds nothing.

doc_id = next(iter(host.remembered))
memory_forget(doc_id), host.memory_search('duplicates')

research searches the web, reads every page that matches and returns them as one digest, with a source line under each page. It belongs to an opt-in tool group: tools_for(host) leaves it out and tools_for(host, optin='research') includes it. This cell builds the tool directly with shalya.research_tools.

from shalya import research_tools

research, = research_tools(host)
print(research('export')[:220])

inspect_python lets the agent look at the user’s live Python namespace. It has two scopes. The isolated scope runs the code against a copy of the namespace. The overlay scope runs it against the real namespace, so it sees every object as it is, and afterwards puts back every name that existed before. An IDE host enforces the same rule by checking the code’s syntax tree before running it; FullHost restores the bindings after.

Restoring bindings undoes an assignment such as kept = None, but it does not undo a change made inside an existing object. The cell below appends to kept in the overlay scope, and the list stays one item longer.

host.run_python('kept = [1, 2, 3]')
host.inspect_python('kept.append(99)\nlen(kept)', scope='overlay'), host.run_python('len(kept)')

The isolated scope copies the namespace dict, not the objects in it. A new name defined there disappears, but kept is the same list in both, so this append also lands on the real list, as the output shows.

host.inspect_python('kept.append(4)\nlen(kept)', scope='isolated'), host.run_python('len(kept)')

read_terminal is the tool that reads the terminal beside the agent. On FullHost it returns the transcript passed as terminal. note is how a host leaves a message for the agent, such as a report that compaction ran, and FullHost also keeps every note in host.notes so a test can check it.

read_terminal = {t.__name__: t for t in session_tools(host)}['read_terminal']
print(read_terminal())

host.note('compacted 12 messages')
host.notes

A backend with no model

FakeBackend replies from a list of strings, in order, and says (done) once the list runs out. It subclasses the real Backend, so these parts are production code:

• the lock that serialises calls
• the usage accounting
• the list of problems shown to the user
• the history replacement that compaction uses

Only the replies are made up. Every message sent is kept in sent, and every sub-agent backend made by spawn in spawned. Its one-shot calls, used for short jobs such as summaries, return ONESHOT: followed by the start of the prompt.

---

source

FakeBackend

def FakeBackend(
    spec:ramabana.core.ModelSpec=ModelSpec(name='fake', runtime='fake', model_id='fake/model', ctx=1000, note='', opts={}, missing=''),
    replies:tuple=(), **kw
):

A backend over a scripted list of replies, for turns with no model.

be = FakeBackend(replies=['the first answer', 'the second'])
be.send('a question'), be.send('another'), be.send('one more than were scripted')

The history holds both the user and the assistant messages, so compaction has a real conversation to shorten.

be.hist[:2]

stream yields the same reply one word at a time. That is enough to test a frontend that draws text as it arrives.

''.join(FakeBackend(replies=['streamed one word at a time']).stream('go'))

use is a Usage record of tokens in and out, the same type a live backend reports, and two of them add up.

be.use, be.use + be.use

fake_agent builds a real Agent and points all of its jobs at one FakeBackend. A job is one kind of model call the agent makes: the turn itself, a summary, a classification, a sub-agent. Normally each job can be routed to a different model. Routing every one to the fake means no call can reach a real model.

The host is a MemHost with one file unless you pass host, and other keyword arguments go to Agent. It returns (agent, backend), so a test can script the replies and then read what was sent. Use it whenever a test needs a whole turn, and FakeBackend alone when it only needs a backend.

---

source

fake_agent

def fake_agent(
    host:NoneType=None, replies:tuple=(), **kw
):

Build an Agent whose every job runs on one FakeBackend, and return (agent, backend).

agent, backend = fake_agent(replies=['`threshold` lives in `ramabana/runtime.py`.'])
agent.ask('where is the compaction threshold?')

The turn ran through the real agent: it built the tools, recorded the activity lines a frontend shows, and counted the tokens.

len(agent.tools), agent.turn_lines(), repr(agent.use)

A real model, recorded

The doubles above show the control flow with no model at all. A docs page that shows what a real model does is harder. Running a multi-gigabyte model in CI is too slow, and pasting a transcript means nobody checks it still holds. recorded sits between the two. Inside with recorded(): every chat the agent opens is urai’s CachedChat, which stores what the model said the first time, keyed by the model, the briefing, the tools and the conversation so far. Later runs replay the stored answer and start no model. Every runtime counts as available inside it, so a page replays Claude’s answers on a machine without Claude Code.

Answers live in CHATS, the chatcache folder beside the package, unless you pass path. record=False only replays. record=True lets a miss reach the real model and stores the answer. With record=None the URAI_RECORD_CHAT environment variable decides. The example uses an empty cache with record=False: the one-shot call gets an empty answer, and the backend’s problems say no recording was found.

---

source

recorded

def recorded(
    path:NoneType=None, record:NoneType=None
):

Replay recorded model answers for the duration, instead of calling a model; every runtime counts as available.

import tempfile
from ramabana.core import ModelSpec
from ramabana.runtime import make_backend

with recorded(Path(tempfile.mkdtemp())/'chats', record=False) as p:
    be = make_backend(ModelSpec('unrecorded', 'remote', 'openai/gpt-x', 8000))
    test_eq(be.oneshot('nothing was ever recorded for this'), '')
    assert 'no recording' in ' '.join(be.problems)
p.name

'chats'

from ramabana.core import resolve

with patch('ramabana.core._claude_available', lambda: False):
    with recorded(Path(tempfile.mkdtemp())/'chats', record=False): test_eq(resolve('claude/claude-sonnet-4-5').backend, 'claude')
    assert 'unavailable' in resolve('claude/claude-sonnet-4-5').missing

VaultHost, the host with durable memory from the vault notebook, rebuilds its memory graph in a background thread with a chat called graph_chat. no_model returns a stand-in chat that answers {} to every prompt, so that thread starts no model. It also works as mk_chat. This matters under nbdev’s test runner, which runs several notebooks in each worker process: two of them loading a model onto the GPU at once can crash the worker.

---

source

no_model

def no_model(
    model:NoneType=None, **kw
):

Return a stand-in chat that answers {} to everything, for use as graph_chat or mk_chat in tests.

test_eq(no_model('any/model', sp='x').oneshot('extract the entities'), '{}')
test_eq(no_model()('hello')['content'], '{}')

Compaction, end to end

Compaction replaces the older part of a conversation with a summary followed by the most recent messages. The agent compacts on its own when the history nears the window, and agent.compact() does it on demand. After six more questions the fake_agent history is still well under its 1,000-token threshold, so compactor.due is False. The second cell compacts anyway, with the summary written by the fake’s one-shot answer.

for i in range(6): agent.ask(f'question {i} about the exporter')
len(backend.hist), agent.compactor.due(backend)

summary = agent.compact()
agent.compactor.note

The history now begins with the summary. The text after it tells the model what happened: its conversation was rewritten, but the Python kernel was not, so the user’s variables are still there.

print(backend.hist[0]['content'][:200])

test_eq(backend.hist[0]['content'].startswith('Previous conversation summary:'), True)
test_eq(agent.compactor.count, 1)
agent.compactor.last[:60]

A backend that mutters

Local model engines sometimes fail without raising. The engine prints its error to its own output stream, returns an empty string, and the caller sees an empty reply with no reason. Ramabana’s local backend keeps what the engine printed and adds it to the backend’s problems. MutteringBackend reproduces the failure so tests can check that handling. GEMMA is a real error line from the LiteRT engine, kept exactly as printed, so the test sees the real shape of it.

---

source

MutteringBackend

def MutteringBackend(
    spec:NoneType=None, mode:str='empty', **kw
):

A backend that fails the way litert does: it prints, returns nothing, and raises nothing.

mute = MutteringBackend()
mute.send('anything'), mute.problems

That is the contract. send returns a sentence saying the model returned nothing and why, instead of an empty string, and last_native keeps the engine’s own words for the status line. Sending again does not add the same problem twice.

test_eq(mute.last_native, GEMMA.strip())
mute.send('again')

test_eq(len(mute.problems), 1)
mute.spec.ctx

A backend that performs

ScriptedBackend plays a list of Steps. A step can call a tool, say some text, or both, and can pause first. The text streams out one word at a time, token_delay seconds apart. The delay exists for screenshots: at zero the turn finishes before the screen repaints, and every capture shows a finished answer where it should show one in progress.

A tool step looks the tool up by name in the backend’s tool list, which refresh sets, and calls it. With the agent’s own tools in that list, the activity feed, the approval gate and the file snapshots all run as they do in production. A step naming a tool the list lacks is skipped. Use FakeBackend to test logic, and ScriptedBackend for demos where the tool calls have to happen.

---

source

ScriptedBackend

def ScriptedBackend(
    spec:ramabana.core.ModelSpec=ModelSpec(name='scripted', runtime='scripted', model_id='scripted/model', ctx=8000, note='', opts={}, missing=''),
    steps:tuple=(), token_delay:float=0.02, **kw
):

Play a list of Steps, streaming words one at a time every token_delay.

---

source

Step

def Step(
    text:str='', tool:NoneType=None, pause:float=0.0
):

One scripted model action: call a tool through the real tool list, or say something.

script = [Step('Looking at the file.', tool=('view_file', {'path': '/proj/a.py'})),
          Step('It returns 1, so the caller gets an int.')]
play = ScriptedBackend(steps=script, token_delay=0)
play.refresh('', [t for t in agent.tools if t.__name__ == 'view_file'])
play.send('what does a() return?')

The tool really ran. Its result is in the history as a tool message, where compaction or a UI would find it.

[m['role'] for m in play.hist], play.hist[2]['content'][:40]

spawn returns the backend for a sub-agent, and its answer depends on the question: it looks for a keyword from SUB_ANSWERS in it. When a demo fans out three sub-agents, three different answers show that they are separate calls. Its usage is fixed, as the last cell shows.

sub = play.spawn()
''.join(sub._run('which files import fastllm?')), ''.join(sub._run('what shape is df?'))

test_eq(play.use.total, 1936)
play.use

A real model with every tool

The doubles above check control flow. This last section checks the assembled product: a FullHost, an Agent over it on a made-up remote model, and every tool the host supports handed to the model’s conversation.

Everything runs inside recorded(), so it starts no model and makes no network call. The two cells that would talk to the model are marked #| eval: false, so the test suite skips them. Run by hand, they check that the model compiled every tool schema and that it can answer.

from contextlib import ExitStack

replay = ExitStack()
replay.enter_context(recorded())

# `remote`: `resolve` refuses a runtime that is not installed
register_model('release-model', 'openai/release-model', 'remote', 32_768)
real_host = FullHost(
    files={'pkg/sizes.py': 'RESERVE = 16_384\n\ndef threshold(ctx):\n    return ctx - RESERVE\n'},
    pages={'https://example.test/nbdev': '# nbdev export\n\nOne module belongs to one notebook.'},
    terminal='$ pytest -q\n69 passed')
real_agent = Agent(real_host, model='release-model', extensions=False)
for job in ('classify', 'summary', 'completion', 'subagent'):
    real_agent.routing.set('release-model', job)
real_agent.start() is not None, real_agent.note

(False,
 'turn job: release-model needs OPENAI_API_KEY; set it in the environment. `ramabana setup` lists the providers and adds one')

Agent.tools is the complete tool list: code, files, notebooks, web, memory, the live session, the shell, skills, delegation, the plan and git, plus image generation when a key for it is set. None of them is a stub that always fails. The next cell checks the list group by group.

real_tool_names = [tool.__name__ for tool in real_agent.tools]
real_tool_names

FILES = {'search_code', 'outline', 'grep', 'ls',
         'view_file', 'replace_text', 'create_file', 'add_root'}
NOTEBOOK = {'notebook_cells', 'view_cell', 'edit_cell', 'add_cell'}
WEB = {'web_search', 'read_url'}
MEMORY = {'memory_search', 'memory_read', 'remember', 'memory_forget'}
SESSION = {'run_python', 'inspect_python', 'restart_kernel', 'read_terminal'}
SHELL = {'run_shell', 'run_shell_bg', 'shell_output', 'shell_stop'}
SKILLS = {'read_skill'}                      # `create_skill` is the `author` opt-in
DELEGATION = {'delegate_search', 'delegate_async', 'delegate_result', 'delegate_cancel'}
PLAN = {'set_plan', 'update_todo'}
# only with a key for the endpoint
IMAGE = {'generate_image'} if image_available() else set()
GIT = {'git_status', 'git_divergence', 'git_remote', 'git_checkout',
       'git_commit', 'git_diff', 'git_log', 'git_stash'}
expected = (FILES | NOTEBOOK | WEB | MEMORY | SESSION | SHELL | SKILLS | DELEGATION | PLAN | IMAGE
            | GIT)
test_eq(set(real_tool_names), expected)
len(real_tool_names)

Attaching the tools to Agent does not prove the model accepts them. The backend’s live chat compiles a schema for each tool, so comparing the number of compiled schemas with the number of tools catches a missing type annotation, a lost functools.wraps, or a tool shape the runtime rejects.

real_backend = real_agent.backend
test_eq(real_backend.ready, True)
test_eq(len(real_backend.chat.toolspecs), len(real_agent.tools))
test_eq(len(real_backend.chat.toolspecs), len(expected))
(real_backend.spec.model_id, len(real_backend.chat.toolspecs))

Then the model is asked for a one-word reply on the classify job, which is routed to the same model. A non-empty reply shows the model really generates text, which accepting the tool schemas alone would not show.

real_reply = real_agent.oneshot(
    'Reply with exactly READY and nothing else.',
    'Return only the requested answer.', job='classify', max_tokens=192)
assert real_reply.strip()
real_reply

real_agent.close()
replay.close()
real_agent.status()['model'], real_agent.use.total