#!/usr/bin/env python3
"""Exercise local TCP, immutable byte extents, and the existing workbench CLI."""
import argparse
from contextlib import contextmanager
from concurrent.futures import ThreadPoolExecutor
import fcntl
import hashlib
import http.client
import json
import platform
from pathlib import Path
import select
import socket
import struct
import subprocess
import sys
import tempfile
import time
from check_edits import check_edits

ROOT = Path(__file__).resolve().parent
WORKBENCH = ROOT.parent / 'workbench'
OBJECT_LIMIT = (2 * 6 + 32) * 32768 + 8192


class Server:
    def __init__(self, binary, storage, budget=4 * 1024 * 1024, timeout=5000, program=None):
        self.process = subprocess.Popen([str(binary), '--store', str(storage), '--port', '0',
            '--program', str(program or ROOT / 'examples/storage.pure'),
            '--budget', str(budget), '--timeout-ms', str(timeout), '--send-buffer', '4096'],
            stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
        if not select.select([self.process.stdout], [], [], 15)[0]:
            self.process.kill()
            raise AssertionError('server readiness timed out')
        line = self.process.stdout.readline()
        if not line:
            raise AssertionError(self.process.communicate(timeout=5))
        self.ready = json.loads(line)
        self.port = self.ready['port']
        self.stats = None

    def close(self):
        if self.process.poll() is None:
            self.process.terminate()
        try:
            stdout, stderr = self.process.communicate(timeout=10)
        except subprocess.TimeoutExpired:
            self.process.kill()
            self.process.communicate()
            raise
        assert self.process.returncode == 0, (stdout, stderr)
        self.stats = json.loads(stderr)
        assert self.stats['active'] == self.stats['retained_payload_bytes'] == 0
        assert self.stats['peak_retained_payload_bytes'] <= self.stats['payload_budget']
        assert self.stats['peak_connections'] <= 8


@contextmanager
def serving(*args, **kwargs):
    server = Server(*args, **kwargs)
    try:
        yield server
    finally:
        server.close()


def rpc(server, method, key, body=b'', expected=None):
    headers = {'Content-Length': str(len(body))}
    if expected is not None:
        headers['If-None-Match' if expected == 0 else 'If-Match'] = '*' if expected == 0 else f'"{expected}"'
    connection = http.client.HTTPConnection('127.0.0.1', server.port, timeout=8)
    try:
        connection.request(method, f'/objects/{key}', body, headers)
        response = connection.getresponse()
        return response.status, dict(response.getheaders()), response.read()
    finally:
        connection.close()


def wire(server, request, chunks=None):
    with socket.create_connection(('127.0.0.1', server.port), timeout=8) as sock:
        if chunks:
            for chunk in chunks:
                sock.sendall(chunk)
        else:
            sock.sendall(request)
        response = http.client.HTTPResponse(sock)
        response.begin()
        return response.status, dict(response.getheaders()), response.read()


def header(method, key, size=0, condition=b''):
    return (f'{method} /objects/{key} HTTP/1.1\r\nHost: localhost\r\n'
            f'Content-Length: {size}\r\n'.encode() + condition + b'\r\n')


def cli(server, commands, source=None, expected_code=0):
    result = subprocess.run(['node', str(WORKBENCH / 'cli.mjs'), *([str(source)] if source else [])],
        input=commands.replace('$URL', f'http://127.0.0.1:{server.port}'), text=True,
        capture_output=True, timeout=30)
    assert result.returncode == expected_code, (result.stdout, result.stderr)
    return [json.loads(line) for line in result.stdout.splitlines()], result.stderr


def check(binary, directory, editor=None):
    storage = directory / 'objects'
    checks, records = [], []
    with serving(binary, storage) as server:
        assert rpc(server, 'GET', 'absent')[0] == 404
        payload = b'\x00\xffbinary\r\n\r\n' + bytes(range(256)) * 41
        request = header('PUT', 'binary', len(payload), b'If-None-Match: *\r\n') + payload
        status, headers, _ = wire(server, request, [request[:7], request[7:51], request[51:57], request[57:]])
        assert status == 200 and headers['ETag'] == '"1"'
        assert rpc(server, 'GET', 'binary')[2] == payload
        checks.append('fragmented request and binary bytes round trip')
        assert rpc(server, 'PUT', 'binary', b'new', 0)[0] == 412
        assert rpc(server, 'PUT', 'binary', b'new', 1)[0] == 200
        assert rpc(server, 'PUT', 'binary', b'stale', 1)[0] == 412
        assert rpc(server, 'GET', 'binary')[2] == b'new'
        checks.append('conditional create and stale revision preserve committed bytes')

        with (storage / '.writer.lock').open('r+b') as lock:
            fcntl.lockf(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
            assert rpc(server, 'PUT', 'binary', b'blocked', 2)[0] == 503
            fcntl.lockf(lock, fcntl.LOCK_UN)
        broken = storage / 'obj-broken'
        broken.write_bytes(b'incomplete')
        assert rpc(server, 'GET', 'broken')[0] == 500
        assert rpc(server, 'PUT', 'broken', b'failure', 0)[0] == 500
        assert broken.read_bytes() == b'incomplete'
        assert rpc(server, 'GET', 'binary')[2] == b'new'
        checks.append('writer contention and invalid extent fail without overwriting data')

        with serving(binary, storage) as second_writer:
            with ThreadPoolExecutor(max_workers=2) as pool:
                tasks = [pool.submit(rpc, target, 'PUT', 'race', value, 0)
                         for target, value in [(server, b'first'), (second_writer, b'second')]]
                statuses = [task.result()[0] for task in tasks]
            assert statuses.count(200) == 1 and all(status in [200, 412, 503] for status in statuses)
            status, headers, body = rpc(server, 'GET', 'race')
            assert status == 200 and headers['ETag'] == '"1"' and body in [b'first', b'second']
        records.append(second_writer.stats)
        checks.append('two native writers cannot both create the same expected revision')

        sock = socket.create_connection(('127.0.0.1', server.port), timeout=8)
        sock.sendall(header('PUT', 'incomplete', 100, b'If-None-Match: *\r\n') + b'partial')
        sock.shutdown(socket.SHUT_WR)
        assert sock.recv(1) == b''
        sock.close()
        assert rpc(server, 'GET', 'incomplete')[0] == 404
        checks.append('incomplete request never commits and releases its allocation')

        # Force genuine partial socket sends. A write on another connection must
        # not alter the immutable file whose bytes are still being transmitted.
        large = bytes(range(256)) * 4096
        assert rpc(server, 'PUT', 'large', large, 0)[0] == 200
        sock = socket.create_connection(('127.0.0.1', server.port), timeout=8)
        sock.sendall(header('GET', 'large'))
        response = http.client.HTTPResponse(sock)
        response.begin()
        assert response.status == 200 and response.getheader('ETag') == '"1"'
        assert response.read(17) == large[:17]
        assert rpc(server, 'PUT', 'large', b'changed', 1)[0] == 200
        assert response.read() == large[17:]
        sock.close()
        assert rpc(server, 'GET', 'large')[2] == b'changed'
        checks.append('borrowed read remains the same snapshot during another commit')

        assert rpc(server, 'PUT', 'large', large, 2)[0] == 200
        sock = socket.create_connection(('127.0.0.1', server.port), timeout=8)
        sock.sendall(header('GET', 'large'))
        assert sock.recv(1)
        sock.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER, struct.pack('ii', 1, 0))
        sock.close()
        assert rpc(server, 'GET', 'binary')[2] == b'new'
        checks.append('receiver reset aborts an outstanding borrowed transfer')

        assert wire(server, header('PUT', 'too_large', OBJECT_LIMIT + 1, b'If-None-Match: *\r\n'))[0] == 413
        malformed = [
            b'PUT /objects/bad HTTP/1.1\r\nHost: x\r\nContent-Length: 1\r\nContent-Length: 1\r\nIf-None-Match: *\r\n\r\nx',
            b'PUT /objects/bad HTTP/1.1\r\nHost: x\r\nTransfer-Encoding: chunked\r\n\r\n',
            b'GET /objects/bad HTTP/1.1\r\nHost: x\x00hidden\r\n\r\n',
        ]
        for request in malformed:
            assert wire(server, request)[0] == 400
        assert rpc(server, 'GET', 'bad')[0] == 404
        checks.append('unsupported framing and oversized payload are rejected before effects')

        # Existing authoring path: define, manipulate, save remotely, reconstruct
        # in a fresh runtime. No fabricated project fixture or mock storage.
        source = WORKBENCH / 'examples' / 'direct.pure'
        saved = directory / 'restored.json'
        lines, _ = cli(server, 'point drag down 80 180\npoint drag move 210 210\npoint drag up 210 210\n'
            'connect $URL authored\npush 0\nquit\n', source)
        snapshots = [line for line in lines if 'states' in line]
        assert snapshots
        expected_states = snapshots[-1]['states']
        assert expected_states['x'] == 210 and expected_states['y'] == 190 and expected_states['dragging'] == 0
        lines, _ = cli(server, f'connect $URL authored\npull\nsave-project {saved}\nquit\n')
        assert [line for line in lines if 'states' in line][-1]['states'] == expected_states
        document = json.loads(saved.read_text())
        assert document['version'] == 2 and document['source'] == source.read_text()
        lines, error = cli(server, 'connect $URL authored\npush 0\nquit\n', source, 1)
        assert '412' in error
        checks.append('existing workbench manipulation restores in a fresh runtime and reports conflict')

        assert rpc(server, 'PUT', 'invalid_project', b'{"format":"pure-workbench","version":2,"source":"state x=1","states":[]}', 0)[0] == 200
        lines, error = cli(server, 'show\nconnect $URL invalid_project\npull\nshow\nquit\n', source, 1)
        states = [line['states'] for line in lines if 'states' in line]
        assert states[0] == states[-1] and error
        lines, error = cli(server, 'point drag down 90 170\nconnect $URL authored\npull\npush\nshow\nquit\n', source, 1)
        states = [line['states'] for line in lines if 'states' in line]
        assert states[0] == states[-1] and states[-1]['dragging'] == 1
        assert len(error.strip().splitlines()) == 2
        checks.append('invalid or non-idle restore preserves local state and does not adopt remote revision')

        assert rpc(server, 'PUT', 'wide_revision', rpc_bytes(saved), 0)[0] == 200
        wide = storage / 'obj-wide_revision'
        raw = wide.read_bytes()
        wide.write_bytes(raw[:8] + struct.pack('<Q', 9007199254740993) + raw[16:])
        lines, _ = cli(server, 'connect $URL wide_revision\npull\npush\nquit\n')
        revisions = [line['revision'] for line in lines if 'remote' in line and 'revision' in line]
        assert revisions == ['9007199254740993', '9007199254740994']
        checks.append('revision above JavaScript safe integer round trips without numeric coercion')

        sock = socket.create_connection(('127.0.0.1', server.port), timeout=8)
        sock.sendall(header('PUT', 'unread_ack', 4, b'If-None-Match: *\r\n') + b'done')
        try:
            deadline = time.monotonic() + 3
            while True:
                committed = storage / 'obj-unread_ack'
                if committed.exists():
                    assert rpc(server, 'GET', 'unread_ack')[2] == b'done'
                    break
                assert time.monotonic() < deadline
                time.sleep(0.01)
        finally:
            sock.close()  # Client never consumes the successful response.
        assert rpc(server, 'PUT', 'unread_ack', b'again', 0)[0] == 412
        assert rpc(server, 'GET', 'unread_ack')[2] == b'done'
        checks.append('unread acknowledgment does not undo commit or permit a stale retry')

        # A disconnect after commit can hide the ACK; retries remain conditional.
        assert rpc(server, 'PUT', 'empty', b'', 0)[0] == 200
        assert rpc(server, 'GET', 'empty')[2] == b''
        checks.append('empty body has a valid lifetime and stored revision')
    records.append(server.stats)

    with serving(binary, storage) as restarted:
        assert rpc(restarted, 'GET', 'binary')[2] == b'new'
        assert rpc(restarted, 'GET', 'authored')[2] == rpc_bytes(saved)
        checks.append('committed authoring state survives process restart')
    records.append(restarted.stats)

    # Fixed small payload budget is an admission test, not a throughput workload.
    with serving(binary, storage, budget=1024 * 1024, timeout=1200) as limited:
        sock = socket.create_connection(('127.0.0.1', limited.port), timeout=8)
        sock.sendall(header('GET', 'large'))
        response = http.client.HTTPResponse(sock)
        response.begin()
        assert response.status == 200
        assert response.read(1) == large[:1]
        status, headers, _ = rpc(limited, 'GET', 'large')
        assert status == 503 and headers['X-Pure-Error'] == 'payload_budget'
        # Leave the first receiver stalled until the absolute connection deadline.
        time.sleep(1.4)
        assert rpc(limited, 'GET', 'binary')[2] == b'new'
        response.close()
        sock.close()
        checks.append('slow consumer retains its lease; budget rejects another; deadline releases it')

        idle = [socket.create_connection(('127.0.0.1', limited.port), timeout=8) for _ in range(8)]
        try:
            for s in idle:
                s.sendall(b'G')
            for s in idle:
                assert s.recv(1) == b''
        finally:
            for s in idle:
                s.close()
        assert rpc(limited, 'GET', 'binary')[0] == 200
        checks.append('connection slots stay bounded and recover after header deadlines')
    assert limited.stats['budget_rejections'] >= 1 and limited.stats['expired'] >= 9
    records.append(limited.stats)
    prepared = authored_connections(binary, directory, storage, records, checks)
    fields = field_connections(binary, directory, storage, records, checks)
    edits = check_edits(editor, binary, ROOT, directory, serving, rpc, header, records, checks) if editor else None
    return {'kind': 'prepared-connection-correctness-not-performance', 'checks': checks,
            'prepared_connections': prepared,
            'field_connections': fields,
            'document_edits': edits,
            'server_runs': records, 'storage': server.ready['storage']}


def authored_connections(binary, directory, storage, records, checks):
    get = 'on get { let o = store.read(request.key); reply(o, o.body); }'
    put = 'on put { let s = store.write(request.key, request.revision, request.body); reply(s); }'

    def author(name, source):
        path = directory / name
        path.write_text(source)
        return path

    def explain(path):
        return json.loads(subprocess.check_output([str(binary), '--check-program', str(path)], text=True))

    with serving(binary, storage) as writer:
        assert rpc(writer, 'PUT', 'short_parent', bytes(range(48)), 0)[0] == 200
        assert rpc(writer, 'PUT', 'short_unused', bytes(range(80)), 0)[0] == 200
    records.append(writer.stats)
    excerpt = ROOT / 'examples/excerpt.pure'
    prepared = {'excerpt': explain(excerpt)}
    assert prepared['excerpt']['read'] == {
        'to_end': False, 'offset': 32, 'length': 16, 'minimum_source_bytes': 80}
    with serving(binary, storage, budget=16, program=excerpt) as selected:
        status, headers, body = rpc(selected, 'GET', 'large')
        assert status == 200 and body == bytes(range(32, 48)) and headers['ETag'] == '"3"'
        # Final [32,48) fits, but the authored parent [16,80) does not.
        assert rpc(selected, 'GET', 'short_parent')[0] == 413
        assert rpc(selected, 'GET', 'missing')[0] == 404
    assert selected.stats['peak_retained_payload_bytes'] == 16
    records.append(selected.stats)
    checks.append('authored nested read maps only final demand while preserving parent bounds and revision')

    unused = author('unused.pure', 'on get { let o = store.read(request.key); '
        'let unused = range(o.body, 0, 96); let body = range(o.body, 32, 16); reply(o, body); } ' + put)
    with serving(binary, storage, budget=16, program=unused) as guarded:
        assert rpc(guarded, 'GET', 'short_unused')[0] == 413
        assert rpc(guarded, 'GET', 'large')[2] == bytes(range(32, 48))
    records.append(guarded.stats)
    checks.append('unused checked view still fails even when response range would be valid')

    select_write = author('write.pure', get + ' on put { let a = range(request.body, 2, 10); '
        'let b = range(a, 3, 4); let alias = b; '
        'let saved = store.write(request.key, request.revision, alias); reply(saved, b); }')
    prepared['write'] = explain(select_write)
    with serving(binary, storage, program=select_write) as writer:
        status, headers, body = rpc(writer, 'PUT', 'selected', b'0123456789AB', 0)
        assert status == 200 and body == b'5678' and headers['ETag'] == '"1"'
        assert rpc(writer, 'GET', 'selected')[2] == b'5678'
        assert rpc(writer, 'PUT', 'selected', b'012345678', 1)[0] == 413
        status, headers, body = rpc(writer, 'GET', 'selected')
        assert status == 200 and body == b'5678' and headers['ETag'] == '"1"'
        assert rpc(writer, 'PUT', 'selected', b'0123456789AB', 0)[0] == 412
    records.append(writer.stats)
    checks.append('authored write and reply share selected input; invalid parent prevents the storage effect')

    metadata = author('metadata.pure', 'on get { let o = store.read(request.key); reply(o); } ' + put)
    with serving(binary, storage, budget=1, program=metadata) as meta:
        status, headers, body = rpc(meta, 'GET', 'large')
        assert status == 200 and headers['ETag'] == '"3"' and body == b''
        assert rpc(meta, 'GET', 'absent')[0] == 404
        assert rpc(meta, 'GET', 'broken')[0] == 500
    assert meta.stats['peak_retained_payload_bytes'] == 0
    records.append(meta.stats)
    checks.append('body-free response preserves read existence and malformed-object errors')

    echo = author('echo.pure', get + ' on put { let body = request.body; '
        'let saved = store.write(request.key, request.revision, body); reply(saved, body); }')
    payload = bytes(range(256)) * 4096
    with serving(binary, storage, budget=len(payload), program=echo) as echoed:
        with socket.create_connection(('127.0.0.1', echoed.port), timeout=8) as sock:
            sock.sendall(header('PUT', 'echo', len(payload), b'If-None-Match: *\r\n') + payload)
            response = http.client.HTTPResponse(sock)
            response.begin()
            assert response.status == 200 and response.read(1) == payload[:1]
            assert rpc(echoed, 'GET', 'binary')[0] == 503
            assert response.read() == payload[1:]
            response.close()
        assert rpc(echoed, 'GET', 'binary')[2] == b'new'
        with socket.create_connection(('127.0.0.1', echoed.port), timeout=8) as sock:
            sock.sendall(header('PUT', 'echo_abort', len(payload), b'If-None-Match: *\r\n') + payload)
            assert sock.recv(1)
            assert rpc(echoed, 'GET', 'binary')[0] == 503
            sock.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER, struct.pack('ii', 1, 0))
        deadline = time.monotonic() + 3
        while rpc(echoed, 'GET', 'binary')[0] == 503:
            assert time.monotonic() < deadline
            time.sleep(0.01)
        assert rpc(echoed, 'GET', 'echo_abort')[2] == payload
    records.append(echoed.stats)
    checks.append('input-backed response survives partial send and releases on finish or abort without undoing commit')

    bad = author('invalid.pure', get + get + put)
    untouched = directory / 'must_not_exist'
    rejected = subprocess.run([str(binary), '--program', str(bad), '--store', str(untouched)],
        capture_output=True, text=True, timeout=10)
    assert rejected.returncode == 2 and 'duplicate handler' in rejected.stderr and not untouched.exists()
    checks.append('invalid authored program fails preparation before creating storage or accepting requests')
    return prepared


