host

The application an agent works inside, and how it says what it can do.

Saying what a host can do

Each kind of support is a capability group, named by a short string such as 'code', 'web' or 'shell'. A capability class names one group and lists the methods its tools need as abstract methods. A host declares a group by inheriting the class. Python’s ABC machinery then refuses to create an instance of a host that inherits a capability class without writing all of its methods, so a host cannot claim a group it only half supports.

provides returns the groups a host’s classes declare, minus the instance’s without set. without covers groups this particular instance cannot serve even though its class can, for example because web access was switched off or the fossick package is not installed. can(group) asks about one group. tools_for in shalya.tools uses provides to decide which tools a host gets.

---

source

Capability

def Capability(
    *args, **kwargs
):

One capability group. A host declares the group by inheriting the class that names it.

Host is the base class of every host. The folder boundary is not a capability group of its own, because every tool needs it: roots (the open folders), check (resolve a path and refuse it if it falls outside them), walk, read, write and text_at are abstract methods of Host itself, and every host provides the file group. add_root, close, approvals and note have defaults a host may override.

Two attributes let an application screen what reaches the model. read_guard is described in shalya.core, and model_local is the function that tells it which model the turn uses. writes = False marks a host that cannot write, which gets no write tools.

---

source

Host

def Host(
    *args, **kwargs
):

The application under an agent: the folders it may touch, and what it declares it can do.

The groups

Each class below names one group and the methods its tools call. Methods with a body are optional and have a default; abstract ones must be written by any host that declares the group.

Notebook tools list cells by id and then edit a cell by sending its id back, so ids must not change between two reads. fastcore’s read_nb invents a random id for each cell saved without one, which would break that. nb_read is the notebook reader every host uses instead. It gives cells without ids stable ones based on their position: c0, c1 and so on.

---

source

nb_read

def nb_read(
    path
):

A notebook whose every cell has an id: c{i} by position for the ones saved without (nbformat 4.4), so a listed id survives the next read.

---

source

GitHost

def GitHost(
    *args, **kwargs
):

Git operations on a working tree inside roots.

---

source

ApiHost

def ApiHost(
    *args, **kwargs
):

Reading an API specification and calling what it describes.

---

source

ShellHost

def ShellHost(
    *args, **kwargs
):

Running a command on the machine.

---

source

SessionHost

def SessionHost(
    *args, **kwargs
):

The live namespace, and the terminal beside it.

---

source

WatchHost

def WatchHost(
    *args, **kwargs
):

Recurring watches and reminders.

---

source

AskHost

def AskHost(
    *args, **kwargs
):

Model-backed answers from memory.

---

source

MemoryHost

def MemoryHost(
    *args, **kwargs
):

Durable memory organized as document sections.

---

source

NotebookHost

def NotebookHost(
    *args, **kwargs
):

Notebook cells: listing, one-cell reads and writes, insertion.

---

source

BrowserHost

def BrowserHost(
    *args, **kwargs
):

A browser tab the agent drives: open, look, read, and act on a page.

---

source

WebHost

def WebHost(
    *args, **kwargs
):

Current web search and page retrieval.

---

source

CodeHost

def CodeHost(
    *args, **kwargs
):

Code search and structure.

The session and shell groups have contracts the tools rely on but cannot check, so a host must keep them:

• run_python may read anything in the user’s live namespace and bind its results to new names, but never rebinds or deletes a name the user made. The agent is told this, and the host enforces it.
• inspect_python runs code without changing the user’s names, in one of two scopes. isolated, the default, runs on a copy of the namespace in a sandbox that allows attribute reads and builtins and refuses most library method calls, so it needs no trust. overlay runs in the real interpreter against the real namespace, under rules that let the agent bind names in its own layer but never delete, rebind or change the user’s. list(df.columns) works in overlay and not in isolated. A host may refuse overlay, and scopes says which scopes it honours.
• run_cmd resolves cwd through check. Confining the working directory does not confine what the command does, which is why the run_shell tool counts as a write and goes to a person for approval. Standard output and standard error come back together in one string, in the order they were written. timeout is enforced by killing the command’s whole process group. A failing command returns a non-zero exit code; it does not raise.

There are eleven groups. The group name in each class is the key tools_for looks up, so the class and the toolset must agree on it, or the group’s tools never appear and nothing reports why. The test below pins the names.

GROUP_CLASSES = (CodeHost, WebHost, BrowserHost, NotebookHost, MemoryHost, AskHost, WatchHost, SessionHost,
                 ShellHost, ApiHost, GitHost)
sorted(c.group for c in GROUP_CLASSES)

test_eq(sorted(c.group for c in GROUP_CLASSES),
        ['api', 'ask', 'browser', 'code', 'git', 'memory', 'notebook', 'session', 'shell', 'watch', 'web'])
assert all(issubclass(c, Capability) for c in GROUP_CLASSES)
test_eq(Capability.group, '')                             # the base names no group
test_eq(Host.group, 'file')                               # and the path boundary is its own
test_eq(len({c.group for c in GROUP_CLASSES}), len(GROUP_CLASSES))   # no two share a name

provides is the only way the toolset asks what a host supports. This example host declares the code and web groups, but its without removes web, as if fossick were not installed.

class Demo(Host, CodeHost, WebHost):
    without = {'web'}                                    # fossick is not installed here
    @property
    def roots(self): return ['/proj']
    def check(self, path, must_exist=False, reading=False): return Path('/proj')/path
    def walk(self): return []
    def read(self, path): return None
    def write(self, path, text): return str(path)
    def text_at(self, path): return ''
    def search(self, query, limit=20): return [Hit('/proj/a.py', 1, 'a', 'def a(): ...')]
    def symbols(self, path): return []
    def peers(self, path, line, limit=20): return []
    def public_api(self, package, limit=MAX_API): return []
    def web_search(self, query, n=20): return []
    def read_url(self, url, remember=True): return None
    def research(self, query): return ''

d = Demo()
sorted(d.provides), d.can('code'), d.can('web')

test_eq(sorted(d.provides), ['code', 'file'])   # every host has the file group
test_eq(d.can('code'), True)
test_eq(d.can('web'), False)
test_eq(d.can('memory'), False)

A host that declares a group without writing all its methods cannot be created, and the error names the missing method:

class Broken(Demo, NotebookHost):
    def nb_cells(self, path): return []                  # and no `nb_add_cell`

test_fail(Broken, contains='nb_add_cell')

A host over real folders

LocalHost is the reference host. It works on folders on the local machine and declares every group except browser. A group whose backend is missing is put in without: web when web access is off or fossick is not installed, memory, ask and watch when no vault is given, and api when no API backend is given. A kernel can be attached later without building a new host.

SKIP_DIRS and SKIP_SUFFIXES name what a walk or a grep leaves out: version-control folders, virtual environments, caches, build output, and binary files such as images, archives and model weights. They are not part of the project, and reading them would waste the model’s context.

assert {'.git', '.venv', '__pycache__', 'node_modules'} <= SKIP_DIRS
assert {'.pyc', '.so', '.png', '.gguf'} <= SKIP_SUFFIXES
assert MAX_VARS > 0 and LD_CHARS > 0

---

source

ld_json

def ld_json(
    html
):

Parse schema.org JSON-LD blocks from html.

---

source

fuse

def fuse(
    legs, limit
):

Fuse ranked Hit lists by path:line.

