host

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

Saying what a host can do

A capability class names a group and its required methods. A host declares the group by inheriting the class. ABCMeta rejects an incomplete declaration.

provides returns declared groups minus the instance’s without set. without records unavailable optional backends, such as disabled web access or a missing fossick installation.

/opt/hostedtoolcache/Python/3.12.14/x64/lib/python3.12/site-packages/usearch/__init__.py:131: UserWarning: Will download `usearch_sqlite` binary from GitHub.
  warnings.warn("Will download `usearch_sqlite` binary from GitHub.", UserWarning)

---

source

Capability

def Capability(
    *args, **kwargs
):

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

The path boundary is not a capability. Every group needs it, so it is Host itself.

---

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 capability class names its group and the methods its tools require.

---

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 cell listing and insertion.

---

source

WebHost

def WebHost(
    *args, **kwargs
):

Current web search and page retrieval.

---

source

CodeHost

def CodeHost(
    *args, **kwargs
):

Code search and structure.

Nine of them, and each carries the name tools_for looks up. A group whose class and whose table entry disagree is a group that silently never arrives.

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

['api',
 'ask',
 'code',
 'git',
 'memory',
 'notebook',
 'session',
 'shell',
 'watch',
 'web']

test_eq(sorted(c.group for c in GROUP_CLASSES),
        ['api', 'ask', '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 answer to every “can you?” the toolset asks. Read it against a host that declares two groups and has lost one of them.

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')

(['code', 'file'], True, False)

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 with an incomplete capability declaration cannot be constructed. 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 implements every capability group. Optional dependencies and constructor arguments add unavailable groups to without. A kernel can be attached without replacing the host.

The refusals and the skip lists are the host’s, not a tool’s. A tool that reported “outside the open folders” in its own words would be a second place to change when the rule changes.

SANDBOX, SECRET, NO_ROOTS

('path is outside the open folders',
 'path holds credentials and is never read',
 'no folders are open, so no path is inside them')

assert all(isinstance(s, str) and s for s in (SANDBOX, SECRET, NO_ROOTS))
assert len(DENY) >= 15 and '*/.env' in DENY and '*.pem' in DENY
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

denied

def denied(
    path,
    patterns:tuple=('*/.ssh/*', '*/.aws/*', '*/.gnupg/*', '*/.config/gcloud/*', '*/.netrc', '*/.git-credentials', '*/.codex/auth.json', '*/.claude/.credentials.json', '*/.env', '*/.env.*', '*/id_rsa*', '*/id_ed25519*', '*.pem', '*.key', '*.p12')
):

Whether path matches a refused credential path.

denied is what read_outside still refuses: credential-shaped paths, matched by fnmatch on the resolved path. A read tool is not a way to exfiltrate a key.

[p for p in ('/home/k/.ssh/id_rsa', '/proj/.env', '/proj/key.pem', '/proj/app.py') if denied(p)]

['/home/k/.ssh/id_rsa', '/proj/.env', '/proj/key.pem']

for p in ('/home/k/.ssh/id_rsa', '/home/k/.aws/credentials', '/proj/.env', '/proj/.env.local',
          '/proj/id_ed25519', '/proj/key.pem', '/proj/cert.p12', '/home/k/.netrc'):
    assert denied(p), p
for p in ('/proj/app.py', '/proj/README.md', '/proj/environment.yml'):
    assert not denied(p), p
test_eq(denied('/proj/.env', patterns=()), False)         # an empty deny list denies nothing

ld_json pulls the structured data out of a page, which is how a product or an article survives a conversion to markdown that would otherwise throw its fields away.

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

[{'@type': 'Product', 'name': 'a kettle'}]

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(''), [])

Construction resolves the roots and starts one Kosha sync per root. search uses ripgrep until an index is ready.

without is computed once from constructor arguments and available dependencies.

---

source

LocalHost

def LocalHost(
    roots:tuple=('.',), # the folders the agent is confined to
    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
):

Reference host for local folders.

The examples use one temporary source tree. index=False keeps boundary checks independent of Kosha; the retrieval section creates an indexed host.

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)

