core

What every tool result looks like, and the limits it is held to.

Limits

Tool results use up the model’s context window, so each kind of result has a cap. MAX_TOOL_CHARS caps one result, and is sized for the smallest model in use. Tool factories take the cap as mx, so a harness running a larger model can raise it. MAX_HITS caps search hits, MAX_GREP_HITS exact grep matches, MAX_API the names a public_api listing returns, and MAX_FILE the size of a file worth reading as source at all.

# a grep answers with more rows than a search, because an exact match is cheaper to read
assert MAX_GREP_HITS > MAX_HITS
assert MAX_TOOL_CHARS < 8_000 and MAX_FILE > MAX_TOOL_CHARS
test_eq((MAX_TOOL_CHARS, MAX_HITS, MAX_GREP_HITS, MAX_API, MAX_FILE), (6000, 20, 60, 200, 2_000_000))

Every search backend returns its results as Hits: the path, line, symbol and matching text.

Hit

def Hit(
    path, line:int=1, symbol:str='', text:str=''
):

One search hit: path, line, symbol, text.

h = Hit('shalya/core.py', 12, 'clip', 'def clip(s, n=MAX_TOOL_CHARS, more=\'\'):')
h.path, h.line, h.symbol

test_eq(repr(h), "shalya/core.py:12  clip  def clip(s, n=MAX_TOOL_CHARS, more=''):")

Failure

A tool must not raise, because an exception ends the agent’s turn. A tool that fails returns a string starting with ERROR: instead, so the model can read what went wrong and try again. err builds that string, failed checks for it, and host_err formats an exception as Type: message. HostError is the exception a host raises to refuse something.

attempt(what, f) is the try/except every tool would otherwise repeat. It runs f and turns any exception into an err, except NotImplementedError, which it lets through: that means the host cannot do this at all, which the caller handles differently from a call that failed. Media is a result string that also carries pictures, as paths or bytes, which the agent’s loop sends to the model after the text.

attempt

def attempt(
    what, f
):

Run f; a host that cannot is left to raise, and anything else is one tool failure.

Media

def Media(
    *args, **kwargs
):

A tool result’s text carrying pictures, as paths or bytes, that the agent’s loop sends the model after it.

failed

def failed(
    result
):

Whether a tool result starts with ERROR:.

err

def err(
    what, e:NoneType=None
):

A tool failure message: ERROR: and what, followed by exception e when given.

host_err

def host_err(
    e
):

Exception e as Type: message, for showing to a user or a model.

HostError

def HostError(
    *args, **kwargs
):

A host refusal.

try: raise HostError('path is outside the open folders: /etc/passwd')
except HostError as e: shown = host_err(e)
shown

test_eq(shown, 'HostError: path is outside the open folders: /etc/passwd')
test_eq(host_err(KeyError('nope')), "KeyError: 'nope'")
test_eq(err('could not read', HostError('no such file')), 'ERROR: could not read: HostError: no such file')
test_eq(ERR, 'ERROR: ')

try: json.loads('{not json}')
except Exception as e: msg = err('cannot read the edits', e)
msg

test_eq(failed(msg), True)
test_eq(failed('ERROR was not the problem'), False)
test_eq(err('no such file'), 'ERROR: no such file')
# attempt: a host that cannot is left to raise; anything else is one tool failure
def _cannot(): raise NotImplementedError
def _broke(): raise RuntimeError('vault unavailable')
test_fail(lambda: attempt('x', _cannot), exc=NotImplementedError)
test_eq(attempt('memory search failed', _broke), 'ERROR: memory search failed: RuntimeError: vault unavailable')
test_eq(attempt('x', lambda: 'fine'), 'fine')

m = Media('saved /tmp/a.png', ['/tmp/a.png'])
test_eq((m, m.media, str(m) == 'saved /tmp/a.png'), ('saved /tmp/a.png', ['/tmp/a.png'], True))
test_eq(failed(m), False)

The read guard

An application may need to control what reaches the model, for example to keep personal data away from a hosted model while letting a local one see it. A host’s read_guard, when set, is a function every read tool calls before sending content. It receives a ReadCheck describing the read: the path (a URL or page id for web reads), the tool, the kind of content (text, hits, cell, diff, url, page or image), the turn’s model, whether that model runs locally, and text, the content the tool would send. text is loaded only if the guard asks for it, so a guard that decides by path reads nothing. local is None when the host cannot tell, and a guard should then treat the model as hosted.

