tools

Hosts, tool lists and sub-agents: what Ramabana adds on top of the shalya toolset.

A model can only write text. To read a file or run a command, it calls a tool. A tool is a Python function. The backend turns its name, docstring and typed parameters into a schema that the model sees. Ramabana’s tools come from shalya.

The tools act through a host. A host is the object that decides which folders are open (its roots), whether files may be written and which other services exist. Such services include a shell, a Python kernel, git and the web. Each service is a capability group. A host declares a group by inheriting the class that names it, such as CodeHost or WebHost. Host.provides lists the groups a host declares. Every host has the file group. The tools that need one capability group form a tool group. The shell group, for example, is run_shell, run_shell_bg, shell_output and shell_stop.

Shalya holds the host classes, the functions that build each tool group, the conventions tool results follow (err writes a failure that failed recognises), and the skills registry. This module re-exports all of them. Code that imports them from ramabana.tools keeps working. The alias sys.modules[__name__] = ramabana.tools, which Leela (the notebook editor) uses, keeps resolving. New code imports them from shalya.

What this module adds depends on parts of the agent that shalya does not know. A sub-agent needs a Backend to talk to and a Run to record it. draws_itself needs a ModelSpec. The sections below cover a host that refuses everything, the image tools, tools_for, which builds the tool list a turn gets, delegation to sub-agents, and attachments.

---

source

getattr

def __getattr__(
    name
):

Resolve the deprecated Shalya compatibility surface without mirroring it here.

The re-exported names are shalya’s own objects, not copies. A copy could drift into a second implementation. The cell below checks that every name in _all_ resolves and is the same object that shalya holds. Shalya’s own notebooks test what the objects do.

len(_all_), sorted(_all_)[:8]

import shalya.core, shalya.host, shalya.skills, shalya.tools
here = sys.modules[__name__] if False else __import__('ramabana.tools', fromlist=['x'])
missing, different = [], []
for n in _all_:
    if not hasattr(here, n): missing.append(n); continue
    src = next((m for m in (shalya.core, shalya.host, shalya.skills, shalya.tools) if hasattr(m, n.lstrip('_')) or hasattr(m, n)), None)
    if src is None: continue
    theirs = getattr(src, n, None) or getattr(src, n.lstrip('_'), None)
    if getattr(here, n) is not theirs: different.append(n)
test_eq(missing, [])
test_eq(different, [])
test_eq((_cmds, _edits, _apply_edits, _diff),
        (shalya.core.cmds, shalya.core.edits, shalya.core.apply_edits, shalya.core.diff_text))

WRITE_TOOLS names every tool with an effect: tools that change files, the repository, the kernel, the shell, the vault or a browser page. It comes from shalya unchanged.

sorted(WRITE_TOOLS - shalya.core.WRITE_TOOLS)

assert {'replace_text', 'run_shell', 'git_commit'} <= WRITE_TOOLS

The examples below run against a real LocalHost, the host the terminal uses, over a small project in a temporary folder. index=False skips building the semantic code index. provides lists the capability groups the host declares.

root = Path(tempfile.mkdtemp()).resolve()/'proj'
(root/'pkg').mkdir(parents=True)
(root/'pkg'/'sizes.py').write_text('def threshold(n):\n    "Half of n."\n    return n // 2\n')
local = LocalHost([root], index=False)
sorted(local.provides)

test_eq(local.can('file'), True)

A host with nothing behind it

NullHost is a host with no folders and no services. Checking a path raises HostError, reading returns nothing, and writing or deleting raises. It declares no capability group beyond file. Its writes is False, and tools_for gives it view_file and nothing else. Use it in a test, or wherever code needs a host object that must not touch the disk. The examples under it, and two further down, show each refusal.

The same cell adds delete and exists to LocalHost.

---

source

LocalHost.exists

def exists(
    path
):

Whether path is a file inside the roots (an undecodable file counts).

---

source

LocalHost.delete

def delete(
    path
):

Remove one file inside the roots; one already gone is not an error.

---

source

NullHost

def NullHost(
    roots:tuple=()
):

A host with no folders and no services, which refuses every file operation.

test_eq([t.__name__ for t in tools_for(NullHost())], ['view_file'])
test_fail(lambda: NullHost().write('/a', 'x'))

Drawing

Shalya’s generate_image tool makes pictures. Some models can draw themselves when they are given the image tool. For any other model the prompt goes to a separate image model. Which case applies is a fact about the turn’s model, so shalya asks for it as a callable, and this module supplies the callable.

draws_itself looks up a model’s capabilities with spec_caps and returns True when image is among the tools the model supports (Caps.tools). The supported_output_modalities field cannot answer this: it says no for every chat model, because it describes what a model returns without being asked.

---

source

image_tools

def image_tools(
    host, mx:int=6000, session:str='', get_spec:NoneType=None, on_media:NoneType=None
):

Shalya’s image tools for host, checking the turn’s model on every call.

---

source

draws_itself

def draws_itself(
    spec
):

Whether the model behind spec can draw pictures itself when given the image tool.

The example fakes the capability table with mock_caps, so it runs without a model. The first spec lists image among its tools and the second does not. No spec at all gives False.

from types import SimpleNamespace
draws = ModelSpec('drawer', 'remote', 'openai/gpt-5.6-luna')
plain = ModelSpec('plain', 'remote', 'openai/gpt-5.6')
caps = {'openai/gpt-5.6-luna': SimpleNamespace(tools=('image', 'search')), 'openai/gpt-5.6': SimpleNamespace(tools=('search',))}
with mock_caps(caps): got = draws_itself(draws), draws_itself(plain), draws_itself(None)
got

