tools

Every tool an agent can be given, and how to pick the ones a host can support.

Every factory takes mx, the largest number of characters one tool result may hold, because the model in use sets that budget. Results longer than mx are clipped with the helpers in shalya.core.

A tool must never raise, since an exception would end the agent’s turn. resolved checks a path against the host and returns (path, ''), or (None, error) with an ERROR: message the tool can return as is. readable resolves a path a tool is only going to read, which lets a host with read_outside allow reads beyond its open folders.

---

source

resolved

def resolved(
    host, path, writing:bool=False, reading:bool=False, must_exist:bool=False
):

Resolve path on host as (path, ''), or (None, error) when the host refuses it.

---

source

readable

def readable(
    host, path, must_exist:bool=False
):

Resolve path for reading only, which lets a host with read_outside allow paths beyond its open folders.

Every read-only tool resolves its paths through readable, so the host alone decides whether reads may go outside its folders.

Searching the code

code_tools builds the tools for finding code. search_code searches the open folders and the installed packages, by meaning when kosha’s index is ready and literally until then. grep finds exact lines matching a pattern, ls lists a folder, and outline lists the functions and classes in one file. When the host has an index, two more appear: similar_code, which finds code like the function at a given line, and public_api, which lists a package’s public names.

A search can return hits the agent cannot open, for example in an installed package outside the open folders. reach_hint adds a line under each hit saying which tools can reach that path. screen_hits passes hits through the host’s read_guard, asking once per file.

---

source

code_tools

def code_tools(
    host, mx:int=6000
):

Tools to search code and see its structure: search_code, grep, ls, outline, and with an index similar_code and public_api.

---

source

outside_note

def outside_note(
    host, path, what
):

Why what cannot cover path, an absolute path outside every open folder; '' when it is inside one or not absolute.

---

source

screen_hits

def screen_hits(
    host, hits, tool
):

hits as host.read_guard lets each file’s lines go: whole, masked line for line, or one refusal row per file.

---

source

reach_hint

def reach_hint(
    host, path
):

What a model may do with a search hit at path on host: view and edit, view only, or neither.

The examples use a real folder and a real host. Without a kosha index, search_code uses ripgrep and outline parses the file.

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')
(root/'pkg'/'use.py').write_text('from .sizes import threshold\n\ndef budget(): return threshold(8192)\n')
host = LocalHost([root], index=False)
ct = {t.__name__: t for t in code_tools(host)}
sorted(ct)

test_eq(sorted(ct), ['grep', 'ls', 'outline', 'search_code'])          # no index: nothing similar, and no public surface, to show
assert 'threshold' in ct['search_code']('threshold')
assert 'sizes.py' in ct['ls']('pkg') and 'sizes.py' in ct['ls'](pattern='sizes', recursive=True)
assert 'sizes.py' in ct['ls']('pkg', pattern='sizes') and 'use.py' not in ct['ls']('pkg', pattern='sizes')
assert 'sizes.py' in ct['ls'](pattern='*.py', recursive=True) and 'use.py' not in ct['ls'](pattern='sizes', recursive=True)
test_eq(ct['ls'](pattern='zzz', recursive=True), 'no matching files')
assert 'recursive' in ct['ls'].__doc__
assert 'threshold' in ct['outline']('pkg/sizes.py')
hit = ct['search_code']('threshold')
assert 'view_file/replace_text' in hit and 'edit_file' not in hit, hit                # the default editor, not the opt-in one
assert 'pkg/' in ct['ls'](pattern='*.py') and 'pkg/' in ct['ls'](pattern='zzz')       # folders always show, so the model knows where to descend
test_eq(ct['ls'](pattern=root.parent.name, recursive=True), 'no matching files')     # an ancestor's name matches nothing below the root
assert 'sizes.py' in ct['ls'](pattern='pkg/*.py', recursive=True)                    # a path below the root does

A search covers installed packages, but the read and write tools may refuse a path outside the open folders. Each hit’s hint names only the tools that can reach it on this host: view and edit inside the open folders, view only in a read-only folder or with read_outside, and neither anywhere else.

from shalya.core import Hit

lib = Path(tempfile.mkdtemp()).resolve()/'site-packages'; lib.mkdir()
(lib/'sizes.py').write_text('def threshold(n): return n\n')
class Spread(LocalHost):
    def search(self, query, limit=8): return [Hit(str(root/'pkg'/'sizes.py'), 1, 'threshold'), Hit(str(lib/'sizes.py'), 1, 'threshold')]
def hints(**kw):
    "The hint under each hit, for a host built with `kw`."
    sc = {t.__name__: t for t in code_tools(Spread([root], index=False, **kw))}['search_code']
    return sc('threshold').split('\n')[2::2]   # note, then a hit line and its hint, per hit
mine, theirs = hints()
assert 'view_file/replace_text' in mine, mine
assert 'view_file' not in theirs and 'outside the open folders' in theirs, theirs     # nothing here can read it
for kw in (dict(read_outside=True), dict(read_roots=[lib])):
    mine, theirs = hints(**kw)
    assert 'view_file/replace_text' in mine and 'view_file' in theirs and 'replace_text' not in theirs, (kw, theirs)   # readable, never written

When a search finds nothing, the result names the engine that searched, so the model can tell “no matches” from “the index is not ready”.

ct['search_code']('nonexistent_symbol_xyz')

miss = ct['search_code']('nonexistent_symbol_xyz')
assert 'no matches' in miss.lower(), miss
assert 'ripgrep' in miss or 'fallback' in miss, miss

The literal fallback has to work before indexing finishes. The host below builds a real kosha index over this repository, to check search by meaning, the public API listing, and similar_code. A fresh machine spends minutes indexing the environment, so nbdev-test skips these cells; nbdev-test --flags notest runs them.

from litesearch import repo_root

repo_root()
indexed_host = LocalHost([repo_root()], rerank=False)
assert indexed_host.wait_index(180), indexed_host.search_note
indexed_tools = {t.__name__: t for t in code_tools(indexed_host)}
assert {'similar_code', 'public_api'} <= set(indexed_tools), sorted(indexed_tools)   # an index makes both worth offering
sorted(indexed_tools)

indexed_tools['search_code']('code tools')

indexed_tools['public_api']('shalya.host')

indexed_tools['similar_code']('shalya/host.py', 32)

Files

file_tools builds the tools for reading and editing files. view_file shows a file as lineno|hash|content lines, where the hash is a short fingerprint of the line, and a picture comes back as an image. replace_text edits a file with a list of {oldText, newText} pairs: each oldText must match the file exactly once, and either every edit applies or none does. create_file writes a whole file, and add_root opens another folder. A host that cannot write gets only view_file.

exhash_tools builds edit_file, an editor that addresses lines by the lineno|hash| prefixes view_file shows and refuses an edit when the line has changed since it was viewed. It is opt-in: ask for it with tools_for(host, optin=('exhash',)).

---

source

file_tools

def file_tools(
    host, mx:int=6000
):

Tools to read files and edit them by exact text: view_file, replace_text, create_file and add_root, or view_file alone where the host cannot write.

---

source

picture_mime

def picture_mime(
    head
):

The MIME type of a PNG, JPEG, GIF or WebP picture from its first bytes, or None; the file name is ignored.

---

source

exhash_tools

def exhash_tools(
    host, mx:int=6000
):

The opt-in edit_file tool, which edits lines addressed by the hashes view_file shows.

edits is declared as a list in the tool’s schema, so a model sends a list. A JSON string of the list is still accepted in this release. A string that is not JSON, or an edit with the wrong keys, gets an error naming the shape wanted.

ft = {t.__name__: t for t in file_tools(host)}
test_eq(sorted(ft), ['add_root', 'create_file', 'replace_text', 'view_file'])
p = root/'pkg'/'greet.py'; p.write_text('def greet(name):\n    return "hello " + name\n')
out = ft['replace_text'](str(p), [{'oldText': 'hello', 'newText': 'howdy'}])
assert out.startswith('replaced 1 block(s)') and '-    return "hello " + name' in out
test_eq(p.read_text(), 'def greet(name):\n    return "howdy " + name\n')
edits_schema = get_schema(ft['replace_text'])['input_schema']['properties']['edits']
test_eq((edits_schema['type'], edits_schema['items']), ('array', {'type': 'object'}))
assert 'oldText' in edits_schema['description'] and 'newText' in edits_schema['description'], edits_schema   # the keys are in the schema a model reads
# Review Focus 1: a JSON string still lands; a non-JSON string and wrong keys name the shape
assert ft['replace_text'](str(p), '[{"oldText": "howdy", "newText": "hey"}]').startswith('replaced')
bad = ft['replace_text'](str(p), 'oldText=hey newText=hi'); assert failed(bad) and 'list of {"oldText"' in bad, bad
assert "'oldText' and 'newText'" in ft['replace_text'](str(p), [{'old_text': 'hey', 'new_text': 'hi'}])
test_eq(p.read_text(), 'def greet(name):\n    return "hey " + name\n')
# Review Focus 2: overlap, ambiguity, emptiness, no-op: nothing is written
for es, msg in (([{'oldText': 'name', 'newText': 'who'}], 'matches 2 places'),
                ([{'oldText': 'def greet', 'newText': 'def hi'}, {'oldText': 'greet(name)', 'newText': 'hi(n)'}], 'overlap'),
                ([], 'no edits given'), ([{'oldText': 'hey', 'newText': 'hey'}], 'changed nothing'),
                ([{'oldText': 'nowhere', 'newText': 'x'}], 'not found')):
    r = ft['replace_text'](str(p), es); assert failed(r) and msg in r, (es, r)