assert _fuse is fuse   # the old name still imports

ld_json pulls the structured data out of a web page: the schema.org JSON-LD blocks that sites embed for search engines. A product page’s price and rating usually live there and not in the visible text, so read_page puts these blocks on top of the page’s markdown.

page = '<html><script type="application/ld+json">{"@type": "Product", "name": "a kettle"}</script></html>'
ld_json(page)

test_eq(ld_json(page), [{'@type': 'Product', 'name': 'a kettle'}])
test_eq(ld_json('<html>nothing structured here</html>'), [])
test_eq(ld_json('<script type="application/ld+json">not json</script>'), [])
test_eq(ld_json(''), [])

These are the constructor’s arguments. roots are the folders the agent may read and write, and read_roots folders it may only read. With index=True, the default, the host starts indexing every open folder with kosha in a background thread, and search uses ripgrep until that index is ready. web, memory, apis and kernel attach the optional backends. read_outside=True lets read tools open any file on the machine except credential files. without is computed once, from these arguments and the packages installed.

---

source

LocalHost

def LocalHost(
    roots:tuple=('.',), # the folders the agent is confined to; each must exist
    ns:NoneType=None, # live namespace; fresh dict when None
    approvals:NoneType=None, # write approval handler
    note:NoneType=None, # status callback
    web:bool=True, # use fossick when installed
    index:bool=True, # start a Kosha sync for every open root
    graph:bool=False, # build Kosha's call graph during sync
    rerank:bool=True, # rerank Kosha hits with flashrank
    rerank_model:NoneType=None, # flashrank model; None uses its default
    memory:NoneType=None, # memory, ask and watch backend
    apis:NoneType=None, # API specification backend
    kernel:NoneType=None, # live kernel; None runs in process
    read_outside:bool=False, # let read-only tools name any path on this machine
    deny:tuple=('*/.ssh/*', '*/.aws/*', '*/.gnupg/*', '*/.config/gcloud/*', '*/.netrc', '*/.git-credentials', '*/.codex/auth.json', '*/.claude/.credentials.json', '*/.env', '*/.env.*', '*/id_rsa*', '*/id_ed25519*', '*.pem', '*.key', '*.p12'), # what `read_outside` still refuses to open
    read_roots:tuple=(), # folders the read tools search and no write may touch
    tmux:NoneType=None, # None detects `$TMUX`; False off; True on; or this host's own pane (target or `Pane`)
    read_guard:NoneType=None, # `(ReadCheck) -> None | str | Refusal`: what the read tools may send
    model_local:NoneType=None, # `() -> (local, model)` for the turn's model
):

Reference host for local folders.

The examples below use one small temporary project. index=False keeps the folder checks independent of kosha; the search section builds an indexed host of its own.

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')
local = LocalHost([root], index=False)
sorted(local.provides), sorted(local.without)

test_eq(sorted(local.provides), ['code', 'file', 'git', 'notebook', 'session', 'shell', 'web'])
test_eq(sorted(local.without), ['api', 'ask', 'memory', 'watch'])

The folder boundary

roots lists the open folders as absolute paths, and read_roots the folders open for reading only.

---

source

LocalHost.read_roots

def read_roots():

Folders open for reading only: searched by every read tool, refused to every write.

---

source

LocalHost.roots

def roots():

A relative path is taken against the first open folder.

test_eq(local.roots, [str(root)])
test_eq(len(local.roots), 1)

---

source

LocalHost.added_roots

def added_roots():

Roots opened after construction. /resume lapses these, and says that it did.

added_roots lists folders opened after the host was built. A resumed session does not reopen them automatically.

test_eq(local.added_roots, [])

---

source

LocalHost.add_root

def add_root(
    path, readonly:bool=False, # open it for reads only: searched, never written
):

Open an existing folder for reads and writes, or for reads only. Never creates one.

add_root opens another existing folder, for reads and writes or, with readonly=True, for reads only, and starts indexing it. It never creates a folder. Open folders never overlap: a folder inside one already open is not opened again, and opening a parent closes the folders inside it.

extra = Path(tempfile.mkdtemp()).resolve()/'extra'
(extra/'sub').mkdir(parents=True)
(extra/'notes.md').write_text('read me\n')
local.add_root(str(extra)), local.added_roots

test_eq(local.added_roots, [str(extra)])
test_eq(local.roots[-1], str(extra))                     # the new root is open for reads and writes
test_eq(local.add_root(str(extra)), str(extra))          # already open: no second entry
test_eq(len(local.added_roots), 1)
test_eq(local.add_root(str(extra/'sub')), str(extra))    # inside an open root: nothing to open
test_eq(local.added_roots, [str(extra)])
test_fail(lambda: local.add_root(str(extra/'nope')), contains='no such folder')
test_fail(lambda: local.add_root(str(extra/'notes.md')), contains='not a folder')

parent = Path(tempfile.mkdtemp()).resolve()/'parent'
(parent/'child').mkdir(parents=True)
nest = LocalHost(roots=[str(parent/'child')], index=False)
test_eq(nest.add_root(str(parent)), str(parent))
test_eq(nest.roots, [str(parent)])                       # opening a parent closes the root inside it
test_eq([r for r in nest._pending if r not in nest.roots], [])   # and nothing waits on a closed one

---

source

LocalHost.check

def check(
    path, must_exist:bool=False, reading:bool=False
):

Resolve path through this host’s one sandbox. reading is the read_outside allowance.

check resolves a path and refuses it if it lands outside the open folders. The host keeps one Sandbox (from shalya.core) and asks it, so the boundary check enforces is the same one add_root widens.

test_fail(lambda: local.check('../../etc/passwd'), contains='outside the open folders')
test_fail(lambda: local.check('pkg/nope.py', must_exist=True), contains='no such file')
test_eq(local.check('pkg/sizes.py').name, 'sizes.py')
test_eq(local.check('pkg/nope.py').name, 'nope.py')      # a path to write need not exist yet
test_eq(local.roots, [str(r) for r in local.sandbox.roots])
test_eq((local.read_outside, local.deny), (False, DENY))
test_fail(lambda: LocalHost([Path(root)/'not-a-folder'], index=False), contains='no such folder')

---

source

LocalHost.roots_note

def roots_note():

A summary of root and read access, naming the folders open for reading only.

roots_note is a one-line summary of the boundary for the agent’s briefing: how many folders are open, whether reads may go outside them, and which folders are read-only.

local.roots_note, LocalHost([root], index=False, read_outside=True).roots_note

assert 'nothing outside them is readable' in local.roots_note
assert 'reads may name any path' in LocalHost([root], index=False, read_outside=True).roots_note

---

source

LocalHost.walk

def walk():

Every readable file under the open folders.

walk lists every readable file under the open folders, leaving out SKIP_DIRS, SKIP_SUFFIXES and files larger than MAX_FILE. It uses rgapi’s fd when it is installed and Python’s os.walk otherwise.

got = {p.name for p in local.walk()}
assert {'sizes.py', 'use.py'} <= got, got
(root/'__pycache__').mkdir(exist_ok=True); (root/'__pycache__'/'x.pyc').write_text('junk')
assert 'x.pyc' not in {p.name for p in local.walk()}

---

source

LocalHost.read

def read(
    path
):

One file’s text, or None when it cannot be read.

read returns a file’s text, or None when the file does not exist, lies outside the open folders or cannot be read.