(True, False, False)

with mock_caps(caps):
    test_eq(draws_itself(draws), True)
    test_eq(draws_itself(plain), False)
test_eq(draws_itself(None), False)
with mock_caps({}): test_eq(draws_itself(draws), False)

image_tools builds shalya’s image group for a host, with draws_itself asked about the turn’s model each time the tool is called. Pass get_spec, a callable returning the current ModelSpec. The tool then sees a model switch mid-session. Without an image API key the tool returns a message saying so and makes no request. This example therefore runs anywhere.

gen = image_tools(local, get_spec=lambda: draws)[0]
gen.__name__, image_available()

if not image_available(): assert failed(gen('a kettle'))
assert failed(gen('a kettle', size='3x3'))
test_eq(len(image_tools(local, get_spec=lambda: None)), 1)

Assembling the tool list

tools_for decides which tools a host gets. It asks Host.provides for the capability groups and adds the tool group for each. A get_skills callable adds read_skill, and extra adds extension tools. drop removes groups by name, such as 'shell' or 'image'. optin adds groups that are off by default: exhash (the hash-addressed edit_file), research, author, legacy (tool names from older releases) and browser.

Shalya has a tools_for of its own. Use this one in Ramabana code, because it also builds the image group from the turn’s model through get_spec and on_media, which a host cannot know. The image group comes only when an image key is set and the host can write, since generate_image saves files. On a LocalHost without litesearch installed, search_code is left out.

small_tool marks an extension tool to keep in the small profile, the short catalog a local model or a small context window gets. ToolCatalog wraps a tool list once and answers questions about it: each tool’s group, which tools write, which have effects, and a read-only copy. ToolEntry is one row of it.

---

source

ToolCatalog

def ToolCatalog(
    tools:tuple=()
):

An immutable list of tools that answers each one’s group, whether it writes, and whether it has effects.

---

source

ToolEntry

def ToolEntry(
    tool:object, group:str=''
)->None:

One callable and the Shalya metadata every consumer needs.

---

source

small_tool

def small_tool(
    f
):

Mark an extension tool for the small profile, whose catalog omits unmarked tools.

---

source

tools_for

def tools_for(
    host, get_skills:NoneType=None, extra:tuple=(), mx:int=6000, drop:tuple=(), get_spec:NoneType=None,
    on_media:NoneType=None, optin:tuple=()
):

The tools for host: a group per capability it declares, plus skills, images and extensions, less drop and plus optin.

The list for the real host above. It has the file, code, notebook, shell, git and web groups. The memory and API tools are absent because LocalHost does not declare those groups here, and generate_image appears only when an image key is set.

sorted(t.__name__ for t in tools_for(local))

ts = tools_for(local)
names = {t.__name__ for t in ts}
assert {'search_code', 'view_file', 'run_python', 'run_shell'} <= names, sorted(names)
assert not (names & {'memory_tree', 'api_load', 'ask_memory'})
test_eq('generate_image' in names, image_available())
test_eq(names - {t.__name__ for t in tools_for(local, drop=['shell'])},
        {'run_shell', 'run_shell_bg', 'shell_output', 'shell_stop'})
assert 'read_skill' in {t.__name__ for t in tools_for(local, get_skills=lambda: [])}
assert 'read_skill' not in names
_installed, installed = installed, lambda m: m != 'litesearch'
try: assert 'search_code' not in {t.__name__ for t in tools_for(local)}
finally: installed = _installed

Opt-in groups come only on request. edit_file appears with optin=('exhash',), the old tool names with optin='legacy', and an unknown name raises.

assert 'edit_file' not in names
assert 'edit_file' in {t.__name__ for t in tools_for(local, optin=('exhash',))}
assert 'list_files' in {t.__name__ for t in tools_for(local, optin='legacy')}
test_fail(lambda: tools_for(local, optin=('nosuch',)))

h = NullHost(['/proj'])
sorted(h.provides), [t.__name__ for t in tools_for(h)]

test_fail(lambda: h.check('a.py'))

Every method of NullHost refuses. Checking a path raises because no folders are open, reading gives nothing, and opening another folder raises. It is the reference for how a host says “I cannot”.

h.walk(), h.read('a.py'), h.text_at('a.py')

test_eq(h.walk(), [])
test_eq(h.read('a.py'), None)
test_eq(h.text_at('a.py'), None)
test_fail(lambda: h.write('a.py', 'x'))
test_fail(lambda: h.check('a.py'))
test_fail(lambda: NullHost().add_root('/tmp'))

Sub-agents

A sub-agent is a separate conversation with a model, started to answer one question for the agent that asked. Handing a question over this way is delegation. Delegation saves context. It does not save time. A broad question can take twenty tool calls to answer. If the main agent made them itself, every result would stay in its conversation. Delegated, the same work costs the main conversation one question and one answer. The sub-agent’s conversation is thrown away when it finishes.

A sub-agent gets read-only tools unless the session allows it to write. NO_SUB lists the tools it never gets by default: the five delegation tools and the two watch tools (watch and cancel_watch). A watch would outlive the sub-agent’s task. A session that allows nesting passes its own block. The sub-agent may then delegate in turn. These constants set the limits:

Constant	Value	Meaning
SUB_MAX_STEPS	80	Steps a delegation gets when it names none.
SUB_STEP_CEILING	400	Most steps a delegation may ask for under auto, and the most a numeric session cap may be.
SUB_MIN_STEPS	20	Fewest steps a delegation runs with when it asks for fewer.
SUB_TIMEOUT	1800	Seconds before a delegation is stopped. 0 never stops it.
SUB_STEP_SECONDS	20	Under auto steps, a delegation with the default time limit gets its steps times this many seconds, when that is more than SUB_TIMEOUT.
SUB_TIMEOUT_CEILING	10800	Most seconds that step-based limit can reach.
SUB_UNFINISHED	300	A reply with no tool call and no state: line, shorter than this many characters, counts as a progress line and not a report.
SUB_SLOTS	4	Sub-agents working at once across one agent’s whole delegation tree.

A briefing is the standing instructions a model gets before the conversation starts, its system prompt. sub_briefing builds a sub-agent’s briefing from parts in this order: SUB_SP_HEAD, which every sub-agent gets, then SUB_READ_SP or SUB_WRITE_SP, then an optional note on how much of the answer the caller will read, then notes for the model’s family.

---

source

sub_briefing

def sub_briefing(
    writes:bool=False, answer_max:int=0, # characters the caller reads of the answer; 0 says nothing about it
    notes:str='', # the sub-agent model's family notes, after the rest
):

The sub-agent standing instructions: the shared half, the read-only or the write half, the answer budget and the family notes.

---

source

inbox_note

def inbox_note(
    key
):

The one tag the model may take as the user: it carries key, which no file or page knows.

SUB_SP is the read-only briefing with nothing added, and it is the default. A sub-agent allowed to write gets SUB_SP_HEAD followed by SUB_WRITE_SP.

SUB_MAX_STEPS, len(SUB_SP.split()), len(SUB_WRITE_SP.split())

assert SUB_MIN_STEPS < SUB_MAX_STEPS

read_only, from shalya, filters a tool list down to what an agent may have when it must not change anything. Write tools are dropped, or swapped for a safe version where the tool has one. block drops tools by name, and Ramabana passes NO_SUB. effects=False also drops the acting tools, those that do something without writing a file, such as inspect_python, shell_stop and generate_image. Shalya marks them with acts and lists them in ACTING_TOOLS. Use effects=False for a surface that may look and propose but never act.

A safe version can fix an argument. In the second cell below, read_url still fetches the page, but its remember parameter is gone and it no longer stores the page in the vault.

[t.__name__ for t in read_only(ts, block=NO_SUB)]

import inspect

test_eq(set(t.__name__ for t in read_only(ts, block=NO_SUB)) & WRITE_TOOLS, set())
shut = set(t.__name__ for t in read_only(ts, effects=False, block=NO_SUB))
test_eq(shut & (WRITE_TOOLS | NO_SUB | ACTING_TOOLS), set())
test_eq('search_code' in shut, True)

kept = []
class Page: text = 'page'
class Keeps(MemHost, WebHost):
    def web_search(self, query, n=20): return []
    def research(self, query): return ''
    def read_url(self, url, remember=True): kept.append(remember); return Page()
fetch = next(t for t in read_only(tools_for(Keeps({'/proj/a.py': 'x = 1\n'}))) if t.__name__ == 'read_url')
fetch(url='https://example.com')
test_eq(kept, [False])
test_eq('remember' in inspect.signature(fetch).parameters, False)
sorted(set(t.__name__ for t in ts) & ACTING_TOOLS)

max_calls puts a hard budget on the list. Past it, every tool returns a message telling the model to stop and answer. Local engines run their own tool loop. Ramabana cannot stop them between steps. A budget inside the tools is the one limit that works on every backend. Below, the second search is refused.

budgeted = read_only(ts, max_calls=1, block=NO_SUB)
search = next(t for t in budgeted if t.__name__ == 'search_code')
search('return'), search('return')

delegate asks one question of a sub-agent. It spawns a new conversation on the same engine as backend, hands it the tools that read_only lets through (with writes=True, the write tools too), sends the question, and closes the conversation in a finally block whether or not it succeeded. The tools carry a call budget of four times max_steps. A short reply with no tool call and no state: line gets one nudge (SUB_NUDGE), and the sub-agent answers again. delegate returns the answer text. An empty or repetitive answer is replaced by a notice that the delegation failed. A sub-agent that raised gives an error result. timeout cancels the run after that many seconds, and slots, a shared Slots, makes it wait for a free place among the sub-agents working at once. A sub-agent waiting on sub-agents of its own lends its place back. A full tree of delegations then cannot deadlock.

Use delegate for one question. delegate_many, below, asks several, and subagent_tools wraps both as tools a model can call.

---

source

delegate

def delegate(
    backend, question, tools:tuple=(), sp:NoneType=None, max_steps:int=80, skills:tuple=(),
    writes:bool=False, # hand over WRITE_TOOLS as well
    approve:NoneType=None, # the gate those writes answer to
    run:NoneType=None, # a pre-registered child run
    images:tuple=(), # picture bytes sent with the question
    block:frozenset=frozenset({'delegate_tell', 'delegate_cancel', 'delegate_search', 'delegate_async', 'delegate_result', 'watch', 'cancel_watch'}), # tool names withheld from the sub-agent
    timeout:int=0, # seconds before the run is cancelled; 0 never
    slots:NoneType=None, # a shared `Slots`, held while the sub-agent works
    spawn:NoneType=None, # options for `backend.spawn`, such as `effort` and `tool_mode`
):