test_eq(p.read_text(), 'def greet(name):\n    return "hey " + name\n')
r = ft['replace_text'](str(root/'pkg'/'new.py'), [{'oldText': 'a', 'newText': 'b'}]); assert failed(r) and 'new.py' in r and 'create_file' in r, r
print(ft['view_file'](str(p)))

A host’s read_guard sees what each read tool would send, and its answer is what the model gets: the text, a masked version of it, or a refusal. Search hits are checked once per file.

cp = root/'pkg'/'contact.py'; cp.write_text('owner = "ada@x.org"\n')
host.read_guard = lambda r: Refusal('Local only') if r.path.endswith('sizes.py') else r.text.replace('ada@x.org', '[EMAIL]')
assert '[EMAIL]' in ft['view_file'](str(cp)) and 'ada@x.org' not in ct['grep']('ada@x')
test_eq(ft['view_file'](str(root/'pkg'/'sizes.py')), f"ERROR: withheld: {root/'pkg'/'sizes.py'}: Local only")
host.read_guard = None
assert 'ada@x.org' in ft['view_file'](str(cp))
cp.unlink()

edit_file takes exhash commands addressed by the lineno|hash| prefixes from view_file, and refuses a command whose address no longer matches the file. Every write tool is marked with @writes, and WRITE_TOOLS lists the same names for a caller that has only a name. A host that cannot write gets no editor from either factory.

xt = {t.__name__: t for t in exhash_tools(host)}
test_eq(sorted(xt), ['edit_file'])
line2 = ft['view_file'](str(p)).splitlines()[1]
stale = xt['edit_file'](str(p), [['2|0000|', 's', 'hey', 'hi']]); assert failed(stale)
ok = xt['edit_file'](str(p), [['|'.join(line2.split('|')[:2]) + '|', 's', 'hey', 'hi']]); assert not failed(ok), ok
test_eq(p.read_text(), 'def greet(name):\n    return "hi " + name\n')
test_eq(get_schema(xt['edit_file'])['input_schema']['properties']['commands']['type'], 'array')
writers = {t.__name__ for t in file_tools(host) + exhash_tools(host) if is_write(t)}
test_eq(writers, {'edit_file', 'replace_text', 'create_file', 'add_root'}); assert writers <= WRITE_TOOLS
frozen = LocalHost([root], index=False); frozen.writes = False
test_eq([t.__name__ for t in file_tools(frozen)], ['view_file']); test_eq(exhash_tools(frozen), [])
assert 'greet' in file_tools(frozen)[0](str(p))   # reading still works
assert 'lineno|hash|' in get_schema(xt['edit_file'])['input_schema']['properties']['commands']['description']

Notebooks

notebook_tools builds the notebook tools, which address cells by id, never by line. notebook_cells lists a notebook’s cells with their ids, view_cell shows one cell’s source with line numbers, edit_cell edits a cell with the same {oldText, newText} pairs as replace_text, and add_cell inserts a new cell.

---

source

notebook_tools

def notebook_tools(
    host, mx:int=6000
):

Tools to list, read, edit and add notebook cells, addressed by cell id.

# every write consults the host's optional guard
class Guarded(Host):
    group = 'file'
    def __init__(self): self.seen = []
    @property
    def roots(self): return ['/proj']
    def check(self, path, must_exist=False, reading=False): return Path('/proj')/path
    def check_write(self, p): self.seen.append(str(p)); raise HostError('generated')
    def walk(self): return []
    def read(self, path): return 'x = 1\n'
    def write(self, path, text): raise AssertionError(f'the guard was bypassed for {path}')
    def text_at(self, path): return ''
g = Guarded()
guarded = {t.__name__: t for t in file_tools(g) + exhash_tools(g) + notebook_tools(g)}
for nm, args in (('replace_text', ('a.py', [{'oldText': 'x = 1', 'newText': 'x = 2'}])), ('edit_file', ('a.py', [['1|ab|', 'd']])),
                 ('create_file', ('a.py', 'whole file')), ('add_cell', ('a.ipynb', 'y = 2')),
                 ('edit_cell', ('a.ipynb', 'c1', [{'oldText': 'x', 'newText': 'y'}]))):
    assert failed(guarded[nm](*args)), f'{nm} wrote past the guard'
test_eq(len(g.seen), 5)

A unique prefix of an id also addresses a cell, and an id that matches nothing gets an error listing the ids.

nt = {t.__name__: t for t in notebook_tools(host)}
test_eq(sorted(nt), ['add_cell', 'edit_cell', 'notebook_cells', 'view_cell'])
said = nt['add_cell']('nb/demo.ipynb', 'x = 1\ny = x + 1'); cid = said.split()[2]
assert cid in nt['notebook_cells']('nb/demo.ipynb')
print(nt['view_cell']('nb/demo.ipynb', cid))

test_eq(nt['view_cell']('nb/demo.ipynb', cid), '1: x = 1\n2: y = x + 1')
out = nt['edit_cell']('nb/demo.ipynb', cid, [{'oldText': 'x = 1', 'newText': 'x = 2'}])
assert out.startswith(f'replaced 1 block(s) in cell {cid}') and '+x = 2' in out, out
test_eq(nt['view_cell']('nb/demo.ipynb', cid[:4]), '1: x = 2\n2: y = x + 1')             # a unique prefix addresses too
test_eq(host.nb_cells('nb/demo.ipynb')[0][2], 'x = 2\ny = x + 1')
r = nt['edit_cell']('nb/demo.ipynb', cid, [{'oldText': 'x', 'newText': 'z'}]); assert failed(r) and 'matches 2 places' in r
r = nt['edit_cell']('nb/demo.ipynb', 'nope', [{'oldText': 'x = 2', 'newText': 'x = 3'}]); assert failed(r) and cid in r and 'demo.ipynb' in r, r  # names the notebook and lists the ids
assert failed(nt['view_cell']('nb/demo.ipynb', 'nope'))
assert nt['edit_cell']('nb/demo.ipynb', cid, '[{"oldText": "x = 2", "newText": "x = 3"}]').startswith('replaced')   # a JSON string still lands
test_eq(host.nb_cells('nb/demo.ipynb')[0][2], 'x = 3\ny = x + 1')
edits_schema = get_schema(nt['edit_cell'])['input_schema']['properties']['edits']
test_eq(edits_schema['type'], 'array'); assert 'oldText' in edits_schema['description'], edits_schema
assert 'exhash' not in (nt['edit_cell'].__doc__ + nt['view_cell'].__doc__)

The web, and what was read before

The web tools and the memory tools are separate groups, and the difference matters. web_tools goes to the network now: web_search returns titles and URLs, and read_url reads one page as text and asks the host to keep it in memory unless remember=False. The memory tools recall what earlier reads found, without going out again. research_tools adds the opt-in research tool, which reads the top results for a query into one cited digest and is much slower than a search. browser_tools builds the opt-in browser tools, described at the end of this notebook.

---

source

web_tools

def web_tools(
    host, mx:int=6000
):

Tools to search the web and read one page: web_search and read_url.

---

source

research_tools

def research_tools(
    host, mx:int=6000
):

The opt-in research tool, which reads the top web results for a query into one cited digest.

---

source

browser_tools

def browser_tools(
    host, mx:int=6000
):

Tools to drive a browser tab: open, screenshot, read and reload a page, and click, type or run JavaScript in it.

These checks use a stand-in web backend, so they test the text the model sees and the arguments passed on without touching the network. The host notebook tests fossick itself.

class WebResults:
    research_note = 'fixture'
    def web_search(self, query, n=20):
        return [AttrDict(title='Python', url='https://python.org')] if query else []
    def read_url(self, url, remember=True):
        self.remember = remember
        return AttrDict(text='# Python', url=url)
    def research(self, query): return 'Python is a programming language.'

web_results = WebResults()
wt = {t.__name__: t for t in web_tools(web_results)}

test_eq(sorted(wt), ['read_url', 'web_search'])
test_eq(wt['web_search']('python'), 'Python\n  https://python.org')
test_eq(wt['web_search'](''), 'no results (fixture)')
test_eq(wt['read_url']('https://python.org', remember=False), '# Python')
test_eq(web_results.remember, False)
assert not any(is_write(t) for t in wt.values())
# the slow tool is an opt-in group over the same host, and a host without the web gets none
test_eq([t.__name__ for t in research_tools(web_results)], ['research'])
test_eq(research_tools(web_results)[0]('python'), 'Python is a programming language.')
assert has_effect(research_tools(web_results)[0]) and not is_write(research_tools(web_results)[0])
test_eq(research_tools(LocalHost([root], index=False, web=False)), [])
class Anything:
    def __getattr__(self, n): return lambda *a, **kw: None
test_eq([t.__name__ for t in research_tools(Anything())], ['research'])   # only a host that says `False` to the web is refused

Memory

memory_tools builds the tools over the host’s vault. memory_search searches remembered pages and returns matching sections with their headings, age and whether they are stale. memory_read browses: with no argument it lists every document, with a document id or title it shows its headings, and with a section id such as doc#3 it returns that section. remember files a note, and a later note with the same key replaces it. memory_forget deletes a document, and only at the user’s request.

---

source

memory_tools

def memory_tools(
    host, mx:int=6000
):

Tools to search, read, add to and delete from durable memory, which returns whole document sections.

---

source

lesson_text

