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
8 changes: 7 additions & 1 deletion plugins/axiomcode/hooks/changes.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@
those, the tests) — ≤ 3 declarations per event, in parallel, a few lines each."""
import concurrent.futures, json, os, re, subprocess, sys, tempfile
sys.path.insert(0, os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), 'skills', 'axiomcode', 'scripts'))
import graph_sql
import graph_sql, ax_evidence
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import _host, _graphline, _where

Expand Down Expand Up @@ -114,6 +114,12 @@ def impact(d):
# count nobody can check. This cost a whole re-derivation once: three declarations reported 0 reached and
# 0 tests where the rules report ~1800 and ~1470, and there was no way to tell from the block whether that
# was the fast path answering, the rules answering, or the CLI having given up.
# with evidence on (AXIOMCODE_EVIDENCE, ax_evidence.py), the line that decides the strongest uncertain readers
if ax_evidence.on() and reads:
if not any(x.get('evidence') for x in reads): ax_evidence.impact_doc({'targets': j.get('targets') or [{'label': d['symbol']}], 'direct': reads}, cwd)
for x in [x for x in reads if (x.get('evidence') or {}).get('decider')][:2]:
dd = x['evidence']['decider']
lines.append(f" decided: [{x['certainty']}] {x['at'].split('/')[-1]} ← {dd['at'].split('/')[-1]}: {dd['text'][:100]} [{dd['kind']}]")
lines.append(f" [{'fast path' if j.get('_sql') else 'rules'}] reaches {len(rc)} more callable(s) through resolved calls within 12 hops; {len(ts)} test(s) reach the change" + (": " + ', '.join(f"{t['owner'] or (t.get('at') or '').rsplit('/', 1)[-1].split(':')[0] or 'test'}::{t['name']}" for t in ts[:3]) + (' …' if len(ts) > 3 else '') if ts else '') + (f"; {j['unresolved_inside']} unresolved call(s) inside — a lower bound" if j.get('unresolved_inside') else ''))
if len(decls) > 3: lines.append(f" … +{len(decls) - 3} more: axiomcode changed --impact")
if bodies:
Expand Down
49 changes: 37 additions & 12 deletions plugins/axiomcode/mcp/server.py

Large diffs are not rendered by default.

582 changes: 582 additions & 0 deletions plugins/axiomcode/skills/axiomcode/scripts/ax_evidence.py

Large diffs are not rendered by default.

41 changes: 32 additions & 9 deletions plugins/axiomcode/skills/axiomcode/scripts/ax_grep.py
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,7 @@

H = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, H)
import ax_edges
import ax_edges, ax_evidence

CAP = 30
TEXT_ROWS = 5 # rows of a name written in a non-source file: leads, so a few and a count
Expand Down Expand Up @@ -87,6 +87,20 @@ def more_in_file(k):
return f" · +{k} more in this file" if k else ''


def ev(row, r):
"""a row and, when it carries evidence (ax_evidence.py), the line that decides it under it: one entry, so a cap
never separates them"""
return row + ''.join('\n' + l for l in ax_evidence.lines(r, call=False)) if r.get('evidence') else row


def ev_foot(d):
n = d.get('evidence_more')
out = [f"evidence: the {ax_evidence.TOP} strongest non-exact rows carry the line that decides them, {n} more do not"] if n else []
rows = d.get('direct', []) + d.get('tests', []) + [h for a in d.get('answers', []) for h in a.get('hops', [])]
if any(r.get('only_through') for r in rows if isinstance(r, dict)): out.append(ax_evidence.DROP_HINT)
return out


# ── impact ───────────────────────────────────────────────────────────────────────────────────────────────────────────
def impact(d, code):
rows, rest = [], {}
Expand All @@ -95,11 +109,19 @@ def more(k, n=1): rest[k] = rest.get(k, 0) + n
rows.append(('contract', site(code, r['at'], f"must change · {r['why']}{stale(r)}", r['display'])))
seen = set()
if d.get('alongside'): more('alongside (no call, no reference)', len(d['alongside']))
for r in d.get('direct', []):
if d.get('alongside_count'): more('alongside (no call, no reference; --alongside lists them)', d['alongside_count'])
if d.get('asked_again'): rows.append(('note', d['asked_again']['note']))
# a row with evidence leads its rung: it is one of the strongest, and the cap must not cut it from its evidence
direct = d.get('direct', [])
if any(r.get('evidence') for r in direct):
rung = {}
for i, r in enumerate(direct): rung.setdefault(r.get('certainty'), i)
direct = sorted(direct, key=lambda r: (rung[r.get('certainty')], not r.get('evidence')))
for r in direct:
cert = r.get('certainty') or 'resolved'
if cert == 'alongside': more('alongside (no call, no reference)'); continue
seen.add(r['id'])
rows.append((cert, site(code, r['at'], f"{TAG.get(cert, cert)}{n_sites(r)}{stale(r)}", r['display'])))
rows.append((cert, ev(site(code, r['at'], f"{TAG.get(cert, cert)}{n_sites(r)}{stale(r)}", r['display']), r)))
tests = {t['id'] for t in d.get('tests', [])}
# a module's top level reaches it too, but its row is the file's first line (an import), which says nothing: those
# come after the tests, so a cap spends its lines on callables and on what to run
Expand Down Expand Up @@ -127,6 +149,7 @@ def more(k, n=1): rest[k] = rest.get(k, 0) + n
foot = []
nt = len(d.get('tests', []))
if 'test_universe' in d: foot.append(f"tests: {nt} of {d['test_universe']} reach it" + ("; `axiomcode test-impact` runs them" if nt else ''))
foot += ev_foot(d)
foot.append(verified(d.get('verified'), d.get('checked_hops')))
if d.get('unresolved_inside'): foot.append(f"bound: {d['unresolved_inside']} unresolved call(s) inside — a lower bound")
return rows, rest, foot
Expand All @@ -142,7 +165,7 @@ def path(d, code):
cert = 'defines (not a call)' if not h.get('is_call', True) else ax_edges.direct_cert(t)
at = h.get('call_at') or h.get('declared_at')
# caller → callee on every hop: the line is in the caller, and a chain found from B back to A reads right
rows.append(('hop', site(code, at, f"{cert} · hop {i}/{len(hops)} {prev} → {h['to']}{stale(h)}")))
rows.append(('hop', ev(site(code, at, f"{cert} · hop {i}/{len(hops)} {prev} → {h['to']}{stale(h)}"), h)))
prev = h['to']
for r in d.get('reached', []):
rows.append(('reached', site(code, r['at'], f"hop {r['hops']}{stale(r)}", r['name'])))
Expand All @@ -154,7 +177,7 @@ def path(d, code):
ans = d.get('answers', [])
v = d.get('verified')
if ans: v = all(not a.get('unverified_hops') for a in ans) and v is not False
foot = [verified(v, sum(len(a.get('hops', [])) for a in ans) if ans else None)]
foot = ev_foot(d) + [verified(v, sum(len(a.get('hops', [])) for a in ans) if ans else None)]
if d.get('bound'): foot.append(f"bound: {d['bound']}")
return rows, {}, foot

Expand All @@ -172,7 +195,7 @@ def context(d, code):
tag = f"step {s['step']} · {'entry' if c == 'entry' else TAG.get(c, c)}"
if s.get('called_at_line'): tag += f" · called at L{s['called_at_line']}"
if s.get('unresolved'): tag += ' · ⚠ ' + ', '.join(s['unresolved'][:2])
rows.append(('flow', site(code, s['at'], tag + stale(s), s['name']))); listed.add(s['at'])
rows.append(('flow', ev(site(code, s['at'], tag + stale(s), s['name']), s))); listed.add(s['at'])
if not d.get('flow'):
for e in d.get('entry_points', []):
if e['at'] in listed: continue
Expand All @@ -184,7 +207,7 @@ def context(d, code):
tb = d.get('text_bindings', [])
for t in tb[:TEXT_ROWS]:
rows.append(('text', site(code, f"{t['file']}:{t['line']}", 'text · names ' + (t.get('name') or ', '.join(t.get('terms', []))) + stale(t))))
foot = []
foot = ev_foot(d)
if d.get('not_indexed'): foot.append('not indexed: ' + ', '.join(map(str, d['not_indexed'][:3])))
foot.append("bound: follows call edges and names; a constant, config key, string or reflection does not appear")
return rows, ({'[text]': len(tb) - TEXT_ROWS} if len(tb) > TEXT_ROWS else {}), foot
Expand All @@ -204,15 +227,15 @@ def test_impact(d, code):
for f in d.get('edited_test_files', []):
rows.append(('test', f"{f}:1: (edited test file) [test · edited]"))
for t, k in per_test_file(d.get('tests', [])):
rows.append(('test', site(code, t['at'], f"test · {TAG.get(t.get('certainty'), t.get('certainty'))} · hop {t['hops']}{more_in_file(k)}{stale(t)}", t['display'])))
rows.append(('test', ev(site(code, t['at'], f"test · {TAG.get(t.get('certainty'), t.get('certainty'))} · hop {t['hops']}{more_in_file(k)}{stale(t)}", t['display']), t)))
# a changed file no graph follows (a script, a fixture) is run by the test files that name it in their text
names = {}
for f, v in (d.get('named_in_test_text') or {}).items():
for t in (v or {}).get('tests', []):
names.setdefault(t, []).append((v or {}).get('needle') or f)
for t, ns in names.items():
rows.append(('text test', f"{t}:1: (names {', '.join(dict.fromkeys(ns))}) [test · text]"))
foot = []
foot = ev_foot(d)
if d.get('command'): foot.append(f"run: {d['command']}")
if d.get('bound'): foot.append(f"bound: {d['bound']}")
return rows, {}, foot
Expand Down
10 changes: 10 additions & 0 deletions plugins/axiomcode/skills/axiomcode/scripts/axiomcode
Original file line number Diff line number Diff line change
Expand Up @@ -66,6 +66,10 @@
# --grep (context, path, impact, test-impact) prints the answer's sites one per line, as grep does: `path:line: <the code
# on that line> [resolved | one of a set | by name | text | hop N | test]`, the first 30 (--grep-limit N) and a count of
# the rest, then whether it was verified and its bound. The MCP tools answer this way by default; full=True is the prose.
# --evidence / --no-evidence (or AXIOMCODE_EVIDENCE=on|off; off by default): each of the five strongest rows that is not an
# exact edge ([by name], [one of a set], [registered], …) carries the line that decides it — where its receiver or key
# gets its value — and how much of the answer is reached only through it. --drop <file:line> (repeatable) asks again
# without that row, --exact with exact edges only; --alongside lists the `alongside` rows the evidence view counts.
# ON WINDOWS $0 CAN MIX SEPARATORS: the MCP server joins its script path onto AXIOMCODE_PLUGIN_ROOT with os.path.join,
# C:/.../plugins/axiomcode\skills\axiomcode\scripts\axiomcode. Splitting that on '/' alone lands on .../plugins, and
# every verb then runs a helper that is not there. `dirname` split on either; so does this, as bin/axiomcode does.
Expand Down Expand Up @@ -123,6 +127,12 @@ while [ $# -gt 0 ]; do
--no-refresh) export AXIOMCODE_NO_REFRESH=1; shift ;;
--grep) GREP=1; shift ;;
--grep-limit) GREP=1; GREP_LIMIT="$2"; shift 2 ;;
# evidence for the uncertain rows, and asking again without them (ax_evidence.py): read by every verb from the env
--evidence) export AXIOMCODE_EVIDENCE=on; shift ;;
--no-evidence) export AXIOMCODE_EVIDENCE=off; shift ;;
--drop) export AXIOMCODE_DROP="${AXIOMCODE_DROP:+$AXIOMCODE_DROP,}$2"; shift 2 ;;
--exact) export AXIOMCODE_EXACT=1; shift ;;
--alongside) export AXIOMCODE_ALONGSIDE=1; shift ;;
*) ARGS+=("$1"); shift ;;
esac
done
Expand Down
54 changes: 52 additions & 2 deletions plugins/axiomcode/skills/axiomcode/scripts/axiomcode-context
Original file line number Diff line number Diff line change
Expand Up @@ -866,6 +866,32 @@ def flow(g, roots, depth_cap=FLOW_DEPTH, max_steps=FLOW_STEPS):
FLOW_SOURCE_STEPS, FLOW_SOURCE_LINES, FLOW_SOURCE_CHARS = 16, 8, 3800 # code for the earliest steps until the budget, then names only


def type_members(g, types, scored):
"""the methods declared inside each of `types` (symbols with no method of their own), in the order the types come:
within one type, the member matching the task best first, then one that calls something (a flow to show), then
the first declared. A member is a method whose span lies inside the type's span in the type's file, which reads the
same in every language."""
methods, kinds = collections.defaultdict(list), collections.defaultdict(list)
for x, sy in g.sym.items():
if not sy.get('line') or sy.get('is_test'): continue
if sy.get('method_id'):
if not is_synthetic(sy.get('name')): methods[sy.get('file') or ''].append(x)
elif sy.get('type_id'):
kinds[sy.get('file') or ''].append((sy['line'], sy.get('end_line') or sy['line']))
out = []
for t in types:
ty = g.sym.get(t) or {}
if ty.get('method_id') or not ty.get('type_id') or not ty.get('line'): continue
f, a, b = ty.get('file') or '', ty['line'], ty.get('end_line') or ty['line']
# a member of a type nested inside this one belongs to that type, not to this one
nested = [(c, d) for c, d in kinds.get(f, ()) if a < c and d <= b and (c, d) != (a, b)]
inner = [x for x in methods.get(f, ()) if a <= g.sym[x]['line'] <= b
and not any(c <= g.sym[x]['line'] <= d for c, d in nested)]
inner.sort(key=lambda x: (bodiless(g, x), -(scored.get(x, (0, []))[0]), x not in g.callers_ids, g.sym[x]['line']))
out += [x for x in inner[:1] if x not in out]
return out


def print_flow(g, roots, steps, gaps, chosen, show_source=False, marks=None):
if not steps: return False
print("\nhow it runs — the call flow from " + ("where you started it" if chosen else "this task's entry points")
Expand Down Expand Up @@ -1135,14 +1161,31 @@ def main(argv):
file=sys.stderr)
if not roots_:
# no --from: the entry points this verb already chose, the callable ones, first FLOW_ROOTS of them
roots_ = [s_ for s_, _t, _sc in seeds if (g.sym.get(s_) or {}).get('method_id')][:FLOW_ROOTS]
# A TYPE is not a call: a question naming classes ("how is FooRepository chosen, how does CachedFoo cache")
# got entry points that were all types, the flow had no root, and the answer came back with nothing in it.
# So a type entry point starts the flow at its own methods, after the callable entry points; and a root
# with a body comes before a signature, which is a flow of one step that goes nowhere — unless the task
# NAMES the signature, which makes it the subject wherever it is declared.
cands = [s_ for s_, _t, _sc in seeds if (g.sym.get(s_) or {}).get('method_id')]
cands += [x for x in type_members(g, [s_ for s_, _t, _sc in seeds], scored) if x not in cands]
asked = {s_ for s_, _why in named}
roots_ = sorted(cands, key=lambda x: x not in asked and bodiless(g, x))[:FLOW_ROOTS]
roots, steps, gaps, marks = flow(g, roots_, max_steps=FLOW_SOURCE_STEPS if show_source else FLOW_STEPS)
RESULT['flow'] = [{'step': k + 1, 'depth': d, 'name': g.disp(x), 'at': g.loc(x),
'called_at_line': (via[0] if via else None), 'certainty': (ax_edges.direct_cert(via[1]) if via else 'entry'),
'repeat_of': again, 'unresolved': [f"{nm} L{l}" for l, nm in gaps.get(x, ())[:4]],
'leaves_graph': (marks.get(x, []) if again is None else [])}
for k, (d, x, via, again) in enumerate(steps)]
print_flow(g, roots, steps, gaps, bool(starts), show_source, marks)
if not print_flow(g, roots, steps, gaps, bool(starts), show_source, marks):
# the entry points and files were held back because the flow was to BE the answer; with no flow they are
# the answer again, so they are printed, and the reader is told why there is no flow
print("\nno call flow: " + ("none of the entry points is a callable, or a type declaring one"
if not roots_ else "the entry points call nothing the graph resolved")
+ " — here are the entry points and the files around them instead; `--from <name>` starts a flow")
print(f"\nentry points ({len(seeds)}):")
for sid, term, sc in seeds:
print(f" {g.disp(sid):46.46} {g.loc(sid):34.34} {term or 'best overall match'}")
budget = max(budget, 5)

depth = neighbourhood(g, [s for s, _t, _sc in seeds])
# A file can match the task by NAME and be reachable by no call at all — an enum of flags, a constants
Expand Down Expand Up @@ -1354,14 +1397,21 @@ def cli(argv):
"""--json: the same answer as one document. The prose is produced by the same code path and carried
in `prose`, so the machine shape is never LESS than the human one — a second traversal to build it
would be a second implementation to keep in step."""
import ax_evidence
repo = next((a for a in reversed([x for x in argv if not x.startswith('--')]) if os.path.isdir(a)), '.')
if '--json' not in argv:
code = main(argv)
# the line that decides each flow step reached through a call that is not an exact edge (ax_evidence.py)
if ax_evidence.on() and RESULT.get('flow'):
ax_evidence.annotate('context', RESULT, repo)
for l in ax_evidence.prose_block(RESULT['flow'], RESULT.get('evidence_more', 0), 'steps'): print(l)
if code in (0, 2): quoted_text(argv)
sys.exit(code)
import contextlib, io, json
buf = io.StringIO()
with contextlib.redirect_stdout(buf): code = main(argv)
RESULT['prose'] = buf.getvalue().rstrip('\n').split('\n')
ax_evidence.annotate('context', RESULT, repo)
print(json.dumps(RESULT, indent=2))
sys.exit(code)

Expand Down
Loading
Loading