(['code', 'file', 'git', 'notebook', 'session', 'shell', 'web'],
 ['api', 'ask', 'memory', 'watch'])

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

The path boundary

check resolves paths before enforcing the root boundary. It refuses .. and symlinks that resolve outside a root.

---

source

LocalHost.roots

def roots():

Call self as a function.

The open folders, resolved and absolute. A relative path a caller gives is taken against the first of them.

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 records roots opened after construction. A resumed session does not inherit them.

test_eq(local.added_roots, [])

---

source

LocalHost.add_root

def add_root(
    path
):

return resolved folders for read/write. does not create folders

add_root widens the write boundary. It accepts existing directories and never creates them.

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

('/private/var/folders/kg/9vdw4mdd1fs58svgh4k1qhr09x7dqh/T/tmprswhhqdf/extra',
 ['/private/var/folders/kg/9vdw4mdd1fs58svgh4k1qhr09x7dqh/T/tmprswhhqdf/extra'])

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_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')

---

source

LocalHost.check

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

Resolve path. Refuse outside roots (unless read_outside and reading). Walks stay confined.

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

---

source

LocalHost.roots_note

def roots_note():

A summary of root and read access.

roots_note summarizes the root count and read boundary.

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

('2 folder(s); nothing outside them is readable',
 '1 folder(s); reads may name any path on this machine, writes may not')

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.

Every readable file under the open folders, with the generated directories and binary suffixes SKIP_DIRS and SKIP_SUFFIXES name left out.

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 file text or None when the path 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.

Through the same check, and it makes the parent directory. Returns the path written.

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

'/private/var/folders/kg/9vdw4mdd1fs58svgh4k1qhr09x7dqh/T/tmp38ppko54/proj/pkg/new.py'

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.

One file as a diffable document. A missing file returns '', which gives a new file a diff. A notebook returns its cell sources.

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():

Call self as a function.

approvals exposes the host’s write approval handler.

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