def lesson_text(
    text, # the lesson: what to do the next time the cue fires
    cue:str='', # the shape of situation the lesson applies to
    component:str='', # the part of the harness it concerns, one of `LESSON_COMPONENTS`
    evidence:str='', # what in the conversation shows it
    confidence:str='', # one of `LESSON_CONFIDENCE`
):

Format a lesson for remember: text alone, or one field: value line per filled field, lesson first.

An agent can use remember to record a lesson learned from a mistake, with four extra fields: cue (the kind of situation it applies to), component (the part of the harness it concerns, one of LESSON_COMPONENTS), evidence and confidence (one of LESSON_CONFIDENCE). lesson_text formats such a note as one field: value line per filled field, starting with lesson, so a vault titles the note by its first line. Line breaks inside a value become spaces and empty fields are left out. Without any of the four fields, the text is stored unchanged. An unknown component or confidence, or extra fields with an empty text, raise a ValueError, which remember returns as an ERROR: result.

print(lesson_text('filter on the root-relative path', cue='an empty filtered search', component='tool', confidence='high'))

---

source

lessons

def lessons(
    text
):

Every lesson in a memory document: one per memory-file line, or one per lesson: block of a vault note.

---

source

parse_lesson

def parse_lesson(
    text
):

The fields of a lesson lesson_text wrote, from a vault note or a memory-file line; None for a plain note.

parse_lesson reads a lesson back into its fields, whichever layout stored it. A vault keeps lesson_text’s lines. A host without a vault keeps one memory-file line, - [key] **title**: and the fields joined by |. lessons finds every lesson in a memory file or a vault note and skips plain notes. Ramabana and Dhrona both read lessons this way.

full = dict(cue='an empty filtered search', component='tool', evidence='grep found nothing under src', confidence='high')
t = lesson_text('filter on the root-relative path', **full)
want = dict(lesson='filter on the root-relative path', **full)
test_eq(parse_lesson(t), want)                                                   # the vault's layout
test_eq(parse_lesson('- [k1] **paths**: ' + ' | '.join(t.splitlines())), want)   # a memory-file line
test_eq(parse_lesson('- plain note about paths'), None)
test_eq(parse_lesson(lesson_text('a habit', confidence='low')), dict(lesson='a habit', confidence='low'))
doc = f"# Notes\n- a plain note\n- [k1] **paths**: {' | '.join(t.splitlines())}\n\n{t}\nunrelated: line\n"
test_eq(lessons(doc), [want, want])

m=memory_tools(host)
summarise(m[-2])

---

source

watch_tools

def watch_tools(
    host, mx:int=6000
):

Tools to set up, list and cancel watches: pages, web searches and folders to re-check, and reminders.

watch_tools builds the watch tools. One watch tool covers every kind of watch the vault supports: a page (url), a reminder (remind), a web search (search) and a folder (folder), re-checked every every, such as '1d'. list_watches lists them and cancel_watch deletes one by id. No tool runs the watches; the harness calls the host’s poll for that.

ask_tools builds ask_memory, which answers a question from remembered research with citations. It is its own group because it needs a model to write the answer, and a host can store and search pages without having one.

---

source

ask_tools

def ask_tools(
    host, mx:int=6000
):

The ask_memory tool, which answers a question from remembered research with citations.

These checks use a stand-in memory backend with fixed answers, to test JSON output, citations, watch listings, errors and the write marks. The host notebook tests the real vault.

class MemoryResults:
    watch_actions = ('url', 'remind', 'web', 'folder')
    def memory_search(self, query, limit=8):
        if query == 'fail': raise RuntimeError('vault unavailable')
        return [{'id': 'd1', 'title': 'Kettles', 'text': 'boils at 100C', 'age': 60, 'stale': False}][:limit]
    def memory_tree(self, document=''):
        roots = [{'id': 'd1', 'title': 'Kettles'}]
        return roots if not document else [roots[0] | {'children': [{'id': 'd1#2', 'title': 'Boiling'}]}]
    def memory_read(self, node_id): return {'id': node_id, 'text': 'boils at 100C'}
    def memory_forget(self, doc_id): return doc_id == 'd1'
    def remember(self, text, title=None, tags=(), key=''):
        self.key, self.text = key, text
        return {'doc_id': 'd2', 'title': title or text, 'replaced': bool(key)}
    def watch(self, target, action='remind', every='1d', note=None, **params):
        self.watched = (target, action, every, note, params)
        return {'id': 'w1', 'kind': action, 'action': action, 'target': target, 'every': 86400, 'runs': 0}
    def watches(self, due_only=False):
        return [] if due_only else [{'id': 'w1', 'target': 'renew domain', 'kind': 'remind', 'action': 'remind',
                                     'every': 86400, 'runs': 0, 'last_status': None}]
    def unwatch(self, watch_id): return watch_id == 'w1'
    def poll(self):
        return {'ran': 1, 'checked': 1,
                'results': [{'status': 'ok', 'action': 'remind', 'target': 'renew domain'}]}
    def ask(self, question, ref=None, instruction=''):
        return {'answer': 'Water boils at 100C.', 'policy': 'redact',
                'pii': {'has_pii': question == 'who', 'identifying': {'name': 1}},
                'cited': [{'n': 1, 'breadcrumb': 'Kettles', 'node_id': 'd1'}]}

memory_results = MemoryResults()
mt = {t.__name__: t for t in memory_tools(memory_results)}
wa = {t.__name__: t for t in watch_tools(memory_results)}
at = {t.__name__: t for t in ask_tools(memory_results)}

test_eq(sorted(mt), ['memory_forget', 'memory_read', 'memory_search', 'remember'])
test_eq(sorted(wa), ['cancel_watch', 'list_watches', 'watch']); test_eq(sorted(at), ['ask_memory'])
assert 'boils at 100C' in mt['memory_search']('kettle')
assert '"age"' in mt['memory_search']('kettle') and '"stale"' in mt['memory_search']('kettle')   # 0.1.17 rows pass through
assert failed(mt['memory_search']('fail')) and 'vault unavailable' in mt['memory_search']('fail')
test_eq(json.loads(mt['memory_read']()), [{'id': 'd1', 'title': 'Kettles'}])                 # roots
test_eq(json.loads(mt['memory_read']('d1'))[0]['id'], 'd1')                                  # a document's tree
test_eq(json.loads(mt['memory_read']('d1#2'))['text'], 'boils at 100C')                     # a node
test_eq(mt['memory_forget']('d1'), 'forgot document'); test_eq(mt['memory_forget']('d9'), 'document was not forgotten')
test_eq(mt['remember']('fact', title='Facts', tags='one, two'), "remembered 'Facts' as d2"); test_eq(memory_results.key, '')
test_eq(memory_results.text, 'fact')                                                         # no lesson field: the text as given
mt['remember']('filter on the relative path', cue='an empty filtered search', component='tool',
               evidence='grep with path_filter\nfound nothing', confidence='high')
test_eq(dict(l.split(': ', 1) for l in memory_results.text.splitlines()),
        dict(cue='an empty filtered search', lesson='filter on the relative path', component='tool',
             evidence='grep with path_filter found nothing', confidence='high'))      # one line per field
test_eq(memory_results.text.split(': ')[0], 'lesson')                                       # a vault titles an untitled note by its first line
assert 'cue' not in lesson_text('x', cue='  ', confidence='low')                           # a blank field is not set
r = mt['remember']('  ', cue='c'); assert failed(r) and '`text`' in r
r = mt['remember']('x', component='habit'); assert failed(r) and 'rule, skill, extension, tool, memory' in r
r = mt['remember']('x', confidence='sure'); assert failed(r) and 'low, medium, high' in r
test_eq(mt['remember']('fact', key='kettle-facts'), "remembered 'fact' as d2 (replaced)"); test_eq(memory_results.key, 'kettle-facts')
test_eq(wa['watch']('https://x.test', every='1h'), 'watching url https://x.test as w1, every 1h')
test_eq(memory_results.watched[1], 'url')
wa['watch']('renew domain', kind='remind', every='1w'); test_eq(memory_results.watched[1], 'remind')
wa['watch']('late chunking', kind='search'); test_eq(memory_results.watched[1], 'web')     # `search` is vishalakshi's `web` action
r = wa['watch']('/proj', kind='folder'); assert failed(r) and 'instructions' in r
wa['watch']('/proj', kind='folder', instructions='flag print()', pattern='*.py'); test_eq(memory_results.watched[4], dict(instructions='flag print()', pattern='*.py'))
r = wa['watch']('x', kind='youtube'); assert failed(r) and 'url, remind, search, folder' in r
memory_results.watch_actions = ('remind',); r = wa['watch']('https://x.test'); assert failed(r) and 'remind' in r
memory_results.watch_actions = ('url', 'remind', 'web', 'folder')
assert 'remind' in wa['list_watches']() and 'renew domain' in wa['list_watches']() and 'every 86400s' in wa['list_watches']()
test_eq(wa['list_watches'](due_only=True), 'nothing is being watched')
test_eq(wa['cancel_watch']('w1'), 'cancelled w1')
answer = at['ask_memory']('boiling point')
assert 'Water boils at 100C.' in answer and '[1] Kettles  (d1)' in answer and 'policy' not in answer
assert '[policy redact: withheld name' in at['ask_memory']('who')                          # the footer names the policy applied
test_eq({t.__name__ for t in mt.values() if is_write(t)}, {'memory_forget'})
test_eq({t.__name__ for t in wa.values() if is_write(t)}, {'cancel_watch'}); assert has_effect(wa['watch'])
for t in [*mt.values(), *wa.values(), *at.values()]: test_eq(len((t.__doc__ or '').strip().splitlines()), 1)
test_eq(sorted(WATCH_KINDS), ['folder', 'remind', 'search', 'url'])
test_eq(json.loads(mt['memory_read']('C# notes'))[0]['id'], 'd1')                            # a title with '#' is still a document