The guard returns None or True to send the content unchanged, a string to send instead (a masked version, or even an empty string), or a Refusal; False is a refusal with the default wording. A guard that raises refuses the read with kind guard_failed. A refusal reaches the model as an ERROR: result naming its kind, path and reason, so failed recognises it. model and local come from the host’s model_local(), which returns (local, model).

Tools call the guard through two helpers. guard_read returns None (send as is), a replacement string, or a Refusal. screened returns the text to send, or the Refusal. With no guard set, guard_read returns None without calling anything and screened returns the text unchanged.

screened

def screened(
    host, path, text, tool:str='', kind:str='text', **kw
):

text as host.read_guard lets it go: whole, masked, or the Refusal.

guard_read

def guard_read(
    host, path, load:NoneType=None, tool:str='', kind:str='text', model:NoneType=None, local:NoneType=None
):

What host.read_guard says of one read: None sends it as is, a str is sent instead, a Refusal withholds it.

Refusal

def Refusal(
    why, kind:str='withheld', path:str=''
):

A read_guard withholding a read: why the model is told, kind a caller branches on, the path refused.

ReadCheck

def ReadCheck(
    path, tool:str='', kind:str='text', load:NoneType=None, model:str='', local:NoneType=None
):

One read for a read_guard to decide on; text is loaded only when the guard reads it.

class _H: read_guard = model_local = None
h = _H()
test_eq(guard_read(h, '/a.md', lambda: 1/0), None)              # no guard: nothing is called
test_eq(screened(h, '/a.md', 'secret'), 'secret')
seen = []
def guard(r):
    seen.append(r)
    if r.path.endswith('open.md'): return None
    if r.path.endswith('local.md'): return Refusal('local-only, and this model is hosted') if not r.local else True
    if r.path.endswith('boom.md'): raise RuntimeError('vault closed')
    if r.path.endswith('no.md'): return False
    return r.text.replace('ada@x.org', '[EMAIL]')
h.read_guard, h.model_local = guard, lambda: (False, 'claude-opus')
loads = []
test_eq(screened(h, '/open.md', 'ada@x.org'), 'ada@x.org')
test_eq(guard_read(h, '/open.md', lambda: loads.append(1)), None); test_eq(loads, [])   # decided on the path, read nothing
test_eq(screened(h, '/p.md', 'mail ada@x.org'), 'mail [EMAIL]')
test_eq((seen[-1].model, seen[-1].local, seen[-1].kind), ('claude-opus', False, 'text'))
r = screened(h, '/local.md', 'x')
test_eq((type(r), r.kind, r.path, str(r)), (Refusal, 'withheld', '/local.md', 'ERROR: withheld: /local.md: local-only, and this model is hosted'))
test_eq(failed(str(r)), True)
test_eq(screened(h, '/local.md', 'x', local=True), 'x')          # a per-call model overrides the turn's
test_eq(str(screened(h, '/no.md', 'x')), 'ERROR: withheld: /no.md: this host withholds it from this model')
test_eq(screened(h, '/boom.md', 'x').kind, 'guard_failed')
h.model_local = lambda: 1/0
screened(h, '/p.md', 'x'); test_eq((seen[-1].model, seen[-1].local), ('', None))   # unknown model: local is None

The sandbox

The sandbox keeps tools inside the folders the user opened, called roots. Each path is resolved before it is checked, so .., ~ and symbolic links are judged by where they actually lead. Sandbox does this and raises Unsafe, a HostError, for a path it refuses. Each host keeps one Sandbox and every tool goes through it.

Some paths are refused even when reads outside the roots are allowed. DENY lists them: SSH and cloud credentials, .env files, private keys and similar. denied checks a path against that list, ignoring case, because a case-insensitive disk opens .SSH as .ssh.

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 one of the refused credential patterns, ignoring case.

Unsafe

def Unsafe(
    *args, **kwargs
):

A path that resolves outside every open root, or one refused for safety.

A read tool must not become a way to copy out a key, so denied paths stay refused even with read_outside:

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

Sandbox