(None, 'an 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 never blocks. It appends every message to transcript before calling the optional status callback.

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

(['starting a kernel'], ['starting a kernel'])

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'])

Seeing the code

Three engines answer a search, and which one answered is part of the result. Kosha’s hybrid index when it has synced, ripgrep when it has not, and reading the files when neither is installed.

---

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 daemon thread. Each root publishes its index when its sync completes.

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

(True, 'shalya-kosha-sync')

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 reports whether every open folder has an index. indexed gives search the indexed folders.

local.index_ready, local.indexed

(False, ())

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.

Joins the sync thread and answers whether semantic search is ready.

local.wait_index(1), local.index_ready

(False, False)

test_eq(local.wait_index(1), False)

---

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.

Exact matching through ripgrep. None means this host has no exact matcher, and the tool then reads the files itself rather than reporting no matches.

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.

Kosha’s index and ripgrep fused by rank rather than tried in order, falling back to reading the files when neither is installed. An empty query answers [] rather than everything.

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

---

source

LocalHost.search_note

def search_note():

Call self as a function.

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

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

'Kosha semantic + keyword index over 1 folder(s) and environment fused with ripgrep'

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

"Kosha unavailable (AttributeError: 'LocalHost' object has no attribute 'sync_index'); literal fallback"

---

source

LocalHost.public_api

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

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

public_api lists package exports. It raises without an index, preserving the distinction between an unavailable index and a package with no exports.

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.

From parsing rather than grepping, so a method is reported at its own depth under its qualified name. score is the indent depth.

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 finds references to the definition at a line and excludes the defining line.

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

---

source

LocalHost.nb_cells

def nb_cells(
    path
):

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

(id, cell_type, source) per cell, and nothing else. Shalya owns no notebook representation.

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')

[('c0', 'code', 'x = threshold(4096)')]

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')

---

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.

Writes the notebook when it does not exist, and returns the new cell’s id. -1 appends.

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')]

[('28ab593a', 'code'), ('129c5da3', 'code'), ('c15553ab', 'markdown')]

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')

---

source

LocalHost.run_python

def run_python(
    code
):

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

The live namespace persists between calls, which is what makes “bind results to new names” a rule worth briefing an agent on. A failure comes back as text: a tool that raises ends the turn.

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

('(no output)', '3')

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 rebinding anything the user made. A kernel enforces this; a copy approximates it.

In-process inspection uses a shallow namespace copy. New bindings disappear after the call; mutations to shared objects remain.

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

('3', 'False')

test_eq(local.inspect_python('doubled = [r*2 for r in radii]\nlen(doubled)'), '3')
test_eq(local.run_python('"doubled" in globals()'), 'False')
local.inspect_python('radii.append(99)')
test_eq(local.run_python('len(radii)'), '4')
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 shallow copy alone.

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

local.scopes

('isolated',)

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

---

source

LocalHost.kernel_kind

def kernel_kind():

Call self as a function.

The live namespace runner states whether inspection can run while a cell is busy. The harness uses concurrent to decide whether to inspect now or queue the call.

local.kernel_kind, local.concurrent

('inprocess', False)

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.

Name, type and a short value, one line each. Callables and dunders are left out: an agent asking what is in the namespace 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.terminal_text

def terminal_text(
    lines:int=200
):

What this process has printed, when the application records it in transcript.

What this process has printed, when the application records it. Read-only: it shows what was run, it cannot run anything.

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

'line 1\nline 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(), '')

---

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.

In its own process group under the first open folder, with stdout and stderr interleaved. A failure is an exit code rather than an exception, and the group is killed on timeout.

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.shell_note

def shell_note():

Call self as a function.

How commands are run here, or why they are not.

local.shell_note

'shell, in /private/var/folders/kg/9vdw4mdd1fs58svgh4k1qhr09x7dqh/T/tmp38ppko54/proj'

assert str(root) in local.shell_note

The web

Every page arrives as markdown. A URL that is a paper, a repository file or a video gets the reader that knows its shape. Everything else is fetched. A page that comes back too thin to be real is fetched again the expensive way.

---

source

LocalHost.web_search

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

Search the web through fossick. An empty query answers []: that is how tools_for probes.

Through fossick, and an empty query answers [] without reaching the network.

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

(True, [])

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)

---

source

LocalHost.read_url

def read_url(
    url, remember:bool=True
):

Page as markdown via fossick: READERS, then fetch(auto=True), thin-page escalate, JSON-LD.

---

source

LocalHost.research

def research(
    query
):

The cited corpus fossick assembled: its digest, and not the record it assembled it from.

---

source

LocalHost.research_note

def research_note():

Call self as a function.

research_note reports whether fossick or no web backend serves the group.

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

('fossick', 'web access is switched off')

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 exercises LocalHost against fossick directly. It is excluded from the default suite because it searches and fetches the public web.

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()