A vault host answers a search with vishalakshi’s context object, which holds results and related, not a list. memory_search passes the whole object on. Treating it as a list would leave only its key names.

class VaultShape:
    "What ramabana's `VaultHost.memory_search` returns over vishalakshi 0.1.17's `Vault.context`."
    def memory_search(self, query, limit=8):
        return AttrDict(query=query, encoder='model2vec',
                        results=[{'node_id': 'd1#0', 'doc_id': 'd1', 'breadcrumb': 'Kettles', 'text': 'boils at 100C', 'age': 60, 'stale': False}],
                        related=[{'node_id': 'd2#1', 'text': 'steam', 'age': None, 'stale': True, 'via': 'graph'}])
got = json.loads(memory_tools(VaultShape())[0]('kettle'))
test_eq(got['results'][0]['text'], 'boils at 100C')
test_eq((got['results'][0]['age'], got['results'][0]['stale']), (60, False))
test_eq(got['related'][0]['via'], 'graph')
class LShape:
    def memory_search(self, query, limit=8): return L([{'node_id': 'd1#0', 'text': 'boils'}])   # `Vault.search` answers with an `L`
test_eq(json.loads(memory_tools(LShape())[0]('kettle')), [{'node_id': 'd1#0', 'text': 'boils'}])

The live session and the shell

session_tools builds the tools for the user’s live Python session. run_python runs code in it and binds results to new names. inspect_python looks at variables without changing them, and with no code lists them. restart_kernel clears the session, and read_terminal reads what the terminal has printed.

shell_tools builds the shell tools. run_shell runs one command that finishes and returns its exit code and output; a failing command’s result starts with ERROR:. run_shell_bg starts a long-running command such as a server and returns an id, shell_output reads its latest output, and shell_stop ends it.

---

source

session_tools

def session_tools(
    host, mx:int=6000
):

Tools for the user’s live Python session and terminal: run_python, inspect_python, restart_kernel and read_terminal.

---

source

shell_tools

def shell_tools(
    host, mx:int=6000
):

Tools to run shell commands, in the foreground or in the background.

run_python and the shell tools are write tools, so an application can ask a person before they run. They report failures as text and never raise.

st = {t.__name__: t for t in session_tools(host)}
sh = {t.__name__: t for t in shell_tools(host)}
st['run_python']('total = 6 * 7'), st['run_python']('total'), sh['run_shell']('echo hi')

import time

started = sh['run_shell_bg']('echo one; sleep 30')
rid = started.split()[1].rstrip(';')
time.sleep(0.5)
assert sh['shell_output'](rid).startswith('running\none'), sh['shell_output'](rid)
test_eq(sh['shell_stop'](rid), 'stopped')
assert failed(sh['shell_output']('cmd_nope'))
test_eq({t.__name__ for t in shell_tools(host) if is_write(t)}, {'run_shell', 'run_shell_bg'})
test_eq(sorted(sh), ['run_shell', 'run_shell_bg', 'shell_output', 'shell_stop'])

test_eq(st['run_python']('total'), '42')
assert 'total' in st['inspect_python']() and 'total' in st['inspect_python']('')   # empty code lists the variables
test_eq(st['inspect_python']('total + 1'), '43')
assert 'hi' in sh['run_shell']('echo hi')
host.note('a status line')
assert 'a status line' in st['read_terminal'](), st['read_terminal']()
assert is_write(st['run_python']) and is_write(sh['run_shell'])
assert not is_write(st['inspect_python']) and not is_write(st['read_terminal'])
test_eq(sorted(st), ['inspect_python', 'read_terminal', 'restart_kernel', 'run_python'])
assert is_write(st['restart_kernel']) and 'restart_kernel' in WRITE_TOOLS
test_eq(st['restart_kernel'](), 'the namespace is empty'); assert 'total' not in host.ns
class NoKernel:
    def list_vars(self): raise RuntimeError('no kernel')
listed = {t.__name__: t for t in session_tools(NoKernel())}['inspect_python']('')
assert failed(listed) and 'no kernel' in listed, listed                    # listing fails like any other inspection

An API specification

api_tools lets an agent work with a web API from its OpenAPI specification. api_load loads a specification and reports how many operations it has, api_ops lists operations with their parameters, filtered and paged, and api_call calls one operation with its declared parameters.

---

source

api_tools

def api_tools(
    host, mx:int=6000
):

Tools to load an OpenAPI specification, list its operations and call one: api_load, api_ops and api_call.

These checks use a stand-in API backend with three operations, to test the JSON output, paging, parameter passing and errors. The host notebook shows a real backend.

class ApiResults:
    def api_load(self, src, name=''): return {'name': name or 'petstore', 'operations': 3}
    def api_ops(self, group='', name='', match='', offset=0):
        rows = [{'name': 'listPets'}, {'name': 'getPet'}, {'name': 'createPet'}]
        if match: rows = [r for r in rows if match.lower() in r['name'].lower()]
        return rows[offset:offset + 1]
    def api_count(self, group='', name='', match=''):
        return len([n for n in ('listPets', 'getPet', 'createPet') if not match or match.lower() in n.lower()])
    def api_call(self, operation, name='', **params):
        if operation == 'fail': raise RuntimeError('service unavailable')
        return {'called': operation, 'with': params}

api_results = ApiResults()
apit = {t.__name__: t for t in api_tools(api_results)}

test_eq(sorted(apit), ['api_call', 'api_load', 'api_ops'])
test_eq(json.loads(apit['api_load']('spec.json')), {'name': 'petstore', 'operations': 3})
page = json.loads(apit['api_ops']())
test_eq(page['operations'], [{'name': 'listPets'}])
test_eq((page['matched'], page['showing'], page['more']), (3, '1-1', 'call again with offset=1'))
test_eq(json.loads(apit['api_ops'](match='get'))['operations'], [{'name': 'getPet'}])
test_eq(json.loads(apit['api_call']('getPet', params={'id': 7})),
        {'called': 'getPet', 'with': {'id': 7}})
assert failed(apit['api_call']('fail')) and 'service unavailable' in apit['api_call']('fail')

Skills as tools

A skill is a document of instructions for one kind of task (see shalya.skills). The system prompt lists only skill names and descriptions, and skill_tools builds read_skill, which fetches a skill’s full text when the agent needs it. get_skills is a function that returns the current list of skills. author_tools builds the opt-in create_skill, which writes a new project skill.

---

source

skill_tools

def skill_tools(
    host, get_skills, mx:int=6000
):

The read_skill tool, which returns the full text of a skill the system prompt lists.

---

source

author_tools

def author_tools(
    host, mx:int=6000
):

The opt-in create_skill tool, which writes a new skill under .agents/skills/ in the first open folder.

read_skill finds a skill by name or unique prefix, and a name that matches nothing gets the list of available skills.

from shalya.skills import discover

skill_path = root/'.agents'/'skills'/'house-style'/'SKILL.md'
skill_path.parent.mkdir(parents=True)
skill_path.write_text(
    '---\nname: house-style\ndescription: How code is written here.\n---\n\nDense lines. Short names.\n')
skills = discover(roots=[root])
skt = {t.__name__: t for t in skill_tools(host, lambda: skills)}
skt['read_skill']('house-style')

from shalya.skills import MAX_SKILL_CHARS

test_eq(sorted(skt), ['read_skill'])
body = skt['read_skill']('house-style')
assert body.startswith('<skill name="house-style"') and 'Dense lines.' in body
test_eq(skt['read_skill']('house'), body)                 # unique prefix resolves to the same skill
assert "no skill matching 'nope'" in skt['read_skill']('nope')
assert 'house-style' in skt['read_skill']('nope')
assert not any(is_write(t) for t in skill_tools(host, lambda: skills))
test_eq([t.__name__ for t in author_tools(host)], ['create_skill']); test_eq(author_tools(frozen), [])   # writing one is an opt-in, and needs a writable host
class _Big:
    name, where = 'big', 'here'
    def text(self): return 'x' * (MAX_SKILL_CHARS - 400)
read_big = lambda mx: {t.__name__: t for t in skill_tools(host, lambda: [_Big()], mx)}['read_skill']('big')
big = read_big(MAX_SKILL_CHARS)       # a budget that holds the body gets it whole
assert big.rstrip().endswith('</skill>'), big[-80:]
small = read_big(MAX_TOOL_CHARS)      # a smaller one gets what fits, and the file holding the rest
assert len(small) <= MAX_TOOL_CHARS and 'saved to' in small, small[-200:]

create_skill writes .agents/skills/NAME/SKILL.md under the first open folder, the layout discover reads, and refuses to overwrite a skill that already exists.

au = {t.__name__: t for t in author_tools(host)}
au['create_skill']('notebook-tests', 'when testing', 'Put the readable case in the notebook.')

made = root/'.agents'/'skills'/'notebook-tests'/'SKILL.md'
assert made.exists(), made
assert 'when testing' in made.read_text() and 'readable case' in made.read_text()
assert au['create_skill']('notebook-tests', 'again', 'body').startswith('refusing to overwrite')
assert 'kebab-case' in au['create_skill']('Notebook Tests', 'x', 'y')
assert is_write(au['create_skill'])

Pictures