assert local.read('pkg/sizes.py').startswith('def threshold')
test_eq(local.read('pkg/nope.py'), None)
test_eq(local.read('/etc/passwd'), None)                 # outside the folders, so unreadable

---

source

LocalHost.write

def write(
    path, text
):

Write text to path, through the same sandbox check enforces. Returns the path written.

write checks the path in the same way, creates missing parent folders, writes the text and returns the path written.

local.write('pkg/new.py', 'x = 1\n')

test_eq((root/'pkg'/'new.py').read_text(), 'x = 1\n')
test_eq(local.write('deep/er/f.py', 'y = 2\n'), str(root/'deep'/'er'/'f.py'))
test_fail(lambda: local.write('/etc/passwd', 'no'), contains='outside the open folders')

---

source

LocalHost.text_at

def text_at(
    path
):

One file as a diffable document: a notebook as its cell sources, anything else as text.

text_at returns a file as one document that can be compared before and after an edit. A notebook becomes its cell sources, and a file that does not exist yet becomes '', so creating a file still produces a diff. It returns None on error.

test_eq(local.text_at('pkg/nope.py'), '')
test_eq(local.text_at('pkg/sizes.py'), local.read('pkg/sizes.py'))

---

source

LocalHost.approvals

def approvals():

approvals is the object that asks a person to approve a write, if the application supplied one.

local.approvals, LocalHost([root], index=False, approvals='an Approvals').approvals

test_eq(local.approvals, None)
test_eq(LocalHost([root], index=False, approvals='an Approvals').approvals, 'an Approvals')

---

source

LocalHost.note

def note(
    text
):

Tell the user something out of band. Never blocks. A host may drop it.

note tells the user something without waiting for an answer, such as “starting a kernel”. It adds the message to transcript, then calls the status callback given to the constructor, if any. A callback that raises is ignored, so a broken display cannot end an agent’s turn.

said = []
noisy = LocalHost([root], index=False, note=said.append)
noisy.note('starting a kernel')
said, noisy.transcript

test_eq(said, ['starting a kernel'])
test_eq(noisy.transcript, ['starting a kernel'])         # kept, and `terminal_text` reads it back
boom = LocalHost([root], index=False, note=lambda t: 1/0)
boom.note('swallowed')                                   # a broken note must not end a turn
test_eq(boom.transcript, ['swallowed'])

Searching the code

Three engines can answer a search. Kosha’s index combines keyword and vector search once it has synced. Ripgrep finds literal matches while the index is still building. If neither is installed, the host reads the files itself. Which engine answered is reported to the model with the results.

---

source

LocalHost.sync_index

def sync_index(
    wait:bool=False, force:bool=False
):

Run Kosha.sync for every open root, once, in a daemon thread. Each root publishes as it returns.

sync_index starts one background thread that runs Kosha.sync for each open folder in turn. Each folder’s index becomes searchable as soon as its own sync finishes. Calling it again while the thread is running does nothing, and wait=True blocks until it is done.

quiet = LocalHost([root], index=False)
quiet.sync_index() is quiet, quiet._index_thread.name

test_eq(quiet.sync_index(), quiet)                        # chainable, and starts nothing twice
test_eq(quiet._index_thread.name, 'shalya-kosha-sync')

---

source

LocalHost.index_ready

def index_ready():

Whether all open roots are indexed.

index_ready is true once every open folder has an index, and indexed lists the folders that have one.

local.index_ready, local.indexed

test_eq(local.index_ready, False)                         # this host was built with index=False
test_eq(local.indexed, [])

---

source

LocalHost.indexed

def indexed():

Indexed root paths.

---

source

LocalHost.wait_index

def wait_index(
    timeout:NoneType=None
):

Wait for the automatic Kosha sync. Returns whether semantic search is ready.

wait_index waits up to timeout seconds for the sync thread and returns whether every folder is indexed.

local.wait_index(1), local.index_ready

test_eq(local.wait_index(1), False)

---

source

path_scope

def path_scope(
    pf, root
):

path_filter as a regex over paths relative to root: '' filters nothing, None when it names a place outside root.

path_scope reads a path_filter as a model writes one. A folder or file under the root, named relative to it, absolute, or starting with the root’s own name, matches that place and what is below it. Anything else is a substring of the relative path, in any case, and * and ? work as in a glob. An absolute path outside the root returns None, so grep skips that root.

tree = Path(tempfile.mkdtemp()).resolve()
(tree/'pkg'/'sub').mkdir(parents=True); (tree/'pkg'/'sub'/'mod.py').write_text('x')
test_eq(path_scope('', tree), '')
test_eq(path_scope(str(tree), tree), '')
assert path_scope('/elsewhere/pkg', tree) is None
assert re.search(path_scope('pkg/sub/', tree), 'pkg/sub/mod.py') and not re.search(path_scope('pkg/sub', tree), 'xpkg/sub2/a.py')
assert re.search(path_scope(f'{tree.name}/pkg', tree), 'pkg/sub/mod.py')
assert re.search(path_scope('MOD', tree), 'pkg/sub/mod.py') and re.search(path_scope('*.py', tree), 'pkg/sub/mod.py')

---

source

LocalHost.grep

def grep(
    pattern, path_filter:str='', regex:bool=True, ignore_case:bool=False, limit:int=60
):

Exact matching through ripgrep. None when rgapi is unavailable. The tool then reads files itself.

grep finds every line matching a pattern, as a regular expression by default, through ripgrep. It returns None when rgapi is not installed, and the grep tool then reads the files itself instead of reporting that nothing matched.

hits = local.grep('threshold')
assert hits and all(h.line >= 1 for h in hits), hits
assert any('sizes.py' in str(h.path) for h in hits)
test_eq(local.grep('no_such_symbol_anywhere'), [])
assert local.grep('def .*old', regex=True), 'a regex must reach ripgrep as a regex'

---

source

LocalHost.search

def search(
    query, limit:int=20
):

The code index and the literal scan, fused by rank rather than tried in order.

search runs kosha’s index and ripgrep together and merges their hits by rank with reciprocal rank fusion, so a hit both engines find rises to the top. If neither returns anything and ripgrep is not installed, it falls back to reading the files. An empty query returns [], not everything.

search_note says which engine answered, and is shown to the model when a search finds nothing.

assert local.search('threshold'), 'ripgrep found nothing'
test_eq(local.search(''), [])
test_eq(local.search('   '), [])

---

source

LocalHost.search_note

def search_note():

This cell builds a host with a real index, to check wait_index, indexed, search_note and public_api.

indexing = LocalHost([root])
indexing.wait_index(180)
indexing.search_note

assert 'fallback' in local.search_note, local.search_note
# the sync runs concurrently; read the note once
note = indexing.search_note
assert (note.startswith('Kosha semantic + keyword index over')
        or note.startswith('Kosha unavailable')
        or note == 'Kosha sync in progress; literal fallback via ripgrep'), note
if indexing.index_ready: test_eq(indexing.indexed, [str(root)])
assert indexing.search('threshold'), 'ripgrep answers whether or not the index has landed'

local.search_note

---

source

LocalHost.public_api

def public_api(
    package, limit:int=200
):

Kosha’s public surface for package, @patch-added methods included.

public_api lists a package’s public names through kosha, including functions that fastcore’s @patch adds to its classes. Without an index it raises an error, so the caller can tell “no index yet” from “this package exports nothing”.