def Sandbox(
    roots:tuple=(), # folders to open for reads and writes; each must already exist
    read_outside:bool=False, # let reads name any path on this machine
    deny:tuple=('*/.ssh/*', '*/.aws/*', '*/.gnupg/*', '*/.config/gcloud/*', '*/.netrc', '*/.git-credentials', '*/.codex/auth.json', '*/.claude/.credentials.json', '*/.env', '*/.env.*', '*/id_rsa*', '*/id_ed25519*', '*.pem', '*.key', '*.p12'), # what `read_outside` still refuses to open
    read_roots:tuple=(), # folders to open for reads only
):

The open folders, and the one place that decides whether a path is inside them.

spill

def spill(
    text
):

Save text whole in spill_dir, and return its path.

spill_dir

def spill_dir():

The private folder, one per process, where results too long to send whole are saved; every Sandbox may read it and none may write it.

Sandbox(roots) opens folders for reading and writing, and read_roots folders for reading only. check(path) resolves a path, relative to the first root if it is relative, and returns it or raises Unsafe. A path that does not exist yet still resolves, since a tool creating a file needs that; readers pass must_exist=True. reading=True marks a read, which may use read-only roots and read_outside. covers and reads test a path against the two kinds of root, add_root and drop_root open and close folders, root_of names the root holding a path, and bare and under handle a file name given by a person or a model.

A result too long to send whole can be saved with spill, which writes it to the folder spill_dir names and returns the path. Every sandbox may read that folder, so the model can page through the full result with view_file, and none may write to it.

tmp = TemporaryDirectory()
root = Path(tmp.name).resolve()/'proj'
(root/'pkg').mkdir(parents=True)
(root/'pkg'/'a.py').write_text('x = 1\n')

sb = Sandbox([root])
sb.check('pkg/a.py').name, sb.check('pkg/new.py').name, sb.covers(root/'pkg'), sb.root_of('pkg/a.py') == root

Resolving before checking is what stops a symbolic link from escaping: checking the unresolved path would accept link/secret.txt because it looks inside the root. With no open folders nothing resolves, so a host cannot be tricked into reading /etc/passwd.

outside = Path(tmp.name).resolve()/'outside'
outside.mkdir()
(outside/'secret.txt').write_text('token\n')
(root/'link').symlink_to(outside, target_is_directory=True)

test_fail(lambda: sb.check('link/secret.txt'), contains=SANDBOX)        # a symlink out
test_fail(lambda: sb.check('../outside/secret.txt'), contains=SANDBOX)  # .. traversal
test_fail(lambda: sb.check(outside/'secret.txt'), contains=SANDBOX)     # an absolute path elsewhere
test_fail(lambda: sb.check('pkg/new.py', must_exist=True), contains='no such file')
test_fail(lambda: Sandbox().check('/etc/passwd'), contains=NO_ROOTS)    # nothing open, nothing readable

read_outside=True widens reads only. Writes must still land in an open folder, and credential paths are always refused.

lax = Sandbox([root], read_outside=True)
lax.check(outside/'secret.txt', reading=True).name

Open folders never nest. Opening a parent closes the roots inside it, and opening a folder already covered changes nothing, so each file is walked and indexed once.

sb2 = Sandbox([root/'pkg', root])                         # a folder and its parent, opened together
[r.name for r in sb2.roots]

bare refuses a name that is really a path, such as ../../etc. under places a named entry inside a folder without letting it escape.

sb.under('pkg', 'b.py').name, sb.bare('b.py')

Clipping

Long results have to be cut to fit the budget, and the cut should tell the model what it missed. clip shortens a string to n characters, at a line break when one is close, and appends a note giving how much was shown. The note counts against n. more adds a hint on how to read the rest, and keep=True saves the whole result with spill and names the file in the note. clip_json does the same for JSON while keeping it valid JSON: a list keeps the items that fit plus a last item saying how many more there are, and a dict keeps its plain fields, with the longest strings shortened and a truncated note.

clip_json

def clip_json(
    r, n:int=6000
):

r as indented JSON of at most n characters that is still valid JSON, with the whole saved by spill when it does not fit.

clip

def clip(
    s, n:int=6000, more:str='', # how to read the rest, when the caller has a way
    keep:bool=False, # save the whole with `spill` and name the file, for a result nothing can page
):

Truncate a tool result to n chars, its note included, whenever n is longer than the note.

print(clip('one\ntwo\nthree\nfour\nfive\n', 12))

import re

