#!/usr/bin/env python3 """Trajectory analyzer for pi-skill-lab eval transcripts. Parses pi.jsonl transcripts, reconstructs per-run tool-call sequences, detects failed / retried commands (output-text heuristics — pi's bash tool returns no exit codes), and correlates with grading.json outcomes. Outputs: 1. per-run summary (effort + failure counts + doc-read order) 2. aggregated failure signatures per config 3. first-attempt success rate for the core CLI verbs """ import json, glob, os, re, sys from collections import defaultdict # --- failure heuristics on tool OUTPUT text (stdout only in pi) ------------- # Strict failure patterns: an output line is a FAILURE only when it looks like # an actual error line — NOT when the word "failed"/"error" appears in normal # status output (e.g. `runs list` printing status=failed, or "error": null). FAIL_LINES = [ r'^\s*[✗✘❌]', r'^\s*error\b', r'\bError:', r'Validation error', r'command not found', r'ENOENT', r'ECONNREFUSED', r'ETIMEDOUT', r'unauthorized', r'\bforbidden\b', r'Missing required', r'unknown (option|command)', r'is required', r'HTTP_?[45]\d\d', r'Cannot (GET|POST|PUT|PATCH|DELETE)\b', r'"status":\s*[45]\d\d', r'Invalid (option|value|input|RecordId|List)', r'Unsupported library type', r'spawn npm ENOENT', ] FAIL_RE = re.compile('|'.join(FAIL_LINES)) # outputs that legitimately contain these words but are NOT failures OK_RE = re.compile(r'^(id|name|baseId|workflowId|triggerType|config|next|note|\s*$)', re.M) def classify_output(text, cmd=''): """Return failure signature (short str) or None if output looks successful.""" if not text: return None # `--help` legitimately prints "Usage:" — not a failure is_help = '--help' in cmd head = text[:400] if is_help and re.match(r'\s*(Usage:|inkreach-teable \[options\])', text): return None for ln in text.splitlines(): ln = ln.strip() if ln and FAIL_RE.search(ln): if is_help and ln.startswith('Usage:'): continue return ln[:140] return None def cmd_of(args): if not isinstance(args, dict): return '' return args.get('command') or args.get('path') or '' def load_run(pi_path): """Return ordered tool calls [(tool, cmd, out_head)], plus meta.""" calls, meta = [], {'turns': 0, 'settled': False} starts = {} for line in open(pi_path): try: e = json.loads(line) except Exception: continue t = e.get('type') if t == 'tool_execution_start': starts[e['toolCallId']] = e elif t == 'tool_execution_end': s = starts.get(e['toolCallId'], {}) args = s.get('args', {}) r = e.get('result') or {} c = r.get('content') txt = '' if isinstance(c, list): txt = '\n'.join(x.get('text', '') for x in c if isinstance(x, dict)) elif isinstance(c, str): txt = c calls.append((e['toolName'], cmd_of(args), txt[:600])) elif t == 'turn_start': meta['turns'] += 1 elif t == 'agent_settled': meta['settled'] = True return calls, meta def cli_verb(cmd): m = re.search(r'(?:^|\s|&&|;|npx\s+(?:git+\S+\s+)?)inkreach-teable\s+([a-z-]+)(?:\s+([a-z-]+))?', cmd) if not m: return None return '.'.join(x for x in m.groups() if x) def analyze(pi_path): calls, meta = load_run(pi_path) run = { 'path': pi_path, 'total': len(calls), 'bash': sum(1 for t, c, o in calls if t == 'bash'), 'reads': sum(1 for t, c, o in calls if t == 'read'), 'turns': meta['turns'], 'settled': meta['settled'], 'cli': [], # (verb, cmd, fail_sig_or_None) 'other_fail': [], # non-CLI failed bash (npm, curl, node...) 'docs': [], # skill files read, in order } seen_verbs = {} for tool, cmd, out in calls: if tool == 'read': if '/skills/' in cmd and cmd.endswith('.md'): run['docs'].append(cmd.split('/skills/')[-1]) continue if tool != 'bash': continue sig = classify_output(out, cmd) v = cli_verb(cmd) if v: run['cli'].append((v, cmd, sig)) elif sig: run['other_fail'].append((cmd[:100], sig)) return run def attempts_to_success(cli_calls): """Per verb: list of (#failed attempts before first success, succeeded?).""" stats = defaultdict(lambda: {'fails_before': [], 'never': 0, 'first_try': 0, 'total_sessions': 0}) cur = defaultdict(lambda: {'fails': 0, 'succeeded': False}) for verb, cmd, sig in cli_calls: st = cur[verb] if sig is None: if not st['succeeded']: stats[verb]['total_sessions'] += 1 stats[verb]['fails_before'].append(st['fails']) if st['fails'] == 0: stats[verb]['first_try'] += 1 st['succeeded'] = True st['fails'] = 0 else: if not st['succeeded']: st['fails'] += 1 # verbs that never succeeded for verb, st in cur.items(): if not st['succeeded'] and st['fails']: stats[verb]['total_sessions'] += 1 stats[verb]['never'] += 1 return stats def main(): rows = [] for ws in ['inkreach-teable-workspace', 'plugin-workspace']: for pi in sorted(glob.glob(f'/root/pi-skill-lab/{ws}/iteration-*/eval-*/*/run-*/pi.jsonl')): parts = pi.split('/') run = analyze(pi) gp = os.path.join(os.path.dirname(pi), 'grading.json') passed = None if os.path.exists(gp): try: g = json.load(open(gp)) exps = g.get('expectations') or [] if exps: passed = all(x.get('passed') for x in exps) else: passed = g.get('passed') except Exception: pass rows.append({ 'ws': ws.replace('-workspace',''), 'iter': parts[4], 'eval': parts[5], 'cfg': parts[6], 'run': parts[7].replace('run-',''), 'passed': passed, **{k: run[k] for k in ('total','bash','reads','turns','settled','cli','other_fail','docs')}, }) # ---- 1. per-run summary ---- print('=' * 110) print('PER-RUN TRAJECTORY SUMMARY') print('=' * 110) hdr = f"{'eval':38} {'cfg':12} {'ok':4} {'tcs':4} {'rd':3} {'trn':4} {'cli':4} {'f':3} {'retry':5} docs" print(hdr) for r in rows: fails = sum(1 for v,c,s in r['cli'] if s) retries = sum(1 for v,c,s in r['cli'] if s) # failed call == at least one retry if task later passed cli_ok = 'Y' if r['cli'] else '-' print(f"{r['eval'][:38]:38} {r['cfg']:12} {str(r['passed'])[:3]:4} {r['total']:4} {r['reads']:3} {r['turns']:4} " f"{len(r['cli']):4} {fails:3} {retries:5} {len(r['docs'])}d {'>' if not r['settled'] else ''}") # ---- 2. failure signatures ---- print() print('=' * 110) print('FAILURE SIGNATURES (failed CLI / other-bash calls, per config)') print('=' * 110) sigs = defaultdict(list) for r in rows: for verb, cmd, sig in r['cli']: if sig: sigs[(r['cfg'], verb, sig[:70])].append(r['eval'][:30]) for cmd, sig in r['other_fail']: sigs[(r['cfg'], 'bash:'+cmd[:40], sig[:70])].append(r['eval'][:30]) for (cfg, verb, sig), evals in sorted(sigs.items()): print(f"[{cfg:12}] {verb:32} n={len(evals):2} {sig}") # ---- 3. first-attempt success per verb, per config ---- print() print('=' * 110) print('ATTEMPTS-TO-SUCCESS PER CLI VERB (per config)') print('=' * 110) agg = defaultdict(lambda: defaultdict(lambda: {'sess':0,'first':0,'never':0,'retry_hist':[]})) for r in rows: st = attempts_to_success(r['cli']) for verb, s in st.items(): a = agg[r['cfg']][verb] a['sess'] += s['total_sessions']; a['first'] += s['first_try']; a['never'] += s['never'] a['retry_hist'] += [x for x in s['fails_before'] if x] for cfg in agg: print(f'--- {cfg} ---') for verb, a in sorted(agg[cfg].items(), key=lambda kv: -kv[1]['sess']): ft = f"{a['first']}/{a['sess']}" if a['sess'] else '-' print(f" {verb:34} sessions={a['sess']:3} first-try={ft:7} never-ok={a['never']:2} retry-dist={sorted(set(a['retry_hist'])) if a['retry_hist'] else []}") # ---- 4. failed runs: what happened ---- print() print('=' * 110) print('FAILED/TIMEOUT RUNS — trajectory tail') print('=' * 110) for r in rows: if r['passed'] is False or not r['settled']: fails = [(v, s) for v, c, s in r['cli'] if s] + [(c[:50], s) for c, s in r['other_fail']] print(f"\n{r['ws']} {r['iter']}/{r['eval']}/{r['cfg']} passed={r['passed']} settled={r['settled']}") for v, s in fails[-6:]: print(f" FAIL {v}: {s}") if __name__ == '__main__': main()