Check pure-serve in Pure, not in aup

The HTTP host's end-to-end checks (documents against the earlier server,
lending, kill -9 durability, JSON, timers, links, webhooks and answers) sat
in aup's tests, so a change to the host was verified from the toolchain.
They now live in tests/check.py; make check builds their apps with the
sibling aup and runs them after the unit tests, which stay beside their
parts. Design 009 states the rule for all three repositories.

10 files changed, 1255 insertions, 9 deletions

diff --git a/Makefile b/Makefile
index aed243e..d9033a1 100644
--- a/Makefile
+++ b/Makefile
@@ -14,6 +14,8 @@ BUILD ?= build
 #   cli/    command-line hosts.
 # libpure.a is the app model with the tree store: what a host embeds (Vega's
 # surfaces link it). prototypes/ keeps the earlier document runtime (000-077).
+# Unit tests sit by their part (core/test-module.c, store/test-tree-store.c);
+# tests/check.py checks the hosts end to end.
 INCLUDES = -Icore
 MODEL = core/module.c core/module-list.c core/module-save.c core/module-text.c core/module-store.c core/flow.c
 STORE = store/store-tree.c
@@ -54,6 +56,12 @@ $(BUILD)/test-tree-store: store/test-tree-store.c $(STORE) $(HEADERS)
 	@mkdir -p $(BUILD)
 	$(CC) $(CPPFLAGS) $(CFLAGS) $(INCLUDES) -DPURE_STORE_TEST store/test-tree-store.c $(STORE) $(LDFLAGS) $(LDLIBS) -o $@
 
+# The HTTP host's checks build their apps with aup (the sibling checkout) and
+# compare documents with the earlier document server.
+AUP ?= ../aup
 check: all $(BUILD)/test-module $(BUILD)/test-tree-store
 	$(BUILD)/test-tree-store
 	$(BUILD)/test-module
+	$(MAKE) -C $(AUP) build/aup
+	$(MAKE) -C prototypes/connections BUILD=build/tree STORE=store-tree.c build/tree/pure-connections
+	python3 tests/check.py --aup $(AUP)/build/aup --aup-root $(AUP)
diff --git a/README.md b/README.md
index c7c7958..681c230 100644
--- a/README.md
+++ b/README.md
@@ -23,8 +23,10 @@ The source follows the principle below, a small core with the rest selected arou
 | `serve/` | `pure-serve`: the HTTP host, outbound delivery, JSON read in place | `build/pure-serve` |
 | `cli/` | command-line hosts | `build/pure-run`, `build/pure-edit-module` |
 | `prototypes/` | the earlier document runtime (000-077), built against `core/` and `store/` | `make -C prototypes/connections` |
+| `tests/` | the hosts checked end to end, with apps built by aup (the sibling checkout) | `make check` |
 
-`make` builds the library and the hosts, `make check` runs the store's and the app model's tests. Interfaces for
+`make` builds the library and the hosts; `make check` runs the store's and the app model's unit tests (beside
+their parts) and then `tests/check.py`. Interfaces for
 people are [Vega](/srv/git/vega.git), which links `libpure.a`. The text below records the earlier document runtime
 this grew from.
 
diff --git a/docs/design/008-module-format.md b/docs/design/008-module-format.md
index 84aa97b..b70b8f0 100644
--- a/docs/design/008-module-format.md
+++ b/docs/design/008-module-format.md
@@ -358,7 +358,7 @@ text·리스트가 남을 때만 위치를 유지한다.
   응답만 복사한다. PUT 응답은 요청 본문에서 빌려야 하므로 `edit` 핸들러는 메시지 필드만 응답할 수
   있다(`reply edit.insert`). 시작할 때 확인한다.
 
-성능과 비교는 [연구 075](../research/075-http-host-against-axum.md). 검증(aup `make check`): 무작위 요청에 기존 서버와
+성능과 비교는 [연구 075](../research/075-http-host-against-axum.md). 검증(Pure `make check`의 `tests/check.py`, 앱은 aup가 빌드): 무작위 요청에 기존 서버와
 상태 코드·ETag·오류·본문이 같고, 동시 편집이 커밋을 나누어 쓰며, `kill -9` 뒤에도 확인 응답한 편집이 모두 남는다.
 앱 라우트는 키 있는 리스트 앱에 무작위 POST(쿼리 필드와 본문을 가진 메시지, text, 키)와 GET(수, text, 표의 쪽)을
 800번 보내 상태 코드·오류·본문·JSON 행이 파이썬 모델과 같고, 멈췄다 다시 띄워도, 여섯 클라이언트가 보내는 중에
diff --git a/docs/design/009-layout.md b/docs/design/009-layout.md
index 17d6a48..bcaf64e 100644
--- a/docs/design/009-layout.md
+++ b/docs/design/009-layout.md
@@ -33,6 +33,11 @@
 | `cli/` | 명령행 호스트 | `build/pure-run`, `build/pure-edit-module` |
 | `prototypes/` | 이전 문서 런타임. `core/`와 `store/`를 가져다 빌드한다 | `make -C prototypes/connections` |
 
+검사도 부품을 따른다. 부품의 단위 검사는 그 옆에(`core/test-module.c`, `store/test-tree-store.c`), 저장소 전체를 끝에서
+끝까지 보는 검사는 `tests/check.py`에 둔다. aup와 Vega도 같다(aup `compiler/`, Vega `lib/`·`surfaces/`·`symbols/`, 둘 다
+`tests/check.py`). pure-serve의 HTTP 검사는 전에 aup의 검사 안에 있었다. 호스트를 바꾸면 aup에서 확인해야 했던 것이다.
+이제 Pure의 `make check`가 aup로 앱을 빌드해 돌린다(Vega가 aup와 pure-serve를 쓰는 것과 같은 방향이다).
+
 저장소 계약(`state.h`, `store.h`)은 코어 쪽에 둔다. 코어가 필요한 것을 정하고 저장소 구현이 그것을 채운다. 트리
 저장소는 그 계약의 한 구현이다(이전 세대의 `store.c`, `store-paged.c`도 `store.h`의 구현이다).
 
diff --git a/docs/research/077-apps-as-http-services.md b/docs/research/077-apps-as-http-services.md
index 27e69fd..63904a5 100644
--- a/docs/research/077-apps-as-http-services.md
+++ b/docs/research/077-apps-as-http-services.md
@@ -68,7 +68,7 @@ Raspberry Pi 5, 서버는 CPU 0–2, 부하는 CPU 3, 저장소는 SD 카드의
 
 ## 검증
 
