Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
269 changes: 137 additions & 132 deletions plugins/axiomcode/hooks/direct.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,172 +15,177 @@
· SILENT WHEN THE AGENT IS ALREADY DOING IT. A Bash call that IS an axiomcode verb gets nothing.
· NEVER BLOCKS. additionalContext, exit 0. The agent keeps its own judgement; the point is that the
judgement is made with the option in view, not that the option wins.
· ONCE PER SESSION. Said in full before the first search, then never again. It used to repeat as one line
before every later Read, Grep and Bash, and in a measured run those lines were seven of the plugin's
injections, re-read on every turn after, for an agent that had already called the graph twice. A
reminder only has to be visible once; after that it is a context tax on every turn.
· ONLY WHERE THE CHOICE IS. A Bash call that searches or reads source competes with the graph; `git`,
`ls`, a build or a test run does not, and a directive in front of one is noise.
· ONCE PER SESSION, WHEREVER THE SESSION GOES. Said once, then never again. It used to repeat as one line before
every later Read, Grep and Bash; then it was stamped once per session PER REPOSITORY, inside that repository's
.axiomcode/, so an agent moving between a worktree, its fixtures and a corpus app heard the full text in each,
and one whose .axiomcode/ it could not write heard it before every search. The stamp now lives in the temp
directory, keyed by the session alone.
· ONLY WHEN IT CAN NAME WHAT THIS SEARCH IS ABOUT. The generic text was spoken before the first search whatever
it searched for, and a feed audit found it acted on 0% of the time: advice about "a name" in front of a search
for `TODO` or `'use strict'` is not advice about this search. It now speaks only before a search whose pattern
names a class, function or method the graph declares, names it with where it is, and gives the one call that
answers what the grep is asking. A Read of a file already chosen and a Glob for files are not that question.
· ONLY WHERE THE CHOICE IS. A Bash call that searches source competes with the graph; `git`, `ls`, a build, a
test run, or a grep filtering another command's output does not, and a directive in front of one is noise.
"""
import json, os, re, shlex, sys
import hashlib, json, os, re, sys
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import _host, _where

SRC = _where.SOURCE_EXT # one table for every hook (_where.py)
# the shell commands that are the search or read this directive competes with
SEARCH = re.compile(r'(^|[;&|(]\s*|\s)(grep|egrep|rg|ag|ack|git\s+grep|find|fd|cat|head|tail|sed|awk|less)\b')
# words a search pattern is made of that name no declaration (a keyword written into the pattern to anchor it)
KEYWORDS = {'def', 'class', 'public', 'private', 'protected', 'static', 'final', 'void', 'function', 'import', 'from',
'return', 'new', 'async', 'await', 'self', 'this', 'interface', 'struct', 'record', 'override', 'virtual',
'const', 'let', 'var', 'string', 'int', 'bool', 'true', 'false', 'null', 'None', 'True', 'False', 'package',
'namespace', 'using', 'extends', 'implements'}
CALLABLE = ('method', 'function', 'constructor')

# The MCP tools are named first and the shell verbs second (#1425): `axiomcode install` pre-approves only the tools,
# so an agent that takes the first spelling it reads should be taking the one it is already allowed to call.
# Short on purpose: it is read once, and 13 generic lines were re-read on every later turn.
FULL = (
"graph: this repository has a resolved call graph; ask it before a raw search. axiomcode_impact targets=[<name>]:\n"
" who uses a declaration and what a change breaks, with callers that never spell the name (an interface, an override,\n"
" a callback, DI). axiomcode_path: how A reaches B. axiomcode_context: where a task lands. axiomcode_changed /\n"
" axiomcode_test_impact: what an edit moved, which tests reach it. Shell: `axiomcode impact <name>`, `axiomcode path\n"
" <A> <B>`, `axiomcode context \"<task>\"`, `axiomcode changed`, `axiomcode test-impact`. Pass this to a subagent."
)

# after a line that already names the verb for THIS search, the rest of the menu in one line
REST = (" Also: axiomcode_path (how A reaches B), axiomcode_context (a task in words), axiomcode_test_impact (the tests an"
" edit reaches); in a shell `axiomcode <verb>`. Pass this to a subagent.")


def _stamp(cwd, sid):
"""ONCE PER SESSION, not once per (session, repository): keyed on the session in the temp directory, so an agent
that moves between repositories (a dev copy, a second service) hears it the first time only. A session with no id
keeps the old per-repository stamp, since every such session would otherwise share one."""
if not sid:
return os.path.join(cwd, '.axiomcode', '.directed-x')
# the shell commands that are the search this directive competes with, and the pattern each one searches for
SEARCH = re.compile(r'(^|[;&|(]\s*|\s)(grep|egrep|rg|ag|ack|git\s+grep)\b')
PATTERN = re.compile(r'\b(?:grep|egrep|rg|ag|ack|git\s+grep)\b((?:\s+-[-\w=]+)*)\s+(?:-e\s+)?([\'"]?)(.+?)\2(?:\s|$)')
# the declarations a search can be asking about: a callable or a type. A field, a constant or a local shares its name
# with too much prose and config to be the thing a grep for that word is after.
KINDS = ('class', 'interface', 'enum', 'record', 'struct', 'trait', 'function', 'method', 'constructor')
SPREAD = 5
COMMON = re.compile(r'(the|and|for|new|return|null|true|false|this|self|void|int|String|public|private|static|import|export|const|function|class|def)')


def stamp_path(sid, cwd=''):
"""one file per session, outside every repository: a repository's .axiomcode/ is per tree, gets rebuilt, and is not
always writable, and each of those made the directive say itself again. A host that sends no session id is keyed
on the directory it runs in instead, so every such session does not share one stamp and hear it once, ever."""
tmp = os.environ.get('TMPDIR') or os.environ.get('TEMP') or os.environ.get('TMP') or ('/tmp' if os.name != 'nt' else os.path.expanduser('~'))
return os.path.join(tmp, 'axiomcode-hooks', 'directed-' + (re.sub(r'[^\w.-]', '_', str(sid))[:80] or 'x'))
key = str(sid) if sid else 'cwd-' + hashlib.sha1(os.path.realpath(cwd or '.').encode()).hexdigest()[:16]
return os.path.join(tmp, 'axiomcode-hooks', 'directed-' + (re.sub(r'[^\w.-]', '_', key)[:80] or 'x'))


def _pattern(tool, inp):
"""the text a search looks for: Grep's pattern, or the first argument after grep / rg / ag / ack in a shell command"""
if tool == 'Grep':
return str(inp.get('pattern') or '')
if tool == 'Bash':
try: toks = shlex.split(str(inp.get('command') or ''))
except ValueError: toks = str(inp.get('command') or '').split()
for i, t in enumerate(toks):
if t in ('grep', 'egrep', 'rg', 'ag', 'ack'):
j = i + 1
while j < len(toks) and toks[j].startswith('-'):
if toks[j] in ('-e', '--regexp'): j += 1; break
j += 2 if toks[j] in ('-f', '-A', '-B', '-C', '-m', '--include', '--exclude', '-g', '-t', '--type') else 1
if j < len(toks): return toks[j]
return ''