test_fail(lambda: local.public_api('fastcore'), contains='no code index')

test_eq(local.public_api(''), [])
test_eq(local.indexed, [])
test_fail(lambda: local.public_api('fastcore'), contains='no code index')
if indexing.index_ready: assert isinstance(indexing.public_api('shalya'), list)

---

source

LocalHost.symbols

def symbols(
    path
):

The defs and classes in one file, as Hits whose score is the indent depth.

symbols lists the functions and classes in one file by parsing it, so a method is reported under its qualified name, such as A.m, with its nesting depth in score.

test_eq([h.symbol for h in local.symbols('pkg/sizes.py')], ['threshold'])
test_eq(local.symbols('pkg/sizes.py')[0].score, 0)
(root/'pkg'/'cls.py').write_text('class A:\n    def m(self): pass\n')
test_eq([(h.symbol, h.score) for h in local.symbols('pkg/cls.py')], [('A', 0), ('A.m', 1)])
test_eq(local.symbols('pkg/nope.py'), [])

---

source

LocalHost.peers

def peers(
    path, line, limit:int=20
):

Find references to the symbol defined at path:line.

peers takes a file and a line, finds the definition at or above that line, and searches for other places that use its name, leaving out the definition itself.

peers = local.peers('pkg/sizes.py', 1)
assert any('use.py' in str(h.path) for h in peers), peers
assert not any(str(h.path) == str(root/'pkg'/'sizes.py') and h.line == 1 for h in peers)
assert local.peers('pkg/sizes.py', 999), 'the last def at or above the line is still a def'
test_eq(local.peers('pkg/sizes.py', 0), [])
test_eq(local.peers('pkg/nope.py', 1), [])

Notebooks, the session and the shell

nb_cells lists a notebook’s cells as (id, cell_type, source) tuples, which is all shalya needs to know about notebooks. nb_cell and nb_set_cell are written once on NotebookHost in terms of nb_cells and check, so any host that writes nb_cells also gets one-cell reads and writes.

---

source

LocalHost.nb_cells

def nb_cells(
    path
):

[(id, cell_type, source)] for one notebook.

nb_cell finds a cell by its exact id or a unique prefix of it, and an id that matches nothing, or more than one cell, raises an error listing the ids. nb_set_cell replaces one cell’s source and writes the notebook back.

local.write('nb/demo.ipynb', json.dumps(
    {'cells': [{'cell_type': 'code', 'id': 'c0', 'source': ['x = threshold(4096)'],
                'metadata': {}, 'outputs': [], 'execution_count': None}],
     'metadata': {}, 'nbformat': 4, 'nbformat_minor': 5}))
local.nb_cells('nb/demo.ipynb')

test_eq(len(local.nb_cells('nb/demo.ipynb')), 1)
test_eq(local.nb_cells('nb/demo.ipynb')[0][:2], ('c0', 'code'))
assert 'threshold(4096)' in local.nb_cells('nb/demo.ipynb')[0][2]
test_fail(lambda: local.nb_cells('nb/nope.ipynb'), contains='no such file')
# one cell by exact id or unique prefix; a miss names the ids; a set writes the notebook back
test_eq(local.nb_cell('nb/demo.ipynb', 'c0')[1], 'code')
test_eq(local.nb_cell('nb/demo.ipynb', 'c')[0], 'c0')
test_fail(lambda: local.nb_cell('nb/demo.ipynb', 'zz'), contains='c0')
test_fail(lambda: local.nb_cell('nb/demo.ipynb', ''), contains='cell_id')     # '' is not a prefix of every cell
class Ghost(LocalHost):
    "Reports a cell the file does not hold."
    def nb_cells(self, path): return super().nb_cells(path) + [('ghost', 'code', '')]
test_fail(lambda: Ghost([root], index=False).nb_set_cell('nb/demo.ipynb', 'ghost', 'x'), contains="no cell 'ghost'")
test_eq(local.nb_set_cell('nb/demo.ipynb', 'c0', 'x = threshold(8192)'), 'c0')
test_eq(local.nb_cells('nb/demo.ipynb')[0][2], 'x = threshold(8192)')

A notebook saved without cell ids (format 4.4) gets ids from position, c0, c1 and so on, so the id the notebook_cells tool lists is the id edit_cell can send back. The first write saves those ids into the file.

local.write('nb/old.ipynb', json.dumps(
    {'cells': [{'cell_type': 'markdown', 'source': ['# old'], 'metadata': {}},
               {'cell_type': 'code', 'source': ['a = 1'], 'metadata': {}, 'outputs': [], 'execution_count': None}],
     'metadata': {}, 'nbformat': 4, 'nbformat_minor': 4}))
ids = [c[0] for c in local.nb_cells('nb/old.ipynb')]
test_eq(ids, ['c0', 'c1']); test_eq(ids, [c[0] for c in local.nb_cells('nb/old.ipynb')])          # the same on every read
test_eq(local.nb_set_cell('nb/old.ipynb', 'c1', 'a = 2'), 'c1')
test_eq(local.nb_cells('nb/old.ipynb')[1][2], 'a = 2')
saved = json.loads(local.read('nb/old.ipynb'))
test_eq([c['id'] for c in saved['cells']], ['c0', 'c1']); assert saved['nbformat_minor'] >= 5       # ids persisted, format says so

---

source

LocalHost.nb_add_cell

def nb_add_cell(
    path, source, index:int=-1, cell_type:str='code'
):

Insert a cell (-1 appends), creating the notebook if needed. Returns the new cell’s id.

nb_add_cell inserts a cell at index, or appends it when index is -1, creating the notebook if it does not exist, and returns the new cell’s id. Existing cells keep their ids.

cid = local.nb_add_cell('nb/new.ipynb', 'y = 1')          # writes the notebook
second = local.nb_add_cell('nb/new.ipynb', '# a heading', cell_type='markdown')
first = local.nb_add_cell('nb/new.ipynb', 'z = 0', index=0)
[c[:2] for c in local.nb_cells('nb/new.ipynb')]

ids = [c[0] for c in local.nb_cells('nb/new.ipynb')]
test_eq(ids, [first, cid, second])                       # index=0 went first, -1 appended
test_eq([c[1] for c in local.nb_cells('nb/new.ipynb')], ['code', 'code', 'markdown'])
assert 'a heading' in local.text_at('nb/new.ipynb')
assert local.nb_add_cell('nb/old.ipynb', 'b = 3', index=0)
test_eq([c[0] for c in local.nb_cells('nb/old.ipynb')][1:], ['c0', 'c1'])   # a new cell never renames the others

---

source

LocalHost.run_python

def run_python(
    code
):

Run code in the live namespace. Failures come back as text: a tool cannot usefully raise.

run_python runs code in the live namespace, which persists between calls, and returns what it printed followed by the value of the last expression. That persistence is why agents are told to bind results to new names. An error comes back as text, because a tool that raises ends the agent’s turn. With a kernel attached, the kernel runs the code instead.

local.run_python('import math\nradii = [1, 2, 3]'), local.run_python('areas = [math.pi*r*r for r in radii]\nlen(areas)')

test_eq(local.run_python('len(areas)'), '3')
test_eq(local.run_python('print("a side effect")'), 'a side effect')
assert 'NameError' in local.run_python('no_such_name + 1')
test_eq(local.run_python(''), '(no output)')

---