def field_connections(binary, directory, storage, records, checks):
    program = ROOT / 'examples/selection.pure'
    plans = {'selection': json.loads(subprocess.check_output(
        [str(binary), '--check-program', str(program)], text=True))}
    assert plans['selection']['runtime_view_steps'] == 1
    assert plans['selection']['steps'][0] == {
        'source_step': 0, 'source': {'offset': 8, 'length': 0, 'to_end': True},
        'offset': {'value': 0, 'from_u32le': True}, 'length': {'value': 4, 'from_u32le': True}}
    document = '처음🙂 Pure는 필드를 연결한다.\n끝'
    document_file = directory / 'document.txt'
    document_file.write_bytes(document.encode('utf-8'))
    with serving(binary, storage, program=program) as selected:
        command = [sys.executable, str(ROOT / 'examples/select-text.py'), str(document_file),
            '--key', 'text_selection', '--start', '2', '--end', '12', '--url', f'http://127.0.0.1:{selected.port}']
        result = subprocess.run(command, capture_output=True, text=True, timeout=15)
        assert result.returncode == 0, (result.stdout, result.stderr)
        expected = document[2:12].encode('utf-8')
        assert json.loads(result.stdout) == {'revision': '1', 'selected': document[2:12], 'stored_bytes': len(expected)}
        assert rpc(selected, 'GET', 'text_selection')[2] == expected
        checks.append('text client uses character selection; named numeric fields select and persist the matching UTF-8 bytes')

        # Validation precedes the write: even a stale header cannot expose or
        # commit partial input, and prior content/revision survive all failures.
        invalid = [b'\0' * n for n in range(8)] + [
            struct.pack('<II', 0xffffffff, 0) + b'abc',
            struct.pack('<II', 0, 0xffffffff) + b'abc',
            struct.pack('<II', 2, 2) + b'abc',
            struct.pack('<II', 4, 0) + b'abc']
        for payload in invalid:
            assert rpc(selected, 'PUT', 'text_selection', payload, 1)[0] == 413
        status, headers, body = rpc(selected, 'GET', 'text_selection')
        assert status == 200 and headers['ETag'] == '"1"' and body == expected
        for key, payload in [('selection_empty', struct.pack('<II', 0, 0)),
                             ('selection_end', struct.pack('<II', 3, 0) + b'abc')]:
            assert rpc(selected, 'PUT', key, payload, 0)[0] == 200
            assert rpc(selected, 'GET', key)[2] == b''
        assert rpc(selected, 'PUT', 'text_selection', struct.pack('<II', 0, 3) + b'abc', 0)[0] == 412
        checks.append('truncated layouts and invalid data-dependent ranges never change storage; empty and end selections remain valid')
    records.append(selected.stats)

    declaration = 'layout E { tag: bytes[3]; count: u32le; start: u32le; document: bytes; } '
    get = 'on get { let o = store.read(request.key); reply(o, o.body); } '
    put = 'on put { let s = store.write(request.key, request.revision, request.body); reply(s); } '
    reordered = directory / 'reordered.pure'
    reordered.write_text(declaration + get + 'on put { let e = view(E, request.body); let alias = e; '
        'let part = range(alias.document, alias.start, alias.count); let again = part; '
        'let s = store.write(request.key, request.revision, again); reply(s, part); }')
    plans['reordered'] = json.loads(subprocess.check_output([str(binary), '--check-program', str(reordered)], text=True))
    assert plans['reordered']['runtime_view_steps'] == 1
    with serving(binary, storage, program=reordered) as moved:
        status, headers, body = rpc(moved, 'PUT', 'reordered', b'tag' + struct.pack('<II', 4, 2) + b'abcdefgh', 0)
        assert status == 200 and body == b'cdef' and headers['ETag'] == '"1"'
        assert rpc(moved, 'GET', 'reordered')[2] == b'cdef'
    records.append(moved.stats)
    checks.append('changed declaration order and unaligned numeric fields determine access; aliases add no dynamic steps')

    unused = directory / 'unused-dynamic.pure'
    unused.write_text(declaration + get + 'on put { let e = view(E, request.body); '
        'let unused = range(e.document, e.start, e.count); '
        'let s = store.write(request.key, request.revision, e.document); reply(s); }')
    with serving(binary, storage, program=unused) as guarded:
        assert rpc(guarded, 'PUT', 'unused_dynamic', b'tag' + struct.pack('<II', 4, 2) + b'abc', 0)[0] == 413
        assert rpc(guarded, 'GET', 'unused_dynamic')[0] == 404
        assert rpc(guarded, 'PUT', 'unused_dynamic', b'tag' + struct.pack('<II', 1, 1) + b'abc', 0)[0] == 200
        assert rpc(guarded, 'GET', 'unused_dynamic')[2] == b'abc'
    records.append(guarded.stats)
    checks.append('unused dynamic field selection retains its validation before the write effect')

    field_get = directory / 'field-get.pure'
    field_get.write_text(declaration + 'on get { let o = store.read(request.key); '
        'let e = view(E, o.body); reply(o, e.document); } ' + put)
    plans['field_get'] = json.loads(subprocess.check_output([str(binary), '--check-program', str(field_get)], text=True))
    tail = '다른 길이의 문서'.encode('utf-8')
    with serving(binary, storage, program=field_get) as writer:
        assert rpc(writer, 'PUT', 'tail', b'tag' + struct.pack('<II', 0, 0) + tail, 0)[0] == 200
        assert rpc(writer, 'PUT', 'bad_layout', b'short', 0)[0] == 200
        assert rpc(writer, 'PUT', 'empty_tail', b'tag' + struct.pack('<II', 0, 0), 0)[0] == 200
    records.append(writer.stats)
    with serving(binary, storage, budget=len(tail), program=field_get) as reader:
        assert rpc(reader, 'GET', 'tail')[2] == tail
        assert rpc(reader, 'GET', 'empty_tail')[2] == b''
        assert rpc(reader, 'GET', 'bad_layout')[0] == 413
    assert reader.stats['peak_retained_payload_bytes'] == len(tail)
    records.append(reader.stats)
    checks.append('named variable tail reads use the same file version and charge only the selected suffix')

    document_bytes = bytes(range(256)) * 4096
    payload = struct.pack('<II', 16, len(document_bytes) - 32) + document_bytes
    with serving(binary, storage, budget=len(payload), program=program) as slow:
        with socket.create_connection(('127.0.0.1', slow.port), timeout=8) as sock:
            sock.sendall(header('PUT', 'dynamic_slow', len(payload), b'If-None-Match: *\r\n') + payload)
            response = http.client.HTTPResponse(sock)
            response.begin()
            assert response.status == 200 and response.read(1) == document_bytes[16:17]
            assert rpc(slow, 'GET', 'binary')[0] == 503
            assert response.read() == document_bytes[17:-16]
            response.close()
        assert rpc(slow, 'GET', 'binary')[2] == b'new'
        assert rpc(slow, 'GET', 'dynamic_slow')[2] == document_bytes[16:-16]
    assert slow.stats['peak_retained_payload_bytes'] == len(payload)
    records.append(slow.stats)
    checks.append('data-dependent response keeps and charges the original input through partial transmission')
    return plans