image_tools builds generate_image, which draws pictures through an OpenAI-shaped image API. route says which: a function returning the base_url, api_key and model, or a sentence saying how to set one up, which the tool returns instead of calling anything. Without one it is openai_route, OpenAI’s own API with OPENAI_API_KEY. An application with its own model configuration, such as Ramabana’s [jobs] image, passes its own, so an Azure OpenAI or other OpenAI-compatible deployment draws and nothing is sent to OpenAI by name. turn_route gives the endpoint and key of the turn’s own model, for when it draws itself. It picks an endpoint in this order. With reference pictures, it uses the image edits endpoint, so faces, style and palette are taken from them. Otherwise, if the turn’s own model can draw (draws_itself) and no model was named, the model draws through the Responses API and from_reply pulls the pictures out of its reply. Otherwise it uses the image generation endpoint. The constants below name the endpoints, the default model and the limits.

---

source

post_responses

def post_responses(
    prompt, model, timeout:int=300, base_url:NoneType=None, api_key:NoneType=None
):

Return a Responses API image-generation reply from base_url, OpenAI’s by default.

---

source

api_model

def api_model(
    model
):

A model id without its openai/ or azure/ prefix.

---

source

image_dest

def image_dest(
    host, path
):

(resolved, '') where a drawn picture may be saved, or (None, err): inside the roots, outside .git and credential paths, not a symlink, not over a file that is not a picture.

---

source

image_targets

def image_targets(
    path, n:int=1
):

The names n pictures saved at path get: path, then stem-2, stem-3, … beside it.

---

source

image_refs

def image_refs(
    host, paths, model:str='gpt-image-1'
):

(refs, '') for pictures inside the host’s roots as (name, bytes, mime), or ([], err) for the first one refused, host.read_guard included.

---

source

post_image

def post_image(
    prompt, size, n, timeout:int=120, model:str='gpt-image-1', base_url:NoneType=None, api_key:NoneType=None
):

Raw data rows from the images endpoint under base_url, OpenAI’s by default.

---

source

openai_route

def openai_route(
    model:str=''
):

The image route OpenAI’s own API gives, {base_url, api_key, model}, or why there is none.

---

source

image_available

def image_available():

---

source

save_media

def save_media(
    m, session:str='', stem:str='image', host:NoneType=None
):

Save one media item as the next numbered file in media_dir(session, host) and return its path.

---

source

mime_for

def mime_for(
    path
):

A file’s MIME type from its first bytes, then from its suffix, defaulting to image/png.

---

source

media_dir

def media_dir(
    session:str='', host:NoneType=None
):

The media folder under session, else under host’s first open folder, created if needed; never the working directory.

assert _post_image is post_image   # the old name still imports

assert _post_responses is post_responses   # the old name still imports

save_media saves a generated picture in the session’s media folder, numbered after the highest number already used, so a front end can find the result from its session. Without a session the folder is media in the host’s first open folder, checked by host.check like any write. With neither it raises rather than fall back to the process’s working directory.

png = {'mime': 'image/png', 'data': b'\x89PNG\r\n\x1a\n' + b'0' * 20}
saved = save_media(png, session=root/'s1', stem='generated')
saved, mime_for(saved)

assert saved.exists() and saved.suffix == '.png'
test_eq(saved.parent, media_dir(root/'s1'))
test_eq(saved.read_bytes(), png['data'])
test_eq(mime_for(saved), 'image/png')
test_eq(saved.name, 'generated-1.png')
test_eq(save_media(png, session=root/'s1', stem='generated').name, 'generated-2.png')
(media_dir(root/'s1')/'generated-1.png').unlink()          # a deleted file frees no number
test_eq(save_media(png, session=root/'s1', stem='generated').name, 'generated-3.png')
mroot = Path(tempfile.mkdtemp()).resolve()
test_eq(save_media(png, host=LocalHost([mroot], index=False)).parent, mroot/'media')   # no session: the host's open folder
test_fail(lambda: save_media(png), contains='no folder for media')                       # never the working directory

---

source

image_tools

def image_tools(
    host, mx:int=6000, session:str='', draws_itself:NoneType=None, from_reply:NoneType=None, model_id:str='',
    on_media:NoneType=None,
    route:NoneType=None, # `() -> {base_url, api_key, model}` for the image endpoints, or a str saying how to set one up; `openai_route` by default
    turn_route:NoneType=None, # `() -> {base_url, api_key}` where the turn's own model is served, for it to draw itself
):

The generate_image tool, which draws pictures with the turn’s own model where it can and through route’s image endpoints otherwise.

The image group needs a credential, not a capability. image_available checks whether OPENAI_API_KEY is set, and without a route the tool says what is missing and calls nothing. tools_for never checks; the application builds the group and passes it in as image.

image_available(), api_model('gpt-image-1'), api_model('openai/gpt-image-1')

test_eq(image_available(), bool(os.environ.get('OPENAI_API_KEY')))
test_eq(api_model('openai/gpt-image-1'), 'gpt-image-1')   # a vendor prefix is stripped
test_eq(api_model('gpt-image-1'), 'gpt-image-1')
assert '1024x1024' in IMAGE_SIZES and 'auto' in IMAGE_SIZES
test_eq(IMAGE_MODEL, 'gpt-image-1')
assert RESPONSES_API.startswith('https://') and IMAGE_API.startswith('https://')
assert all(v.endswith('/') for v in API_VENDORS)
gen = image_tools(host)[0]
test_eq(gen.__name__, 'generate_image')
if not image_available():
    assert 'OPENAI_API_KEY' in gen('a kettle')            # it says so rather than reaching the wire
else: assert failed(gen('a kettle', size='3x3'))          # and an unknown size never gets sent

Reference pictures go to the edits endpoint, so a face or a style is copied from them instead of described. path saves the result where the caller asked, inside the open folders. view_file recognises a picture and returns it as one, and create_file refuses to overwrite a picture with text. The endpoints are replaced with fakes here, so nothing is sent over the network.

from base64 import b64encode

png = b'\x89PNG\r\n\x1a\n' + bytes(32)
iroot = Path(tempfile.mkdtemp()).resolve(); ihost = LocalHost([iroot], index=False)
(iroot/'refs').mkdir(); (iroot/'refs'/'rama.png').write_bytes(png); (iroot/'notes.txt').write_text('hi')
calls, _key, _saved = [], os.environ.get('OPENAI_API_KEY'), (post_image, _post_edit)
def _fake_edit(prompt, size, n, images, model=IMAGE_MODEL, **kw):
    calls.append(('edit', model, [i[0] for i in images], images[0][1]))
    return [{'b64_json': b64encode(png).decode()}] * n
def _fake_image(prompt, size, n, model=IMAGE_MODEL, **kw):
    calls.append(('gen', model))
    return [{'b64_json': b64encode(png).decode()}] * n
post_image, _post_edit = _fake_image, _fake_edit
os.environ['OPENAI_API_KEY'] = _key or 'test'
try:
    gen = image_tools(ihost, session=iroot)[0]
    test_eq(gen('rama in exile', images=['refs/rama.png'], path='refs/rama_exile.png'), str(iroot/'refs'/'rama_exile.png'))
    test_eq((iroot/'refs'/'rama_exile.png').read_bytes(), png)
    test_eq(calls[-1], ('edit', 'gpt-image-1', ['rama.png'], png))
    assert failed(gen('x', images=['refs/missing.png']))
    assert 'not a picture' in gen('x', images=['notes.txt'])
    assert failed(gen('x', path='../outside.png'))
    test_eq(len(calls), 1)                                   # none of the refusals reached the wire
    test_eq(gen('pair', n=2, path='refs/pair.png').splitlines(), [str(iroot/'refs'/'pair.png'), str(iroot/'refs'/'pair-2.png')])
    plain = gen('no refs'); assert Path(plain).parent.name == 'media' and calls[-1][0] == 'gen', plain
    assert 'images' in get_schema(gen)['input_schema']['properties'] and 'path' in get_schema(gen)['input_schema']['properties']
    cwd, away = os.getcwd(), Path(tempfile.mkdtemp()).resolve()
    os.chdir(away)                                           # no session: media goes to the host's folder, not the process's
    try: test_eq(Path(image_tools(ihost)[0]('no session')).parent, iroot/'media'); test_eq(list(away.iterdir()), [])
    finally: os.chdir(cwd)
finally:
    post_image, _post_edit = _saved
    if _key is None: os.environ.pop('OPENAI_API_KEY')
    else: os.environ['OPENAI_API_KEY'] = _key
ift = {t.__name__: t for t in file_tools(ihost)}
v = ift['view_file']('refs/rama.png')
test_eq(v.media, [str(iroot/'refs'/'rama.png')]); assert not failed(v) and 'image/png' in v and str(iroot/'refs'/'rama.png') in v, v
(iroot/'refs'/'blob.bin').write_bytes(b'\xff\xfe\x00\x01' * 8)
v = ift['view_file']('refs/blob.bin'); assert failed(v) and 'not text' in v and not getattr(v, 'media', None), v   # bytes that are not a picture
(iroot/'refs'/'named.png').write_text('not a picture at all')
assert not getattr(ift['view_file']('refs/named.png'), 'media', None)   # the bytes decide, not the name
(iroot/'.env').write_bytes(png); assert failed(ift['view_file']('.env'))   # a credential path stays refused
(out := Path(tempfile.mkdtemp()).resolve()/'x.png').write_bytes(png)
assert failed(ift['view_file'](str(out)))                    # outside the roots
v = file_tools(LocalHost([iroot], index=False, read_outside=True))[0](str(out))
assert failed(v) and 'outside the open folders' in v, v      # `read_outside` reads text, not pictures
c = ift['create_file']('refs/rama.png', ''); assert failed(c) and 'not text' in c, c
test_eq((iroot/'refs'/'rama.png').read_bytes(), png)         # the picture survives
assert not failed(ift['create_file']('notes.txt', 'bye')); test_eq((iroot/'notes.txt').read_text(), 'bye')