source

LocalHost.inspect_python

def inspect_python(
    code, scope:str='isolated'
):

Run code without changing anything the user made. A kernel enforces this; in process it runs on copies.

Without a kernel, inspect_python runs code on a copy of the namespace. New names disappear after the call, and each name the code reads is a deep copy, so radii.append(99) leaves the real radii alone. Two things stay shared: objects deepcopy refuses, such as modules and open files, and the real namespace a user’s function reads through its globals. A kernel protects the namespace fully; see the end of this notebook.

local.inspect_python('doubled = [r*2 for r in radii]\nlen(doubled)'), local.run_python('"doubled" in globals()')

test_eq(local.inspect_python('doubled = [r*2 for r in radii]\nlen(doubled)'), '3')
test_eq(local.run_python('"doubled" in globals()'), 'False')
test_eq(local.inspect_python('radii.append(99)\nlen(radii)'), '4')
test_eq(local.run_python('len(radii)'), '3')
assert 'only honours' in local.inspect_python('1+1', scope='overlay')

---

source

LocalHost.scopes

def scopes():

What inspect_python honours. A kernel says for itself; in process it is the copy alone.

scopes lists the scopes inspect_python accepts. An attached kernel reports its own; without one there is only isolated.

local.scopes

test_eq(local.scopes, ('isolated',))       # no kernel, so the copy alone

---

source

LocalHost.kernel_kind

def kernel_kind():

kernel_kind names what runs the live namespace: 'inprocess' without a kernel, or the kernel’s own kind. concurrent is true only for 'ipymini', a kernel that can be inspected while a cell is running; the agent uses it to decide whether to inspect now or wait.

local.kernel_kind, local.concurrent

test_eq(local.kernel_kind, 'inprocess')
test_eq(local.concurrent, False)

---

source

LocalHost.list_vars

def list_vars():

What is in the live namespace: name, type, and a short value, one per line.

list_vars shows what is in the live namespace: name, type and a short value, one line each. Functions, classes, modules and names starting with _ are left out, because an agent asking what is there wants the data, not the imports.

rows = dict(l.split(None, 1) for l in local.list_vars().splitlines())
assert 'radii' in rows and 'list' in rows['radii'], rows
assert 'math' not in rows and not any(k.startswith('_') for k in rows)

---

source

LocalHost.restart_kernel

def restart_kernel():

Restart the kernel, or empty the in-process namespace in place.

restart_kernel empties the in-process namespace in place, so anything else holding it sees it empty too. An attached kernel restarts through its own restart method, and one without that method refuses.

held = local.ns
test_eq(local.restart_kernel(), 'the namespace is empty')
assert held is local.ns and 'radii' not in local.ns and local.list_vars() == ''
test_eq(local.run_python('__name__'), "'__main__'")

When the host runs inside tmux, it can show things to the user and read what the user ran. terminal_text reads the panes beside the host’s own, so the model can see a command you just ran. run_cmd_bg starts a background command in a new pane below, so you can watch it. open_pane shows you something in a new pane, such as a sub-agent’s transcript, and close_pane closes it. These need tmux and the fastmux package. Without them, background commands write to a log file, and only the view is lost.

---

source

LocalHost.close_pane

def close_pane(
    pane_id
):

Close a pane open_pane made.

---

source

LocalHost.open_pane

def open_pane(
    command, title:str='', where:str='right'
):

Show command to the user in a new pane beside this one and return its id.

---

source

LocalHost.tmux_pane

def tmux_pane():

The pane this host runs in, or None when tmux is off.

---

source

LocalHost.terminal_text

def terminal_text(
    lines:int=200
):

The neighbouring tmux panes, else what this process recorded in transcript.

terminal_text returns up to lines lines from the neighbouring tmux panes. Outside tmux it returns the last lines messages from transcript. It is read-only: it shows what was run and cannot run anything.

chatty = LocalHost([root], index=False)
for i in range(3): chatty.note(f'line {i}')
chatty.terminal_text(2)

test_eq(chatty.terminal_text(2), 'line 1\nline 2')
test_eq(chatty.terminal_text(), 'line 0\nline 1\nline 2')
test_eq(LocalHost([root], index=False).terminal_text(), '')
test_eq(chatty.terminal_text(0), '')       # asking for none is not asking for all
test_eq(chatty.terminal_text(99), 'line 0\nline 1\nline 2')

---

source

LocalHost.run_cmd

def run_cmd(
    command, cwd:NoneType=None, timeout:int=120
):

Run a shell command in a new process group. Interleave stdout and stderr. Kill the process group on timeout.

run_cmd runs a shell command in its own process group, in the first open folder unless cwd names another, and returns (exit_code, output). Standard output and standard error are interleaved in order. A failure is an exit code, not an exception. When timeout seconds pass, the whole process group is killed and the exit code is 124.

test_eq(local.run_cmd('echo hi'), (0, 'hi\n'))
test_eq(local.run_cmd('exit 3')[0], 3)
test_eq(local.run_cmd('pwd')[1].strip(), str(root))
test_eq(local.run_cmd('echo out; echo err 1>&2')[1], 'out\nerr\n')   # interleaved, in order
test_eq(local.run_cmd(''), (0, ''))
code, out = local.run_cmd('sleep 5', timeout=1)
test_eq(code, 124); assert 'killed after 1s' in out

---

source

LocalHost.close

def close():

Stop every background command and pane this host opened, and drop the logs.

---

source

LocalHost.cmd_stop

def cmd_stop(
    run_id
):

Kill a background command, or clear its finished pane; returns its state afterwards.

---

source

LocalHost.cmd_output

def cmd_output(
    run_id, tail:int=200
):

(state, text): running, exit N or stopped, and the last tail lines so far.

---

source

LocalHost.run_cmd_bg

def run_cmd_bg(
    command, cwd:NoneType=None
):

Start command in a tmux pane below this one, else in its own process group with a log file; returns a run id.

A background command outlives the call that started it: a server, a file watcher, a slow test suite. run_cmd_bg starts it and returns a run id. cmd_output returns its state (running, exit N or stopped) and the last lines of its output. cmd_stop kills its whole process group, and close stops every background command and pane the host opened.

import time

def wait_for(f, secs=10):
    "Poll `f` until it is truthy or `secs` pass."
    end = time.monotonic()+secs
    while not (r := f()) and time.monotonic() < end: time.sleep(0.05)
    return r
rid = local.run_cmd_bg('for i in 1 2 3; do sleep 0.2; echo tick $i; done; exec sleep 30')
wait_for(lambda: 'tick 3' in local.cmd_output(rid)[1])
state, text = local.cmd_output(rid)
test_eq(state, 'running'); assert 'tick 3' in text, text
test_eq(local.cmd_stop(rid), 'stopped')
done = local.run_cmd_bg('exit 3')
test_eq(wait_for(lambda: local.cmd_output(done)[0] != 'running' and local.cmd_output(done)[0]), 'exit 3')
test_eq(local.cmd_output(rid, 1)[1], 'tick 3')

A killed shell can exit with 137 before its own SIGKILL lands, if the kill reaches its child first; the test stands in a kill that reaches only the child. cmd_stop reports stopped for anything it killed, whatever exit code the shell left:

from unittest.mock import patch as mpatch
import subprocess

