Вы просили AI-помощника исправить баг…

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()