[2026-08-31 13:28:59] INFO: Fetched (200) <GET https://www.python.org/> (referer: https://www.google.com/)
[2026-08-31 13:29:03] INFO: Fetched (200) <GET https://www.python.org/> (referer: https://www.google.com/)
[2026-08-31 13:29:04] INFO: Fetched (200) <GET https://www.python.org/> (referer: https://www.google.com/)
[2026-08-31 13:29:15] INFO: Fetched (200) <GET https://www.python.org/> (referer: https://www.google.com/)
[2026-08-31 13:29:15] INFO: Fetched (200) <GET https://www.python.org/doc/essays/blurb/> (referer: https://www.google.com/)
[2026-08-31 13:29:16] INFO: Fetched (200) <GET https://www.w3schools.com/python/python_intro.asp> (referer: https://www.google.com/)
[2026-08-31 13:29:16] INFO: Fetched (200) <GET https://www.teradata.com/insights/data-platform/what-is-python-programming-language> (referer: https://www.google.com/)
[2026-08-31 13:29:16] INFO: Fetched (200) <GET https://en.wikipedia.org/wiki/Python_(programming_language)> (referer: https://www.google.com/)
[2026-08-31 13:29:17] INFO: Fetched (200) <GET https://aws.amazon.com/what-is/python/> (referer: https://www.google.com/)
[2026-08-31 13:29:17] INFO: Fetched (200) <GET https://www.nist.gov/blogs/taking-measure/programming-language-named-monty-python-evolved-nist> (referer: https://www.google.com/)
[2026-08-31 13:29:20] INFO: Fetched (200) <GET https://www.nist.gov/blogs/taking-measure/programming-language-named-monty-python-evolved-nist> (referer: https://www.google.com/)
[2026-08-31 13:29:24] INFO: Fetched (200) <GET https://www.nist.gov/blogs/taking-measure/programming-language-named-monty-python-evolved-nist> (referer: https://www.google.com/)
[2026-08-31 13:29:27] INFO: Fetched (200) <GET https://www.nist.gov/blogs/taking-measure/programming-language-named-monty-python-evolved-nist> (referer: https://www.google.com/)
[2026-08-31 13:29:29] INFO: Fetched (200) <GET https://pythoninstitute.org/about-python> (referer: https://www.google.com/)

---

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.

---

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=()
):

File text into durable memory as a note. Returns the document record.

---

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.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 an API specification

The memory backend is a vishalakshi Vault, opened by whatever built the host, and the forwards are written in the vault’s own vocabulary: remember calls note, memory_tree calls toc, memory_topics calls topic_tree. The other seven names are the same on both sides.

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)

(['ask',
  'code',
  'file',
  'git',
  'memory',
  'notebook',
  'session',
  'shell',
  'watch',
  'web'],
 ['api'])

# 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')

Each forward is one line. The round trip below is the real one: a note filed into the vault, found by search, and read back by the node_id that 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

([{'node_id': '9cbfcd0e4a1c6a08#1', 'doc_id': '9cbfcd0e4a1c6a08', 'page': 0, 'breadcrumb': 'kettles', 'score': 0.03333333333333333, 'snippet': 'the kettle boils at 100C'}],
 ['kettles'],
 {'doc_id': '9cbfcd0e4a1c6a08',
  'title': 'kettles',
  'kind': 'note',
  'nodes': 2,
  'chunks': 1})

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 re-runs on an interval, and poll is the tick.

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

(['a3cdd4ab285e'],
 {'checked': 1,
  'ran': 1,
  'results': [{'job_id': '813f8173080f',
    'kind': 'watch',
    'status': 'ok',
    'took': 0.002,
    'error': None,
    'result': {'doc_id': '5b442d0d06972d20',
     'title': 'check the kettle',
     'kind': 'note',
     'nodes': 2,
     'chunks': 1}}],
  'reclaimed': 0,
  'dead': 0,
  'next_due': 1788253745})

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'])
test_eq(remembering.watch_actions, ('remind',))        # 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 the same way, to anything answering the four calls. ramabana.spec.SpecHost is one such backend

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')

({'name': 'github', 'operations': 2, 'groups': ['repos', 'zen']},
 [{'name': 'get_zen',
   'group': 'zen',
   'verb': 'GET',
   'path': '/zen',
   'summary': 'A design maxim'}])

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')

@patch runs after ABCMeta computes abstract methods. implemented recomputes them after the patches are applied.

---

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())

A kernel, attached to a host that already exists

run_python and related methods delegate to self.kernel when present. Assigning a kernel preserves the host’s capability groups and attached backends.

class Kernel:
    "The shape `LocalHost` expects of a kernel: four calls and two facts."
    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('__'))

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

(['code', 'file', 'git', 'notebook', 'session', 'shell', 'web'],
 ['code', 'file', 'git', 'notebook', 'session', 'shell', 'web'],
 ('isolated', 'overlay'),
 'ipykernel')

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')

A Dhrishti session supplies the protected overlay behind a concrete kernel adapter.

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})