test_eq(clip('short', 12), 'short')
five = 'one\ntwo\nthree\nfour\nfive\n' * 3
test_eq(clip(five, 60).splitlines()[0], 'one')
assert len(clip(five, 60)) <= 60, 'the note counts against the budget'
kept = clip(five, 60, keep=True)
saved = Path(re.search(r'saved to (\S+?);', kept).group(1))
test_eq(saved.read_text(), five)
assert len(kept) <= 200 and 'view_file' in kept
report = {'undo': 'u1', 'head': 'abc123', 'branches': [f'branch-{i}' for i in range(200)]}
small = json.loads(clip_json(report, 500))                     # still a document a harness can parse
test_eq((small['undo'], small['head'], 'branches' in small), ('u1', 'abc123', False))
test_eq(json.loads(Path(re.search(r'saved to (\S+?);', small['truncated']).group(1)).read_text()), report)
test_eq(json.loads(clip_json({'undo': 'u1'})), {'undo': 'u1'})
long_result = clip_json({'undo': 'u1', 'result': 'r' * 5000, 'branches': ['b']}, 600)   # a long plain field is shortened, not kept whole
assert len(long_result) <= 600 and json.loads(long_result)['undo'] == 'u1', long_result
commits = [{'sha': f'{i:07d}', 'subject': 'x' * 40} for i in range(100)]
log = clip_json(commits, 800)                                    # a list stays a list: what fits, then a note
assert len(log) <= 800 and isinstance(json.loads(log), list) and 'more' in json.loads(log)[-1], log[-200:]
test_eq(clip('hello', 5), 'hello')
assert clip('hello', 0).startswith('h') and '1 of 5' in clip('hello', 0)   # a zero or negative budget
assert clip('hello', -200).startswith('h')

clip_lines renders a list of lines within n characters and says which line to continue from, which is how view_file pages through a file. more is the hint, and {next} in it is replaced by the next line number. empty is returned for an empty list. A single line longer than the budget is cut by characters, and the note gives its length but no line to resume from.

clip_lines

def clip_lines(
    lines, start:int=1, n:int=6000, more:str='', empty:str='(nothing)'
):

Render lines within n chars, the note included whenever n is longer than the note, and say which line to resume from.

print(clip_lines([f'def f{i}(): pass' for i in range(20)], n=200, more='read from line {next}'))

test_eq(clip_lines([], empty='(no matches)'), '(no matches)')
test_eq(clip_lines(['a', 'b'], n=100), 'a\nb')
long = clip_lines(['x' * 500], n=200)
assert long.startswith('x' * 100) and '500 chars' in long and len(long) <= 200, long
paged = clip_lines([f'line {i}' for i in range(100)], n=200, more='read from line {next}')
assert len(paged) <= 200 and 'read from line' in paged, paged
hinted = clip_lines(['x' * 50] * 40, n=1000, more='m' * 600)        # a long hint is reserved whole, not capped at half the budget
assert len(hinted) <= 1000 and hinted.endswith('m]'), len(hinted)

Edits

A model sends edits as lists, because the tool schemas declare them as arrays, and a harness may pass a stray JSON string through. cmds parses commands for exhash, the hash-addressed editor, into the tuples it expects. edits parses exact-text replacements. Both reject input they cannot read with an error that names the shape wanted.

cmds

def cmds(
    commands
):

Parse exhash commands, as JSON or lists, into the tuples exhash expects, nested lists included.

cmds('[["12|a1b2|", "s", "old", "new"], ["30|9f3c|", "a", "appended"]]')

test_eq(cmds([['1|ab|', 'd']]), [('1|ab|', 'd')])
test_fail(lambda: cmds('{"not": "a list"}'), contains='must be a JSON list')
test_fail(lambda: cmds('["bare string"]'), contains='must be an array')

edits accepts a list of {oldText, newText} dicts, a list of [old, new] pairs, or a single dict. A JSON string of any of those is still accepted in this release. apply_edits applies them to a text. Each oldText must appear exactly once, and the edits must not overlap. Otherwise it raises a ValueError naming the edit and how to fix it: give the exact text, add surrounding lines when a match is ambiguous, or merge overlapping edits.

apply_edits

def apply_edits(
    text, es
):

Apply exact-text edits to text, or raise saying which one is wrong and why.

edits

def edits(
    es
):