def specific(cwd, tool, inp, fp):
"""ONE line naming the verb that answers THIS search, when the graph has what it names: a Grep for a declared name
gets `impact` on that declaration, a Read of an indexed file gets `context` scoped to it. '' when the graph has
nothing to say about it (the generic block then stands alone)."""
import sqlite3
db = _where.graph_db(cwd, fp or None)
if not os.path.isfile(db):
return ''
def mark(stamp):
try:
con = sqlite3.connect(f'file:{db}?mode=ro', uri=True)
except sqlite3.Error:
return ''
os.makedirs(os.path.dirname(stamp), exist_ok=True)
open(stamp, 'w').close()
except Exception:
pass


def identifiers(pat):
"""the names a search pattern spells: every branch of an alternation, regex syntax stripped, at most 6"""
out = []
for br in re.split(r'\\\||(?<!\\)\|', str(pat or '')):
for m in re.finditer(r'(?<![\w\\])[A-Za-z_]\w{2,}', br):
n = m.group(0)
if COMMON.fullmatch(n) or re.fullmatch(r'[A-Z0-9_]+', n) or n in out:
continue # a keyword or a CONSTANT_SHAPED word is not a callable's name
# a plain lowercase word (`once`, `path`, `session`) is as likely prose or a flag as a name, unless THAT
# word is written as code: `locate\(`, `def locate`, `.locate`, `function render`. Code elsewhere in the
# pattern does not count: in `expr_child("client"` the name is expr_child, and client is a string.
if n.islower() and '_' not in n and not (
re.match(r'\\?\(', br[m.end():])
or re.search(r'(\b(def|function|fn|func|void|class)\s+|\\?\.|::|->)$', br[:m.start()])):
continue
out.append(n)
return out[:6]