rid = local.run_cmd_bg('echo up; sleep 30; exit $?')
wait_for(lambda: 'up' in local.cmd_output(rid)[1])
with mpatch.object(os, 'killpg', lambda pid, sig: subprocess.run(['pkill', '-KILL', '-P', str(pid)])): test_eq(local.cmd_stop(rid), 'stopped')
local.close()

---

source

LocalHost.shell_note

def shell_note():

---

source

LocalHost.environment

def environment():

This process, the venvs under the open folders, and the commands on PATH; computed once per set of roots.

environment answers the question most commands depend on: which Python. The host runs on sys.executable, a project usually has its own .venv, and a task may bring a third. Listing the interpreters, the virtual environments under the open folders, a few common commands and the tmux pane once saves the agent a round of failed imports. The answer is computed once per set of open folders.

import tempfile

with tempfile.TemporaryDirectory() as d:
    for name, cfg in (('.venv', 'version = 3.13.1\n'), ('.aeneas-venv', None)):
        (Path(d)/name/'bin').mkdir(parents=True); (Path(d)/name/'bin'/'python').touch()
        if cfg: (Path(d)/name/'pyvenv.cfg').write_text(cfg)
    (Path(d)/'.notavenv').mkdir()
    e = LocalHost([d], index=False).environment()
print(e)
assert f'{d}/.venv/bin/python (3.13.1)' in e and f'{d}/.aeneas-venv/bin/python' in e and '.notavenv' not in e
assert e.startswith(f'this process: {sys.executable}') and 'honours isolated' in e and 'python:' in e

shell_note says where commands run, or why they cannot.

local.shell_note

assert str(root) in local.shell_note

The web

web_search searches the web through fossick and returns objects with a title and a url. An empty query returns [] without touching the network.

---

source

LocalHost.web_search

def web_search(
    query, n:int=20
):

Search the web through fossick. An empty query answers [] without reaching the network.

LocalHost raises NotImplementedError from every web method when web access is off or fossick is missing. attempt in shalya.core passes that exception on instead of turning it into an ERROR: result, so “this host cannot” stays distinct from “this call failed”.

local.can('web'), local.web_search('')

test_eq(local.web_search(''), [])                        # the probe must not hit the network
offline = LocalHost([root], index=False, web=False)
test_eq(offline.can('web'), False)
test_fail(lambda: offline.web_search('anything'), NotImplementedError)

Reading one page

read_page(fossick, url) turns one URL into markdown, and LocalHost.read_url calls it. It takes the fossick module as an argument, so a host with web access switched off never imports it, and code outside a host can use it too. It returns None, or a record with text, url, title, kind (page, paper or repo), sections and the strategy that produced the text.

It works in stages. A URL that has a dedicated reader in fossick, such as an arXiv paper, a GitHub file, a YouTube video or a PDF, is read with it. Anything else is fetched. Several extraction methods then compete to find the article in the fetched page: readability, each of the wrappers in CONTENT_SEL, and a selector for headings, paragraphs and lists inside the main area. The longest result wins, reduced by its share of bullet lines, since a page that is mostly bullets is a navigation column. A result shorter than THIN_PAGE characters probably means the site answered with an empty shell that builds the page with JavaScript, so the page is fetched again with a real browser and then in stealth mode. Repeated <article> elements, as on a feed or digest page, are also returned separately in sections, and the page’s JSON-LD goes on top. sel names a CSS selector to try before any guess.

---

source

read_page

def read_page(
    fossick, url, sel:NoneType=None, note:NoneType=None,
    readers:tuple=((re.compile('https?://(www\\.)?github\\.com/[^/]+/[^/]+/(blob|raw)/', re.IGNORECASE), 'read_gh_file', {}, 'repo'), (re.compile('https?://(www\\.)?arxiv\\.org/(abs|pdf)/', re.IGNORECASE), 'read_arxiv', {'save_pdf': False, 'source': True}, 'paper'), (re.compile('https?://(www\\.)?(youtube\\.com/watch|youtu\\.be/)', re.IGNORECASE), 'read_yt', {}, 'page')),
    thin:int=800, mx:int=200000
):

One URL as markdown, with its title, kind, sections and extraction strategy, or None; sel is a CSS selector to try first.

---

source

md_title

def md_title(
    md, url:str=''
):

The first markdown heading, else the url’s last path segment.

---

source

LocalHost.read_url

def read_url(
    url, remember:bool=True, sel:NoneType=None
):

One page as markdown. remember=False keeps sensitive or low-quality results ephemeral.

---

source

LocalHost.research

def research(
    query
):

A cited digest of the top web results for query, written by fossick.

---

source

LocalHost.research_note

def research_note():

read_url returns one page through read_page, and research returns fossick’s cited digest of the top web results for a query, which is slower than a search. research_note says which backend serves research, or that web access is off.

local.research_note, LocalHost([root], index=False, web=False).research_note

test_eq(local.research_note, 'fossick')
test_eq(LocalHost([root], index=False, web=False).research_note, 'web access is switched off')

The live check below runs LocalHost against the real web through fossick. It is marked eval: false and left out of the test run, because it needs the network.

live_host = LocalHost([Path(tempfile.mkdtemp())], index=False)
live_hits = live_host.web_search('Python programming language official site', n=5)
live_doc = live_host.read_url('https://www.python.org/', remember=False)
live_digest = live_host.research('What is the Python programming language?')
assert live_hits and all(h.title and h.url.startswith('http') for h in live_hits)
assert live_doc.url == 'https://www.python.org/' and len(live_doc.text) > 400
assert len(live_digest) > 200 and 'python' in live_digest.lower()

---

source

LocalHost.memory_search

def memory_search(
    query, limit:int=8
):

Search remembered pages as whole tree sections, returning structured rows.

---

source

LocalHost.memory_tree

def memory_tree(
    document:str=''
):

The heading tree for remembered documents. An empty document lists every root.

---

source

LocalHost.memory_read

def memory_read(
    node_id
):

Read one remembered section and its children by stable node id.

---

source

LocalHost.memory_topics

def memory_topics(
    limit:int=12
):

Labelled semantic clusters across remembered research; a host without a topic map says so.

---

source

LocalHost.memory_forget

def memory_forget(
    doc_id
):

Purge one remembered document and all derived tree, chunk and vector data.

---

source

LocalHost.remember

def remember(
    text, title:NoneType=None, tags:tuple=(), key:str=''
):

File text as a note and return the document record; key upserts the same note later.

---

source

LocalHost.ask

def ask(
    question, ref:NoneType=None, instruction:str='', **kw
):

Answer question out of remembered research, with citations, as a dict.

---

source

LocalHost.watch

def watch(
    target, action:str='remind', every:str='1d', note:NoneType=None, **params
):

Register a recurring job. target is a URL, a query, or the text of a reminder.

---

source

LocalHost.watch_actions

def watch_actions():

What watch accepts here: every vault action when a vault is attached, reminders alone otherwise.

---

source

LocalHost.watches

def watches(
    due_only:bool=False
):

Every registered watch, soonest first. due_only keeps the ones that have come due.

---

source

LocalHost.unwatch

def unwatch(
    watch_id
):

Delete one watch. Whatever it already filed stays in memory.

---

source

LocalHost.poll

def poll():

Run every watch that is due and report what fired. One failing watch must not stop the rest.

Memory, watches and API specifications