Ask question of a sub-agent in a separate conversation on backend’s engine, and return its answer text.

---

source

Slots

def Slots(
    n:int=4
):

A bound on sub-agents working at once; one waiting on its own sub-agents lends its slot back.

---

source

status_tool

def status_tool(
    run
):

A display-only status tool for run, or the callable returning it, whose model’s text between calls never reaches us.

---

source

write_targets

def write_targets(
    name, args
):

Paths a call writes where it names them: saved pictures for PATH_WRITES, none for a git write, else path.

---

source

path_write

def path_write(
    name, args
):

Whether this call writes where it names: an acting tool from PATH_WRITES given a path.

---

source

keep_media

def keep_media(
    new, old
):

new, carrying old’s pictures when a wrapper or hook rewrote a tool result as text.

---

source

bad_json

def bad_json(
    e, span:int=120
):

The fragment a JSON error is about.

---

source

sub_sp

def sub_sp(
    sp:str="You are a research sub-agent in a Python IDE. Answer the delegated question only.\n- Use tools as needed. Report findings with file paths and line numbers.\n- Before each tool call, write one line of at most eight words, in the -ing form, saying what you are doing.\n- State plainly when you find nothing. Do not guess.\n- A claim that something ran, exists or failed names the tool call or artifact behind it. Anything else is a hypothesis; say so.\n- End with one line: `state: completed`, `state: blocked` or `state: failed`, and why.\n- A tool result that contradicts what you just saw is likely a harness fault: confirm it once another way, name it in your answer, then carry on.\n- A tool result saying the tool-call budget is spent, and the request to answer that follows it, come from this harness: answer then with what you found.\n- You cannot edit. Describe any required change and stop.\n- Use `inspect_python` for live variables. Its default scope is sandboxed; use `scope='overlay'` for the real interpreter.",
    skills:tuple=()
):

A sub-agent’s briefing: standing instructions, then the bodies of skills its task named.

The tools a sub-agent gets are wrapped. A message sent to a running sub-agent, with /tell in the terminal, arrives appended to its next tool result inside a <user-message> tag that carries the run’s key. inbox_note tells the sub-agent that only a tag with that key is the user speaking. A file or a web page could contain the same tag without the key. A sub-agent that gets the same result from the same call REPEAT_CALLS times in a row is told to change course. The cell below shows a queued message arriving with a result.

r = Run('run_t'); r.tell('and the tests')
test_eq(_inboxed(lambda: 'contents', lambda: None)(), 'contents')
test_eq(_inboxed(lambda: 'contents', lambda: r)(), f'contents\n\n<user-message key="{r.key}">\nand the tests\n</user-message>')
test_eq(r.drain_inbox(), [])
finished = ['\n\n<background-results>\n## run_b (complete): q\nthe answer\n</background-results>']
rn = Run('run_n')
pushed = _inboxed(lambda: 'contents', lambda: rn, notice=lambda: finished.pop() if finished else '')
test_eq(pushed(), 'contents' + '\n\n<background-results>\n## run_b (complete): q\nthe answer\n</background-results>')
test_eq(pushed(), 'contents')   # a finished answer arrives once

Each wrapped call also updates the run’s status (Run.on_call). The now pane, the side panel that lists live runs (see pane), shows what a sub-agent is doing from that status. Some backends, LiteRT for one, never pass on the text a model writes between calls. Their sub-agents get a status tool from status_tool to report one line. STATUS_NOTE in the briefing tells them to use it.

class _Quiet(FakeBackend):
    narrates = False
    def spawn(self, sp='', tools=(), **kw):
        self.sub = FakeBackend(self.spec, sp=sp, tools=tools, shared=True)
        self.sub._send = lambda msg, **kw: next(t for t in self.sub.tools if t.__name__ == 'status')(text='Reading the tests')
        return self.sub

qb, qr = _Quiet(), Run('run_quiet', 'child', 'q')
delegate(qb, 'q', tools=ts, run=qr)
test_eq(qr.status, 'Reading the tests')
assert STATUS_NOTE in qb.sub.sp

---

source

delegate_many

def delegate_many(
    backend, questions, tools:tuple=(), sp:NoneType=None, max_steps:int=80, n_workers:int=4, skills:tuple=(),
    writes:bool=False, approve:NoneType=None, parent:NoneType=None, images:tuple=(),
    block:frozenset=frozenset({'delegate_tell', 'delegate_cancel', 'delegate_search', 'delegate_async', 'delegate_result', 'watch', 'cancel_watch'}),
    timeout:int=0, slots:NoneType=None,
    spawn:NoneType=None, # `spawn` options for every question, or a list with one per question
):

Ask several questions of one sub-agent each, and return the answers in question order. backend, sp and spawn may be lists, one per question, and tools a callable from a backend’s spec to its tools; see lanes for what runs at once.

---

source

lanes

def lanes(
    backends, writes:bool=False
):

Question indices grouped into what runs one after another: every write in one lane, the questions on a local engine in one, each cloud question alone.

Here a fake backend answers a delegated question. The caller gets back only the answer text.

be = FakeBackend(replies=['the caller never sees this'])
delegate(be, 'which files import fastllm?', tools=ts)

from urai import ToolCallError

# a sub-agent whose tool call was cut off reports why in its answer
class _Cut(FakeBackend):
    def spawn(self, sp='', tools=(), **kw):
        s = super().spawn(sp, tools, **kw)
        def _send(msg, **kw): raise ToolCallError("openai/gpt-5.6's tool call was cut off at its 16384-token output limit")
        s._send = _send
        return s