With a path, drawing writes a file, and it is held to the same rules as a write. The destination, and every -2, -3 name derived from it for extra pictures (image_targets), is checked through the host before anything is drawn (image_dest): never inside .git, never a credential path, never through a symlink, and never over a file that is not already a picture. image_refs checks the reference pictures: they are recognised by their bytes alone (PNG, JPEG, WebP or GIF), at most MAX_IMAGES of them, each at most MAX_IMAGE_BYTES. When the API refuses a request, the error includes the body it sent. The read-only version of the tool has no path argument and saves only to the media folder.

import httpx

jpg, gif, webp = b'\xff\xd8\xff\xe0' + bytes(16), b'GIF89a' + bytes(16), b'RIFF\0\0\0\0WEBPVP8 ' + bytes(8)
test_eq([picture_mime(b) for b in (png, jpg, gif, webp, b'<svg xmlns="x"/>', b'all:\n\techo')], ['image/png', 'image/jpeg', 'image/gif', 'image/webp', None, None])
(iroot/'main.py').write_text('print(1)\n'); (iroot/'Makefile').write_text('all:\n'); (iroot/'.env').write_text('KEY=1\n')
(iroot/'fake.png').write_text('not a picture'); (iroot/'logo.svg').write_text('<svg xmlns="http://www.w3.org/2000/svg"/>')
(iroot/'.git').mkdir(); (iroot/'.git'/'config').write_text('[core]\n')
outside = Path(tempfile.mkdtemp()).resolve()/'victim.txt'; outside.write_text('keep')
(iroot/'refs'/'img-2.png').symlink_to(outside)
for bad in ('Makefile', '.env', 'fake.png', 'logo.svg'): assert 'not a picture' in image_refs(ihost, [bad])[1], bad
assert 'at most' in image_refs(ihost, ['refs/rama.png'] * (MAX_IMAGES + 1))[1]
with open(iroot/'big.png', 'wb') as f: f.write(png); f.truncate(MAX_IMAGE_BYTES + 1)
assert 'over' in image_refs(ihost, ['big.png'])[1]
calls, _key, _saved = [], os.environ.get('OPENAI_API_KEY'), (post_image, _post_edit)
post_image, _post_edit = _fake_image, _fake_edit
os.environ['OPENAI_API_KEY'] = _key or 'test'
try:
    gen = image_tools(ihost, session=iroot)[0]
    for dest in ('main.py', '.git/config', 'refs/img.png'):   # a text file, git's own, a planted symlink for the -2 name
        out = gen('x', n=2, path=dest); assert failed(out), (dest, out)
    test_eq((iroot/'main.py').read_text(), 'print(1)\n'); test_eq((iroot/'.git'/'config').read_text(), '[core]\n'); test_eq(outside.read_text(), 'keep')
    test_eq(calls, [])                                                       # refused before anything was drawn
    assert not failed(gen('again', path='refs/rama_exile.png'))             # over a picture is allowed
    assert 'whole number' in gen('x', n='two')
    ro = gen.read_only                                                       # what `read_only` hands a view that must not write
    test_eq(ro.__name__, 'generate_image'); assert 'path' not in get_schema(ro)['input_schema']['properties']
    assert 'images' in get_schema(ro)['input_schema']['properties'] and has_effect(ro)
    assert Path(ro('plain')).parent.name == 'media'
finally:
    post_image, _post_edit = _saved
sent, _hp = [], httpx.post
def _refuse(url, **kw):
    sent.append((url, kw))
    return httpx.Response(400, text='{"error": {"message": "Invalid value for size"}}', request=httpx.Request('POST', url))
httpx.post = _refuse
try:
    gen = image_tools(ihost, session=iroot)[0]
    assert 'Invalid value for size' in gen('x'), gen('x')
    assert 'Invalid value for size' in gen('x', images=['refs/rama.png'])
    test_eq(sent[-1][1]['data']['input_fidelity'], 'high')                   # a reference face is kept, not paraphrased
finally:
    httpx.post = _hp
    if _key is None: os.environ.pop('OPENAI_API_KEY')
    else: os.environ['OPENAI_API_KEY'] = _key
assert 'not a text file' in ift['replace_text']('refs/rama.png', [{'oldText': 'a', 'newText': 'b'}])

# a route of the application's own: Azure's v1 API, with its key, and nothing sent to OpenAI by name
sent, _hp = [], httpx.post
def _ok(url, **kw):
    sent.append((url, kw['headers']['Authorization'], (kw.get('json') or kw.get('data') or {}).get('model')))
    return httpx.Response(200, json={'data': [{'b64_json': b64encode(png).decode()}]}, request=httpx.Request('POST', url))
httpx.post = _ok
az = dict(base_url='https://r.openai.azure.com/openai/v1', api_key='azk', model='my-image')
try:
    gen = image_tools(ihost, session=iroot, route=lambda: az)[0]
    assert not failed(gen('a kettle')) and not failed(gen('rama', images=['refs/rama.png']))
    test_eq(sent, [('https://r.openai.azure.com/openai/v1/images/generations', 'Bearer azk', 'my-image'),
                   ('https://r.openai.azure.com/openai/v1/images/edits', 'Bearer azk', 'my-image')])
    none = image_tools(ihost, session=iroot, route=lambda: 'no image model is set up: add [jobs] image in config.toml')[0]
    assert 'add [jobs] image' in none('a kettle') and len(sent) == 2   # it says how, and calls nothing
finally: httpx.post = _hp
_k = os.environ.pop('OPENAI_API_KEY', None)
test_eq(openai_route(), 'OPENAI_API_KEY is not set')
os.environ['OPENAI_API_KEY'] = 'k'; test_eq(openai_route('m'), {'base_url': OPENAI_BASE, 'api_key': 'k', 'model': 'm'})
if _k is None: os.environ.pop('OPENAI_API_KEY')
else: os.environ['OPENAI_API_KEY'] = _k

test_eq(image_targets('refs/rama.png', 3), ['refs/rama.png', 'refs/rama-2.png', 'refs/rama-3.png'])
test_eq(image_targets('a.png', 1), ['a.png'])
_saved = post_image, _post_edit; post_image, _post_edit = _fake_image, _fake_edit
os.environ['OPENAI_API_KEY'] = _key or 'test'
try:
    gen, calls = image_tools(ihost, session=iroot)[0], []
    for dest in ('.GIT/hooks/pre-commit', '.Git/config', '.git/HOOKS/x.png', '.ENV'):   # macOS opens .git by any case
        assert failed(gen('x', path=dest)), dest
    test_eq(calls, []); test_eq(sorted(p.name for p in (iroot/'.git').iterdir()), ['config'])
finally:
    post_image, _post_edit = _saved
    if _key is None: os.environ.pop('OPENAI_API_KEY')
    else: os.environ['OPENAI_API_KEY'] = _key

The live check calls the real OpenAI image endpoint and checks that the saved file is a non-empty picture. It is marked eval: false and left out of the test run, because it spends credits.

image_root = Path(tempfile.mkdtemp())
image_host = LocalHost([image_root], index=False)
generate_image = image_tools(image_host, session=image_root)[0]
image_result = generate_image('A simple black circle centered on a white background', n=1)
assert not image_result.startswith('ERROR:'), image_result
image_path = Path(image_result.strip())
assert image_path.exists() and image_path.stat().st_size > 1000
assert mime_for(image_path).startswith('image/')

Git

git_tools builds the git tools on top of gheasy. The read tools are git_status, git_divergence (how the branch differs from another, with the outcome of a fast-forward, merge, rebase or reset worked out in advance), git_diff and git_log. The write tools are git_commit, git_checkout, git_stash and git_remote (fetch, pull or push). Every tool takes path to choose the repository, and git_repo finds it: the repository holding path, else the host’s project, else its first open folder. A repository found by walking up from a path must still lie inside the open folders. screen_diff passes each file’s part of a diff through the host’s read_guard.

---

source

git_tools

def git_tools(
    host, mx:int=6000
):

Git tools for repositories inside the host’s open folders; each write reports an undo token and the new state.

---

source

screen_diff

def screen_diff(
    host, root, diff
):

A git diff as host.read_guard lets each file’s part go; a part naming no file is asked about as root.

---

source

git_repo

def git_repo(
    host, path:str=''
):

The repository holding path, else the host’s project, else its first root; refused outside the open folders.

The example uses a real repository with one commit and one untracked file. Every write returns gheasy’s report: result, summary, the undo token when gheasy recorded one ('' for a stash push, a fetch, a push or a new branch), and the repository state.

A manual check, not a test: git_remote('push') on a clone of an https remote with no credential helper returns ERROR: git push: fatal: could not read Username for 'https://github.com': terminal prompts disabled within a second, because building git_tools sets GIT_TERMINAL_PROMPT=0. Without that, git would wait for a password nobody can type.

repo = Path(tempfile.mkdtemp()).resolve()/'repo'
repo.mkdir(parents=True)
(repo/'a.py').write_text('x = 1\n')
for c in (['init', '-q', '-b', 'main'], ['config', 'user.email', 't@e'], ['config', 'user.name', 'T'],
          ['add', 'a.py'], ['commit', '-qm', 'first']):
    subprocess.run(['git', '-C', str(repo)] + c, check=True)