def declared(root, names, scope=()):
"""[(name, display, file, line)] for the names a graph in `root` declares as a callable or a type: the declaration
inside the searched files or directories (`scope`, relative to root) first, then the first in the repository. A
name declared in more than SPREAD places (`get`, `run`, `handle`) says nothing about which one this search is for,
and naming one of them is a guess, so it is left out."""
import sqlite3
ax = os.path.join(root, '.axiomcode')
dbs = [os.path.join(ax, 'out', 'graph.sqlite')]
try:
pat = _pattern(tool, inp)
if pat:
for w in re.findall(r'[A-Za-z_]\w{2,}', pat):
if w in KEYWORDS:
dbs += [os.path.join(ax, 'lang', l, 'out', 'graph.sqlite') for l in sorted(os.listdir(os.path.join(ax, 'lang')))]
except OSError:
pass
found = {}
ph = ','.join('?' * len(KINDS))
for db in dbs:
if len(found) == len(names) or not os.path.isfile(db):
continue
try:
con = sqlite3.connect(f'file:{db}?mode=ro', uri=True)
for n in names:
if n in found:
continue
rows = con.execute("SELECT display, kind, file, line FROM symbols WHERE name = ? AND kind != 'module' "
"ORDER BY is_test, (kind IN ('method','function','constructor')) DESC, file, line LIMIT 4", (w,)).fetchall()
if not rows:
rows = con.execute(f"SELECT display, file, line FROM symbols WHERE name = ? AND kind IN ({ph}) "
f"ORDER BY is_test, file, line LIMIT {SPREAD + 1}", (n, *KINDS)).fetchall()
if not rows or len(rows) > SPREAD:
continue
disp, kind, f, l = rows[0]
target = disp if len(rows) == 1 else w
what = 'Who calls it and what a change breaks' if kind in CALLABLE else \
('Who reads and writes it' if kind in ('field', 'const', 'variable', 'property') else 'Who uses it, its subtypes, what a change breaks')
where = f"{disp}, {f}:{l}" if len(rows) == 1 else f"{len(rows) if len(rows) < 4 else '4+'} declarations, first {disp} {f}:{l}"
return (f"graph: `{w}` is declared here ({where}). {what}: axiomcode_impact targets=[\"{target}\"]"
f" (`axiomcode impact {target}`); a grep misses the callers that never spell it.")
if tool in ('Read', 'Bash') and fp:
rel = os.path.relpath(os.path.realpath(fp), os.path.realpath(cwd)).replace(os.sep, '/')
n = con.execute("SELECT count(*) FROM symbols WHERE file = ? AND kind IN ('method','function','constructor')", (rel,)).fetchone()[0]
if n:
return (f"graph: {rel} has {n} callable(s) in the graph. How it works: axiomcode_context task=\"<question>\""
f" in_path=\"{rel}\" source=True; one of them: axiomcode_impact targets=[\"{rel}:<line>\"].")
except sqlite3.Error:
pass
finally:
con.close()
return ''
inside = [r for r in rows if any(r[1] == s or r[1].startswith(s.rstrip('/') + '/') for s in scope)]
r = (inside or rows)[0]
found[n] = (n, r[0], r[1], r[2])
con.close()
except Exception:
continue
return [found[n] for n in names if n in found]


def directive(hits):
"""what to say before a search for these declarations: each named, with the call that answers what the grep asks"""
first = hits[0]
at = f"{first[2]}:{first[3]}"
named = ', '.join(f"`{h[1]}` ({h[2]}:{h[3]})" for h in hits[:3])
# short on purpose: it is read once and then re-read on every later turn
return (
f"graph: this search is for {named}. Who calls it and what a change breaks,\n"
f" with the callers that never spell the name (an interface, an override, a callback, DI):\n"
f" axiomcode_impact targets=[\"{at}\"] (`axiomcode impact {at}`). Also axiomcode_path (how A reaches B,\n"
f" `axiomcode path A B`), axiomcode_context (a task in words, `axiomcode context \"<task>\"`). Said once this session."
)


def main():
ev = _host.read()
tool = ev.get('tool_name') or ''
inp = ev.get('tool_input') or {}
# the repository whose graph would answer: the one above the file or directory this tool is about to touch, not
# the session's working directory, which in 308 of 309 measured sessions held no graph (_where.py)
cwd = _where.locate(tool, inp, ev.get('cwd') or os.getcwd(), ev.get('session_id'))
if not cwd:
sys.exit(0)