out = delegate(_Cut(), 'edit the notebook', tools=ts)
assert 'tool call was cut off' in out and 'JSONDecodeError' not in out, out

The spawned conversation was separate. It held only the question and the answer, and delegate closed it before returning.

test_eq(len(be.spawned), 1)
be.spawned[0].hist

delegate_many asks several questions at once, one sub-agent each, and returns the answers in the order the questions were asked. It registers every child run with the parent before any starts. Cancelling the parent then reaches all of them. The last example on this page shows that. Questions to cloud models run in parallel, up to n_workers (default 4) at once. Questions run one at a time when sub-agents may write, and questions to local models run one at a time. An engine such as LiteRT holds one conversation at a time, and parallel runs would race for it.

delegate_many(be, ['what imports fastllm?', 'where is compaction triggered?'], tools=ts)

The delegation tools take a callable that returns a backend. Each call asks for the current session’s backend. After a /model switch the next sub-agent runs on the new model.

delegate_async starts a sub-agent in the background and returns at once. The turn can carry on while the sub-agent works. Background keeps track of those runs. start registers a run before its thread begins. A status call therefore always finds the id that start returned. At most mx runs reach a model at once (ASYNC_MAX, 4 by default) and the rest wait as pending. status, result and cancel answer for one run, and status with no id answers for all. The last keep answers are held. close, called when the session ends, cancels every run and refuses new work.

The examples hold two runs behind a Background with one place, cancel a run still waiting, close a Background, catch a run that raised, and show an answer dropped once newer ones push it out.

---

source

Background

def Background(
    mx:int=4, # how many reach a model at once
    keep:int=50, # answers held before the oldest is dropped
    on_done:NoneType=None, # called with `(run, answer)` when a run finishes
):

Delegations running behind the turn that asked for them, and the answers they leave.

bg = Background(mx=1)
gate, seen = threading.Event(), []

def slow(run):
    run.start()
    seen.append(run.id)
    gate.wait(5)
    return f'answered {run.question}'

a = bg.start(slow, Run('run_a', 'child', 'first'))
b = bg.start(slow, Run('run_b', 'child', 'second'))
time.sleep(.2)
[(r['id'], r['state']) for r in bg.status()]

test_eq({r['id'] for r in bg.status()}, {'run_a', 'run_b'})
test_eq(bg.status('run_a')[0]['state'], 'running')
test_eq(bg.status('run_b')[0]['state'], 'pending')
test_eq(len(seen), 1)

gate.set()
for _ in range(200):
    if bg.result('run_b').startswith('answered'): break
    time.sleep(.02)
test_eq(bg.result('run_a'), 'answered first')
test_eq(bg.result('run_b'), 'answered second')
test_eq(len(seen), 2)

held = Background(mx=1)
block, ran = threading.Event(), []
held.start(lambda r: (r.start(), block.wait(5), ran.append(r.id))[2], Run('run_1', 'child', 'holds the slot'))
held.start(lambda r: ran.append(r.id), Run('run_2', 'child', 'never runs'))
time.sleep(.2)
held.cancel('run_2')
block.set()
for _ in range(200):
    if 'stopped' in held.result('run_2'): break
    time.sleep(.02)
test_eq(ran, ['run_1'])

shut = Background()
shut.start(lambda r: 'done', Run('run_x', 'child', 'q'))
test_eq(shut.close(), 1)
test_fail(lambda: shut.start(lambda r: 'nope', Run('run_y', 'child', 'q')))
assert 'run_y' not in {r['id'] for r in shut.status()}

boom = Background()
boom.start(lambda r: 1/0, Run('run_z', 'child', 'q'))
for _ in range(200):
    if ERR in boom.result('run_z'): break
    time.sleep(.02)
assert 'ZeroDivisionError' in boom.result('run_z'), boom.result('run_z')

small = Background(keep=1)
for n in ('run_p', 'run_q'): small.start(lambda r: f'ans {r.id}', Run(n, 'child', 'q'))
for _ in range(200):
    if small.result('run_q') == 'ans run_q': break
    time.sleep(.02)
assert small.result('run_p') != 'ans run_p', small.result('run_p')

result(run_id, wait) waits up to wait seconds for the answer, at most WAIT_MAX. A run that has finished but has not yet saved its answer gets SAVE_WAIT seconds. While a run works, result says that the answer comes by itself, so the model does not poll.

slow, go = Background(), threading.Event()
slow.start(lambda r: (go.wait(5), 'slow answer')[1], Run('run_s', 'child', 'q'))
assert 'Do not ask again' in slow.result('run_s') and 'ask again later' not in slow.result('run_s')
threading.Timer(.2, go.set).start()
test_eq(slow.result('run_s', wait=5), 'slow answer')

---

source

subagent_tools