(repo/'b.py').write_text('y = 2\n')
ghost = LocalHost([repo], index=False)
_gtp = os.environ.pop('GIT_TERMINAL_PROMPT', None)   # the assertion below is about building the group, not the outer shell
git_tools(None); assert 'GIT_TERMINAL_PROMPT' not in os.environ                # naming the tools touches nothing; a host building them does
gt = {t.__name__: t for t in git_tools(ghost)}
status = json.loads(gt['git_status']())
status['branch'], status['clean'], [c['path'] for c in status['changes']]

test_eq(status['branch'], 'main')
test_eq(status['clean'], False)
test_eq([c['path'] for c in status['changes']], ['b.py'])
test_eq(sorted(gt), sorted(GIT_TOOLS)); assert 'git_rebase_preview' not in gt      # divergence carries the rehearsal now
test_eq({t.__name__ for t in git_tools(ghost) if is_write(t)}, set(GIT_WRITE_TOOLS))
test_eq(os.environ.get('GIT_TERMINAL_PROMPT'), '0')                                # building the group disarms credential prompts
if _gtp is not None: os.environ['GIT_TERMINAL_PROMPT'] = _gtp
for t in gt.values(): test_eq(len((t.__doc__ or '').strip().splitlines()), 1)

assert failed(gt['git_divergence']()) and 'tracks nothing' in gt['git_divergence']()
test_eq(subprocess.run(['git', '-C', str(repo), 'status', '--porcelain'],
                       capture_output=True, text=True).stdout.strip(), '?? b.py')

assert failed(gt['git_checkout']('no-such-branch'))
assert 'op must be one of' in gt['git_remote']('nonsense')   # never reaches the network
fetched = gt['git_remote']('fetch')                          # no remote, so it is a no-op not a lie
assert failed(fetched) or json.loads(fetched)['branch'] == 'main', fetched
test_eq(sorted(status), sorted(STATE_KEYS))                                        # state only: no empty write report

(repo/'a.py').write_text('x = 2\n')
assert '+x = 2' in gt['git_diff']() and gt['git_diff'](staged=True) == '(no changes)'
assert '+x = 2' in gt['git_diff'](path='a.py') and gt['git_diff'](path='b.py') == '(no changes)'
# a branch with its own commit: divergence against main carries the rebase rehearsal the old preview gave
made = json.loads(gt['git_checkout']('feature', create=True))
test_eq((made['branch'], made['result']), ('feature', 'created and switched to feature'))
committed = json.loads(gt['git_commit']('second', 'a.py b.py'))
test_eq((committed['clean'], committed['moved'], committed['undoes'] != ''), (True, True, True))
assert committed['undo'] and committed['summary'], committed
div = json.loads(gt['git_divergence'](against='main'))
test_eq((div['relation'], div['ahead'], div['behind']), ('ahead', 1, 0))
assert 'stops_at' in next(o for o in div['options'] if o['op'] == 'rebase')
amended = json.loads(gt['git_commit']('second, amended', amend=True)); assert amended['undo']
test_eq([c['subject'] for c in json.loads(gt['git_log'](5))], ['second, amended', 'first'])
assert failed(gt['git_commit'](''))
(repo/'a.py').write_text('x = 3\n'); stashed = json.loads(gt['git_stash']())
test_eq((stashed['clean'], stashed['undo']), (True, ''))                    # stash push is not journaled
test_eq((repo/'a.py').read_text(), 'x = 2\n')
popped = json.loads(gt['git_stash']('pop')); test_eq((repo/'a.py').read_text(), 'x = 3\n'); assert popped['undo']
assert 'action must be one of' in gt['git_stash']('nonsense')
# Review Focus 4: a remote that is not there fails fast with an error string, never a hang
subprocess.run(['git', '-C', str(repo), 'remote', 'add', 'origin', 'file:///nonexistent/remote.git'], check=True)
pushed = gt['git_remote']('push', publish=True); assert failed(pushed), pushed
assert json.loads(gt['git_stash']('push', message='wip'))['clean']
test_eq(gt['git_checkout']('main')[:1], '{')

git_commit chooses its repository from path by the same git_repo rule as the other git tools. Its paths are relative to that repository.

other = repo.parent/'other'
other.mkdir()
for c in (['init', '-q', '-b', 'main'], ['config', 'user.email', 't@e'], ['config', 'user.name', 'T'], ['commit', '-q', '--allow-empty', '-m', 'first']):
    subprocess.run(['git', '-C', str(other)] + c, check=True)
(other/'c.py').write_text('z = 1\n')
two = LocalHost([repo, other], index=False)
g2 = {t.__name__: t for t in git_tools(two)}
test_eq(git_repo(two, str(other)).root, other); test_eq(git_repo(two).root, repo)
heads = subprocess.run(['git', '-C', str(repo), 'rev-parse', 'HEAD'], capture_output=True, text=True).stdout
made2 = json.loads(g2['git_commit']('in the second repository', 'c.py', path=str(other)))
assert made2['undo'] and made2['clean'], made2
test_eq(subprocess.run(['git', '-C', str(other), 'log', '-1', '--format=%s'], capture_output=True, text=True).stdout.strip(), 'in the second repository')
test_eq(subprocess.run(['git', '-C', str(repo), 'rev-parse', 'HEAD'], capture_output=True, text=True).stdout, heads)   # the first repository is untouched
test_eq(summarise(g2['git_commit'], {'message': 'm', 'path': 'other'}), 'Git commit: m in other')
assert 'outside the open folders' in g2['git_commit']('x', path=str(repo.parent))

Which tools a host gets

tools_for(host) builds the tools for every group in host.provides, in the order of GROUPS. drop removes groups by name. get_skills adds read_skill. image is a list of already-built image tools, or None: building them needs to know what the turn’s model can do, which the application knows and the host does not. optin adds groups that are off by default, and extra adds any other tools at the end.

Tools nobody gets by default

Five groups are offered only when a caller names them in optin: exhash (edit_file), research, author (create_skill), browser (for a host that has one) and legacy. The legacy tools are shims for one release, for MCP clients that still send the old names: list_files, list_vars, environment, memory_tree, set_reminder, watch_url and git_rebase_preview. Each calls its replacement, and its docstring names it. They are removed in 0.2.0.

---

source

legacy_tools

def legacy_tools(
    host, mx:int=6000
):

Shims, for one release, for old tool names an MCP client may still send; each calls its replacement.

---

source

tools_for

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

The tools for every group host provides, less drop, plus skills, image tools, the optin groups and extra.

Code without a host sometimes needs a tool’s group, for example to label a call read back from a saved conversation, where only the tool’s name survives. tool_groups maps every tool name shalya can build to its group, and group_of(name) looks one up. The table is built by calling every factory without a host. A factory only defines functions that refer to the host, so this costs nothing and calls nothing. Opt-in tools are filed under the group they extend, so edit_file is in file and research in web.

---

source

group_of

def group_of(
    name, default:str=''
):

Which group a tool belongs to, or default for a name shalya does not build.

---

source

tool_groups

def tool_groups():

Every tool name shalya can build, mapped to its group, including the image, skill and opt-in tools.

GROUPS pairs each capability class with its factory. The group name is written only once, on the class, and the tests check that every capability class has a factory.

[cap.group for cap, _ in GROUPS]

from shalya.host import (ApiHost, AskHost, BrowserHost, CodeHost, GitHost, MemoryHost, NotebookHost,
                         SessionHost, ShellHost, WatchHost, WebHost)

declared = {c.group for c in (CodeHost, WebHost, BrowserHost, NotebookHost, MemoryHost, AskHost, WatchHost,
                              SessionHost, ShellHost, ApiHost, GitHost)} | {'file'}
test_eq({cap.group for cap, _ in GROUPS} | {'browser'}, declared)   # every capability has a factory; `browser`'s is the opt-in
test_eq(len(GROUPS), len(declared) - 1)                   # named once each
assert all(callable(f) for _, f in GROUPS)

A host over a git repository with no vault and no index gets seven groups: code, file, notebook, web, session, shell and git, 30 tools in all. A vault would add the memory, watch and ask groups, and an index would add similar_code and public_api. Groups the host cannot serve are simply absent, and nothing opt-in appears.

ts = tools_for(ghost)
sorted(t.__name__ for t in ts)

names = {t.__name__ for t in ts}
assert {'search_code', 'ls', 'view_file', 'replace_text', 'run_shell', 'inspect_python', 'git_status', 'git_divergence'} <= names, names
assert not (names & {'list_files', 'list_vars', 'environment', 'edit_file', 'research', 'create_skill', 'git_rebase_preview',
                     'memory_search', 'api_load', 'watch'}), 'no vault, no specs, nothing opt-in'
test_eq(len(names), 4 + 4 + 4 + 2 + 4 + 4 + 8)            # code, file, notebook, web, session, shell, git: 30 with no vault or index
test_eq(names, {t.__name__ for t in tools_for(ghost)})

fewer = {t.__name__ for t in tools_for(ghost, drop=['shell', 'git'])}
test_eq(names - fewer, {'run_shell', 'run_shell_bg', 'shell_output', 'shell_stop', *GIT_TOOLS})
assert 'search_code' in fewer

optin adds the groups that are off by default. Each opt-in factory returns nothing when the host cannot back it, such as research without web access, and an unknown name raises an error instead of being ignored. Tools in extra still come last.