The memory, ask and watch groups are served by a vishalakshi Vault, opened by whatever built the host and passed in as memory. A vault stores pages and notes as documents divided into sections. Each host method forwards to one vault method: memory_search to search, memory_tree to toc, memory_read to read, memory_topics to topic_tree, memory_forget to forget and remember to note, while ask, watch, watches, unwatch and poll keep their names. Without a vault each of them raises an error saying there is no vault.

The first example is marked eval: false because it needs a vault on disk.

from vishalakshi import Vault

vault = Vault(Path(tempfile.mkdtemp())/'vault.db', offline=True)   # hashed embeddings, so no model to fetch
remembering = LocalHost([root], index=False, memory=vault)
test_eq(sorted(remembering.without), ['api'])              # one vault answers memory, ask and watch
assert remembering.can('memory') and remembering.can('ask') and remembering.can('watch')
sorted(remembering.provides), sorted(remembering.without)

# a host with no vault refuses by naming the missing backend, rather than failing inside a forward
test_eq(local.can('memory'), False)
for nm in ('memory_tree', 'memory_search', 'ask', 'remember', 'memory_topics', 'memory_read', 'memory_forget'):
    try: getattr(local, nm)('x')
    except HostError as e: assert 'no vault' in str(e), (nm, e)
    else: assert False, f'{nm} did not refuse'
for call in (lambda: local.watch('x'), lambda: local.watches(), lambda: local.unwatch('w1'), lambda: local.poll()):
    test_fail(call, contains='no vault')
test_eq(local.watch_actions, ('remind',))                    # no vault: reminders alone
# `memory_topics` is no longer abstract: a memory host without a topic map inherits a refusal that says so
class Topicless(Host, MemoryHost):
    roots = ['/proj']
    def check(self, path, must_exist=False, reading=False): return Path('/proj')/path
    def walk(self): return []
    def read(self, path): return None
    def write(self, path, text): return str(path)
    def text_at(self, path): return ''
    def memory_search(self, query, limit=8): return []
    def memory_tree(self, document=''): return []
    def memory_read(self, node_id): return {}
    def memory_forget(self, doc_id): return False
    def remember(self, text, title=None, tags=(), key=''): return {}
test_fail(lambda: Topicless().memory_topics(), contains='no topic map')
# a vault gets the key forwarded only when one is given, so a vault without the parameter still files keyless notes
class FakeVault:
    def note(self, text, title=None, tags=None, key='', meta=None):
        self.key = key
        return {'doc_id': 'n1', 'title': title or text, 'replaced': bool(key)}
fake = FakeVault(); faked = LocalHost([root], index=False, memory=fake)
test_eq(faked.remember('fact', key='k')['replaced'], True); test_eq(fake.key, 'k')
faked.remember('fact'); test_eq(fake.key, '')
assert {'url', 'remind'} <= set(faked.watch_actions)         # a vault: what vishalakshi acquires

The round trip below files a note in the vault, finds it with a search, and reads it back by the node_id the search returned.

doc = remembering.remember('the kettle boils at 100C', title='kettles', tags=['home'])
remembering.memory_search('kettle'), [t['title'] for t in remembering.memory_tree()], doc

from inspect import signature

hits = remembering.memory_search('kettle')
test_eq(hits[0]['doc_id'], doc['doc_id'])
assert 'kettle' in hits[0]['snippet']
test_eq([t['doc_id'] for t in remembering.memory_tree(doc['doc_id'])], [doc['doc_id']])
assert 'the kettle boils at 100C' in remembering.memory_read(hits[0]['node_id'])['text']
test_eq(len(remembering.memory_topics()), 0)     # topics come from the graph, and connect() builds it
assert {'ref', 'instruction'} <= set(signature(vault.ask).parameters)   # what `ask` forwards, verbatim
remembering.memory_forget(doc['doc_id'])
assert doc['doc_id'] not in [t['doc_id'] for t in remembering.memory_tree()]

A watch is a job the host runs again at an interval, such as a reminder, a page to check or a search to repeat, and poll runs every watch that is due. watch_actions lists the kinds of watch the vault accepts.

w = remembering.watch('check the kettle', action='remind', every='1d')
due = remembering.watches(due_only=True)     # an interval watch is due the moment it is registered
fired = remembering.poll()
[x['id'] for x in due], fired

test_eq([x['id'] for x in remembering.watches()], [w['id']])
test_eq(w['every'], 86400)                            # '1d' reaches the vault as seconds
test_eq([x['id'] for x in due], [w['id']])
test_eq(fired['ran'], 1)
test_eq([r['status'] for r in fired['results']], ['ok'])
assert {'url', 'remind', 'folder'} <= set(remembering.watch_actions)   # what this host declares it will accept
remembering.unwatch(w['id'])
test_eq(len(remembering.watches()), 0)

---

source

LocalHost.api_load

def api_load(
    src, name:str=''
):

Load an OpenAPI specification from a path or a URL. Returns what it is now called.

---

source

LocalHost.api_ops

def api_ops(
    group:str='', name:str='', match:str='', limit:NoneType=None, offset:int=0
):

The operations a loaded specification describes, filtered and paged.

---

source

LocalHost.api_count

def api_count(
    group:str='', name:str='', match:str=''
):

How many operations that filter matches, without listing them.

---

source

LocalHost.api_call

def api_call(
    operation, name:str='', **params
):

Call one operation from a loaded specification.

The api group forwards in the same way to any backend with the four methods api_load, api_ops, api_count and api_call. ramabana.spec.SpecHost is one such backend. The example builds a small one over fastspec and httpx.

import httpx
from fastspec.spec import SpecParser

class Specs:
    "An api backend over fastspec: it reads the spec, and httpx makes the call."
    def __init__(self): self.parsed, self.base = {}, {}

    def api_load(self, src, name=''):
        spec = src if isinstance(src, dict) else json.loads(Path(src).read_text())
        p = SpecParser.from_openapi(AttrDict(spec))
        key = name or (spec.get('info') or {}).get('title') or 'api'
        self.parsed[key] = p
        self.base[key] = ((spec.get('servers') or [{}])[0].get('url') or '').rstrip('/')
        return {'name': key, 'operations': len(p.ops), 'groups': sorted({o.group or '' for o in p.ops})}

    def _rows(self, group='', name='', match=''):
        ops = self.parsed[name].ops if name else next(iter(self.parsed.values())).ops
        return [{'name': o.name, 'group': o.group or '', 'verb': o.verb, 'path': o.path, 'summary': o.summary or ''}
                for o in ops if (not group or o.group == group) and (not match or match in o.name)]

    def api_ops(self, group='', name='', match='', limit=None, offset=0):
        rows = self._rows(group, name, match)
        return rows[offset:] if not limit else rows[offset:offset + int(limit)]

    def api_count(self, group='', name='', match=''): return len(self._rows(group, name, match))

    def api_call(self, operation, name='', **params):
        key = name or next(iter(self.parsed))
        op = next(o for o in self.parsed[key].ops if o.name == operation)
        path = op.path.format(**{k: params.pop(k) for k in list(params) if f'{{{k}}}' in op.path})
        r = httpx.request(op.verb, self.base[key] + path, params=params, timeout=30)
        return {'status': r.status_code, 'body': r.text[:200]}