def subagent_tools(
    get_backend, get_tools, get_skills:NoneType=None, get_cloud_backend:NoneType=None,
    get_writes:NoneType=None, # the session's sub-agent write toggle, read per call
    get_approve:NoneType=None, # the gate those writes answer to
    background:NoneType=None, # the `Background` for async delegations; made if None
    get_log_dir:NoneType=None, # callable -> the folder run transcripts go in, or None
    get_pictures:NoneType=None, # callable paths -> `(bytes, err)`; see `read_pictures`
    get_steps:NoneType=None, # callable -> the step budget, which also caps a call's `max_steps`
    get_timeout:NoneType=None, # callable -> seconds each delegation may run; 0 never stops it
    get_max:NoneType=None, # callable -> chars one result may spend, notes included; `MAX_TOOL_CHARS` when None
    get_notes:NoneType=None, # callable spec -> the briefing notes for that model's family, or ''
    slots:NoneType=None, # a shared `Slots` bounding sub-agents at once
    block:frozenset=frozenset({'delegate_tell', 'delegate_cancel', 'delegate_search', 'delegate_async', 'delegate_result', 'watch', 'cancel_watch'}), # names withheld from the sub-agent, when `get_tools` does not decide
    get_sub:NoneType=None, # callable (model, effort) -> `(backend, spawn options, note)`, raising why it cannot; None uses `get_backend` and `get_cloud_backend`
    get_tools_for:NoneType=None, # callable spec -> the tools for a sub-agent on that model; None gives every sub-agent `get_tools()`
):

The five delegation tools. The callable arguments are read on each call. The tools follow the session’s current settings.

---

source

sub_steps

def sub_steps(
    n:int=0, setting:str='auto'
):

Resolve a delegation step request against the session setting.

---

source

renamed

def renamed(
    **old
):

A tool that also takes each keyword old names, as the one it now names; its schema shows only the new.

---

source

read_pictures

def read_pictures(
    host, paths
):

(bytes, '') for pictures inside the host’s roots, or ([], err) for the first one refused.

---

source

picture_mime

def picture_mime(
    head
):

The MIME of a PNG, JPEG, GIF or WebP from its first bytes, else None.

---

source

named_skills

def named_skills(
    get_skills, names
):

The skills a delegated task named, and a note about any name that matched nothing.

A skill is a markdown document of instructions that an agent reads when the work calls for it, and a delegated task can name the skills its sub-agent needs. named_skills finds them by name. When a name matches nothing, it returns a note that lists the skills that do exist. It does not raise. The model can then correct the name. sub_sp appends the full text of the found skills to the sub-agent’s briefing.

---

source

parse_plan_items

def parse_plan_items(
    items
):

The item texts from newline text, a JSON list, or a Python list.

parse_plan_items turns a list a model sent into a list of strings. Models send a list as JSON text, as one item per line (with or without - or * bullets), or as a real list. The function accepts all three. The agent’s set_plan tool reads its todo items through it.

The next cell returns to skills. It asks named_skills for one skill that exists and one that does not. It then checks what sub_sp adds to the briefing.

skills = [Skill('house-style', 'md', 'How code is written here.', _text='Dense lines.'),
          Skill('deploy', 'md', 'How this ships.', _text='Push the tag.')]
named_skills(lambda: skills, 'house-style'), named_skills(lambda: skills, 'nope')

got, note = named_skills(lambda: skills, 'house-style')
test_eq([s.name for s in got], ['house-style'])
test_eq(note, '')
test_eq([s.name for s in named_skills(lambda: skills, 'house-style,deploy')[0]],
        ['house-style', 'deploy'])
test_eq(named_skills(lambda: skills, ''), ([], ''))
missed, note = named_skills(lambda: skills, 'nope')
test_eq(missed, [])
assert note

briefed = sub_sp(skills=got)
assert 'Dense lines.' in briefed and briefed.startswith(SUB_SP)
assert 'Push the tag.' not in briefed
test_eq(sub_sp(), SUB_SP)

A pyskill is a skill shipped as a Python module whose docstring is the skill text. Listing skills needs each one’s description, and shalya imports the module to read it. Some of those imports are slow: vishalakshi.skill and kosha.skill pull in litesearch and model2vec, about 0.7 seconds before the first prompt. mod_doc reads a module’s docstring from its source file without importing anything, and this cell swaps shalya’s pyskill loader for one that uses it. The module is imported only when the skill’s full text is read.

---

source

mod_doc

def mod_doc(
    modpath
):

A module’s docstring read from its source, loading nothing; None when that cannot be done.

test_eq(mod_doc('ramabana.theory'), sys.modules['ramabana.theory'].__doc__ if 'ramabana.theory' in sys.modules else __import__('ramabana.theory').theory.__doc__)
assert mod_doc('no_such.module') is None and 'no_such' not in sys.modules

subagent_tools builds the five tools a model uses to delegate. delegate_search asks one or more questions and waits for the answers. delegate_async starts one question in the background and returns a run id. delegate_result collects a background answer, or with no id lists every background run, delegate_cancel stops one, and delegate_tell sends a running one a message. Most arguments are callables that subagent_tools reads on each call. The backend, the step budget, the time limit and the write setting follow the session as they change. An answer longer than the tool-result limit is cut. The whole text is saved to a file, and the cut answer names the path.

subs = {t.__name__: t for t in subagent_tools(lambda: be, lambda: ts)}
delegate_search = subs['delegate_search']
list(subs)

With no model to delegate to, the tools return a message saying so instead of raising into the turn. With a model, several questions come back with one heading each.

nomodel = {t.__name__: t for t in subagent_tools(lambda: None, lambda: ts)}
assert isinstance(nomodel['delegate_search'](['anything']), str)
assert isinstance(nomodel['delegate_async']('anything'), str)
delegate_search(['what imports fastllm?', 'which tests cover it?'])

from fastcore.funccall import get_schema

