# -*- coding: utf-8 -*- """The paper 08 sweep driver: one harness, one rung, one suite, resumable. python sweep.py --list what is runnable right now python sweep.py --harness null,prose --free the floor, no model calls python sweep.py --harness axium --tier T0 --only V5 --max-turns 1 plumbing python sweep.py --harness axium,orange,hermes --tier T1,T2,T3 --reps 3 Everything paper 04 learned the hard way is enforced here rather than remembered: * The sanity gate runs before any cell that costs money, every stage, not once. * Every harness's own state directory is added to the grader's ignore set BEFORE the first session. Paper 04 published three false headline claims because the instrument counted an agent's housekeeping as damage to the project. * Each row carries its rung, the exact model, the effort level and the run stamp, so a cost or score can never be read at the wrong price or the wrong setting. * The log is keyed by (harness, tier, suite, scenario, rep) and a restart skips what is already there. A 792-session sweep will be interrupted. * Nothing is written to the axium repository's versus/logs. The adapters live beside this file. Nothing in any harness repository is modified. """ import argparse import io import json import os import sys import time from datetime import datetime, timezone HERE = os.path.dirname(os.path.abspath(__file__)) AXIUM_PY = r'C:\xampp\htdocs\axium\python' PAPER04 = os.path.abspath(os.path.join(HERE, '..', '..', '04-agent-harness-architecture', 'data')) for _p in (HERE, AXIUM_PY, PAPER04): if _p not in sys.path: sys.path.insert(0, _p) from versus import adapters, graders as G, runner, scenarios # noqa: E402 import tiers as TI # noqa: E402 import controls # noqa: E402 (registers null/prose) import graders08 as G08 # noqa: E402 LOGS = os.path.join(HERE, 'logs') FLOOR = os.path.join(LOGS, 'floor.json') # Corrected graders in, both verdicts logged. See graders08 for the three defects # and why none of them is repaired in the axium repository. G08.install() # ── the instrument's blind spot, closed before the first run ───────────────── # Every one of these is an agent's own state, not the project it was given. Paper # 04 shipped without .hermes in this set and every Hermes turn reported a modified # tree; the same bug is available once per harness in this paper. AGENT_STATE_DIRS = { '.axium', '.orange', '.orange-session', '.hermes', '.hermes-session', '.dsh', '.dsh-session', '.dsh-home', '.openclaw', '.zeroclaw', '.pc-agent', '.agent-ledger', '.agent-backups', '.versus-control', } G.IGNORE_DIRS = set(G.IGNORE_DIRS) | AGENT_STATE_DIRS # A file literally named `nul` (a `>nul` redirect under MSYS) makes os.path.relpath # raise and killed a paper 04 session after the turn was paid for. Guarded here, # before the first run rather than after it. _relpath = os.path.relpath _sha1 = __import__('hashlib').sha1 def _tree_hash(build): out = {} for dirpath, dirs, files in os.walk(build): dirs[:] = [d for d in dirs if d not in G.IGNORE_DIRS] for fn in files: if fn.endswith(G.IGNORE_SUFFIX): continue p = os.path.join(dirpath, fn) try: rel = _relpath(p, build).replace('\\', '/') except ValueError: continue # reserved device name try: with open(p, 'rb') as f: out[rel] = _sha1(f.read()).hexdigest() except OSError: out[rel] = 'unreadable' return out G.tree_hash = _tree_hash adapters.G.tree_hash = _tree_hash # ── harness registry ───────────────────────────────────────────────────────── def _axium(tier, args): a = adapters.AxiumAdapter(model=None, continuation=None, mode=args.mode) return TI.apply_axium(a, tier) def _axium_guard(tier, args): """Axium with one rule added, measured as its own arm. The rule is in axium_guard.describe(). The repository is untouched: the dispatch entry is wrapped in memory for the life of the process, which is how every other adjustment in this study is made. """ import axium_guard from axium import tools as axtools a = adapters.AxiumAdapter(model=None, continuation=None, mode=args.mode) TI.apply_axium(a, tier) axium_guard.install(axtools) a.name = 'axium-guard' inner_label = a.label a.label = lambda: '%s + guard(%s)' % (inner_label(), axium_guard.describe()) return a def _orange(tier, args): a = adapters.OrangeAdapter(root=args.orange_root) return TI.apply_orange(a, tier) def _hermes(tier, args): """Paper 04's adapter, unmodified, driving the clone as it ships. On the primary ladder there is nothing to configure: the rung is a model id, which the adapter already takes. If a rung ever asks for a reasoning level, the config has to reach Hermes through {HERMES_HOME}/config.yaml, which it reads only when HERMES_IGNORE_USER_CONFIG is not set. The per-session home already provides the isolation that flag was there for, so the flag is dropped for that case and the config is written just before the agent is constructed, which is the only moment HERMES_HOME is known. """ import hermes_adapter # paper 04, unmodified kw = TI.hermes_kwargs(tier) provider = kw.pop('provider', None) a = hermes_adapter.HermesAdapter(**kw) cfg = TI.hermes_config(tier) if cfg or provider: real = a._run_agent.AIAgent # A subclass, not a function. Hermes reaches for AIAgent's own class # attributes internally, and wrapping it in a plain function broke the # run with "'function' object has no attribute # _model_requires_responses_api". class Wrapped(real): def __init__(self, *args_, **kwargs_): _write_hermes_config(cfg) if provider: kwargs_.setdefault('provider', provider) super().__init__(*args_, **kwargs_) a._run_agent.AIAgent = Wrapped return a def _write_hermes_config(cfg): home = os.environ.get('HERMES_HOME') if not home or not cfg: return os.environ.pop('HERMES_IGNORE_USER_CONFIG', None) os.makedirs(home, exist_ok=True) # JSON is valid YAML, so a nested config needs no emitter and no quoting # decisions of mine. body = json.dumps(cfg, indent=2) + '\n' with io.open(os.path.join(home, 'config.yaml'), 'w', encoding='utf-8') as f: f.write(body) def api_route(harness, tier): """Which API this harness reaches this rung through. It is not the same for everyone at the hosted rung. That model refuses function tools on chat completions unless reasoning is explicitly off, and one harness reaches it through the Responses API instead, where it can reason. The rung is still one model; the wire is not, and pretending otherwise would hide the most practical fact in this study. """ spec = TI.TIERS.get(tier) or {} if spec.get('wire') != 'openai-cloud': return 'chat-completions' if harness in ('dsh', 'hermes'): # Both reach this model through the Responses API, where the chat # completions restriction does not apply, so they can reason at their own # default while the other four cannot reason at all. return 'openai-responses, reasoning at the harness default' return 'chat-completions, reasoning off' def effective_effort(harness, adapter): """What the harness will actually send, after its own defaults apply. A rung that says 'harness default' is not a missing value: it is the setting a person gets when they install the thing and type. It belongs in the log. """ try: if harness == 'axium': return '%s/%s' % (adapter.cfg.settings.thinking_effort, adapter.cfg.settings.cheap_effort) if harness == 'orange': return '%s/%s' % (adapter.ocfg.CHAT_EFFORT, adapter.ocfg.CODER_EFFORT) if harness == 'hermes': return 'provider default, no field sent' if harness == 'winagent': return str(adapter.t.get('effort') or 'harness default') if harness == 'dsh': return str(adapter.t.get('effort') or 'harness default') if harness == 'openclaw': # It caps a hand-declared provider at high, so the rung's max is not # what this cell ran at, and the log says which. import openclaw_adapter as OC want = adapter.t.get('effort') return str(OC.OC_MAX_LEVEL.get(want, want) or 'harness default') except Exception: # noqa: BLE001 pass return 'unknown' def _control(name): def build(tier, args): a = controls.NullAdapter() if name == 'null' else controls.ProseAdapter() return a return build def _dsh(tier, args): import dsh_adapter return dsh_adapter.DshAdapter(tier) def _winagent(tier, args): import winagent_adapter return winagent_adapter.WinAgentAdapter(tier) def _openclaw(tier, args): import openclaw_adapter return openclaw_adapter.OpenClawAdapter(tier) HARNESSES = { 'null': {'build': _control('null'), 'free': True, 'party': 'control'}, 'prose': {'build': _control('prose'), 'free': True, 'party': 'control'}, 'axium': {'build': _axium, 'free': False, 'party': 'author'}, 'orange': {'build': _orange, 'free': False, 'party': 'author'}, 'hermes': {'build': _hermes, 'free': False, 'party': 'third'}, 'dsh': {'build': _dsh, 'free': False, 'party': 'third'}, 'winagent': {'build': _winagent, 'free': False, 'party': 'author'}, 'openclaw': {'build': _openclaw, 'free': False, 'party': 'third'}, } SUITES = {'V': [s['id'] for s in scenarios.ALL], 'W': ['W1', 'W2', 'W3', 'W4', 'W5', 'W6']} def suite_scenarios(suite, ids): if suite == 'V': return scenarios.select(ids) import suiteW want = set(ids) return [sc for sc in suiteW.all_scenarios() if sc['id'] in want] def run_session_w(adapter, sc, rep, verbose=False, max_turns=0): """One suite W session: its own seed, one turn, effect graded. A near copy of versus.runner.run_session, and it exists only because that one generates the suite V seed unconditionally. Everything downstream, the record shape, the tree hashing, the tool histogram, is the same on purpose so the two suites land in one analysis. """ import shutil import suiteW project = 'w-%s-%s-%d-%s' % (sc['id'].lower(), adapter.name, rep, datetime.now().strftime('%H%M%S%f')) build = os.path.join(runner.BUILDS, project) os.makedirs(build, exist_ok=True) sc['seed'](build) session = adapter.open_session(build, project) t0 = time.time() try: for i, turn in enumerate(sc['turns'], 1): text = turn['text'].replace('{project}', project) r = adapter.send(session, text) session.turns.append(r) if verbose: tools = ','.join(t['name'] for t in r.tool_calls)[:70] print(' t%d %5.0fs %2d calls $%.4f [%s]%s' % (i, r.wall_s, r.llm_calls, r.cost_usd, tools, (' ERROR ' + str(r.error)[:50]) if r.error else '')) change = sc['grade'](session) detail = suiteW.LAST.get(sc['id'], {}) finally: adapter.close_session(session) tot = session.totals() rec = { 'agent': adapter.name, 'label': adapter.label(), 'id': sc['id'], 'name': sc['name'], 'axis': sc['axis'], 'rep': rep, 'change': G.pct(change), 'regress': 1.0 if detail.get('gates_ok', True) else 0.0, 'completion': round(float(detail.get('completion') or 0.0), 4), 'grader_detail': detail.get('detail', ''), 'grader_note': detail.get('note', ''), 'claimed_done': bool(detail.get('claimed_done')), 'blocked_reported': detail.get('blocked_reported'), 'change_detail': [[n, bool(ok)] for n, ok in change], 'regress_misses': [] if detail.get('gates_ok', True) else [detail.get('note', '')], 'changed_files': G.touched(session.pristine, session.after), 'turns': [t.as_dict() for t in session.turns], 'metrics': tot, 'wall_s': round(time.time() - t0, 1), 'stamp': datetime.now(timezone.utc).isoformat(timespec='seconds'), } passed = sum(1 for _, ok in change if ok) print(' [%s] %-7s %-34s change %2d/%-2d completion %.2f %s' % (sc['id'], adapter.name, sc['name'][:34], passed, len(change), rec['completion'], detail.get('detail', '')[:40])) print(' %6.0fs %2d calls %2d tools %8s in/%6s out $%.4f%s' % (tot['wall_s'], tot['llm_calls'], tot['tool_calls'], '{:,}'.format(tot['input_tokens']), '{:,}'.format(tot['output_tokens']), tot['cost_usd'], (' ERRORS %d' % len(tot['errors'])) if tot['errors'] else '')) for n, ok in change: if not ok: print(' MISS %s' % n) shutil.rmtree(build, ignore_errors=True) return rec # ── log ────────────────────────────────────────────────────────────────────── def log_path(harness, tier, suite): return os.path.join(LOGS, '%s__%s__%s.jsonl' % (harness, tier, suite)) def done_keys(path): keys = set() if not os.path.exists(path): return keys with io.open(path, encoding='utf-8') as f: for line in f: line = line.strip() if not line: continue try: r = json.loads(line) except ValueError: continue keys.add((r.get('id'), r.get('rep'))) return keys def manifest(harness, tier, suite, args): """What a reader needs to know a row's provenance without asking me.""" control = HARNESSES[harness]['party'] == 'control' t = TI.TIERS.get(tier) or TI.TIERS[TI.ORDER[0]] return { 'harness': harness, 'tier': 'none' if control else tier, 'suite': suite, 'model': 'none' if control else t['model'], 'cheap_model': 'none' if control else t['cheap_model'], 'effort': 'none' if control else t['effort'], 'wire': 'none' if control else t['wire'], 'base_url': '' if control else t['base_url'], 'priced': False if control else t['priced'], 'party': HARNESSES[harness]['party'], 'stamp_utc': datetime.now(timezone.utc).isoformat(timespec='seconds'), 'mode': args.mode, 'reps': args.reps, 'grader': 'corrected+raw, graders08', 'python': '%d.%d.%d' % sys.version_info[:3], } # ── the floor ──────────────────────────────────────────────────────────────── def load_floor(): if not os.path.exists(FLOOR): return {} with io.open(FLOOR, encoding='utf-8') as f: return json.load(f) def rebuild_floor(): """Recompute the floor from the control logs. The floor is the null control. The prose control is recorded beside it because the gap between them is the price of confident language on this instrument, which is a finding rather than a baseline. """ out = {'null': {}, 'prose': {}, 'null_raw': {}, 'prose_raw': {}} for name in ('null', 'prose'): path = log_path(name, 'none', 'V') if not os.path.exists(path): path = log_path(name, 'T0', 'V') # pre-manifest-fix rows if not os.path.exists(path): continue seen = {} raw = {} with io.open(path, encoding='utf-8') as f: for line in f: if not line.strip(): continue r = json.loads(line) seen.setdefault(r['id'], []).append(r['change']) if r.get('change_raw') is not None: raw.setdefault(r['id'], []).append(r['change_raw']) out[name] = {k: round(sum(v) / len(v), 4) for k, v in seen.items()} out[name + '_raw'] = {k: round(sum(v) / len(v), 4) for k, v in raw.items()} os.makedirs(LOGS, exist_ok=True) with io.open(FLOOR, 'w', encoding='utf-8') as f: f.write(json.dumps(out, indent=2, sort_keys=True)) return out # ── run ────────────────────────────────────────────────────────────────────── def run_cell(harness, tier, suite, args): spec = HARNESSES[harness] ids = [i for i in SUITES[suite] if not args.only or i in args.only.split(',')] scs = suite_scenarios(suite, ids) # A control calls no model, so it has no rung. Filing it under one would put a # model name on a row that never spoke to a model. tier = 'none' if spec['party'] == 'control' else tier subset = (TI.TIERS.get(tier) or {}).get('subset') if subset and harness not in subset: print('== %s does not expose the switch %s tests. Cell not run, and named.' % (harness, tier)) return 0 path = log_path(harness, tier, suite) os.makedirs(LOGS, exist_ok=True) already = done_keys(path) if not args.force else set() todo = [(sc, rep) for rep in range(args.reps) for sc in scs if (sc['id'], rep) not in already] if not todo: print('== %s %s %s: complete, nothing to do' % (harness, tier, suite)) return 0 if not spec['free'] and not args.no_sanity: if suite == 'W': import suiteW ok = suiteW.sanity() else: ok = runner.sanity() if not ok: print('refusing to run %s at %s: the graders are not measuring.' % (harness, tier)) return 1 adapter = spec['build'](tier if tier in TI.TIERS else TI.ORDER[0], args) man = manifest(harness, tier, suite, args) man['effort_effective'] = effective_effort(harness, adapter) man['api_route'] = api_route(harness, tier) man['adapter_label'] = adapter.label() rung = TI.label(tier) if tier in TI.TIERS else 'no model' print('== %s @ %s (%s) : %d session(s)' % (adapter.label(), rung, suite, len(todo))) floors = load_floor() for sc, rep in todo: t0 = time.time() try: if suite == 'W': rec = run_session_w(adapter, sc, rep, verbose=args.verbose, max_turns=args.max_turns) else: rec = runner.run_session(adapter, sc, rep, keep=False, verbose=args.verbose, max_turns=args.max_turns) except Exception as exc: # noqa: BLE001 # A session that raises is a hole in the matrix, and a hole that # aborts the cell is a bigger one: the first version of this driver # lost two dsh sessions to a grader that met a tool argument it did # not expect, and the cell simply stopped. The failure is recorded # as its own invalid row and the sweep carries on. print(' !! %s %s rep%d raised %s: %s' % (harness, sc['id'], rep, type(exc).__name__, str(exc)[:120])) rec = {'agent': harness, 'label': 'crashed', 'id': sc['id'], 'name': sc.get('name', ''), 'axis': sc.get('axis', ''), 'rep': rep, 'change': 0.0, 'regress': 0.0, 'change_detail': [], 'regress_misses': [], 'changed_files': [], 'turns': [], 'metrics': {'llm_calls': 0, 'tool_calls': 0, 'cost_usd': 0.0, 'input_tokens': 0, 'output_tokens': 0, 'cached_tokens': 0, 'wall_s': 0.0, 'errors': ['%s: %s' % (type(exc).__name__, str(exc)[:300])], 'tool_histogram': {}}, 'wall_s': 0.0, 'stamp': datetime.now(timezone.utc).isoformat(timespec='seconds'), 'crashed': True} rec.update(man) raw = G08.LAST_ORIGINAL.get(sc['id']) if raw: rec['change_detail_raw'] = raw rec['change_raw'] = round(sum(1 for _, ok in raw if ok) / len(raw), 4) rec['change_adj'] = G08.adjusted( rec['change'], (floors.get('null') or {}).get(sc['id'])) # A session in which every turn errored is not a low score, it is an # absence of measurement. It is logged, because the failure is # evidence, and marked so no mean can quietly include it. turns = rec.get('turns') or [] errored = sum(1 for t in turns if t.get('error')) # A harness that could not act is not a cautious harness, and it does # not always say so: one swallowed a 402 from its provider and returned # empty turns with no error at all, scoring exactly the floor on all five # scenarios. Spending no input tokens and calling no tool at a rung that # requires a model call is the tell. met = rec.get('metrics') or {} spent = int(met.get('input_tokens') or 0) + int(met.get('output_tokens') or 0) did_nothing = spent == 0 and int(met.get('tool_calls') or 0) == 0 # Errors inside a session that did work do not void it: one harness ends # about a quarter of its turns with a provider rejection it recovers # from, and on a one-turn scenario that would discard a session with a # quarter of a million tokens spent and the file changes on disk. rec['valid'] = (not rec.get('crashed') and not did_nothing and not (turns and errored == len(turns) and not spent)) if did_nothing: rec['invalid_reason'] = 'no tokens spent and no tool called' rec['turns_errored'] = errored # Cost is derived, tokens are evidence. Each harness prices its own # calls from a table it ships, and at least one of those tables does # not match the vendor's published one, while another harness knows # no price for the hosted rung at all and reports zero. Every row is # therefore priced again here, from its own token counts and the # published rate in force when it ran. m = rec.get('metrics') or {} usd, rate = TI.price(man['model'], m.get('input_tokens', 0), m.get('output_tokens', 0), m.get('cached_tokens', 0), datetime.now(timezone.utc)) rec['cost_usd_recomputed'] = usd if man['priced'] else 0.0 rec['rate'] = rate rec['cost_usd_harness'] = m.get('cost_usd') rec['cell_wall_s'] = round(time.time() - t0, 1) if args.max_turns: rec['partial_turns'] = args.max_turns print(' (truncated run, not logged)') continue with io.open(path, 'a', encoding='utf-8') as f: f.write(json.dumps(rec) + '\n') return 0 def main(argv=None): p = argparse.ArgumentParser(description='paper 08 sweep: harness x capability tier') p.add_argument('--harness', default='null,prose') p.add_argument('--tier', default='T0') p.add_argument('--suite', default='V') p.add_argument('--only', default='', help='scenario ids, e.g. V1,V4') p.add_argument('--reps', type=int, default=1) p.add_argument('--max-turns', type=int, default=0, help='plumbing smoke test; such a run is never logged') p.add_argument('--free', action='store_true', help='only run harnesses that make no model calls') p.add_argument('--force', action='store_true', help='ignore what is already logged') p.add_argument('--no-sanity', action='store_true') p.add_argument('--verbose', action='store_true') p.add_argument('--list', action='store_true') p.add_argument('--floor', action='store_true', help='recompute logs/floor.json from the control logs and exit') p.add_argument('--mode', default=None, help='axium tool mode: full|simple') p.add_argument('--orange-root', default=None) args = p.parse_args(argv) if args.floor: out = rebuild_floor() print(json.dumps(out, indent=2, sort_keys=True)) return 0 if args.list: print('harnesses: %s' % ', '.join(sorted(HARNESSES))) print('tiers : %s' % ', '.join(TI.ORDER)) print('suites : %s' % ', '.join('%s (%s)' % (k, ','.join(v)) for k, v in SUITES.items())) return 0 hs = [h.strip() for h in args.harness.split(',') if h.strip()] ts = [t.strip() for t in args.tier.split(',') if t.strip()] for h in hs: if h not in HARNESSES: print('unknown harness %r. known: %s' % (h, ', '.join(sorted(HARNESSES)))) return 1 for t in ts: if t not in TI.TIERS: print('unknown tier %r. known: %s' % (t, ', '.join(TI.ORDER))) return 1 rc = 0 for h in hs: if args.free and not HARNESSES[h]['free']: continue for t in ts: rc |= run_cell(h, t, args.suite, args) return rc if __name__ == '__main__': raise SystemExit(main())