GITHUB = {'openapi': '3.0.0', 'info': {'title': 'github', 'version': '1.0'},
          'servers': [{'url': 'https://api.github.com'}],
          'paths': {'/zen': {'get': {'operationId': 'getZen', 'summary': 'A design maxim'}},
                    '/repos/{owner}/{repo}': {'get': {'operationId': 'getRepo', 'summary': 'One repository',
                        'parameters': [{'name': 'owner', 'in': 'path', 'required': True, 'schema': {'type': 'string'}},
                                       {'name': 'repo', 'in': 'path', 'required': True, 'schema': {'type': 'string'}}]}}}}

speccy = LocalHost([root], index=False, apis=Specs())
speccy.api_load(GITHUB), speccy.api_ops(match='zen')

test_eq(speccy.can('api'), True)
test_eq(speccy.api_load(GITHUB)['operations'], 2)
test_eq(speccy.api_count(), 2)
test_eq([o['name'] for o in speccy.api_ops(match='zen')], ['get_zen'])   # fastspec snake_cases operationId
test_eq([o['name'] for o in speccy.api_ops(group='zen')], ['get_zen'])  # and takes the group from the path
test_eq(speccy.api_ops(group='nothing'), [])
test_eq(len(speccy.api_ops(limit=1)), 1)
test_eq(speccy.api_ops(limit=1, offset=1)[0]['name'], 'get_repo')
test_eq(speccy.api_call('get_zen')['status'], 200)

# and a host with no api backend refuses by naming what is missing
for call in (lambda: local.api_load('x'), lambda: local.api_ops(), lambda: local.api_count(),
             lambda: local.api_call('x')):
    test_fail(call, contains='no API specifications')

LocalHost’s methods are added with @patch after the class is created, but Python’s ABCMeta records the missing abstract methods when the class is created. implemented recomputes that record once the methods are in place, so the host can be instantiated, and a host that is still missing a method still cannot.

---

source

implemented

def implemented(
    cls
):

Recompute what cls is still missing, after @patch filled some of it in.

class Hollow(Host, ApiHost):
    "Declares the api group and writes none of it."
    @property
    def roots(self): return []
    def check(self, path, must_exist=False, reading=False): return Path(path)
    def walk(self): return []
    def read(self, path): return None
    def write(self, path, text): return str(path)
    def text_at(self, path): return ''

Hollow.__abstractmethods__ = frozenset()      # as the LocalHost cell does, to allow the patches
test_fail(implemented(Hollow), contains='api_load')

@patch
def api_load(self:Hollow, src, name=''): return {}
@patch
def api_ops(self:Hollow, group='', name='', match='', limit=None, offset=0): return []
@patch
def api_count(self:Hollow, group='', name='', match=''): return 0
@patch
def api_call(self:Hollow, operation, name='', **params): return {}

implemented(Hollow)
test_eq(Hollow().provides, {'api', 'file'})   # written, so it builds, and it declares what it wrote
test_eq(Hollow.__abstractmethods__, frozenset())

Attaching a kernel to an existing host

When kernel is set, run_python, inspect_python, list_vars, restart_kernel, scopes and kernel_kind go to the kernel instead of the in-process namespace. The example shows the shape a kernel must have. Assigning one to an existing host keeps its groups and backends.

class Kernel:
    "The shape `LocalHost` expects of a kernel: four calls and two facts, and a `restart` if it can."
    scopes, kind = ('isolated', 'overlay'), 'ipykernel'
    def __init__(self): self.ns = {}
    def run(self, code): exec(code, self.ns); return 'ran on the kernel'
    def inspect(self, code, scope='isolated'): return f'inspected under {scope}'
    def list_vars(self): return ', '.join(k for k in self.ns if not k.startswith('__'))
    def restart(self): self.ns.clear(); return 'restarted on the kernel'

k = LocalHost([root], index=False)
before = sorted(k.provides)
k.kernel = Kernel()
before, sorted(k.provides), k.scopes, k.kernel_kind

test_eq(sorted(k.provides), before)                 # attaching a kernel takes nothing away
test_eq(k.scopes, ('isolated', 'overlay'))
test_eq(k.kernel_kind, 'ipykernel')
test_eq(k.run_python('answer = 42'), 'ran on the kernel')
test_eq(k.list_vars(), 'answer')
test_eq(k.inspect_python('answer', 'overlay'), 'inspected under overlay')
test_eq((k.restart_kernel(), k.list_vars()), ('restarted on the kernel', ''))
del Kernel.restart
test_fail(k.restart_kernel, contains='cannot be restarted')   # a kernel without `restart` refuses rather than pretending

A dhrishti AgentSession provides the protected overlay scope. This adapter wraps one as a kernel.

from dhrishti.agent import AgentSession

class DhrishtiKernel:
    scopes, kind = ('isolated', 'overlay'), 'dhrishti'
    def __init__(self, owner): self.session = AgentSession(owner=lambda: owner, rules=[], log=False)
    def _text(self, result):
        if not result.ok: return result.error
        value = getattr(result.result, 'value', None)
        return (result.stdout + (str(value) if value is not None else '')).strip() or '(no output)'
    def run(self, code): return self._text(self.session.run(code, 'overlay'))
    def inspect(self, code, scope='isolated'): return self._text(self.session.run(code, scope))
    def list_vars(self): return '\n'.join(f'{x.name} {x.type} {x.value}' for x in self.session.snapshot())

owner = {'base': 21}
dhrishti_host = LocalHost([Path(tempfile.mkdtemp())], index=False, kernel=DhrishtiKernel(owner))
assert dhrishti_host.run_python('answer = base * 2\nanswer') == '42'
assert 'answer int 42' in dhrishti_host.list_vars()
assert dhrishti_host.inspect_python('base + 1', 'isolated') == '22'
dhrishti_host.run_python('base = 4')
test_eq(owner, {'base': 21})

With tmux on the machine, the same host reads a neighbouring pane, runs a command in a pane below, opens a pane for the user and closes everything it made. The test is skipped where tmux is not installed.

import time

if shutil.which('tmux'):
    import fastmux
    sess = fastmux.new_session(name=f'shalya-test-{uuid.uuid4().hex[:6]}', width=140, height=40, cwd='.')
    try:
        me = list(list(sess.windows)[0].panes)[0]
        user = me.split('right', cwd='.')
        h = LocalHost(['.'], index=False, tmux=me)
        user.send('echo hello-from-user'); user.send_keys('Enter', wait_ms=2000)
        tt = h.terminal_text(50)
        assert tt.startswith(f'## pane {user.id}') and 'hello-from-user' in tt, tt
        rid = h.run_cmd_bg("printf 'a\\nb\\n'; sleep 30"); time.sleep(1)
        st, txt = h.cmd_output(rid, 0); assert st == 'running' and txt.startswith('a\nb'), (st, txt)
        assert h.cmd_stop(rid) == 'stopped' and h.cmd_output(rid, 0) == ('stopped', txt)
        rid2 = h.run_cmd_bg('exit 3'); time.sleep(1.5); assert h.cmd_output(rid2)[0] == 'exit 3'
        pid = h.open_pane('tail -f /dev/null', title='watch'); assert pid.startswith('%')
        e = h.environment(); assert e.endswith(f'pane {me.id}'), e
        h.close()
        ids = {p.id for w in sess.windows for p in w.panes}
        assert ids == {me.id, user.id}, ids
    finally: sess.kill()
assert LocalHost(['.'], index=False, tmux=False).tmux_pane is None
assert LocalHost(['.'], index=False, tmux=False).environment().endswith('tmux: off')