-aup `make check`: 키 있는 리스트 앱에 무작위 요청 800개(쿼리 필드와 본문을 가진 메시지, text, 키 누름, 수·text 보기, 표의
+aup `make check`(지금은 Pure `tests/check.py`의 `server_app`): 키 있는 리스트 앱에 무작위 요청 800개(쿼리 필드와 본문을 가진 메시지, text, 키 누름, 수·text 보기, 표의
 쪽, 없는 이름과 잘못된 쿼리)를 보내 상태 코드·오류 이름·본문·JSON 행이 파이썬 모델과 같다. 한 스레드와 연결 스레드 셋 +
 소유자에서 모두 확인한다. 여섯 클라이언트가 보내는 중에 `kill -9`로 죽여도, 확인 응답한 이벤트(각 3,500–4,200개)가 다시
 시작한 뒤 모두 남는다. ASan·UBSan 빌드에서 같은 검사와, 클라이언트 열둘이 쓰기·읽기·실패를 섞어 보내는 부하(요청 3,600개)를
diff --git a/docs/research/079-lent-views.md b/docs/research/079-lent-views.md
index 1cb8453..668a832 100644
--- a/docs/research/079-lent-views.md
+++ b/docs/research/079-lent-views.md
@@ -38,7 +38,7 @@ Raspberry Pi 5, 서버는 CPU 0–2, 부하는 CPU 3. 1 MiB 메모를 `GET /view
 
 ## 회수 확인
 
-aup `make check`의 `server_lent_views`(스레드 1, 연결 스레드 3 + 소유자):
+aup `make check`의 `server_lent_views`(지금은 Pure `tests/check.py`)(스레드 1, 연결 스레드 3 + 소유자):
 
 - 한 연결에 1 MiB 보기 32개를 이어 보내면 커널 버퍼가 받지 못한 답이 빌린 채 기다린다. 그동안 메모를 다섯 번 바꿔도
   32개 답은 각각 한 버전 전체로 나온다(첫 답은 바꾸기 전 버전).
diff --git a/docs/research/080-snapshot-reads.md b/docs/research/080-snapshot-reads.md
index 1300632..82bf7f7 100644
--- a/docs/research/080-snapshot-reads.md
+++ b/docs/research/080-snapshot-reads.md
@@ -117,7 +117,7 @@ Raspberry Pi 5, 서버는 CPU 0–2, 부하는 CPU 3, 저장소는 SD 카드의
   저장을 기다리는 변경이 있으면 스냅숏이 없고, 실패한 저장 뒤의 스냅숏은 실패 전과 같다. 상태를 메모리에만 두는 앱은
   스냅숏을 만들지 않는다. 두 스레드가 최신 스냅숏을 읽는 동안(2만–4만 번) 앱이 계속 적용하고 게시한다. ASan·UBSan·
   LeakSanitizer 빌드에서 같다.
-- aup `make check`: 무작위 요청 800개의 파이썬 모델 대조가 연결 스레드 하나와 셋에서 같다. 여섯 클라이언트가 쓰는 동안
+- aup `make check`(지금은 Pure `tests/check.py`의 `server_app`): 무작위 요청 800개의 파이썬 모델 대조가 연결 스레드 하나와 셋에서 같다. 여섯 클라이언트가 쓰는 동안
   `kill -9`하는 검사에 다른 연결의 읽기 둘을 더했다. 읽기마다 요청 전에 확인 응답된 점수를 모두 보고, 합계가 뒤로
   가지 않으며, 다시 띄운 서버의 합계는 읽기가 본 어떤 합계 이상이다(저장되지 않은 것을 보지 않았다). 빌려준 보기의
   회수(079)도 그대로다. `pure-serve`의 ASan·UBSan 빌드로 같은 서버 검사를 통과하고, 섞인 부하 뒤 정상 종료에서 누수가
diff --git a/docs/research/082-json-in-place.md b/docs/research/082-json-in-place.md
index 6a456b8..aab6afb 100644
--- a/docs/research/082-json-in-place.md
+++ b/docs/research/082-json-in-place.md
@@ -61,7 +61,7 @@ Raspberry Pi 5, 서버는 CPU 0–2(연결 스레드 하나), 부하는 CPU 3. �
   시작하거나 끝나는 자리는 `limit`, UTF-8이 아닌 자리는 `encoding`으로 실패하고 아무것도 바꾸지 않는다. 경로 이름
   `user.name`은 받고, `user..name`, `.user`, `user.`, `user.1st`, `user name`, 그리고 `user`와 `user.name`을 함께 가진
   모듈은 거절한다.
-- aup `make check`의 `server_json`: 전의 무작위 평평한 JSON 300개와 틀린 요청 14가지에 더해, 그룹이 두 단계인 메시지에
+- aup `make check`의 `server_json`(지금은 Pure `tests/check.py`): 전의 무작위 평평한 JSON 300개와 틀린 요청 14가지에 더해, 그룹이 두 단계인 메시지에
   무작위 JSON 100개(멤버 순서를 단계마다 섞고, 문자열에 이스케이프·제어 바이트·한글·이모지·U+2028)를 보내 답과 표가
   파이썬 모델과 같다. 그룹의 고정 필드를 쿼리 경로로 준 요청, 틀린 그룹 요청 10가지(모르는 멤버, 그룹 자리에 값, 필드
   자리에 객체, 빈 그룹, 두 번 온 필드, 배열, `null`, 외톨이 서로게이트, 닫히지 않은 객체), 쿼리와 본문에 함께 준 필드도
diff --git a/docs/research/083-delivery-over-http.md b/docs/research/083-delivery-over-http.md
index eedc5f8..5f51b2b 100644
--- a/docs/research/083-delivery-over-http.md
+++ b/docs/research/083-delivery-over-http.md
@@ -74,14 +74,14 @@ Raspberry Pi 5. 받는 서버는 CPU 0–1(연결 스레드 2), 보내는 서버
 
 ## 검증
 
-- aup `make check`의 `server_links`: 클라이언트 넷이 보내는 앱에 주문하는 동안 받는 서버를, 그다음 보내는 서버를
+- aup `make check`의 `server_links`(지금은 Pure `tests/check.py`): 클라이언트 넷이 보내는 앱에 주문하는 동안 받는 서버를, 그다음 보내는 서버를
   `kill -9`로 죽이고 같은 저장소로 다시 띄운다. 그 뒤 수량 150(받는 앱이 거절한다)과 7의 주문을 더 넣는다. 받는 앱의 건수와
   합계가 보내는 앱이 저장한 것(거절한 한 건을 빼고)과 같아지고, 확인 응답한 주문은 모두 그 안에 있다. 다섯 번(각 약
   7,300건)과 ASan·UBSan 빌드로 한 번 통과했다. 거절한 주문 뒤의 주문도 전해졌다(막히지 않는다).
-- aup `make check`의 `server_webhook`: 파이썬 HTTPS 서버(자체 인증서)를 제3자 웹훅으로 두고, 주문 다섯 건이 순서대로
+- aup `make check`의 `server_webhook`(지금은 Pure `tests/check.py`): 파이썬 HTTPS 서버(자체 인증서)를 제3자 웹훅으로 두고, 주문 다섯 건이 순서대로
   출력의 JSON 모양과 헤더(`X-Pure-Source`, `Idempotency-Key`, 묶음 헤더 없음)로 도착한다. 처음에 503으로 답한 2번은 다시 온다.
   그 인증서를 믿지 않는 보내는 쪽에서는 아무것도 도착하지 않고 기록이 남는다. ASan·UBSan 빌드로도 통과했다.
-- aup `make check`의 `server_answers`: 파이썬 "결제 API"(1000 이하는 200 `succeeded`, 넘으면 402 `declined`, 같은
+- aup `make check`의 `server_answers`(지금은 Pure `tests/check.py`): 파이썬 "결제 API"(1000 이하는 200 `succeeded`, 넘으면 402 `declined`, 같은
   `Idempotency-Key`에는 첫 답을 되풀이)에 출력을 잇고 응답을 `charged` 입력으로 받는다. 클라이언트 셋이 사는 동안 보내는 서버를
   `kill -9`로 죽였다 살린다. 끝에 앱의 결제·거절 수의 합이 저장된 구매 수와 API가 받은 키 수와 같고, 결제 수와 결제 합계가 API의
   200 응답과 같다(각 3,774–3,979건, 네 번, ASan·UBSan 빌드로 한 번 더).
diff --git a/tests/check.py b/tests/check.py
new file mode 100644
index 0000000..8352578
--- /dev/null
+++ b/tests/check.py
@@ -0,0 +1,1231 @@
+#!/usr/bin/env python3
+"""Pure's end-to-end checks: pure-serve, the HTTP host, running modules that
+aup builds (the sibling checkout, --aup-root). The core's and the store's own
+tests are core/test-module.c and store/test-tree-store.c.
+
+1. Documents: pure-serve running the document editor written in aup
+   (aup's examples/editor.aup) answers random requests exactly as the earlier
+   document server running editor.pure, with one and with three connection
+   threads, on persistent and pipelined connections too.
+2. Lending: a read lends the stored extents and still carries its version
+   while the document changes; lent answers count against the payload budget;
+   a text view is lent while its answer waits and let go when it is read or
+   its connection closes, the store keeping one mapping again.
+3. Durability: concurrent edits share commits, and every acknowledged edit
+   survives kill -9, with one and with three threads; so do the app's
+   acknowledged posts (keyed lists, JSON bodies with groups).
+4. JSON: request bodies are read in place, fields by path, and malformed or
+   mistyped ones are refused with the field named.
+5. Timers run on the owner and their beats are saved like events.
+6. Links: sends are saved with their event and delivered in order and in
+   batches to another app, which applies each once, across kill -9 of either
+   side; webhooks receive them over HTTPS only from a trusted receiver; an
+   API's answers become the app's events (--reply).
+"""
+import argparse
+import json
+import random
+import subprocess
+import sys
+import tempfile
+from pathlib import Path
+
+HERE = Path(__file__).resolve().parent
+ROOT = HERE.parent
+# The app that exercises documents: the editor written in aup, from aup's examples.
+EDITOR = ROOT.parent / 'aup' / 'examples' / 'editor.aup'
+
+
+def run(*args, stdin=None, check=True):
+    return subprocess.run([str(a) for a in args], input=stdin, capture_output=True, text=True,
+                          check=check, timeout=60)
+
+
+def compile_module(aup, source_path, out):
+    run(aup, 'build', source_path, '-o', out)
+    return out.read_bytes()
+
+
+class HttpHost:
+    """A Pure HTTP host on a fresh port; stops with its statistics."""
+
+    def __init__(self, args):
+        import select
+        self.process = subprocess.Popen([str(a) for a in args] + ['--port', '0'], stdout=subprocess.PIPE,
+                                        stderr=subprocess.PIPE)
+        ready, _, _ = select.select([self.process.stdout], [], [], 10)
+        assert ready, 'HTTP host did not start'
+        line = self.process.stdout.readline()
+        assert line, self.process.communicate(timeout=5)
+        self.port = json.loads(line)['port']
+
+    def request(self, method, key, body=b'', expected=None):
+        import http.client
+        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', self.port, timeout=30)
+        try:
+            connection.request(method, f'/objects/{key}', body, headers)
+            response = connection.getresponse()
+            return response.status, response.getheader('ETag'), response.getheader('X-Pure-Error'), response.read()
+        finally:
+            connection.close()
+
+    def stop(self):
+        self.process.terminate()
+        _, error = self.process.communicate(timeout=30)
+        assert self.process.returncode == 0, error[-1000:]
+        return [json.loads(line) for line in error.decode().splitlines() if line.startswith('{')]
+
+
+def server_parity(aup, pure, work, seeds, operations=150, threads=1):
+    """The document server running editor.aup answers every request exactly
+    as the server running editor.pure: status, revision, error and body."""
+    import struct
+    module = work / 'editor-server.pmod'
+    compile_module(aup, EDITOR, module)
+    compared = 0
+    for seed in seeds:
+        rng = random.Random(seed + 13000)
+        old = HttpHost([pure / 'prototypes/connections/build/tree/pure-connections', '--program', pure / 'prototypes/connections/examples/editor.pure',
+                        '--store', work / f'http-old-{seed}-{threads}'])
+        new = HttpHost([pure / 'build/pure-serve', '--module', module, '--store', work / f'http-new-{seed}-{threads}',
+                        '--threads', str(threads)])
+        try:
+            revision, size = {}, {}
+            for _ in range(operations):
+                key = rng.choice(['a', 'b', 'long-key_1'])
+                current, length = revision.get(key, 0), size.get(key, 0)
+                if rng.random() < 0.35:
+                    request = ('GET', key, b'', None)
+                else:
+                    pick = rng.random()
+                    base = current if pick < 0.75 else max(current - 1, 0) if pick < 0.9 else current + 3
+                    if rng.random() < 0.05:
+                        body = bytes(rng.randrange(256) for _ in range(rng.randint(0, 7)))  # too short
+                    else:
+                        insert = bytes(rng.randrange(256) for _ in range(rng.choice([0, 1, 7, 300, 5000])))
+                        body = struct.pack('<II', rng.randint(0, length + 2), rng.choice([0, 1, 3, length])) + insert
+                    request = ('PUT', key, body, base)
+                theirs, ours = old.request(*request), new.request(*request)
+                assert ours == theirs, (seed, request[:2], request[3], ours[:3], theirs[:3])
+                if theirs[0] == 200 and theirs[1]:
+                    revision[key] = int(theirs[1].strip('"'))
+                    if request[0] == 'GET':
+                        size[key] = len(theirs[3])
+                    else:
+                        size[key] = len(old.request('GET', key)[3])
+                        assert new.request('GET', key) == old.request('GET', key)
+                compared += 1
+        finally:
+            old.stop()
+            new.stop()
+    return compared
+
+
+def server_pipelining(aup, pure, work):
+    """One connection, many requests: a keep-alive client and a pipelined
+    burst (all requests in one write, one split across writes) get the same
+    answers, in order, as separate connections would."""
+    import http.client
+    import socket
+    import struct
+    module = work / 'editor-pipeline.pmod'
+    compile_module(aup, EDITOR, module)
+    host = HttpHost([pure / 'build/pure-serve', '--module', module, '--store', work / 'pipeline'])
+    try:
+        connection = http.client.HTTPConnection('127.0.0.1', host.port, timeout=30)
+        answers = []
+        for body, headers in [(struct.pack('<II', 0, 0) + b'hello', {'If-None-Match': '*'}),
+                              (struct.pack('<II', 5, 0) + b'!', {'If-Match': '"1"'}),
+                              (struct.pack('<II', 0, 0) + b'x', {'If-Match': '"1"'})]:
+            connection.request('PUT', '/objects/p', body, headers)
+            response = connection.getresponse()
+            answers.append((response.status, response.getheader('ETag'), response.read()))
+        connection.request('GET', '/objects/p')
+        response = connection.getresponse()
+        answers.append((response.status, response.getheader('ETag'), response.read()))
+        connection.close()
+        assert answers == [(200, '"1"', b'hello'), (200, '"2"', b'!'), (412, '"2"', b''), (200, '"2"', b'hello!')], answers
+
+        def request(method, key, body=b'', condition=''):
+            return (f'{method} /objects/{key} HTTP/1.1\r\nHost: x\r\nContent-Length: {len(body)}\r\n{condition}\r\n'
+                    .encode() + body)
+        pipeline = (request('GET', 'p') + request('PUT', 'q', struct.pack('<II', 0, 0) + b'q1', 'If-None-Match: *\r\n') +
+                    request('GET', 'q') + request('GET', 'missing') +
+                    request('PUT', 'q', struct.pack('<II', 2, 0) + b'!', 'If-Match: "1"\r\n') + request('GET', 'q'))
+        expected = [(200, b'hello!'), (200, b'q1'), (200, b'q1'), (404, b''), (200, b'!'), (200, b'q1!')]
+        for split in (len(pipeline), 7, 61):  # one write, and small writes that cut heads and bodies
+            with socket.create_connection(('127.0.0.1', host.port), timeout=30) as sock:
+                for at in range(0, len(pipeline), split):
+                    sock.sendall(pipeline[at:at + split])
+                reader = sock.makefile('rb')
+                for status, body in expected:
+                    code = int(reader.readline().split()[1])
+                    length = 0
+                    for line in iter(reader.readline, b'\r\n'):
+                        name, _, value = line.decode().partition(':')
+                        if name.lower() == 'content-length':
+                            length = int(value)
+                    assert (code, reader.read(length)) == (status, body), (split, code, status, body)
+            # The next round starts from the same documents.
+            host.request('PUT', 'q', struct.pack('<II', 0, 3) + b'q1', int(host.request('GET', 'q')[1].strip('"')))
+            expected = [(200, b'hello!'), (412, b''), (200, b'q1'), (404, b''), (412, b''), (200, b'q1')]
+            pipeline = (request('GET', 'p') + request('PUT', 'q', struct.pack('<II', 0, 0) + b'q1', 'If-None-Match: *\r\n') +
+                        request('GET', 'q') + request('GET', 'missing') +
+                        request('PUT', 'q', struct.pack('<II', 2, 0) + b'!', 'If-Match: "1"\r\n') + request('GET', 'q'))
+    finally:
+        stats = host.stop()
+    assert stats[0]['active'] == 0 and stats[0]['retained_payload_bytes'] == 0, stats
+    return len(answers) + 3 * len(expected)
+
+
+def server_lent_reads(aup, pure, work):
+    """A read lends the stored document's extents instead of copying them: a
+    1 MiB answer sent slowly still carries the version read while the
+    document is replaced five times meanwhile."""
+    import socket
+    import struct
+    module = work / 'editor-lent.pmod'
+    compile_module(aup, EDITOR, module)
+    host = HttpHost([pure / 'build/pure-serve', '--module', module, '--store', work / 'lent'])
+    big = bytes(range(256)) * 4096
+    try:
+        assert host.request('PUT', 'big', struct.pack('<II', 0, 0) + big, 0)[0] == 200
+        with socket.create_connection(('127.0.0.1', host.port), timeout=30) as sock:
+            sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 4096)
+            sock.sendall(b'GET /objects/big HTTP/1.1\r\nHost: x\r\nContent-Length: 0\r\n\r\n')
+            data = sock.recv(100)
+            for revision in range(1, 6):
+                edit = struct.pack('<II', 0, len(big) if revision == 1 else 1) + b'Z'
+                assert host.request('PUT', 'big', edit, revision)[0] == 200
+            while b'\r\n\r\n' not in data or len(data) < data.index(b'\r\n\r\n') + 4 + len(big):
+                chunk = sock.recv(65536)
+                assert chunk, len(data)
+                data += chunk
+        assert data[data.index(b'\r\n\r\n') + 4:] == big
+        assert host.request('GET', 'big') == (200, '"6"', 'none', b'Z')
+    finally:
+        stats = host.stop()
+    assert stats[0]['active'] == 0 and stats[0]['retained_payload_bytes'] == 0, stats
+    return len(big)
+
+
+def server_lent_budget(aup, pure, work):
+    """Lent extents count against the payload budget while an answer waits
+    on its socket. Two readers pipeline 1 MiB reads without reading: the
+    kernel takes what it can and each reader leaves one answer waiting. A
+    third request finds no room (503 payload_budget) until a reader goes."""
+    import socket
+    import struct
+    import time
+    module = work / 'editor-budget.pmod'
+    compile_module(aup, EDITOR, module)
+    big = bytes(range(256)) * 4096
+    budget = 2 * len(big) + len(big) // 2
+    host = HttpHost([pure / 'build/pure-serve', '--module', module, '--store', work / 'lent-budget',
+                     '--budget', budget])
+    readers = []
+    try:
+        assert host.request('PUT', 'big', struct.pack('<II', 0, 0) + big, 0)[0] == 200
+        for _ in range(2):
+            sock = socket.create_connection(('127.0.0.1', host.port), timeout=30)
+            sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 4096)
+            sock.sendall(b'GET /objects/big HTTP/1.1\r\nHost: x\r\nContent-Length: 0\r\n\r\n' * 32)
+            readers.append(sock)
+        time.sleep(0.3)
+        refused = host.request('GET', 'big')
+        assert refused[0] == 503 and refused[2] == 'payload_budget', refused[:3]
+        readers.pop().close()
+        for _ in range(100):
+            answered = host.request('GET', 'big')
+            if answered[0] == 200:
+                break
+            time.sleep(0.05)
+        assert answered[0] == 200 and answered[3] == big, answered[:3]
+    finally:
+        for sock in readers:
+            sock.close()
+        stats = host.stop()
+    assert stats[0]['active'] == 0 and stats[0]['retained_payload_bytes'] == 0, stats
+    return budget
+
+
+NOTE_SOURCE = """state note: text
+input write = message {
+  text: text
+}
+on write { note = write.text }
+view page = note
+view size = note.bytes
+"""
+
+
+def server_lent_views(aup, pure, work, threads=1):
+    """A view that shows a text state as it is lends the state's stored
+    extents. Answers pipelined on one connection and left waiting hold the
+    store mapping they were read from while the note is replaced: each comes
+    out whole in one version. When a reader goes without reading, the
+    mappings its answers held are unmapped and nothing stays charged. With
+    several threads the owner lends and a connection thread gives back."""
+    import socket
+    import time
+    module, source = work / 'note.pmod', work / 'note.aup'
+    source.write_text(NOTE_SOURCE)
+    compile_module(aup, source, module)
+    store = work / f'lent-views-{threads}'
+    host = HttpHost([pure / 'build/pure-serve', '--module', module, '--store', store, '--threads', threads])
+
+    def write(text):
+        body = json.dumps({'text': text}).encode()
+        import http.client
+        connection = http.client.HTTPConnection('127.0.0.1', host.port, timeout=30)
+        connection.request('POST', '/inputs/write', body, {'Content-Type': 'application/json'})
+        assert connection.getresponse().status == 200
+        connection.close()
+
+    def mappings():
+        # A compaction replaces the file: a mapping of the old one shows as "(deleted)".
+        path = str((store / 'pure.tree').resolve())
+        return sum(1 for line in open(f'/proc/{host.process.pid}/maps') if path in line)
+
+    def settle(to):
+        for _ in range(100):  # an answer's ticket is released on its thread
+            if mappings() <= to: return mappings()
+            time.sleep(0.02)
+        return mappings()
+
+    def pipeline(count):
+        # Far more than the kernel's buffers take (a few MiB): an answer waits, lent.
+        sock = socket.create_connection(('127.0.0.1', host.port), timeout=30)
+        sock.sendall(b'GET /views/page HTTP/1.1\r\nHost: x\r\n\r\n' * count)
+        time.sleep(0.3)
+        return sock
+
+    size = 1 << 20
+    try:
+        write('x' * size)
+        assert http(host.port, 'GET', '/views/size')[3] == str(size).encode()
+        alone = mappings()
+        # Whole versions, however the note changes while answers wait.
+        sock = pipeline(32)
+        for round_ in range(5):
+            write(chr(ord('A') + round_) * size)
+        held = mappings()
+        data = b''
+        while data.count(b'HTTP/1.1 200') < 32 or len(data) < 32 * size:
+            chunk = sock.recv(1 << 20)
+            assert chunk, len(data)
+            data += chunk
+        sock.close()
+        answers = [part.split(b'\r\n\r\n', 1) for part in data.split(b'HTTP/1.1 200 OK\r\n')[1:]]
+        bodies = [body[:size] for _, body in answers]
+        assert all(b'text/plain' in head for head, _ in answers) and bodies[0] == b'x' * size, [b[:1] for b in bodies]
+        assert all(len(set(body)) == 1 and len(body) == size for body in bodies), [len(b) for b in bodies]
+        after_reading = settle(alone)
+        # A reader that goes without reading gives back what its answers held.
+        sock = pipeline(32)
+        for round_ in range(3):
+            write(chr(ord('a') + round_) * size)
+        held_again = mappings()
+        sock.close()
+        after_leaving = settle(alone)
+        assert held > alone and held_again > alone and after_reading <= alone and after_leaving <= alone, \
+            (alone, held, after_reading, held_again, after_leaving)
+        assert http(host.port, 'GET', '/views/page')[3] == b'c' * size
+    finally:
+        stats = host.stop()
+    assert stats[0]['active'] == 0 and stats[0]['retained_payload_bytes'] == 0, stats
+    return {'alone': alone, 'while_lent': max(held, held_again), 'after': after_leaving}
+
+
+def server_bursts(aup, pure, work, clients=8, edits=30, shards=1):
+    """Concurrent edits share commits, and every acknowledged edit survives a
+    kill -9 of the server."""
+    import http.client
+    import signal
+    import struct
+    import threading
+    import time
+    module = work / 'editor-bursts.pmod'
+    compile_module(aup, EDITOR, module)
+
+    def load(host, rounds, acknowledged):
+        def client(number):
+            key, base = f'c{number}', acknowledged[number][1]
+            for _ in range(rounds):
+                try:
+                    status, tag, _, _ = host.request('PUT', key, struct.pack('<II', 0, 0) + b'x', base)
+                except (OSError, http.client.HTTPException):  # the server died mid-answer: not acknowledged
+                    return
+                if status != 200:
+                    return
+                base = int(tag.strip('"'))
+                acknowledged[number] = (acknowledged[number][0] + 1, base)
+        threads = [threading.Thread(target=client, args=(n,)) for n in range(clients)]
+        start = time.perf_counter()
+        for thread in threads:
+            thread.start()
+        return threads, start
+
+    timings = {}
+    for name, args in [('program', [pure / 'prototypes/connections/build/tree/pure-connections', '--program', pure / 'prototypes/connections/examples/editor.pure']),
+                       ('module', [pure / 'build/pure-serve', '--module', module, '--threads', str(shards)])]:
+        host = HttpHost(args + ['--store', work / f'bursts-{name}-{shards}'])
+        acknowledged = [(0, 0)] * clients
+        threads, start = load(host, edits, acknowledged)
+        for thread in threads:
+            thread.join()
+        timings[name] = time.perf_counter() - start
+        assert all(count == edits for count, _ in acknowledged), acknowledged
+        stats = host.stop()
+        if name == 'module':
+            bursts = stats[0]['bursts']
+            assert stats[0]['burst_events'] == clients * edits and (clients == 1 or bursts < clients * edits), stats
+    # Kill the module server while clients edit; what it acknowledged is stored.
+    store = work / f'bursts-module-{shards}'
+    host = HttpHost([pure / 'build/pure-serve', '--module', module, '--store', store, '--threads', str(shards)])
+    acknowledged = [(edits, edits)] * clients
+    threads, _ = load(host, 1000, acknowledged)
+    time.sleep(0.5)
+    host.process.send_signal(signal.SIGKILL)
+    host.process.wait()
+    for thread in threads:
+        thread.join()
+    host = HttpHost([pure / 'build/pure-serve', '--module', module, '--store', store, '--threads', str(shards)])
+    try:
+        for number, (count, base) in enumerate(acknowledged):
+            status, tag, _, body = host.request('GET', f'c{number}')
+            stored = int(tag.strip('"'))
+            assert status == 200 and stored in (base, base + 1) and len(body) == stored, (number, count, base, tag)
+    finally:
+        host.stop()
+    return {'writes': clients * edits, 'bursts': bursts, 'program_seconds': round(timings['program'], 3),
+            'module_seconds': round(timings['module'], 3),
+            'acknowledged_before_kill': sum(count for count, _ in acknowledged) - clients * edits}
+
+
+SERVED_SOURCE = """record Entry {
+  name: text
+  points = 0
+}
+state board: list of Entry by name
+state at: index of board
+input score = message {
+  points: u32
+  name: bytes
+}
+input join = text
+input drop = key Delete
+on score {
+  if !board.has(text(score.name)) { board.add(Entry { name: text(score.name) }) }
+  board[text(score.name)].points = board[text(score.name)].points + score.points
+  at = board[text(score.name)]
+  reply score.name
+}
+on join {
+  board.add(Entry { name: join })
+  at = board[join]
+}
+on drop.down { board.remove(at) }
+view total = sum(e.points for e in board)
+view current = at.name
+view leaders = each e in board {
+  name = e.name
+  points = e.points
+}
+"""
+
+
+def http(port, method, path, body=b''):
+    """One request on its own connection: status, X-Pure-Error, content type, body."""
+    import http.client
+    connection = http.client.HTTPConnection('127.0.0.1', port, timeout=30)
+    try:
+        connection.request(method, path, body, {'Content-Length': str(len(body))} if method != 'GET' else {})
+        response = connection.getresponse()
+        return response.status, response.getheader('X-Pure-Error'), response.getheader('Content-Type'), response.read()
+    finally:
+        connection.close()
+
+
+JSON_SOURCE = """record Note {
+  title: text
+  body: text
+  points = 0
+}
+state notes: list of Note by title
+state counted = 0
+input add = message {
+  title: text
+  points: number
+  body: text
+}
+input count = message {
+  n: u32
+}
+record Shipment {
+  title: text
+  city: text
+  note: text
+  zip = 0
+}
+state shipped: list of Shipment
+input ship = message {
+  title: text
+  to: {
+    city: text
+    zip: u32
+    door: {
+      note: text
+    }
+  }
+}
+on add {
+  notes.add(Note { title: add.title, body: add.body, points: add.points })
+  reply add.title
+}
+on count { counted = counted + count.n }
+on ship {
+  shipped.append(Shipment { title: ship.title, city: ship.to.city, note: ship.to.door.note, zip: ship.to.zip })
+  reply ship.to.door.note
+}
+view counted_view = counted
+view all = each n in notes {
+  title = n.title
+  body = n.body
+  points = n.points
+}
+view shipments = each s in shipped {
+  title = s.title
+  city = s.city
+  note = s.note
+  zip = s.zip
+}
+"""
+
+
+def server_json(aup, pure, work, rounds=300):
+    """Messages of text and number fields posted as JSON objects: strings
+    come out as Python's json module reads them (escapes, control bytes,
+    Korean, surrogate pairs), numbers as it reads them; a missing, unknown,
+    doubled or mistyped field, broken JSON and a lone surrogate answer 400."""
+    module, source = work / 'json.pmod', work / 'json.aup'
+    source.write_text(JSON_SOURCE)
+    compile_module(aup, source, module)
+    host = HttpHost([pure / 'build/pure-serve', '--module', module, '--store', work / 'json-store'])
+    rng = random.Random(77)
+    pieces = ['a', 'Z', ' ', '"', '\\', '/', '\n', '\t', '\x01', '\x1f', 'é', '가', '😀', '\u2028', 'x' * 20]
+    notes = {}
+
+    def post(path, body, json_body=True):
+        import http.client
+        connection = http.client.HTTPConnection('127.0.0.1', host.port, timeout=30)
+        try:
+            headers = {'Content-Length': str(len(body))}
+            if json_body: headers['Content-Type'] = 'application/json'
+            connection.request('POST', path, body, headers)
+            response = connection.getresponse()
+            return response.status, response.getheader('X-Pure-Error'), response.read()
+        finally:
+            connection.close()
+
+    try:
+        for _ in range(rounds):
+            title = ''.join(rng.choice(pieces) for _ in range(rng.randint(0, 6)))
+            body = ''.join(rng.choice(pieces) for _ in range(rng.randint(0, 4)))
+            points = rng.choice([0, 1, -2.5, 1e3, 3.25e-2, 12345678.5])
+            members = [('title', title), ('points', points), ('body', body)]
+            rng.shuffle(members)
+            text = json.dumps(dict(members), ensure_ascii=rng.random() < 0.5)
+            text = text.replace(': ', ':' + ' ' * rng.randint(0, 2)).replace(', ', ',\n' if rng.random() < 0.3 else ', ')
+            got = post('/inputs/add', text.encode())
+            if title in notes:
+                assert got[:2] == (409, 'exists'), (text, got)
+                continue
+            assert got == (200, 'none', title.encode()), (text, got)
+            notes[title] = (body, points)
+        rows = json.loads(http(host.port, 'GET', '/views/all')[3])['items']
+        want = [{'title': t, 'body': notes[t][0], 'points': notes[t][1]} for t in sorted(notes, key=lambda t: t.encode())]
+        assert rows == want, [(a, b) for a, b in zip(rows, want) if a != b][:2]
+        # The query gives fixed fields; a body without JSON fills a message's one variable field only.
+        assert post('/inputs/count?n=3', b'', json_body=False)[:2] == (200, 'none')
+        assert post('/inputs/count', b'{"n": 4}')[:2] == (200, 'none')
+        assert http(host.port, 'GET', '/views/counted_view')[3] == b'7'
+        assert post('/inputs/add?points=2', b'{"title": "q-from-query", "body": ""}')[:2] == (200, 'none')
+        wrong = [
+            (b'{"title": "m", "points": 1}', 'missing_field'),
+            (b'{"title": "m", "points": 1, "body": "", "extra": 1}', 'json_field'),
+            (b'{"title": "m", "title": "n", "points": 1, "body": ""}', 'json_field'),
+            (b'{"title": 1, "points": 1, "body": ""}', 'json_type'),
+            (b'{"title": "m", "points": "1", "body": ""}', 'json_type'),
+            (b'{"title": "m", "points": 1e999, "body": ""}', 'json_type'),
+            (b'{"title": "m", "points": 01, "body": ""}', 'json'),
+            (b'{"title": "m", "points": NaN, "body": ""}', 'json'),
+            (b'{"title": "m", "points": 1, "body": "",}', 'json'),
+            (b'{"title": "\\ud800", "points": 1, "body": ""}', 'json'),
+            (b'{"title": "\xff", "points": 1, "body": ""}', 'json'),
+            (b'{"title": "m", "points": 1, "body": ""} x', 'json'),
+            (b'[1]', 'json'),
+            (b'{"title": "m\n", "points": 1, "body": ""}', 'json'),  # a raw newline in a string
+        ]
+        for body, error in wrong:
+            assert post('/inputs/add', body)[:2] == (400, error), (body, post('/inputs/add', body))
+        assert post('/inputs/add?points=2', b'{"title": "m", "points": 1, "body": ""}')[:2] == (400, 'json_field')
+        assert post('/inputs/add', b'plain', json_body=False)[:2] == (400, 'json')  # two text fields need JSON
+        assert post('/inputs/count', b'{"n": -1}')[:2] == (400, 'json_type')
+        assert post('/inputs/count', b'{"n": 4294967296}')[:2] == (400, 'json_type')
+        # Groups: objects inside the body, members in any order, strings
+        # read where they are (escaped ones unescaped over themselves).
+        shipments = []
+        for _ in range(rounds // 3):
+            title, city, note = (''.join(rng.choice(pieces) for _ in range(rng.randint(0, 5))) for _ in range(3))
+            zip_code = rng.randint(0, 99999)
+            door = [('note', note)]
+            to = [('city', city), ('zip', zip_code), ('door', dict(door))]
+            rng.shuffle(to)
+            members = [('title', title), ('to', dict(to))]
+            rng.shuffle(members)
+            text = json.dumps(dict(members), ensure_ascii=rng.random() < 0.5)
+            got = post('/inputs/ship', text.encode())
+            assert got == (200, 'none', note.encode()), (text, got)
+            shipments.append({'title': title, 'city': city, 'note': note, 'zip': zip_code})
+        # A group's fixed field from the query, by its path.
+        assert post('/inputs/ship?to.zip=7', b'{"title": "q", "to": {"city": "c", "door": {"note": "n"}}}')[:2] == \
+            (200, 'none')
+        shipments.append({'title': 'q', 'city': 'c', 'note': 'n', 'zip': 7})
+        rows = json.loads(http(host.port, 'GET', f'/views/shipments?count={len(shipments)}')[3])['items']
+        assert rows == shipments, [(a, b) for a, b in zip(rows, shipments) if a != b][:2]
+        grouped = '{"title": "t", "to": {"city": "c", "zip": 1, "door": {"note": "n"}}}'
+        wrong_groups = [
+            (grouped.replace('"zip": 1', '"zip": 1, "lat": 2'), 'json_field'),         # unknown member of a group
+            (grouped.replace('"note": "n"', '"note": "n", "bell": {"x": 1}'), 'json_field'),
+            (grouped.replace('"door": {"note": "n"}', '"door": "n"'), 'json_type'),    # a group as a value
+            (grouped.replace('"title": "t"', '"title": {"a": "t"}'), 'json_type'),    # a field as a group
+            (grouped.replace('"door": {"note": "n"}', '"door": {}'), 'missing_field'),
+            (grouped.replace('"zip": 1', '"zip": 1, "zip": 2'), 'json_field'),
+            (grouped.replace('"zip": 1', '"zip": [1]'), 'json_type'),
+            (grouped.replace('"zip": 1', '"zip": null'), 'json_type'),
+            (grouped.replace('"note": "n"', '"note": "n\\ud800"'), 'json'),
+            (grouped.replace('"door": {"note": "n"}}', '"door": {"note": "n"}'), 'json'),  # unclosed
+        ]
+        for body, error in wrong_groups:
+            assert post('/inputs/ship', body.encode())[:2] == (400, error), (body, post('/inputs/ship', body.encode()))
+        assert post('/inputs/ship?to.zip=7', grouped.encode())[:2] == (400, 'json_field')  # given twice
+    finally:
+        host.stop()
+    return rounds + len(wrong) + rounds // 3 + len(wrong_groups)
+
+
+LINK_SENDER = """state n = 0
+state total = 0
+input buy = message {
+  item: text
+  count: u32
+}
+output ordered = message {
+  item: text
+  count: u32
+  to: {
+    city: text
+  }
+}
+on buy {
+  n = n + 1
+  total = total + buy.count
+  send ordered { item: buy.item, count: buy.count, to: { city: "Seoul" } }
+}
+view sent = n
+view sent_total = total
+"""
+
+LINK_RECEIVER = """state stock = 0
+state orders = 0
+input order = message {
+  item: text
+  count: u32
+  to: {
+    city: text
+  }
+}
+on order {
+  require order.count < 100
+  stock = stock + order.count
+  orders = orders + 1
+}
+view total = stock
+view count = orders
+"""
+
+
+def server_links(aup, pure, work):
+    """Two served apps, one's output wired to the other's input over HTTP
+    (--send). While four clients buy from the sender, the receiver and then
+    the sender are killed (-9) and started again on their stores. In the end
+    the receiver has applied every order the sender saved exactly once (its
+    total and count equal the sender's), a refused order (count 150) was
+    delivered once and did not hold back the order after it, and every
+    acknowledged order is among the saved ones."""
+    import select
+    import signal
+    import socket
+    import threading
+    import time
+    sender_module, receiver_module = work / 'link-sender.pmod', work / 'link-receiver.pmod'
+    for module, text in ((sender_module, LINK_SENDER), (receiver_module, LINK_RECEIVER)):
+        source = module.with_suffix('.aup')
+        source.write_text(text)
+        compile_module(aup, source, module)
+
+    def free_port():
+        with socket.socket() as probe:
+            probe.bind(('127.0.0.1', 0))
+            return probe.getsockname()[1]
+
+    def start(args, port):
+        process = subprocess.Popen([str(a) for a in args] + ['--port', str(port)], stdout=subprocess.PIPE,
+                                   stderr=subprocess.PIPE)
+        ready, _, _ = select.select([process.stdout], [], [], 10)
+        assert ready and process.stdout.readline(), 'host did not start'
+        return process
+
+    def post(port, path, body):
+        import http.client
+        connection = http.client.HTTPConnection('127.0.0.1', port, timeout=10)
+        try:
+            data = json.dumps(body).encode()
+            connection.request('POST', path, data, {'Content-Length': str(len(data)), 'Content-Type': 'application/json'})
+            response = connection.getresponse()
+            response.read()
+            return response.status
+        finally:
+            connection.close()
+
+    def number(port, view):
+        return int(http(port, 'GET', f'/views/{view}')[3])
+
+    port_a, port_b = free_port(), free_port()
+    receiver = [pure / 'build/pure-serve', '--module', receiver_module, '--store', work / 'link-b', '--threads', '2']
+    sender = [pure / 'build/pure-serve', '--module', sender_module, '--store', work / 'link-a',
+              '--send', f'ordered=http://127.0.0.1:{port_b}/inputs/order']
+    hosts = {'b': start(receiver, port_b), 'a': start(sender, port_a)}
+    acknowledged = [0] * 4
+    stop = threading.Event()
+
+    def client(which):
+        rng = random.Random(which + 7700)
+        while not stop.is_set():
+            count = rng.randint(1, 20)
+            try:
+                status = post(port_a, '/inputs/buy', {'item': f'c{which}', 'count': count})
+            except OSError:
+                time.sleep(0.02)
+                continue
+            if status == 200:
+                acknowledged[which] += count
+            else:
+                time.sleep(0.02)
+
+    clients = [threading.Thread(target=client, args=(n,)) for n in range(len(acknowledged))]
+    for thread in clients:
+        thread.start()
+    try:
+        for name, args, port in (('b', receiver, port_b), ('a', sender, port_a)):
+            time.sleep(0.6)
+            hosts[name].send_signal(signal.SIGKILL)
+            hosts[name].wait()
+            time.sleep(0.3)
+            hosts[name] = start(args, port)
+        time.sleep(0.6)
+    finally:
+        stop.set()
+        for thread in clients:
+            thread.join()
+    try:
+        assert post(port_a, '/inputs/buy', {'item': 'big', 'count': 150}) == 200
+        assert post(port_a, '/inputs/buy', {'item': 'after', 'count': 7}) == 200
+        deadline = time.time() + 30
+        while True:
+            sent, sent_total = number(port_a, 'sent'), number(port_a, 'sent_total')
+            count, total = number(port_b, 'count'), number(port_b, 'total')
+            if (count, total) == (sent - 1, sent_total - 150) or time.time() > deadline:
+                break
+            time.sleep(0.1)
+        assert (count, total) == (sent - 1, sent_total - 150), (count, total, sent, sent_total)
+        assert sent_total >= sum(acknowledged) + 157 and sum(acknowledged) > 0, (sent_total, acknowledged)
+    finally:
+        stats = {}
+        for name, process in hosts.items():
+            process.terminate()
+            _, error = process.communicate(timeout=30)
+            assert process.returncode == 0, error[-1000:]
+            stats[name] = [json.loads(line) for line in error.decode().splitlines() if line.startswith('{"active"')][-1]
+    return {'orders': sent, 'acknowledged_total': sum(acknowledged), 'delivered_after_restart': stats['a']['delivered'],
+            'attempts_after_restart': stats['a']['delivery_attempts']}
+
+
+def server_webhook(aup, pure, work):
+    """An output wired to a third-party HTTPS endpoint (a Python server with
+    a certificate of its own): five orders arrive in order as JSON with the
+    sender's headers, one answered 503 at first arrives again, and none goes
+    to a receiver whose certificate the sender does not trust."""
+    from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+    import ssl
+    import threading
+    import time
+    module = work / 'hook-sender.pmod'
+    module.with_suffix('.aup').write_text(LINK_SENDER)
+    compile_module(aup, module.with_suffix('.aup'), module)
+    for name in ('trusted', 'other'):
+        subprocess.run(['openssl', 'req', '-x509', '-newkey', 'ec', '-pkeyopt', 'ec_paramgen_curve:P-256', '-nodes',
+                        '-keyout', work / f'{name}.key', '-out', work / f'{name}.pem', '-days', '1',
+                        '-subj', '/CN=127.0.0.1', '-addext', 'subjectAltName=IP:127.0.0.1'],
+                       check=True, capture_output=True)
+    received, refused_once = [], set()
+
+    class Hook(BaseHTTPRequestHandler):
+        protocol_version = 'HTTP/1.1'
+
+        def do_POST(self):
+            body = self.rfile.read(int(self.headers['Content-Length']))
+            received.append((dict(self.headers), json.loads(body)))
+            status = 200
+            if self.headers['X-Pure-Sequence'] == '2' and '2' not in refused_once:
+                refused_once.add('2')
+                status = 503
+            self.send_response(status)
+            self.send_header('Content-Length', '0')
+            self.end_headers()
+
+        def log_message(self, *args):
+            pass
+
+    server = ThreadingHTTPServer(('127.0.0.1', 0), Hook)
+    context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
+    context.load_cert_chain(work / 'trusted.pem', work / 'trusted.key')
+    server.socket = context.wrap_socket(server.socket, server_side=True)
+    threading.Thread(target=server.serve_forever, daemon=True).start()
+    url = f'ordered=https://127.0.0.1:{server.server_address[1]}/hook'
+    try:
+        def buy(host, count):
+            import http.client
+            connection = http.client.HTTPConnection('127.0.0.1', host.port, timeout=10)
+            data = json.dumps({'item': f'i{count}', 'count': count}).encode()
+            connection.request('POST', '/inputs/buy', data, {'Content-Length': str(len(data)), 'Content-Type': 'application/json'})
+            assert connection.getresponse().status == 200
+            connection.close()
+
+        host = HttpHost([pure / 'build/pure-serve', '--module', module, '--store', work / 'hook-a', '--send', url,
+                         '--send-ca', work / 'trusted.pem'])
+        try:
+            for count in range(1, 6):
+                buy(host, count)
+            deadline = time.time() + 20
+            while len({h['X-Pure-Sequence'] for h, _ in received}) < 5 and time.time() < deadline:
+                time.sleep(0.05)
+        finally:
+            stats = host.stop()[-1]
+        firsts = []
+        for headers, body in received:
+            seq = int(headers['X-Pure-Sequence'])
+            assert headers['X-Pure-Source'] == 'hook-sender.ordered' and headers['Content-Type'] == 'application/json'
+            assert headers['Idempotency-Key'] == f'hook-sender.ordered-{seq}' and 'X-Pure-Count' not in headers
+            assert body == {'item': f'i{seq}', 'count': seq, 'to': {'city': 'Seoul'}}, body
+            if seq not in firsts:
+                firsts.append(seq)
+        assert firsts == [1, 2, 3, 4, 5] and len(received) == 6 and stats['delivered'] == 5, (firsts, len(received), stats)
+        # A receiver the sender cannot trust gets nothing; the send waits in the outbox.
+        before = len(received)
+        host = HttpHost([pure / 'build/pure-serve', '--module', module, '--store', work / 'hook-b', '--send', url,
+                         '--send-ca', work / 'other.pem'])
+        try:
+            buy(host, 1)
+            time.sleep(1)
+        finally:
+            host.process.terminate()
+            _, error = host.process.communicate(timeout=30)
+        assert len(received) == before and b'not trusted' in error, error[-500:]
+    finally:
+        server.shutdown()
+    return {'hooks': len(received), 'delivered': stats['delivered']}
+
+
+PAY_SOURCE = """state sent = 0
+state paid = 0
+state declined = 0
+state paid_total = 0
+input buy = message {
+  order: text
+  amount: u32
+}
+output charge = message {
+  order: text
+  amount: u32
+}
+input charged = message {
+  order: text
+  amount: u32
+  status: text
+  http_status: u32
+}
+on buy {
+  sent = sent + 1
+  send charge { order: buy.order, amount: buy.amount }
+}
+on charged {
+  if charged.http_status == 200 {
+    paid = paid + 1
+    paid_total = paid_total + charged.amount
+  } else {
+    declined = declined + 1
+  }
+}
+view sent_view = sent
+view paid_view = paid
+view declined_view = declined
+view paid_total_view = paid_total
+"""
+
+
+def server_answers(aup, pure, work):
+    """An output wired to a payments API whose answers come back as events
+    (--reply): the API answers 200 or 402 and gives a key's first answer to
+    every request with that Idempotency-Key. While clients buy, the sender is
+    killed (-9) and started again; in the end every saved order was answered
+    exactly once: the app's paid and declined counts and paid total match the
+    API's answers, one per key."""
+    from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+    import select
+    import signal
+    import socket
+    import threading
+    import time
+    module = work / 'pay.pmod'
+    module.with_suffix('.aup').write_text(PAY_SOURCE)
+    compile_module(aup, module.with_suffix('.aup'), module)
+    answers, lock = {}, threading.Lock()
+
+    class Payments(BaseHTTPRequestHandler):
+        protocol_version = 'HTTP/1.1'
+
+        def do_POST(self):
+            body = json.loads(self.rfile.read(int(self.headers['Content-Length'])))
+            with lock:
+                key = self.headers['Idempotency-Key']
+                if key not in answers:
+                    ok = body['amount'] <= 1000
+                    answers[key] = (body, 200 if ok else 402,
+                                    {'status': 'succeeded' if ok else 'declined', 'id': f'ch_{len(answers)}', 'x': [1, {}]})
+                _, status, answer = answers[key]
+            data = json.dumps(answer).encode()
+            self.send_response(status)
+            self.send_header('Content-Type', 'application/json')
+            self.send_header('Content-Length', str(len(data)))
+            self.end_headers()
+            self.wfile.write(data)
+
+        def log_message(self, *args):
+            pass
+
+    server = ThreadingHTTPServer(('127.0.0.1', 0), Payments)
+    server.handle_error = lambda request, address: None  # a sender killed mid-request
+    threading.Thread(target=server.serve_forever, daemon=True).start()
+    with socket.socket() as probe:
+        probe.bind(('127.0.0.1', 0))
+        port = probe.getsockname()[1]
+    args = [pure / 'build/pure-serve', '--module', module, '--store', work / 'pay-store', '--port', port,
+            '--send', f'charge=http://127.0.0.1:{server.server_address[1]}/charge', '--reply', 'charge=charged']
+
+    def start():
+        process = subprocess.Popen([str(a) for a in args], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+        ready, _, _ = select.select([process.stdout], [], [], 10)
+        assert ready and process.stdout.readline(), 'host did not start'
+        return process
+
+    def number(view):
+        return int(http(port, 'GET', f'/views/{view}')[3])
+
+    host = start()
+    stop = threading.Event()
+
+    def client(which):
+        import http.client
+        rng = random.Random(which + 8800)
+        while not stop.is_set():
+            data = json.dumps({'order': f'o{which}-{rng.randrange(10**6)}', 'amount': rng.randint(1, 1500)}).encode()
+            try:
+                connection = http.client.HTTPConnection('127.0.0.1', port, timeout=10)
+                connection.request('POST', '/inputs/buy', data, {'Content-Length': str(len(data)), 'Content-Type': 'application/json'})
+                connection.getresponse().read()
+                connection.close()
+            except OSError:
+                time.sleep(0.02)
+
+    clients = [threading.Thread(target=client, args=(n,)) for n in range(3)]
+    for thread in clients:
+        thread.start()
+    try:
+        time.sleep(0.8)
+        host.send_signal(signal.SIGKILL)
+        host.wait()
+        host = start()
+        time.sleep(0.8)
+    finally:
+        stop.set()
+        for thread in clients:
+            thread.join()
+    try:
+        deadline = time.time() + 30
+        while number('paid_view') + number('declined_view') < number('sent_view') and time.time() < deadline:
+            time.sleep(0.1)
+        sent, paid, declined, total = (number(v) for v in ('sent_view', 'paid_view', 'declined_view', 'paid_total_view'))
+        with lock:
+            keys = list(answers.values())
+        assert paid + declined == sent == len(keys), (paid, declined, sent, len(keys))
+        assert paid == sum(1 for _, status, _ in keys if status == 200), (paid, len(keys))
+        assert total == sum(body['amount'] for body, status, _ in keys if status == 200)
+    finally:
+        host.terminate()
+        _, error = host.communicate(timeout=30)
+        server.shutdown()
+    stats = [json.loads(line) for line in error.decode().splitlines() if line.startswith('{"active"')][-1]
+    return {'orders': sent, 'paid': paid, 'declined': declined, 'answers_after_restart': stats['answers']}
+
+
+BEAT_SOURCE = """state beats = 0
+state last = 0
+state stamped = 0
+input beat = every 50 ms
+input stamp = message {
+  note: text
+}
+on beat {
+  beats = beats + 1
+  last = now
+}
+on stamp { stamped = now }
+view beats_view = beats
+view last_view = last
+view stamped_view = stamped
+"""
+
+
+def server_timers(aup, pure, work):
+    """A served app's timers tick on the server's clock and its events see
+    the time they were applied, on one thread and with an owner; the ticks
+    are saved, so a restarted server goes on from them."""
+    import time
+    module, source = work / 'beat.pmod', work / 'beat.aup'
+    source.write_text(BEAT_SOURCE)
+    compile_module(aup, source, module)
+    seen = {}
+    for threads in (1, 3):
+        store = work / f'beat-{threads}'
+        beats = 0
+        for run_number in range(2):
+            host = HttpHost([pure / 'build/pure-serve', '--module', module, '--store', store, '--threads', threads])
+            try:
+                posted = time.time() * 1000
+                assert http(host.port, 'POST', '/inputs/stamp', b'x')[0] == 200
+                time.sleep(0.6)
+                before = time.time() * 1000
+                read = lambda view: float(http(host.port, 'GET', f'/views/{view}')[3])
+                count, last, stamped = read('beats_view'), read('last_view'), read('stamped_view')
+                after = time.time() * 1000 + 1  # the server's clock reads whole milliseconds
+                assert count >= beats + 5, (threads, run_number, count, beats)
+                assert before - 1000 < last <= after and posted - 50 <= stamped <= before, (last, stamped, posted, before)
+                beats = count
+            finally:
+                stats = host.stop()
+            assert stats[0]['ticks'] >= 5, stats
+        seen[threads] = beats
+    return seen
+
+
+def server_app(aup, pure, work, seeds, operations=200, threads=1):
+    """An app served by its views and inputs, no routes written: random
+    POSTs (a message with query fields and a body, a text, a key) and GETs
+    (a number, a text, a table's pages) answer exactly as a Python model
+    says, failures with their fault; what was acknowledged is there after
+    the server stops, and after it is killed while clients post. The app's
+    owner thread applies the inputs; connection threads (one or several)
+    show views from the snapshot of its last save."""
+    import signal
+    import threading
+    import time
+    module = work / 'served.pmod'
+    source = work / 'served.aup'
+    source.write_text(SERVED_SOURCE)
+    compile_module(aup, source, module)
+    names = ['ann', 'bob', 'zed', '가', 'a"b', 'back\\slash', 'tab\t', '', 'bob2']
+    compared = 0
+    for seed in seeds:
+        rng, board, at = random.Random(seed + 9100), {}, 0
+        store = work / f'served-{seed}-{threads}'
+        serve = [pure / 'build/pure-serve', '--module', module, '--threads', threads]
+        host = HttpHost(serve + ['--store', store])
+        order = lambda: sorted(board, key=lambda name: name.encode())
+
+        def place(name):
+            nonlocal at
+            keys = order()
+            if name not in board:
+                position = sum(1 for k in keys if k.encode() < name.encode())
+                board[name] = 0
+                if len(board) > 1 and at >= position: at += 1
+
+        try:
+            for _ in range(operations):
+                roll, keys = rng.random(), order()
+                if roll < 0.35:
+                    name, points = rng.choice(names), rng.randint(0, 9)
+                    if rng.random() < 0.05:
+                        request = ('POST', '/inputs/score', b'x')  # no points
+                        want = (400, 'missing_field', b'')
+                    elif rng.random() < 0.05:
+                        request = ('POST', f'/inputs/score?points={points}', b'\xff')
+                        want = (422, 'encoding', b'')
+                    else:
+                        request = ('POST', f'/inputs/score?points={points}', name.encode())
+                        place(name)
+                        board[name] += points
+                        at = order().index(name)
+                        want = (200, 'none', name.encode())
+                elif roll < 0.5:
+                    name = rng.choice(names)
+                    request = ('POST', '/inputs/join', name.encode())
+                    if name in board: want = (409, 'exists', b'')
+                    else:
+                        place(name)
+                        at = order().index(name)
+                        want = (200, 'none', b'')
+                elif roll < 0.58:
+                    request = ('POST', '/inputs/drop', b'')
+                    if not board: want = (404, 'not_found', b'')
+                    else:
+                        del board[keys[at]]
+                        at = min(at, max(len(board) - 1, 0))
+                        want = (200, 'none', b'')
+                elif roll < 0.68:
+                    request = ('GET', '/views/total', b'')
+                    want = (200, 'none', str(sum(board.values())).encode())
+                elif roll < 0.78:
+                    request = ('GET', '/views/current', b'')
+                    want = (200, 'none', keys[at].encode()) if board else (404, 'no_value', b'')
+                elif roll < 0.96:
+                    first, count = rng.randint(0, 4), rng.choice([1, 2, 5, 100])
+                    request = ('GET', f'/views/leaders?from={first}&count={count}', b'')
+                    shown = min(first, len(keys))
+                    rows = [{'name': k, 'points': board[k]} for k in keys[shown:shown + count]]
+                    want = (200, 'none', {'rows': len(keys), 'from': shown, 'items': rows})
+                else:
+                    request, want = rng.choice([(('GET', '/views/nope', b''), (404, 'no_view', b'')),
+                                                (('POST', '/inputs/nope', b''), (404, 'no_input', b'')),
+                                                (('POST', '/inputs/join?x=1', b'q'), (400, 'query', b'')),
+                                                (('GET', '/views/total?from=1', b''), (400, 'query', b''))])
+                status, error, kind, body = http(host.port, *request)
+                if isinstance(want[2], dict):
+                    assert kind == 'application/json', kind
+                    body = json.loads(body)
+                assert (status, error, body) == want, (seed, request, (status, error, body), want)
+                compared += 1
+        finally:
+            stats = host.stop()
+        assert stats[0]['bursts'] <= stats[0]['requests'] and stats[0]['owner'], stats
+        assert stats[0]['views_from_snapshots'] > 0, stats
+        # A graceful stop kept every acknowledged event.
+        host = HttpHost(serve + ['--store', store])
+        try:
+            got = json.loads(http(host.port, 'GET', '/views/leaders')[3])
+            assert got['items'] == [{'name': k, 'points': board[k]} for k in order()], (seed, got)
+        finally:
+            host.stop()
+
+    # Clients post while the server is killed: each acknowledged point is
+    # stored. Readers on other connections meanwhile see every point
+    # acknowledged before they asked, never a total going back, and never one
+    # that was not saved: the total after the kill is at least any they saw.
+    store = work / f'served-kill-{threads}'
+    serve = [pure / 'build/pure-serve', '--module', module, '--threads', threads]
+    host = HttpHost(serve + ['--store', store])
+    acknowledged = [0] * 6
+    seen = [0, 0]
+
+    def client(number):
+        for _ in range(10000):
+            try:
+                status, _, _, _ = http(host.port, 'POST', '/inputs/score?points=1', f'c{number}'.encode())
+            except OSError:
+                return
+            if status != 200:
+                return
+            acknowledged[number] += 1
+
+    def reader(number):
+        for _ in range(100000):
+            floor = sum(acknowledged)
+            try:
+                status, _, _, body = http(host.port, 'GET', '/views/total')
+            except OSError:
+                return
+            if status != 200:
+                return
+            total = int(body)
+            assert floor <= total and seen[number] <= total, (floor, seen[number], total)
+            seen[number] = total
+
+    clients = [threading.Thread(target=client, args=(n,)) for n in range(len(acknowledged))]
+    clients += [threading.Thread(target=reader, args=(n,)) for n in range(len(seen))]
+    for thread in clients:
+        thread.start()
+    time.sleep(0.8)
+    host.process.send_signal(signal.SIGKILL)
+    host.process.wait()
+    for thread in clients:
+        thread.join()
+    host = HttpHost(serve + ['--store', store])
+    try:
+        stored = {row['name']: row['points'] for row in json.loads(http(host.port, 'GET', '/views/leaders')[3])['items']}
+        for number, count in enumerate(acknowledged):
+            assert count > 0 and stored.get(f'c{number}', 0) in (count, count + 1), (number, count, stored)
+        assert min(seen) > 0 and max(seen) <= sum(stored.values()), (seen, stored)
+    finally:
+        host.stop()
+    return compared, sum(acknowledged)
+
+
+def main():
+    global EDITOR
+    parser = argparse.ArgumentParser()
+    parser.add_argument('--aup', type=Path, required=True, help='the aup compiler')
+    parser.add_argument('--aup-root', type=Path, default=ROOT.parent / 'aup', help='its checkout, for examples/editor.aup')
+    parser.add_argument('--seeds', type=int, default=4)
+    args = parser.parse_args()
+    aup, pure = args.aup.resolve(), ROOT
+    EDITOR = args.aup_root.resolve() / 'examples' / 'editor.aup'
+    report = {}
+    with tempfile.TemporaryDirectory(prefix='pure-check-') as directory:
+        work = Path(directory)
+        report['server_parity_requests'] = sum(server_parity(aup, pure, work, range(args.seeds), threads=threads)
+                                               for threads in (1, 3))
+        report['server_pipelined_answers'] = server_pipelining(aup, pure, work)
+        report['server_lent_read_bytes'] = server_lent_reads(aup, pure, work)
+        report['server_lent_budget_bytes'] = server_lent_budget(aup, pure, work)
+        report['server_lent_views'] = [server_lent_views(aup, pure, work, threads) for threads in (1, 3)]
+        report['server_bursts'] = server_bursts(aup, pure, work)
+        report['server_bursts_three_threads'] = server_bursts(aup, pure, work, shards=3)
+        report['server_app_requests'], report['server_app_acknowledged_before_kill'] = \
+            server_app(aup, pure, work, range(args.seeds))
+        report['server_app_owner_requests'], report['server_app_owner_acknowledged_before_kill'] = \
+            server_app(aup, pure, work, range(args.seeds), threads=3)
+        report['server_json_requests'] = server_json(aup, pure, work)
+        report['server_timer_beats'] = server_timers(aup, pure, work)
+        report['server_links'] = server_links(aup, pure, work)
+        report['server_webhook'] = server_webhook(aup, pure, work)
+        report['server_answers'] = server_answers(aup, pure, work)
+    print(json.dumps(report))
+
+
+if __name__ == '__main__':
+    sys.exit(main())