dt = {t.__name__: t for t in subagent_tools(lambda: be, lambda: [], None, background=Background())}
assert 'sub answer' in dt['delegate_search'](['one question'])
out = dt['delegate_search'](['q1', 'q2']); test_eq(out.count('### '), 2)
n0 = len(be.spawned)
dt['delegate_search']([]); dt['delegate_search'](['', '  '])
test_eq(len(be.spawned), n0)
assert 'sub answer' in dt['delegate_search']('a bare string still works')
test_eq(dt['delegate_search']('["q1", "q2"]').count('### '), 2)
test_eq(dt['delegate_search']('one question\nthat runs onto a second line').count('### '), 0)
rid = dt['delegate_async']('later').split()[1]
for _ in range(300):
    if rid in dt['delegate_result']() and 'completed' in dt['delegate_result'](): break
    time.sleep(.02)
assert rid in dt['delegate_result']()

delegate_async, delegate_result and delegate_cancel share one Background. The run id comes back at once and the answer later. delegate_result with no id lists every background delegation and its state.

bg_be = FakeBackend()
bgs = {t.__name__: t for t in subagent_tools(lambda: bg_be, lambda: ts)}

started = bgs['delegate_async']('survey the tree')
rid = started.split()[1]
assert started.startswith('started run_') and 'read-only' in started, started
assert 'arrives by itself' in started and 'Collect' not in started, started

for _ in range(300):
    if 'completed' in bgs['delegate_result']().splitlines()[0]: break
    time.sleep(.02)
test_eq(bgs['delegate_result'](rid).split('\nstate:')[0], 'sub answer')
test_eq(len(bg_be.spawned), 1)
assert rid in bgs['delegate_result']()
assert bgs['delegate_result']('run_nope').startswith('no delegation named')
assert bgs['delegate_cancel'](rid).startswith(f'{rid} had already finished')

delegate_tell sends a message to a sub-agent that is still running, in the background or in the calling turn’s tree. The sub-agent reads it attached to its next tool result, and an unknown or finished run is refused with an error.

tell_bg = Background()
child = Run('run_kid', 'background', 'survey the tree')
tell_bg.runs[child.id] = child
tt = {t.__name__: t for t in subagent_tools(lambda: bg_be, lambda: ts, background=tell_bg)}
assert tt['delegate_tell']('run_kid', 'skip the tests') == 'told run_kid'
nxt = _inboxed(lambda: 'contents', lambda: child)()
assert '<user-message' in nxt and 'skip the tests' in nxt
assert 'no sub-agent named' in tt['delegate_tell']('run_nope', 'hi')
child.finish()
tt['delegate_tell']('run_kid', 'too late')

A sub-agent sees pictures only when they come with its first message. images names them by path inside the open folders, and read_pictures reads them, refusing any path outside the folders or any file that is not a picture. The question then ends with a note naming the pictures in order, so the sub-agent knows which is which. A model that cannot take images gets an error rather than a question without its pictures, and so does a session with no way to read them. A cloud_model that names no reachable model gets an error saying to leave it empty and use the sub-agent model.

pic_root = Path(tempfile.mkdtemp()).resolve(); pic_host = LocalHost([pic_root], index=False)
png = b'\x89PNG\r\n\x1a\n' + bytes(32)
(pic_root/'refs').mkdir(); (pic_root/'refs'/'rama.png').write_bytes(png); (pic_root/'notes.txt').write_text('hi')
test_eq(read_pictures(pic_host, ['refs/rama.png']), ([png], ''))
test_eq(read_pictures(pic_host, []), ([], ''))
assert 'not a picture' in read_pictures(pic_host, ['notes.txt'])[1]
assert failed(read_pictures(pic_host, ['refs/missing.png'])[1]) and failed(read_pictures(pic_host, ['../x.png'])[1])

pic_be = FakeBackend()
pt = {t.__name__: t for t in subagent_tools(lambda: pic_be, lambda: [], None, lambda m: None, background=Background(),
                                            get_pictures=lambda ps: read_pictures(pic_host, ps))}
assert 'sub answer' in pt['delegate_search'](['match this face'], images=['refs/rama.png'])
sent = pic_be.spawned[-1].sent[0]
test_eq(sent[0], png); assert 'match this face' in sent[-1] and 'refs/rama.png' in sent[-1], sent[-1]
assert 'not a picture' in pt['delegate_search'](['q'], images=['notes.txt']); test_eq(len(pic_be.spawned), 1)
test_eq(pic_be.spawned[-1].sent[0], sent)
pt['delegate_search'](['plain']); test_eq(pic_be.spawned[-1].sent[0], 'plain')
rid = pt['delegate_async']('draw from it', images=['refs/rama.png']).split()[1]
for _ in range(300):
    if pt['delegate_result'](rid).startswith('sub answer'): break
    time.sleep(.02)
test_eq(pic_be.spawned[-1].sent[0][0], png)
blind = {t.__name__: t for t in subagent_tools(lambda: pic_be, lambda: [])}
assert failed(blind['delegate_search'](['q'], images=['refs/rama.png']))
missing = pt['delegate_search'](['q'], cloud_model='gpt-6-typo')
assert 'gpt-6-typo' in missing and 'leave `model` empty' in missing, missing
assert 'images' in get_schema(pt['delegate_search'])['input_schema']['properties']

model and effort name a model and an effort level for every question, or give a list with one per question. A list of another length is refused before any sub-agent runs. get_sub(model, effort) turns each pair into a backend, the options spawn takes and a note for the answer. It runs in the caller’s thread, before any sub-agent starts. A pair named twice is resolved once. A question whose model cannot run gets the error as its answer. The other questions still run. cloud_model is still taken as model and is not in the schema. delegate_many runs the questions in lanes: every write in one lane, the questions on a local engine in one, and each cloud question in its own. Sub-agents on different cloud models therefore answer at once.