stamp = _stamp(cwd, ev.get('session_id'))
stamp = stamp_path(ev.get('session_id'), ev.get('cwd') or os.getcwd())

# the agent is already reaching for the graph -- saying it again is the nag this is trying not to be,
# and once it has, the directive has nothing left to say this session
# and once it has, the directive has nothing left to say this session, in any repository
if 'axiomcode' in tool or (tool == 'Bash' and 'axiomcode' in (inp.get('command') or '')):
try:
os.makedirs(os.path.dirname(stamp), exist_ok=True)
open(stamp, 'w').close()
except Exception:
pass
mark(stamp)
sys.exit(0)
if os.path.exists(stamp):
sys.exit(0)

# a Read of something that is not source is not the decision this is about (a log, a lockfile, a README)
if tool == 'Read':
fp = inp.get('file_path') or ''
if not _where.is_source(_where._abs(fp, ev.get('cwd') or os.getcwd())):
# only a search has a pattern that can name what it is looking for; a Read of a chosen file or a Glob does not
here = ev.get('cwd') or os.getcwd()
if tool == 'Grep':
pat = inp.get('pattern') or ''
paths = [inp['path']] if inp.get('path') else []
elif tool == 'Bash':
cmd = inp.get('command') or ''
m = SEARCH.search(cmd) and PATTERN.search(cmd)
if not m:
sys.exit(0)

if tool == 'Bash' and not SEARCH.search(inp.get('command') or ''):
sys.exit(0)
# the same rule as a Read: a shell read or search whose every file is not source (`cat NOTES.md`, `tail run.log`)
# is not the decision this is about. Of the 3 times this hook spoke in 190 measured agents, one was a `cat` of a
# Markdown file. A search with no file argument (a directory, or none) still searches source.
if tool == 'Bash':
_, paths = _where.bash_where(inp.get('command'), ev.get('cwd') or os.getcwd())
# after a `|` a grep filters another command's output (`npm test | grep FAIL`), which is not a code search
seps = re.findall(r'\|\||&&|;|\n|\|', cmd[:m.start()])
if seps and seps[-1] == '|':
sys.exit(0)
# a search whose every file is not source (`grep x NOTES.md`, `rg y run.log`) is not the decision this is about
_, paths = _where.bash_where(cmd, here)
files = [p for p in paths if os.path.isfile(p)]
if files and not any(_where.is_source(p) for p in files):
sys.exit(0)
pat = m.group(3)
else:
sys.exit(0)

if os.path.exists(stamp):
names = identifiers(pat)
if not names:
sys.exit(0)
# the repository whose graph would answer: the one above the file or directory this tool is about to touch, not
# the session's working directory, which in 308 of 309 measured sessions held no graph (_where.py)
cwd = _where.locate(tool, inp, here, ev.get('session_id'))
if not cwd:
sys.exit(0)
real = os.path.realpath(cwd)
scope = [os.path.relpath(os.path.realpath(os.path.join(here, p)), real) for p in paths]
hits = declared(cwd, names, [s for s in scope if s != '.' and not s.startswith('..')])
if not hits:
sys.exit(0)
try:
os.makedirs(os.path.dirname(stamp), exist_ok=True)
open(stamp, 'w').close()
except Exception:
pass

fp = _where._abs(inp['file_path'], ev.get('cwd') or os.getcwd()) if tool == 'Read' and inp.get('file_path') else \
(next((p for p in files if _where.is_source(p)), '') if tool == 'Bash' else '')
try:
one = specific(cwd, tool, inp, fp)
except Exception:
one = ''
text = (one + '\n' + REST) if one else FULL
mark(stamp)
text = directive(hits)
# stream-json does not carry additionalContext, so a run cannot show from its transcript that this
# fired or what it said -- and #1100 is a question about exactly that. enrich.py already logs itself
# next to the graph for the same reason; this writes the same file, so one reader sees both halves.
try:
with open(os.path.join(cwd, '.axiomcode', 'hooks.jsonl'), 'a') as f:
f.write(json.dumps({'hook': 'direct', 'tool': tool, 'chars': len(text),
f.write(json.dumps({'hook': 'direct', 'tool': tool, 'chars': len(text), 'named': [h[1] for h in hits],
'input': {k: v for k, v in inp.items()
if k in ('file_path', 'pattern', 'command')}}) + '\n')
except OSError:
Expand Down
Loading
Loading