# -*- coding: utf-8 -*- """versus adapter for OpenClaw 2026.7.1-2, the most adopted harness in the set. One subprocess per turn, all turns sharing one `--session-id`, which is how OpenClaw itself carries a conversation: its session store is the memory, and the CLI is a client of it. That makes the multi-turn scenarios honest without any scaffolding beyond passing the same id. Two things had to be arranged, both recorded because both change what is measured: 1. **Workspace, not working directory.** OpenClaw's tools operate on the configured workspace and ignore the process cwd. Measured, not assumed: asked to read a file sitting in the process cwd it answered NOTFOUND and named the workspace it had searched instead. So the workspace is pointed at the build for the life of each session, which is the same treatment every other harness gets by being started inside it. 2. **A provider entry for the local rung.** The DeepSeek rungs use the provider plugin its own onboarding installed. The floor rung is a hand-declared OpenAI-compatible route to the local server, written into the same config the onboarding wrote. Tool calls and their arguments come from OpenClaw's own session log rather than from the JSON summary, which reports only the names and a count. Token usage comes from the command's JSON reply. Nothing in the OpenClaw package is modified. The home directory is scaffolding created for this paper and is published with it. """ import io import json import os import subprocess import time from datetime import datetime, timezone import tiers as TI from versus import adapters, graders as G OC_ROOT = r'C:\agent-eval-thirdparty\openclaw' OC_BIN = os.path.join(OC_ROOT, 'node_modules', 'openclaw', 'openclaw.mjs') OC_HOME = r'C:\agent-eval-thirdparty\openclaw-home' CONFIG = os.path.join(OC_HOME, '.openclaw', 'openclaw.json') SESSIONS = os.path.join(OC_HOME, '.openclaw', 'agents', 'main', 'sessions') NODE = 'node' TURN_TIMEOUT_S = int(os.environ.get('OPENCLAW_TURN_TIMEOUT', '600')) LOCAL_PROVIDER = 'ollama-local' CLOUD_PROVIDER = 'openai-chat' # What this harness will accept on a hand-declared provider. OC_MAX_LEVEL = {'max': 'high', 'xhigh': 'high'} def _load_config(): with io.open(CONFIG, encoding='utf-8') as f: return json.load(f) def _save_config(cfg): tmp = CONFIG + '.tmp' with io.open(tmp, 'w', encoding='utf-8') as f: json.dump(cfg, f, indent=1) os.replace(tmp, CONFIG) def ensure_local_provider(): """Declare the local endpoint as a provider, once. contextWindow is the model's real window rather than the 32,768 a smaller model advertises, because a harness that trusts the advertised number will plan its compaction around it. """ cfg = _load_config() providers = cfg.setdefault('models', {}).setdefault('providers', {}) # Onboarding allow-lists exactly the models it configured, and a run naming # anything else is refused with "Model override is not allowed for agent". The # flagship rung is refused by that list too, not just the local one, so every # rung this paper uses is registered here rather than discovered one failure # at a time. allowed = cfg.setdefault('agents', {}).setdefault('defaults', {}).setdefault( 'models', {}) for mid, alias in (('deepseek/deepseek-v4-flash', 'DeepSeek Flash'), ('deepseek/deepseek-v4-pro', 'DeepSeek Pro'), ('%s/%s' % (LOCAL_PROVIDER, TI.OLLAMA_MODEL), 'Local'), ('%s/%s' % (CLOUD_PROVIDER, TI.TIERS['T3']['model']), 'Hosted')): allowed.setdefault(mid, {'alias': alias}) # The hosted rung, hand declared on the Responses API. Chat completions # cannot carry tools and a reasoning level at the same time for this model # family, and a rung with reasoning off is not the rung this study wants. if CLOUD_PROVIDER not in providers: providers[CLOUD_PROVIDER] = { 'baseUrl': TI.TIERS['T3']['base_url'], 'api': 'openai-responses', # A hand-declared provider carries its own credential reference: the # auth profiles its onboarding writes belong to the plugins it # installed, and a route it does not know refuses with # missing-provider-auth. ${VAR} is its own indirection, so no key is # written into the config file. 'apiKey': '${OPENAI_API_KEY}', 'api': 'openai-completions', 'models': [{ 'id': TI.TIERS['T3']['model'], 'name': TI.TIERS['T3']['model'], 'reasoning': True, 'input': ['text'], 'cost': {'input': 2.0, 'output': 12.0, 'cacheRead': 0.2, 'cacheWrite': 0}, 'contextWindow': 400000, 'maxTokens': 32768, 'compat': {'supportsReasoningEffort': True, 'supportsUsageInStreaming': True}, 'api': 'openai-responses', }], } # The provider plugin its own onboarding installed is trusted explicitly, so # the run does not carry an auto-load warning on every turn. cfg.setdefault('plugins', {}).setdefault('allow', ['deepseek']) if LOCAL_PROVIDER in providers: _save_config(cfg) return providers[LOCAL_PROVIDER] = { 'baseUrl': TI.OLLAMA_BASE, 'api': 'openai-completions', 'models': [{ 'id': TI.OLLAMA_MODEL, 'name': TI.OLLAMA_MODEL, 'reasoning': False, 'input': ['text'], 'cost': {'input': 0, 'output': 0, 'cacheRead': 0, 'cacheWrite': 0}, 'contextWindow': 131072, 'maxTokens': 8192, 'compat': {'supportsReasoningEffort': False, 'supportsUsageInStreaming': True, 'maxTokensField': 'max_tokens'}, 'api': 'openai-completions', }], } _save_config(cfg) def set_workspace(path): cfg = _load_config() cfg.setdefault('agents', {}).setdefault('defaults', {})['workspace'] = path _save_config(cfg) class OpenClawAdapter: name = 'openclaw' def __init__(self, tier): if not os.path.exists(OC_BIN): raise SystemExit('openclaw is not installed at %s' % OC_BIN) self.tier = tier self.t = TI.TIERS[tier] ensure_local_provider() provider = {'openai': LOCAL_PROVIDER, 'openai-cloud': CLOUD_PROVIDER}.get(self.t['wire'], 'deepseek') self.model = '%s/%s' % (provider, self.t['model']) self.sid = None self.session_file = None self._offset = 0 self._workspace0 = _load_config().get('agents', {}).get('defaults', {}).get( 'workspace') def label(self): return 'openclaw[%s]' % self.model def open_session(self, build, project): home = os.path.join(build, '.openclaw') os.makedirs(home, exist_ok=True) self._lock(build) set_workspace(build) self.sid = 'versus-%s' % project self.session_file = os.path.join(SESSIONS, self.sid + '.jsonl') self._offset = 0 return adapters.Session(self.name, build, project, home, time.time()) # ── one turn ───────────────────────────────────────────────────────────── def send(self, session, text): before = G.tree_hash(session.build) env = dict(os.environ) env['OPENCLAW_HOME'] = OC_HOME env[self.t['key_env']] = TI.key_for(self.tier) env['NO_COLOR'] = '1' cmd = [NODE, OC_BIN, 'agent', '--local', '--model', self.model, '--session-id', self.sid, '--json', '--timeout', str(TURN_TIMEOUT_S)] if self.t['effort'] is not None and self.t['wire'] == 'openai-cloud': # Its vocabulary for a hand-declared provider stops at high: asking # for max is refused with the list of levels it will take, and # declaring more levels on the model entry is refused by its config # schema. So this harness runs the rung one level below the others, # and the row records that rather than the rung pretending otherwise. cmd += ['--thinking', OC_MAX_LEVEL.get(self.t['effort'], self.t['effort'])] elif self.t['effort'] is not None and self.t['wire'] != 'openai': # It takes a thinking level on the command line, which is why it is in # the side study subset. Not sent on the primary ladder, where every # harness runs at its own default, and not sent to the local model, # which has no thinking to set. cmd += ['--thinking', self.t['effort']] cmd += ['--message', text] t0 = time.time() error, reply = None, {} try: p = subprocess.run(cmd, cwd=session.build, env=env, text=True, encoding='utf-8', errors='replace', capture_output=True, timeout=TURN_TIMEOUT_S + 60) reply = self._parse(p.stdout) if not reply: error = 'no JSON reply (rc=%s): %s' % ( p.returncode, (p.stderr or '')[-300:]) except subprocess.TimeoutExpired: error = 'turn exceeded %ds' % (TURN_TIMEOUT_S + 60) except Exception as e: # noqa: BLE001 error = '%s: %s' % (type(e).__name__, e) meta = (reply.get('meta') or {}).get('agentMeta') or {} usage = meta.get('usage') or {} tools, calls, asked = self._read_session_tail() when = datetime.now(timezone.utc) usd, rate = TI.price(self.t['model'], usage.get('input', 0), usage.get('output', 0), usage.get('cacheRead', 0), when) payloads = reply.get('payloads') or [] out = adapters.TurnResult( text=(payloads[0].get('text') if payloads else '') or '', tool_calls=tools, llm_calls=calls, cost_usd=usd if self.t['priced'] else 0.0, input_tokens=int(usage.get('input') or 0), output_tokens=int(usage.get('output') or 0), cached_tokens=int(usage.get('cacheRead') or 0), wall_s=time.time() - t0, asked=asked, error=error) out.rate = rate out.before, out.after = before, G.tree_hash(session.build) return out @staticmethod def _parse(stdout): """The JSON reply, from a stream that also carries the harness's own log.""" if not stdout: return {} start = stdout.find('{') while start != -1: try: obj = json.loads(stdout[start:]) if isinstance(obj, dict) and ('payloads' in obj or 'meta' in obj): return obj except ValueError: pass start = stdout.find('{', start + 1) return {} def _read_session_tail(self): """Tool calls with their arguments, from OpenClaw's own session log. Only the lines appended since the previous turn are read, so a call is attributed to the turn that made it. """ tools, calls, asked = [], 0, [] if not self.session_file or not os.path.exists(self.session_file): return tools, calls, asked with io.open(self.session_file, encoding='utf-8', errors='replace') as f: lines = f.readlines() fresh = lines[self._offset:] self._offset = len(lines) for line in fresh: line = line.strip() if not line: continue try: ev = json.loads(line) except ValueError: continue if ev.get('type') != 'message': continue msg = ev.get('message') or {} if msg.get('role') != 'assistant': continue calls += 1 for block in (msg.get('content') or []): if not isinstance(block, dict) or block.get('type') != 'toolCall': continue name = block.get('name') or '' args = block.get('arguments') or {} tools.append({'name': name, 'args': dict(args) if isinstance(args, dict) else {'raw': str(args)[:400]}}) if 'ask' in name.lower() or 'question' in name.lower(): asked.append(json.dumps(args)[:400]) return tools, calls, asked LOCK = os.path.join(OC_HOME, 'versus.lock') def _lock(self, build): """One OpenClaw session at a time, because the workspace is global. The workspace this harness works in is a single config key, so two sessions running at once would point one agent at the other's build and the scores would look perfectly ordinary. A stale lock older than an hour is broken, since a killed sweep cannot clean up after itself. """ if os.path.exists(self.LOCK): age = time.time() - os.path.getmtime(self.LOCK) if age < 3600: with io.open(self.LOCK, encoding='utf-8') as f: who = f.read()[:200] raise SystemExit( 'another OpenClaw session holds the workspace (%ds old): %s' % (int(age), who)) with io.open(self.LOCK, 'w', encoding='utf-8') as f: f.write('pid %d, build %s' % (os.getpid(), build)) def _unlock(self): try: os.remove(self.LOCK) except OSError: pass def close_session(self, session): if self._workspace0: set_workspace(self._workspace0) self._unlock()