def rpc_bytes(path):
    # CLI export is pretty JSON; wire export is compact JSON with the same model.
    return json.dumps(json.loads(path.read_text()), ensure_ascii=False, separators=(',', ':')).encode()


def source(name):
    """A source file of this document runtime; the core's live in ../../core."""
    here = ROOT / name
    return here if here.exists() else ROOT.parent.parent / 'core' / name


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--server', type=Path, default=ROOT / 'build/pure-connections')
    parser.add_argument('--output', type=Path)
    parser.add_argument('--runtime-test', type=Path)
    parser.add_argument('--splice-test', type=Path)
    parser.add_argument('--editor', type=Path)
    parser.add_argument('--store-source', type=Path, default=ROOT / 'store.c')
    args = parser.parse_args()
    if args.output and args.output.exists():
        parser.error('evidence file already exists')
    binary = args.server.resolve()
    with tempfile.TemporaryDirectory(prefix='pure-connections-check-') as temporary:
        result = check(binary, Path(temporary), args.editor.resolve() if args.editor else None)
    if args.runtime_test:
        runtime_test = args.runtime_test.resolve()
        tested = subprocess.run([str(runtime_test)], cwd=ROOT, capture_output=True, text=True, timeout=30)
        assert tested.returncode == 0, (tested.stdout, tested.stderr)
        result['execution_core'] = json.loads(tested.stdout)
        result['execution_core_test_sha256'] = hashlib.sha256(runtime_test.read_bytes()).hexdigest()
    if args.splice_test:
        splice_test = args.splice_test.resolve()
        tested = subprocess.run([str(splice_test)], cwd=ROOT, capture_output=True, text=True, timeout=30)
        assert tested.returncode == 0, (tested.stdout, tested.stderr)
        result['document_edit_units'] = json.loads(tested.stdout)
        result['document_edit_test_sha256'] = hashlib.sha256(splice_test.read_bytes()).hexdigest()
    if args.editor:
        result['editor_sha256'] = hashlib.sha256(args.editor.resolve().read_bytes()).hexdigest()
    result['binary_sha256'] = hashlib.sha256(binary.read_bytes()).hexdigest()
    result['environment'] = {'platform': platform.platform(), 'machine': platform.machine(),
        'python': platform.python_version(), 'node': subprocess.check_output(['node', '--version'], text=True).strip()}
    result['source_sha256'] = {str(path.relative_to(ROOT.parent.parent)): hashlib.sha256(path.read_bytes()).hexdigest()
        for path in [args.store_source.resolve(), *(source(name) for name in ('flow.c', 'flow.h', 'store.h', 'program.c', 'program.h',
                         'runtime.c', 'runtime.h', 'result.h', 'evaluate.c', 'store-port.c', 'store-port.h', 'test-runtime.c',
                         'server.c', 'check.py', 'check_edits.py', 'edit.c', 'test-splice.c',
                         'test-flow.c', 'test-program.c', 'test-fields.c', 'Makefile',
                         'core-layout.c', 'check_core.py', 'core-budget.json', 'examples/editor.pure',
                         'examples/storage.pure', 'examples/excerpt.pure', 'examples/selection.pure', 'examples/select-text.py')),
                     *(WORKBENCH / name for name in ('remote.mjs', 'cli.mjs', 'web/core.wasm', 'web/runtime.mjs',
                         'web/compiler.mjs', 'web/project.mjs', 'examples/direct.pure'))]}
    output = json.dumps(result, ensure_ascii=False, indent=2) + '\n'
    if args.output:
        args.output.parent.mkdir(parents=True, exist_ok=True)
        with args.output.open('x') as destination:
            destination.write(output)
    print(output, end='')


if __name__ == '__main__':
    main()