{'oldText','newText'} dicts or [old, new] pairs, as a list; a JSON string of either still parses this release.

before = 'def greet(name):\n    return "hello " + name\n'
after = apply_edits(before, edits([{'oldText': 'hello', 'newText': 'howdy'}]))
after

test_eq(edits([{'oldText': 'a', 'newText': 'b'}]), [('a', 'b')])
test_eq(edits([['a', 'b']]), [('a', 'b')]); test_eq(edits({'oldText': 'a', 'newText': 'b'}), [('a', 'b')])
test_eq(edits('[{"oldText": "a", "newText": "b"}]'), [('a', 'b')])            # a string still parses this release
test_fail(lambda: edits('not json'), contains='list of {"oldText"')
test_fail(lambda: edits([{'old_text': 'a', 'new_text': 'b'}]), contains="'oldText' and 'newText'")
test_fail(lambda: edits([['a']]), contains='[old, new]')
test_fail(lambda: edits(3), contains='list of')
try: apply_edits(before, [('', 'x')])
except ValueError as e: assert 'oldText is empty' in str(e) and 'create_file' not in str(e), e   # the message is one both editors can carry
test_eq(after, 'def greet(name):\n    return "howdy " + name\n')

Here are the three refusals on real text. Each message tells the model how to correct the edit, not only that it failed.

refusals = [lambda: apply_edits(before, [('name', 'who')]), lambda: apply_edits(before, [('nowhere', 'x')]), lambda: apply_edits(before, [('def greet', 'def hi'), ('greet(name)', 'hi(n)')])]
for r in refusals:
    try: r()
    except Exception as ex: print(host_err(ex))

diff_text makes the unified diff that edit tools return and that approval prompts show a person.

diff_text

def diff_text(
    before, after, path:str='file'
):

A unified diff from before to after, labelled with path.

print(diff_text(before, after, 'greet.py'))

assert '-    return "hello " + name' in diff_text(before, after)
assert diff_text(before, before) == ''

Which tools write

An application usually asks a person before a tool changes something. @writes marks a tool as one that does, and is_write checks the mark. WRITE_TOOLS lists the same tools by name, for a caller that only has a name; the tests check that the two agree. @acts and has_effect mark tools that act without writing a file the user owns, such as running an API call or starting a watch, and ACTING_TOOLS names them. Approval does not gate these, but a sub-agent told not to act loses them too. The git tools are listed in GIT_READ_TOOLS and GIT_WRITE_TOOLS.

@summary attaches a function that turns a call’s arguments into the one line a person reads after the tool runs, such as “Edit pkg/a.py”. summarise produces that line for a tool or a bare tool name, falling back to name(arg=value, ...). It finds summaries for bare names in SUMMARIES, which shalya.tools fills when it is imported. one_line squeezes any value onto one short line.

summarise

def summarise(
    tool, args:NoneType=None
):

The imperative one-liner for a call: what a person would say they just did.

summary

def summary(
    fn
):

Mark the one line a person reads after this tool runs. fn is given the call’s arguments.

one_line

def one_line(
    v, n:int=90
):

One line of a value, short enough to sit in a list.

has_effect

def has_effect(
    t
):

Whether tool t acts without writing a user file, the kind of effect approval does not gate.

acts

def acts(
    f
):

Mark a tool that acts without writing a file the user owns.

is_write

def is_write(
    t
):

Whether t is a tool that changes something.

writes

def writes(
    f
):

Mark tool f as one that changes something, so an approval step can put it in front of a person.

@writes
def create_file(path, text): return f'wrote {path}'
def view_file(path): return 'contents'
is_write(create_file), is_write(view_file)

test_eq(is_write(create_file), True)
test_eq(is_write(view_file), False)
test_eq(create_file.__name__ in WRITE_TOOLS, True)
test_eq(sorted(GIT_TOOLS), sorted(set(GIT_READ_TOOLS) | GIT_WRITE_TOOLS))
test_eq(GIT_READ_TOOLS, ('git_status', 'git_divergence', 'git_diff', 'git_log'))
assert {'watch', 'research', 'inspect_python'} <= ACTING_TOOLS and not ({'watch_url', 'set_reminder'} & ACTING_TOOLS)
assert 'edit_file' in WRITE_TOOLS and 'watch' not in WRITE_TOOLS