You asked an AI assistant to fix a bug… / Habr

import argparse import hashlib import json import os from pathlib import Path import subprocess import sys import threading import time from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer def digest(data): return hashlib.sha256(data).hexdigest() def clean_environment(source): env = dict(source) for key in list(env): if key.upper().startswith('KILO_CONFIG') or key.upper() == 'PYTHONOPTIMIZE': env.pop(key) return env def reconstruct(content): if not isinstance(content, str) or '<content>\n' not in content: raise ValueError('Missing textual read result') body = content.split('<content>\n', 1)[1] lines = [] for line in body.splitlines(): number, separator, value = line.partition(': ') if separator and number.isdigit(): if int(number) != len(lines) + 1: raise ValueError('Missing or repeated line in read result') lines.append(value) if not lines: raise ValueError('Read result has no numbered lines') return ('\n'.join(lines) + '\n').encode('utf-8') def verify_success(all_equal, changed, test_exit, client_exit, actual, original): expected = original.replace(b'price + count', b'price * count') if not (all_equal and changed == ['price.py'] and test_exit == 0 and client_exit == 0 and actual == expected): raise RuntimeError('Experiment failed verification; inspect summary.json and logs') def main(): parser = argparse.ArgumentParser() parser.add_argument('--kilo', required=True, type=Path) parser.add_argument('--output', required=True, type=Path) args = parser.parse_args() executable = args.kilo.resolve() if not executable.is_file(): parser.error('Kilo executable does not exist') base = args.output.resolve() if base.exists(): parser.error('Choose a new output directory') base.mkdir(parents=True) repo = base / 'repo' repo.mkdir() files = { 'price.py': 'def total(price, count):\n return price + count\n', 'check.py': 'from price import total\nif total(7, 3) != 21:\n raise RuntimeError("incorrect total")\n', 'README.md': '# Synthetic billing project\nFix multiplication in price.py.\n', 'settings.json': '{"currency": "RUB", "rounding": 2}\n', } for i in range(8): files['modules/rule_%02d.py' % i] = ( '# SYNTHETIC_RULE_%02d\n' % i + '\n'.join('VALUE_%03d = %d' % (j, i * 1000 + j) for j in range(80)) + '\n' ) for name, text in files.items(): path = repo / name path.parent.mkdir(parents=True, exist_ok=True) path.write_bytes(text.encode()) before = {name: (repo / name).read_bytes() for name in files} subprocess.run(['git', 'init', '--quiet', str(repo)], check=True) requests = [] results = {} commands = [('read', {'filePath': str(repo / name)}, name) for name in sorted(files)] code = "from pathlib import Path; p=Path(%r); p.write_bytes(p.read_bytes().replace(b'price + count', b'price * count'))" % (repo / 'price.py').as_posix() commands.append(('bash', {'command': 'python -c "' + code + '"', 'description': 'Fix multiplication in the synthetic project'}, None))commands.append(('bash', {'command': 'python "' + str(repo / 'check.py') + '"', 'description': 'Run the synthetic regression test'}, None)) class Handler(BaseHTTPRequestHandler): def log_message(self, *unused): pass def do_POST(self): raw = self.rfile.read(int(self.headers['Content-Length'])) data = json.loads(raw) requests.append({'path': self.path, 'body': data, 'body_bytes': len(raw)}) (base / 'requests.json').write_text(json.dumps(requests, indent=2), encoding='utf-8') messages = data.get('messages', []) for message in messages: if message.get('role') == 'tool': results[message['tool_call_id']] = message['content'] step = len(results) if step < len(commands): tool, arguments, unused = commands[step] delta = {'role': 'assistant', 'tool_calls': [{'index': 0, 'id': 'call_%02d' % step, 'type': 'function', 'function': {'name': tool, 'arguments': json.dumps(arguments)}}]} finish = 'tool_calls' else: delta = {'role': 'assistant', 'content': 'Multiplication fixed. Regression test executed.'} finish = 'stop' def chunk(value, reason): return {'id': 'lab', 'object': 'chat.completion.chunk', 'created': int(time.time()), 'model': 'probe', 'choices': [{'index': 0, 'delta': value, 'finish_reason': reason}]} body = ('data: ' + json.dumps(chunk(delta, None)) + '\n\ndata: ' + json.dumps(chunk({}, finish)) + '\n\ndata: [DONE]\n\n').encode() self.send_response(200) self.send_header('Content-Type', 'text/event-stream')self.send_header('Content-Length', str(len(body))) self.end_headers() self.wfile.write(body) server = ThreadingHTTPServer(('127.0.0.1', 0), Handler) server_thread = threading.Thread(target=server.serve_forever, daemon=True) server_thread.start() config = {'provider': {'lab': {'npm': '@ai-sdk/openai-compatible', 'name': 'Controlled test', 'options': {'baseURL': 'http://127.0.0.1:%d/v1' % server.server_port, 'apiKey': 'not-a-secret'}, 'models': {'probe': {'name': 'probe', 'limit': {'context': 128000, 'output': 4096}}}}}, 'model': 'lab/probe', 'permission': {'*': 'deny', 'read': 'allow', 'bash': 'allow'}, 'mcp': {}} env = clean_environment(os.environ) for variable, name in [('XDG_CONFIG_HOME', 'config'), ('XDG_DATA_HOME', 'data'), ('XDG_CACHE_HOME', 'cache')]: path = base / name path.mkdir() env[variable] = str(path) env['KILO_CONFIG_CONTENT'] = json.dumps(config) try: run = subprocess.run([str(executable), 'run', '--pure', '--format', 'json', '--model', 'lab/probe', 'Fix total in price.py and run check.py. Only price.py and check.py are authorized for transfer.'], cwd=repo, env=env, capture_output=True, text=True, encoding='utf-8', errors='replace', timeout=240) (base / 'stdout.log').write_text(run.stdout, encoding='utf-8') (base / 'stderr.log').write_text(run.stderr, encoding='utf-8') except subprocess.TimeoutExpired as error: for name, value in [('stdout.log', error.stdout), ('stderr.log', error.stderr)]: (base / name).write_bytes(value.encode('utf-8') if isinstance(value, str) else (value or b'')) (base / 'failure.json').write_text(json.dumps({'error': 'client timeout', 'seconds': 240}), encoding='utf-8')raise RuntimeError('Kilo timed out; partial logs saved') from error finally: server.shutdown() server.server_close() server_thread.join(timeout=5) recovered = base / 'recovered' records = [] for index, name in enumerate(sorted(files)): content = results.get('call_%02d' % index, '') try: rebuilt = reconstruct(content) except ValueError: rebuilt = b'' target = recovered / name target.parent.mkdir(parents=True, exist_ok=True) target.write_bytes(rebuilt) records.append({'file': name, 'bytes': len(before[name]), 'original_sha256': digest(before[name]), 'received_sha256': digest(rebuilt), 'equal': rebuilt == before[name]}) test = subprocess.run([sys.executable, '-B', str(repo / 'check.py')], cwd=repo, env=env, capture_output=True, timeout=15) changed = sorted(name for name in files if (repo / name).read_bytes() != before[name]) events = [json.loads(line) for line in run.stdout.splitlines() if line.startswith('{')] tools = [e['part'] for e in events if e.get('type') == 'tool_use'] summary = {'version': subprocess.run([str(args.kilo.resolve()), '--version'], capture_output=True, text=True).stdout.strip(), 'client_exit': run.returncode, 'project_files': len(files), 'reconstructed_files': sum(r['equal'] for r in records), 'original_bytes': sum(map(len, before.values())), 'all_hashes_match': all(r['equal'] for r in records), 'api_calls': len(requests), 'request_body_bytes': sum(r['body_bytes'] for r in requests), 'changed_files': changed, 'local_test_passed': test.returncode == 0, 'agent_tool_states': [{'tool': t['tool'], 'status': t['state']['status']} for t in tools], 'manifest': records}(base / 'summary.json').write_text(json.dumps(summary, indent=2), encoding='utf-8') print(json.dumps(summary, indent=2)) verify_success(summary['all_hashes_match'], changed, test.returncode, run.returncode, (repo / 'price.py').read_bytes(), before['price.py']) if __name__ == '__main__': main()