picture_mime recognises a PNG, JPEG, WebP or GIF by its first bytes, whatever the file is called. A fake.png that holds text is therefore refused below. The limits are the terminal’s: at most MAX_ATTACH (8) pictures, each at most MAX_MEDIA (20 MB). Every question’s sub-agent receives every picture. Pictures times questions may not exceed MAX_ATTACH.

A read-only sub-agent’s generate_image refuses a path, because saving there would be a write. With path empty, it saves under media/. PATH_WRITES, path_write and write_targets tell the approval gate about this case. A call to a tool in PATH_WRITES counts as a write when it names a path. write_targets lists the files any call will write.

for name, body in (('Makefile', 'all:\n'), ('.env', 'K=1\n'), ('fake.png', 'text'), ('logo.svg', '<svg xmlns="x"/>')):
    (pic_root/name).write_text(body); assert 'not a picture' in read_pictures(pic_host, [name])[1], name
test_eq(read_pictures(pic_host, ['refs/rama.png'])[0], [png])
assert 'at most' in read_pictures(pic_host, ['refs/rama.png'] * (MAX_ATTACH + 1))[1]
with open(pic_root/'big.png', 'wb') as f: f.write(png); f.truncate(MAX_MEDIA + 1)
assert 'over' in read_pictures(pic_host, ['big.png'])[1]
too_many = pt['delegate_search'](['q%d' % i for i in range(MAX_ATTACH + 1)], images=['refs/rama.png'])
assert 'at most' in too_many, too_many
def generate_image(prompt: str, path: str = '') -> str: return f'drew {path or "media"}'
drawn = FakeBackend()
delegate(drawn, 'draw', tools=[generate_image])
sub_draw = {t.__name__: t for t in drawn.spawned[-1].tools}['generate_image']
assert failed(sub_draw('x', path='main.py')) and 'read-only' in sub_draw('x', path='main.py')
test_eq(sub_draw('x'), 'drew media')
delegate(drawn, 'draw', tools=[generate_image], writes=True)
test_eq({t.__name__: t for t in drawn.spawned[-1].tools}['generate_image']('x', path='a.png'), 'drew a.png')

delegate_many registers every child run before any work starts. Cancelling the parent stops the child that is running, and the children still queued never spawn. The fake backend below blocks on its first question until it is cancelled.

class _Sub:
    max_steps = 0
    def __init__(self, owner): self.owner, self.release = owner, threading.Event()
    def send(self, question, run=None):
        self.owner.started.append(question); self.release.wait(); return 'late'
    def cancel(self): self.owner.cancelled += 1; self.release.set(); return True
    def close(self): pass

class _ParentBackend:
    def __init__(self):
        self.spec = AttrDict(name='fake-child', local=False)
        self.started, self.spawned, self.cancelled = [], 0, 0
    def spawn(self, **kw): self.spawned += 1; return _Sub(self)

backend, parent, box = _ParentBackend(), Run('run_parent', grace=.03), []
parent.start()
t = threading.Thread(target=lambda: box.extend(delegate_many(backend, ['a', 'b', 'c'], n_workers=1, parent=parent)), daemon=True)
t.start()
while not backend.started: time.sleep(.001)
test_eq(len(parent.children), 3)
parent.cancel(); t.join(.2)
test_eq((backend.spawned, backend.cancelled), (1, 1))
test_eq(len(box), 3)
assert all((isinstance(x, dict) and x['state'] in ('cancelled', 'detached'))
           or (isinstance(x, str) and 'cancelled' in x) for x in box)

Attachments

A prompt can carry files: pictures and sound. These helpers live here and not in the terminal module. A host that only sends files then does not load the terminal UI. media_paths reads a paste or a dropped file in the shapes terminals deliver it, quoted, backslash-escaped or as a file:// URL, and returns the media files it names, or an empty list for any other text. attach_refs finds @path references in a prompt. clipboard_png reads a picture from the clipboard with pngpaste, wl-paste or xclip, whichever is installed. Attachment holds one file’s bytes, read once. media_parts and media_note turn a list of attachments into what the message sends: the bytes the model accepts, and a note listing every file. Audio goes to the model only when it accepts audio; otherwise the note gives the file’s path.

---

source

media_note

def media_note(
    atts, spec:NoneType=None
):

What the message says about the files attached to it.

---

source

media_parts

def media_parts(
    atts, spec:NoneType=None
):

The attachments that go out as model content parts.

---

source

sendable

def sendable(
    atts, spec:NoneType=None
):

The attachment kinds spec’s model accepts: images always, and audio when it takes audio input or no spec is given.

---

source

Attachment

def Attachment(
    path
):

One media file for the next prompt, its bytes read once when attached.

---

source

clipboard_png

def clipboard_png():

The clipboard picture as PNG bytes, or None.

---

source

attach_refs

def attach_refs(
    text
):

Media named @path or @"path" in a prompt, trailing punctuation stripped from a bare one.

---

source

ref_tokens

def ref_tokens(
    text
):

Each @path in text as (path, quoted); a bare one keeps whatever punctuation follows it.

---

source

media_paths

def media_paths(
    text
):

Every media file a paste names, or [] when it is anything else.

---

source

is_media

def is_media(
    p
):

Whether p names a media file that exists.

---

source

media_path

def media_path(
    s
):

A Path from one path as a terminal delivered it, quoted, escaped or as a file:// URL, or None.