with_all = {t.__name__ for t in tools_for(ghost, optin=('exhash', 'research', 'author', 'legacy'))}
assert {'edit_file', 'research', 'create_skill', 'list_files', 'list_vars', 'environment', 'git_rebase_preview'} <= with_all, with_all
noweb = LocalHost([repo], index=False, web=False)
test_eq({t.__name__ for t in tools_for(noweb, optin=('research',))} & {'research'}, set())   # an opt-in the host cannot back is empty
test_fail(lambda: tools_for(ghost, optin=('nope',)), contains='optin')
test_eq(tools_for(ghost, optin=('exhash',))[-1].__name__, 'edit_file')
test_eq([t.__name__ for t in tools_for(ghost, optin=('exhash',), extra=[ct['grep']])][-2:], ['edit_file', 'grep'])
# a vault host stays under the budget the spec sets
vh = LocalHost([repo], index=False, memory=memory_results)       # `MemoryResults` stands in for a vault
vnames = {t.__name__ for t in tools_for(vh)}
assert {'memory_search', 'memory_read', 'remember', 'memory_forget', 'watch', 'list_watches', 'cancel_watch', 'ask_memory'} <= vnames, vnames
test_eq(len(vnames), 30 + 4 + 3 + 1); assert len(vnames) <= 45
test_eq(tools_for(ghost, optin='exhash')[-1].__name__, 'edit_file')                        # one name as a bare string

The legacy shims call their replacements and name them in their docstrings. A host without a vault gets no memory or watch shims.

lg = {t.__name__: t for t in legacy_tools(ghost)}
test_eq(sorted(lg), ['environment', 'git_rebase_preview', 'list_files', 'list_vars'])
assert 'a.py' in lg['list_files']('a.py') and lg['list_files']('zzz') == 'no matching files'
assert all('use `' in (t.__doc__ or '').lower() for t in lg.values())
assert not failed(lg['list_vars']()) and not failed(lg['environment']())
assert 'relation' in json.loads(lg['git_rebase_preview']('feature'))
test_eq(sorted(t.__name__ for t in legacy_tools(vh)), ['environment', 'git_rebase_preview', 'list_files', 'list_vars', 'memory_tree', 'set_reminder', 'watch_url'])
class VaultLike:                                                                 # the host API over the fixture, declaring only the vault groups
    def can(self, group): return group in ('memory', 'watch')
    def __getattr__(self, name): return getattr(memory_results, name)
lgv = {t.__name__: t for t in legacy_tools(VaultLike())}
test_eq(sorted(lgv), ['memory_tree', 'set_reminder', 'watch_url'])
test_eq(json.loads(lgv['memory_tree']()), [{'id': 'd1', 'title': 'Kettles'}])
test_eq(lgv['set_reminder']('renew domain', every='1w'), 'watching remind renew domain as w1, every 1w'); test_eq(memory_results.watched[1], 'remind')
lgv['watch_url']('https://x.test'); test_eq(memory_results.watched[1], 'url')
assert has_effect(lgv['set_reminder']) and has_effect(lgv['watch_url']) and not any(is_write(t) for t in lgv.values())
test_eq(len(legacy_tools(None)), 7)                                              # buildable without a host, as `tool_groups` needs

file is not a capability class a host inherits. Every host has the folder boundary, so every host gets the file tools. GROUPS still lists it, so drop=['file'] can remove them.

What a sub-agent may have

A sub-agent that only looks things up should not change anything. read_only(tools) returns the tools it may have. Every write tool is removed, unless the tool provides a read-only version of itself in its read_only attribute, which is then used instead. writes=True keeps the write tools. block removes tools by name. max_calls puts the result behind a hard budget: once the sub-agent has made that many calls in total, every tool answers that the budget is spent.

---

source

read_only

def read_only(
    tools, max_calls:NoneType=None, writes:bool=False, effects:bool=True, block:tuple=()
):

tools without write tools (or without any tool that acts, with effects=False), optionally behind a budget of max_calls calls.

Write tools are removed whatever else the host offers. effects=False also removes the tools that act without writing a file the user owns: inspecting the live session, an API call that may change something on a server, drawing a picture that costs money, a watch that keeps running after the turn, a research digest, stopping a background command, and the browser’s looking tools. ACTING_TOOLS names them all. An agent asked to propose, not act, should not have them either. read_url stays, through its read-only version, which reads the page without saving it to memory.

[t.__name__ for t in read_only(ts)]

ro = {t.__name__ for t in read_only(ts)}
test_eq(ro & WRITE_TOOLS, set())
assert 'search_code' in ro and 'git_status' in ro
test_eq({t.__name__ for t in read_only(ts, writes=True)} & {'create_file'}, {'create_file'})
safe_web = {t.__name__: t for t in read_only(wt.values())}
test_eq(wt['read_url'].read_only, safe_web['read_url'])
safe_web['read_url']('https://python.org')
test_eq(web_results.remember, False)
test_eq(set(get_schema(safe_web['read_url'])['input_schema']['properties']), {'url'})
full_web = {t.__name__: t for t in read_only(wt.values(), writes=True)}
full_web['read_url']('https://python.org')
test_eq(web_results.remember, True)
draw = image_tools(host)[0]                                          # a path is a write: the read-only view has none
test_eq([t for t in read_only([draw])], [draw.read_only]); test_eq(read_only([draw], writes=True), [draw])
assert 'path' not in get_schema(read_only([draw])[0])['input_schema']['properties']

# `effects=False` also withholds what acts without writing a file the user owns
acting = {t.__name__ for t in ts if has_effect(t)}
test_eq(acting - ACTING_TOOLS, set())              # nothing is marked without being named
assert acting <= ro                                # and the default keeps every one of them
propose = {t.__name__ for t in read_only(ts, effects=False)}
test_eq(propose & (WRITE_TOOLS | ACTING_TOOLS), set())
assert 'search_code' in propose                    # looking is still allowed
# Review Focus 5: what a sub-agent keeps of the opt-ins
ro_opt = {t.__name__ for t in read_only(tools_for(ghost, optin=('exhash', 'research', 'legacy')))}
assert 'research' in ro_opt and 'list_files' in ro_opt and 'edit_file' not in ro_opt, ro_opt
assert 'research' not in {t.__name__ for t in read_only(tools_for(ghost, optin=('research',)), effects=False)}

With a budget, the tools themselves stop the loop. A local model engine runs its own tool loop, so a limit the harness keeps outside the tools could not be enforced. In this example the second call is refused.

budgeted = {t.__name__: t for t in read_only(ts, max_calls=1)}
budgeted['search_code']('threshold'), budgeted['ls']('.')

budgeted = {t.__name__: t for t in read_only(ts, max_calls=1)}
budgeted['ls']('.')
assert 'budget exhausted' in budgeted['ls']('.'), 'the second call should have been refused'

The browser

The browser group is a tab the agent drives. Looking at a page (browse, screenshot, page_text, page_reload) acts without writing. Acting on it (page_click, page_type, page_eval) counts as a write and goes to a person. A screenshot comes back as Media: text naming the saved file, carrying the picture for an agent loop that can send it to the model. The example uses a stand-in browser host.

class Tab(Host, BrowserHost):
    "A browser host over a dict of pages."
    roots, check, walk, read, write, text_at = (), None, None, None, None, None
    def __init__(self): self.seen = []
    def browse(self, url): self.seen.append(('browse', url)); return f'p1 Example'
    def screenshot(self, page=''): return '/tmp/p1.png'
    def page_text(self, page=''): return '# Example'
    def page_reload(self, page=''): return 'Example'
    def page_click(self, page, x, y): self.seen.append(('click', page, x, y)); return 'clicked'
    def page_type(self, page, text): return 'typed'
    def page_eval(self, page, js): raise ValueError('refused on this origin')
tab = Tab()
assert not any(group_of(t.__name__) == 'browser' for t in tools_for(tab)), 'opt-in: a host that can browse still offers nothing unasked'
bt = {t.__name__: t for t in tools_for(tab, optin=('browser',))}
test_eq(OPTIN['browser'](LocalHost([Path('.')], index=False, web=False)), [])   # named for a host without the group: nothing
test_eq([n for n in bt if group_of(n) == 'browser'], ['browse', 'screenshot', 'page_text', 'page_reload', 'page_click', 'page_type', 'page_eval'])
test_eq(bt['browse'](' https://example.com '), 'p1 Example'); test_eq(tab.seen[0], ('browse', 'https://example.com'))
s = bt['screenshot'](); test_eq(s.media, ['/tmp/p1.png']); assert '/tmp/p1.png' in s and not failed(s)
bt['page_click']('p1', '3', 4.0); test_eq(tab.seen[-1], ('click', 'p1', 3, 4))
assert failed(bt['page_eval']('p1', '1')) and 'refused' in bt['page_eval']('p1', '1')   # a host refusal is one tool failure
test_eq({n for n, t in bt.items() if group_of(n) == 'browser' and is_write(t)}, {'page_click', 'page_type', 'page_eval'})
test_eq({n for n, t in bt.items() if group_of(n) == 'browser' and has_effect(t)}, {'browse', 'screenshot', 'page_text', 'page_reload'})
assert {'page_click', 'page_type', 'page_eval'} <= WRITE_TOOLS and {'browse', 'screenshot', 'page_text', 'page_reload'} <= ACTING_TOOLS
test_eq({t.__name__ for t in read_only(bt.values())} & {'page_click', 'page_type', 'page_eval'}, set())
assert not issubclass(LocalHost, BrowserHost) and not LocalHost([Path('.')], index=False, web=False).can('browser')   # owes none of the methods