# host
## 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
``` python
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`](https://vedicreader.github.io/shalya/host.html#host) itself.
------------------------------------------------------------------------
source
### Host
``` python
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
``` python
def GitHost(
*args, **kwargs
):
```
*Git operations on a working tree inside `roots`.*
------------------------------------------------------------------------
source
### ApiHost
``` python
def ApiHost(
*args, **kwargs
):
```
*Reading an API specification and calling what it describes.*
------------------------------------------------------------------------
source
### ShellHost
``` python
def ShellHost(
*args, **kwargs
):
```
*Running a command on the machine.*
------------------------------------------------------------------------
source
### SessionHost
``` python
def SessionHost(
*args, **kwargs
):
```
*The live namespace, and the terminal beside it.*
------------------------------------------------------------------------
source
### WatchHost
``` python
def WatchHost(
*args, **kwargs
):
```
*Recurring watches and reminders.*
------------------------------------------------------------------------
source
### AskHost
``` python
def AskHost(
*args, **kwargs
):
```
*Model-backed answers from memory.*
------------------------------------------------------------------------
source
### MemoryHost
``` python
def MemoryHost(
*args, **kwargs
):
```
*Durable memory organized as document sections.*
------------------------------------------------------------------------
source
### NotebookHost
``` python
def NotebookHost(
*args, **kwargs
):
```
*Notebook cell listing and insertion.*
------------------------------------------------------------------------
source
### WebHost
``` python
def WebHost(
*args, **kwargs
):
```
*Current web search and page retrieval.*
------------------------------------------------------------------------
source
### CodeHost
``` python
def CodeHost(
*args, **kwargs
):
```
*Code search and structure.*
Nine of them, and each carries the name
[`tools_for`](https://vedicreader.github.io/shalya/tools.html#tools_for)
looks up. A group whose class and whose table entry disagree is a group
that silently never arrives.
``` python
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']
``` python
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.
``` python
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)
``` python
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.
``` python
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`](https://vedicreader.github.io/shalya/host.html#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.
``` python
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')
``` python
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
``` python
def ld_json(
html
):
```
*Parse `schema.org` JSON-LD blocks from `html`.*
------------------------------------------------------------------------
source
### denied
``` python
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`](https://vedicreader.github.io/shalya/host.html#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.
``` python
[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']
``` python
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`](https://vedicreader.github.io/shalya/host.html#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.
``` python
page = ''
ld_json(page)
```
[{'@type': 'Product', 'name': 'a kettle'}]
``` python
test_eq(ld_json(page), [{'@type': 'Product', 'name': 'a kettle'}])
test_eq(ld_json('nothing structured here'), [])
test_eq(ld_json(''), [])
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
``` python
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.
``` python
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'])
``` python
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
``` python
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.
``` python
test_eq(local.roots, [str(root)])
test_eq(len(local.roots), 1)
```
------------------------------------------------------------------------
source
### LocalHost.added_roots
``` python
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.
``` python
test_eq(local.added_roots, [])
```
------------------------------------------------------------------------
source
### LocalHost.add_root
``` python
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.
``` python
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'])
``` python
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
``` python
def check(
path, must_exist:bool=False, reading:bool=False
):
```
*Resolve `path`. Refuse outside `roots` (unless `read_outside` and
`reading`). Walks stay confined.*
``` python
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
``` python
def roots_note():
```
*A summary of root and read access.*
`roots_note` summarizes the root count and read boundary.
``` python
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')
``` python
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
``` python
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.
``` python
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
``` python
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.
``` python
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
``` python
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.
``` python
local.write('pkg/new.py', 'x = 1\n')
```
'/private/var/folders/kg/9vdw4mdd1fs58svgh4k1qhr09x7dqh/T/tmp38ppko54/proj/pkg/new.py'
``` python
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
``` python
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.
``` python
test_eq(local.text_at('pkg/nope.py'), '')
test_eq(local.text_at('pkg/sizes.py'), local.read('pkg/sizes.py'))
```
------------------------------------------------------------------------
source
### LocalHost.approvals
``` python
def approvals():
```
*Call self as a function.*
`approvals` exposes the host’s write approval handler.
``` python
local.approvals, LocalHost([root], index=False, approvals='an Approvals').approvals
```
(None, 'an Approvals')
``` python
test_eq(local.approvals, None)
test_eq(LocalHost([root], index=False, approvals='an Approvals').approvals, 'an Approvals')
```
------------------------------------------------------------------------
source
### LocalHost.note
``` python
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.
``` python
said = []
noisy = LocalHost([root], index=False, note=said.append)
noisy.note('starting a kernel')
said, noisy.transcript
```
(['starting a kernel'], ['starting a kernel'])
``` python
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
``` python
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.
``` python
quiet = LocalHost([root], index=False)
quiet.sync_index() is quiet, quiet._index_thread.name
```
(True, 'shalya-kosha-sync')
``` python
test_eq(quiet.sync_index(), quiet) # chainable, and starts nothing twice
test_eq(quiet._index_thread.name, 'shalya-kosha-sync')
```
------------------------------------------------------------------------
source
### LocalHost.index_ready
``` python
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.
``` python
local.index_ready, local.indexed
```
(False, ())
``` python
test_eq(local.index_ready, False) # this host was built with index=False
test_eq(local.indexed, [])
```
------------------------------------------------------------------------
source
### LocalHost.indexed
``` python
def indexed():
```
*Indexed root paths.*
------------------------------------------------------------------------
source
### LocalHost.wait_index
``` python
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.
``` python
local.wait_index(1), local.index_ready
```
(False, False)
``` python
test_eq(local.wait_index(1), False)
```
------------------------------------------------------------------------
source
### LocalHost.grep
``` python
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.
``` python
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
``` python
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.
``` python
assert local.search('threshold'), 'ripgrep found nothing'
test_eq(local.search(''), [])
test_eq(local.search(' '), [])
```
------------------------------------------------------------------------
source
### LocalHost.search_note
``` python
def search_note():
```
*Call self as a function.*
This cell builds a real index to check `wait_index`, `indexed` and
`public_api`.
``` python
indexing = LocalHost([root])
indexing.wait_index(180)
indexing.search_note
```
'Kosha semantic + keyword index over 1 folder(s) and environment fused with ripgrep'
``` python
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'
```
``` python
local.search_note
```
"Kosha unavailable (AttributeError: 'LocalHost' object has no attribute 'sync_index'); literal fallback"
------------------------------------------------------------------------
source
### LocalHost.public_api
``` python
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.
``` python
test_fail(lambda: local.public_api('fastcore'), contains='no code index')
```
``` python
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
``` python
def symbols(
path
):
```
*The defs and classes in one file, as
[`Hit`](https://vedicreader.github.io/shalya/core.html#hit)s 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.
``` python
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
``` python
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.
``` python
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
``` python
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.
``` python
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)')]
``` python
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
``` python
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.
``` python
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')]
``` python
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
``` 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.
``` python
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')
``` python
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
``` 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.
``` python
local.inspect_python('doubled = [r*2 for r in radii]\nlen(doubled)'), local.run_python('"doubled" in globals()')
```
('3', 'False')
``` python
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
``` python
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.
``` python
local.scopes
```
('isolated',)
``` python
test_eq(local.scopes, ('isolated',)) # no kernel, so the shallow copy alone
```
------------------------------------------------------------------------
source
### LocalHost.kernel_kind
``` python
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.
``` python
local.kernel_kind, local.concurrent
```
('inprocess', False)
``` python
test_eq(local.kernel_kind, 'inprocess')
test_eq(local.concurrent, False)
```
------------------------------------------------------------------------
source
### LocalHost.list_vars
``` python
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.
``` python
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
``` python
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.
``` python
chatty = LocalHost([root], index=False)
for i in range(3): chatty.note(f'line {i}')
chatty.terminal_text(2)
```
'line 1\nline 2'
``` python
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
``` python
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.
``` python
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
``` python
def shell_note():
```
*Call self as a function.*
How commands are run here, or why they are not.
``` python
local.shell_note
```
'shell, in /private/var/folders/kg/9vdw4mdd1fs58svgh4k1qhr09x7dqh/T/tmp38ppko54/proj'
``` python
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
``` python
def web_search(
query, n:int=20
):
```
*Search the web through fossick. An empty query answers `[]`: that is
how
[`tools_for`](https://vedicreader.github.io/shalya/tools.html#tools_for)
probes.*
Through fossick, and an empty query answers `[]` without reaching the
network.
``` python
local.can('web'), local.web_search('')
```
(True, [])
``` python
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
``` python
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
``` python
def research(
query
):
```
*The cited corpus fossick assembled: its `digest`, and not the record it
assembled it from.*
------------------------------------------------------------------------
source
### LocalHost.research_note
``` python
def research_note():
```
*Call self as a function.*
`research_note` reports whether fossick or no web backend serves the
group.
``` python
local.research_note, LocalHost([root], index=False, web=False).research_note
```
('fossick', 'web access is switched off')
``` python
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`](https://vedicreader.github.io/shalya/host.html#localhost)
against fossick directly. It is excluded from the default suite because
it searches and fetches the public web.
``` python
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) (referer: https://www.google.com/)
[2026-08-31 13:29:03] INFO: Fetched (200) (referer: https://www.google.com/)
[2026-08-31 13:29:04] INFO: Fetched (200) (referer: https://www.google.com/)
[2026-08-31 13:29:15] INFO: Fetched (200) (referer: https://www.google.com/)
[2026-08-31 13:29:15] INFO: Fetched (200) (referer: https://www.google.com/)
[2026-08-31 13:29:16] INFO: Fetched (200) (referer: https://www.google.com/)
[2026-08-31 13:29:16] INFO: Fetched (200) (referer: https://www.google.com/)
[2026-08-31 13:29:16] INFO: Fetched (200) (referer: https://www.google.com/)
[2026-08-31 13:29:17] INFO: Fetched (200) (referer: https://www.google.com/)
[2026-08-31 13:29:17] INFO: Fetched (200) (referer: https://www.google.com/)
[2026-08-31 13:29:20] INFO: Fetched (200) (referer: https://www.google.com/)
[2026-08-31 13:29:24] INFO: Fetched (200) (referer: https://www.google.com/)
[2026-08-31 13:29:27] INFO: Fetched (200) (referer: https://www.google.com/)
[2026-08-31 13:29:29] INFO: Fetched (200) (referer: https://www.google.com/)
------------------------------------------------------------------------
source
### LocalHost.memory_search
``` python
def memory_search(
query, limit:int=8
):
```
*Search remembered pages as whole tree sections, returning structured
rows.*
------------------------------------------------------------------------
source
### LocalHost.memory_tree
``` python
def memory_tree(
document:str=''
):
```
*The heading tree for remembered documents. An empty document lists
every root.*
------------------------------------------------------------------------
source
### LocalHost.memory_read
``` python
def memory_read(
node_id
):
```
*Read one remembered section and its children by stable node id.*
------------------------------------------------------------------------
source
### LocalHost.memory_topics
``` python
def memory_topics(
limit:int=12
):
```
*Labelled semantic clusters across remembered research.*
------------------------------------------------------------------------
source
### LocalHost.memory_forget
``` python
def memory_forget(
doc_id
):
```
*Purge one remembered document and all derived tree, chunk and vector
data.*
------------------------------------------------------------------------
source
### LocalHost.remember
``` python
def remember(
text, title:NoneType=None, tags:tuple=()
):
```
*File `text` into durable memory as a note. Returns the document
record.*
------------------------------------------------------------------------
source
### LocalHost.ask
``` python
def ask(
question, ref:NoneType=None, instruction:str='', **kw
):
```
*Answer `question` out of remembered research, with citations, as a
dict.*
------------------------------------------------------------------------
source
### LocalHost.watch
``` python
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
``` python
def watches(
due_only:bool=False
):
```
*Every registered watch, soonest first. `due_only` keeps the ones that
have come due.*
------------------------------------------------------------------------
source
### LocalHost.unwatch
``` python
def unwatch(
watch_id
):
```
*Delete one watch. Whatever it already filed stays in memory.*
------------------------------------------------------------------------
source
### LocalHost.poll
``` python
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](https://github.com/vedicreader/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.
``` python
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'])
``` python
# 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.
``` python
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})
``` python
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.
``` python
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})
``` python
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
``` python
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
``` python
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
``` python
def api_count(
group:str='', name:str='', match:str=''
):
```
*How many operations that filter matches, without listing them.*
------------------------------------------------------------------------
source
### LocalHost.api_call
``` python
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
``` python
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'}])
``` python
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)
```
``` python
# 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`](https://vedicreader.github.io/shalya/host.html#implemented)
recomputes them after the patches are applied.
------------------------------------------------------------------------
source
### implemented
``` python
def implemented(
cls
):
```
*Recompute what `cls` is still missing, after `@patch` filled some of it
in.*
``` python
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')
```
``` python
@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.
``` python
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')
``` python
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.
``` python
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})
```