Establish native micro runtime checkpoint

1 files changed, 552 insertions, 0 deletions

diff --git a/prototypes/connections/check.py b/prototypes/connections/check.py
new file mode 100644
index 0000000..7719537
--- /dev/null
+++ b/prototypes/connections/check.py
@@ -0,0 +1,552 @@
+#!/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 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(), *(ROOT / 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()