Make an API's answers the app's events (--reply OUTPUT=INPUT)

The final answer to each send of a wired output (a 2xx, or a 4xx but
408 and 429) becomes a message event of the given input: fields from
the answer's JSON object by name, other members skipped; a field the
answer lacks from the send's field of that name and type, so the
request's own id is there; a u32 http_status field is the status. It is
applied with the send's number from the wire's source, before the send
is trimmed and in the same save, so an answer that comes again after a
restart (the API repeating itself for the Idempotency-Key) applies once.
Sends that wait for answers go one per request; answers are saved with
the next round, or at once when none waits.

A server that writes an answer in pieces holds the second until the
first is acknowledged, and a delayed acknowledgement stalled every
answer 40 ms: the client now acknowledges at once (TCP_QUICKACK) after
each read. Against a payments API that answers 200 or 402, with the
sender killed (-9) midway, paid and declined orders matched the API's
answers one to one.

6 files changed, 281 insertions, 35 deletions

diff --git a/CHECKPOINT.md b/CHECKPOINT.md
index e7cb1c0..53c306f 100644
--- a/CHECKPOINT.md
+++ b/CHECKPOINT.md
@@ -18,7 +18,7 @@
 | HTTP 서비스 | 보기가 GET, 입력이 POST다(JSON 본문과 그 안의 그룹 객체를 제자리에서 읽는다, 라우터 코드 없음). 앱의 `send`는 이벤트와 함께 저장되고, `--send 출력=URL`로 다른 앱이나 HTTP 끝점에 순서대로 묶어 전해지며, 받는 앱은 번호마다 한 번 적용한다(서버를 죽였다 살려도 정확히 한 번). 답은 저장 뒤에 나간다. 상태를 가진 앱은 소유자 스레드 하나가 입력을 적용하고, 연결 스레드는 마지막 저장의 스냅숏으로 보기를 답한다(쓰기가 동기화하는 동안에도 읽기는 기다리지 않는다) | 008 HTTP 호스트, [077](docs/research/077-apps-as-http-services.md), [080](docs/research/080-snapshot-reads.md), [082](docs/research/082-json-in-place.md), [083](docs/research/083-delivery-over-http.md) |
 | 성능(Pi 5) | 문서 서버: 작은 GET 151.6k/s, 1 MiB GET 2,794/s(axum 1,372, drogon 708), 충돌 안전 PUT은 axum-atomic의 5.1–5.8배. 앱 서비스(aup 22줄 대 손으로 쓴 서비스): 내구성 있는 쓰기 약 76배(그 뒤 저장소 커밋을 한 구간으로 바꿔 64연결 21–26k/s, axum·drogon 내구 약 180/s), 20행 JSON GET 145k/s(axum 95k, drogon 144k), 쓰기와 섞인 읽기 149k/s·p99 0.6 ms(axum·drogon 내구 1.1–1.3k/s·p99 111–156 ms), 최대 RSS 3.0–4.1 MiB(axum 6.4, drogon 11.6) | [075](docs/research/075-http-host-against-axum.md), [077](docs/research/077-apps-as-http-services.md), [080](docs/research/080-snapshot-reads.md), [081](docs/research/081-tail-records.md) |
 
-남은 것: macOS 검증, `bool`·`i64`, 꼬리 늘리기·압축을 커밋 밖으로와 열린 부하에서의 저장·적용 겹치기(081 다음 결정), 응답을 이벤트로 돌려받기(083 다음 결정), 앱 라우트의 인증·TLS(문서
+남은 것: macOS 검증, `bool`·`i64`, 꼬리 늘리기·압축을 커밋 밖으로와 열린 부하에서의 저장·적용 겹치기(081 다음 결정), 밀린 outbox를 저장마다 다 훑지 않기(083 다음 결정), 앱 라우트의 인증·TLS(문서
 피어의 TLS는 있다), aup·Vega의 배포 묶음.
 
 ## 공개할 작업물과 완료 증거
diff --git a/docs/design/008-module-format.md b/docs/design/008-module-format.md
index 96a77fa..3fd4d52 100644
--- a/docs/design/008-module-format.md
+++ b/docs/design/008-module-format.md
@@ -343,7 +343,9 @@ text·리스트가 남을 때만 위치를 유지한다.
   JSON(필드와 그룹)으로 그 출력의 줄에 넣고, 출력마다 전달 스레드가 순서대로 보낸다(`X-Pure-Source: 이름.출력`,
   `X-Pure-Sequence`, `Idempotency-Key`). 2xx 또는 Pure 받는 쪽의 거절이면 전해진 것이고, 소유자가 덜어 낸다. 그 밖에는 쉬었다가
   다시 보낸다(버리지 않는다). 받는 쪽이 `X-Pure-Batch`를 알리면 밀린 것을 묶어 보낸다([083](../research/083-delivery-over-http.md)).
-  출력이 있는 앱은 상태가 없어도 소유자를 둔다.
+  출력이 있는 앱은 상태가 없어도 소유자를 둔다. `--reply 출력=입력`이면 각 보냄의 최종 응답(2xx, 408·429를 뺀 4xx)이 입력의
+  메시지 이벤트가 된다: 필드는 응답 JSON에서 이름으로(모르는 멤버는 건너뛴다), 없으면 보낸 메시지의 같은 이름·타입 필드에서,
+  u32 `http_status`는 상태 코드로. 응답은 보낸 번호로 한 번만 적용되고, 적용한 뒤에 보냄을 덜어 낸다.
 - **보기는 스냅숏에서.** 소유자는 저장이 성공할 때마다, 답을 돌려주기 전에 저장된 상태의 스냅숏을 게시한다.
   연결 스레드는 라운드마다(돌려받은 답을 처리한 뒤) 최신 스냅숏을 잡아 자기 읽기용 앱으로 그 라운드의 보기를 모두
   평가하고 답을 모아 보낸다. 입력은 그보다 먼저 소유자에게 넘겨, 소유자가 그동안 적용하고 저장한다. 그래서 쓰기의
diff --git a/docs/research/083-delivery-over-http.md b/docs/research/083-delivery-over-http.md
index 5427590..829c666 100644
--- a/docs/research/083-delivery-over-http.md
+++ b/docs/research/083-delivery-over-http.md
@@ -41,6 +41,24 @@
 - **TLS.** `https://` 받는 곳에는 OpenSSL로 TLS 1.2 이상을 쓴다. 인증서는 시스템 CA(또는 `--send-ca 파일`)로 검증하고, 호스트
   이름(SNI와 `SSL_set1_host`, IP 주소면 그 주소)과 맞아야 한다. 맞지 않으면 기록하고 10초마다 다시 시도한다(보냄은 남는다).
 
+## 응답을 이벤트로
+
+결제처럼 답이 필요한 호출은 `--reply 출력=입력`으로 잇는다. 그 출력으로 보낸 요청의 최종 응답(2xx, 그리고 408·429를 뺀
+4xx)이 입력의 메시지 이벤트가 된다.
+
+- 필드는 응답 JSON 객체에서 이름으로 채우고, 모르는 멤버는 값째 건너뛴다(API는 많은 것을 돌려준다). 응답에 없는 필드는
+  보낸 메시지의 같은 이름·타입 필드에서 채운다. 그래서 요청의 `order`를 응답 이벤트에서 그대로 볼 수 있다(API가 되풀이하지
+  않아도). u32 필드 `http_status`는 응답의 상태 코드다. 어느 쪽에도 없는 필드가 있으면 기록하고 10초마다 다시 보낸다.
+- 응답은 보낸 쪽 이름과 보낸 번호로 `pure_app_deliver`된다. 소유자는 응답을 적용한 뒤에 그 보냄을 덜어 내고, 둘은 한 저장에
+  담긴다. 그 사이에 죽으면 보냄이 남아 다시 보내지고(`Idempotency-Key`로 API가 같은 답을 준다), 응답은 번호로 한 번만 적용된다.
+- 응답을 기다리는 출력은 한 요청에 하나씩 보낸다. 요청이 기다리는 라운드가 있으면 그 저장에 응답을 합치고, 쉴 때는 따로
+  저장한다.
+
+**찾은 것: 조각으로 쓰는 서버와 지연 ACK.** 처음에는 응답이 초당 24건쯤밖에 오지 않았다. 파이썬 `http.server`는 헤더와
+본문을 두 번에 나눠 쓰고(Nagle이 둘째 조각을 첫 조각의 ACK까지 붙잡는다), 우리 클라이언트는 ACK를 40 ms 늦췄다. 요청마다
+40 ms다. 읽을 때마다 `TCP_QUICKACK`으로 곧바로 ACK하자 같은 검사가 30초를 넘기던 것에서 1.8초가 됐다. 받는 서버가 어떻게
+쓰든 클라이언트가 막히지 않아야 하므로 클라이언트에서 푼다.
+
 ## 장치가 말한 것
 
 Raspberry Pi 5. 받는 서버는 CPU 0–1(연결 스레드 2), 보내는 서버는 CPU 2, 클라이언트(파이썬 16스레드)는 나머지.
@@ -63,13 +81,15 @@ Raspberry Pi 5. 받는 서버는 CPU 0–1(연결 스레드 2), 보내는 서버
 - aup `make check`의 `server_webhook`: 파이썬 HTTPS 서버(자체 인증서)를 제3자 웹훅으로 두고, 주문 다섯 건이 순서대로
   출력의 JSON 모양과 헤더(`X-Pure-Source`, `Idempotency-Key`, 묶음 헤더 없음)로 도착한다. 처음에 503으로 답한 2번은 다시 온다.
   그 인증서를 믿지 않는 보내는 쪽에서는 아무것도 도착하지 않고 기록이 남는다. ASan·UBSan 빌드로도 통과했다.
+- aup `make check`의 `server_answers`: 파이썬 "결제 API"(1000 이하는 200 `succeeded`, 넘으면 402 `declined`, 같은
+  `Idempotency-Key`에는 첫 답을 되풀이)에 출력을 잇고 응답을 `charged` 입력으로 받는다. 클라이언트 셋이 사는 동안 보내는 서버를
+  `kill -9`로 죽였다 살린다. 끝에 앱의 결제·거절 수의 합이 저장된 구매 수와 API가 받은 키 수와 같고, 결제 수와 결제 합계가 API의
+  200 응답과 같다(각 3,774–3,979건, 네 번, ASan·UBSan 빌드로 한 번 더).
 - `test-module`의 링크 검사([078](078-app-connections.md), 008)가 런타임 쪽(이벤트의 보냄, 저장 실패, 재시작, 중복·거절)을
   맡는다.
 
 ## 다음 결정
 
-- **응답을 이벤트로.** 지금은 보내고 끝이다(알림, 웹훅). 결제처럼 답이 필요한 호출은 응답을 앱의 입력으로 돌려주는
-  배선(`--send 출력=URL --reply 입력`)이 필요하다.
 - 저장이 끝날 때마다 outbox 전체를 훑어 새 보냄을 찾는다. 받는 쪽이 오래 꺼져 수만 건이 밀리면 저장마다 그만큼 훑는다.
   마지막으로 넣은 자리에서 이어 훑게 해야 한다.
 - 덜어 낸 것은 다음 저장에 지워진다. 요청이 없는 앱은 저장하지 않으므로, 다시 시작하면 전한 것을 한 번 더 보낸다(받는
diff --git a/prototypes/connections/deliver.c b/prototypes/connections/deliver.c
index c4136a2..0d3da64 100644
--- a/prototypes/connections/deliver.c
+++ b/prototypes/connections/deliver.c
@@ -1,5 +1,6 @@
 #define _POSIX_C_SOURCE 200809L
 #include "deliver.h"
+#include "message-json.h"
 #include "module-text.h"
 #include <errno.h>
 #include <netdb.h>
@@ -18,10 +19,14 @@
 #include <time.h>
 #include <unistd.h>
 
-enum { TIMEOUT_S = 10, FIRST_PAUSE_MS = 100, LONGEST_PAUSE_MS = 10000, GROUPS = PURE_MODULE_NAME / 2, BATCH_MOST = 256 };
-enum { DELIVERED, AGAIN, REFUSED_BY_PEER, TLS_FAILED };
+enum { TIMEOUT_S = 10, FIRST_PAUSE_MS = 100, LONGEST_PAUSE_MS = 10000, GROUPS = PURE_MODULE_NAME / 2, BATCH_MOST = 256,
+       ANSWER_MOST = 1 << 20 };
+enum { DELIVERED, AGAIN, REFUSED_BY_PEER, TLS_FAILED, ANSWER_UNREAD };
 
-typedef struct { char *body; size_t size; uint64_t seq; } Pending;
+typedef struct { char *body; size_t size; uint64_t seq; unsigned char *sent; size_t sent_size; } Pending;
+/* An answer made into its input's message: a table of the fields' starts and
+ * sizes (u32le), then the payload from its prefix on. */
+typedef struct { uint64_t seq; unsigned char *bytes; size_t size, table; } Answer;
 
 struct PureWire {
     const PureModule *module;
@@ -40,6 +45,11 @@ struct PureWire {
     int fd;   /* the connection kept for the next send, or -1; changed under the lock */
     bool tls; /* https: the connection is TLS */
     SSL *ssl; /* its TLS session */
+    int answer_input; /* the input answers become events of, or -1 */
+    Answer *answers;  /* made, for the owner to take in order */
+    size_t answer_head, answer_count, answer_capacity;
+    Answer taken;     /* the one the owner took last, kept until it takes the next */
+    uint32_t placed[2 * PURE_MODULE_FIELDS];
 };
 
 /* ---- TLS: one client context, verifying the receiver's certificate against
@@ -226,12 +236,42 @@ PureWire *pure_wire_open(const PureModule *m, const char *output, const char *ur
     memcpy(w->source, source, length + 1);
     w->fd = -1;
     w->tls = tls;
+    w->answer_input = -1;
     pthread_mutex_init(&w->lock, NULL);
     pthread_cond_init(&w->changed, NULL);
     return w;
 }
 
 uint32_t pure_wire_output(const PureWire *w) { return w->output; }
+
+bool pure_wire_reply(PureWire *w, uint32_t input, char *error, size_t size) {
+    const PureModule *m = w->module;
+    if (input >= m->inputs || m->input[input].kind != PURE_KIND_MESSAGE || m->input[input].keyed) {
+        snprintf(error, size, "an answer goes to a message input without a key");
+        return false;
+    }
+    w->answer_input = (int)input;
+    return true;
+}
+
+bool pure_wire_answer(PureWire *w, PureWireAnswer *out) {
+    free(w->taken.bytes);
+    w->taken = (Answer){0};
+    pthread_mutex_lock(&w->lock);
+    bool some = w->answer_count > 0;
+    if (some) {
+        w->taken = w->answers[w->answer_head++];
+        --w->answer_count;
+        w->delivered = w->taken.seq; /* trimmed after the owner applies it */
+    }
+    pthread_mutex_unlock(&w->lock);
+    if (!some) return false;
+    const PureInputShape *in = &w->module->input[w->answer_input];
+    for (size_t i = 0; i < 2 * (size_t)in->field_count; ++i) w->placed[i] = read_u32(w->taken.bytes + 4 * i);
+    *out = (PureWireAnswer){(uint32_t)w->answer_input, w->taken.seq,
+                            {w->taken.bytes + w->taken.table, w->taken.size - w->taken.table}, w->placed, w->source};
+    return true;
+}
 uint64_t pure_wire_queued(const PureWire *w) { return w->queued; }
 
 uint64_t pure_wire_delivered(PureWire *w) {
@@ -275,7 +315,14 @@ bool pure_wire_queue(PureWire *w, uint64_t seq, PureSpan payload) {
             w->capacity = capacity;
         }
     }
-    w->queue[w->head + w->count++] = (Pending){t.bytes, t.size, seq};
+    unsigned char *sent = NULL;
+    if (w->answer_input >= 0 && !(sent = malloc(payload.size ? payload.size : 1))) {
+        pthread_mutex_unlock(&w->lock);
+        free(t.bytes);
+        return false;
+    }
+    if (sent && payload.size) memcpy(sent, payload.data, payload.size);
+    w->queue[w->head + w->count++] = (Pending){t.bytes, t.size, seq, sent, payload.size};
     w->queued = seq;
     pthread_cond_signal(&w->changed);
     pthread_mutex_unlock(&w->lock);
@@ -294,6 +341,13 @@ static bool fill(Reader *r) {
     if (w->ssl) n = SSL_read(w->ssl, r->bytes + r->end, (int)(sizeof(r->bytes) - r->end));
     else do n = recv(w->fd, r->bytes + r->end, sizeof(r->bytes) - r->end, 0); while (n < 0 && errno == EINTR);
     if (n <= 0) return false;
+#ifdef TCP_QUICKACK
+    /* Acknowledge at once: a server that writes an answer in pieces holds the
+     * next piece until the first is acknowledged (Nagle), which a delayed
+     * acknowledgement would stall for 40 ms. The flag does not stay set. */
+    int one = 1;
+    setsockopt(w->fd, IPPROTO_TCP, TCP_QUICKACK, &one, sizeof(one));
+#endif
     r->end += (size_t)n;
     return true;
 }
@@ -320,14 +374,17 @@ static bool read_line(Reader *r, char *line, size_t size) {
     }
 }
 
-static bool skip(Reader *r, uint64_t size) {
+/* size bytes of the body: kept in into (up to ANSWER_MOST), or skipped when it is NULL. */
+static bool take(Reader *r, uint64_t size, Text *into) {
+    if (into && size > ANSWER_MOST - into->size) return false;
     while (size) {
         if (r->at == r->end && !fill(r)) return false;
         size_t n = r->end - r->at < size ? r->end - r->at : (size_t)size;
+        if (into) put(into, r->bytes + r->at, n);
         r->at += n;
         size -= n;
     }
-    return true;
+    return !into || !into->failed;
 }
 
 static bool send_all(PureWire *w, const char *bytes, size_t size) {
@@ -397,7 +454,7 @@ static bool refusal(const char *error) {
 
 /* The answer to one request: its status, whether a Pure receiver refused it,
  * and whether the connection may carry the next. */
-static bool read_answer(PureWire *w, int *status, bool *refused, bool *keep, unsigned *batch) {
+static bool read_answer(PureWire *w, int *status, bool *refused, bool *keep, unsigned *batch, Text *body) {
     Reader r = {.wire = w};
     char line[1024];
     if (!read_line(&r, line, sizeof(line)) || strncmp(line, "HTTP/1.", 7) || strlen(line) < 12) return false;
@@ -414,15 +471,18 @@ static bool read_answer(PureWire *w, int *status, bool *refused, bool *keep, uns
                     if (!read_line(&r, line, sizeof(line))) return false;
                     chunk = strtoull(line, NULL, 16);
                     if (!chunk) break;
-                    if (!skip(&r, chunk) || !read_line(&r, line, sizeof(line))) return false;
+                    if (!take(&r, chunk, body) || !read_line(&r, line, sizeof(line))) return false;
                 }
                 while (read_line(&r, line, sizeof(line)) && *line) {}
                 return true;
             }
             if (*status == 204 || *status == 304 || *status < 200) return true;
-            if (sized) return skip(&r, length);
+            if (sized) return take(&r, length, body);
             *keep = false; /* the body runs to the close */
-            while (fill(&r)) r.at = r.end;
+            while (fill(&r)) {
+                if (body && !take(&r, r.end - r.at, body)) return false;
+                r.at = r.end;
+            }
             return true;
         }
         char *value = strchr(line, ':');
@@ -438,9 +498,64 @@ static bool read_answer(PureWire *w, int *status, bool *refused, bool *keep, uns
     return false;
 }
 
+static void write_u32(unsigned char *p, uint32_t v) { for (unsigned i = 0; i < 4; ++i) p[i] = (unsigned char)(v >> (8 * i)); }
+
+/* The answer (status, body) to the send p as a message of the answer input:
+ * fields from the body's JSON object by name, others skipped; a field the
+ * body lacks from the send's field of that name and type; a u32 field named
+ * http_status from the status. False when a field has none of these. */
+static bool form_answer(const PureWire *w, int status, const Text *body, const Pending *p, Answer *out) {
+    const PureModule *m = w->module;
+    const PureInputShape *in = &m->input[w->answer_input], *sent = &m->input[w->output];
+    size_t table = 8 * (size_t)in->field_count, total = table + in->prefix + body->size + p->sent_size;
+    unsigned char *bytes = calloc(1, total ? total : 1);
+    if (!bytes) return false;
+    unsigned char *prefix = bytes + table, *text = prefix + in->prefix, *copy = text + body->size;
+    if (body->size) memcpy(text, body->bytes, body->size);
+    if (p->sent_size) memcpy(copy, p->sent, p->sent_size);
+    PureJsonMessage message = {.module = m, .shape = in, .payload = prefix, .table = bytes, .lenient = true};
+    size_t at = pure_json_space(text, 0, body->size);
+    if (at < body->size && text[at] == '{' && pure_json_message(&message, text, body->size, &at)) {
+        free(bytes);
+        return false;
+    }
+    for (uint32_t f = 0; f < in->field_count; ++f) {
+        if (message.seen[f]) continue;
+        const PureField *field = &m->field[in->field_start + f];
+        if (field->type == PURE_FIELD_U32 && !strcmp(field->name, "http_status")) {
+            write_u32(prefix + field->offset, (uint32_t)status);
+            continue;
+        }
+        /* The send's field of the same name and type, and where its bytes are. */
+        size_t start = sent->prefix, length = 0;
+        const PureField *from = NULL;
+        for (uint32_t g = 0; g < sent->field_count && !from; ++g) {
+            const PureField *other = &m->field[sent->field_start + g];
+            bool variable = other->type == PURE_FIELD_BYTES || other->type == PURE_FIELD_TEXT;
+            if (variable) length = other->offset == PURE_FIELD_REST ? p->sent_size - start : read_u32(p->sent + other->offset);
+            if (!strcmp(other->name, field->name) && other->type == field->type) from = other;
+            else if (variable) start += length;
+        }
+        bool variable = field->type == PURE_FIELD_BYTES || field->type == PURE_FIELD_TEXT;
+        if (!from || (variable && (start > p->sent_size || length > p->sent_size - start))) {
+            free(bytes);
+            return false;
+        }
+        if (field->type == PURE_FIELD_U32 || field->type == PURE_FIELD_NUMBER)
+            memcpy(prefix + field->offset, p->sent + from->offset, field->type == PURE_FIELD_U32 ? 4 : 8);
+        else {
+            write_u32(bytes + 8 * f, (uint32_t)(copy + start - prefix));
+            write_u32(bytes + 8 * f + 4, (uint32_t)length);
+        }
+    }
+    *out = (Answer){p->seq, bytes, total, table};
+    return true;
+}
+
 /* One request for count sends from p (numbered in a row): one object, or
- * an array of them for a receiver that takes batches. */
-static int post(PureWire *w, const Pending *p, unsigned count) {
+ * an array of them for a receiver that takes batches. A final answer to a
+ * send whose answers are wired is made into *answer. */
+static int post(PureWire *w, const Pending *p, unsigned count, Answer *answer) {
     size_t body = count > 1 ? count + 1 : 0;
     for (unsigned i = 0; i < count; ++i) body += p[i].size;
     char head[2048], batch[32] = "";
@@ -473,19 +588,30 @@ static int post(PureWire *w, const Pending *p, unsigned count) {
         int status;
         bool refused, keep, sent = send_all(w, head, (size_t)size);
         unsigned batch_size = 0;
+        Text body = {0};
         for (unsigned i = 0; sent && i < count; ++i)
             sent = (!i || count == 1 || send_all(w, ",", 1)) && (i || count == 1 || send_all(w, "[", 1)) &&
                    send_all(w, p[i].body, p[i].size);
         if (sent && count > 1) sent = send_all(w, "]", 1);
-        if (!sent || !read_answer(w, &status, &refused, &keep, &batch_size)) {
+        if (!sent || !read_answer(w, &status, &refused, &keep, &batch_size, w->answer_input >= 0 ? &body : NULL)) {
+            free(body.bytes);
             drop_connection(w);
             if (fresh) return AGAIN;
             continue;
         }
         if (!keep) drop_connection(w);
-        if (batch_size) w->batch = batch_size;
+        if (batch_size && w->answer_input < 0) w->batch = batch_size;
+        if (status == 408 || status == 429 || status >= 500) {
+            free(body.bytes);
+            return AGAIN;
+        }
+        if (w->answer_input >= 0 && status >= 200 && status < 500) { /* a final answer: the app hears it */
+            bool formed = form_answer(w, status, &body, p, answer);
+            free(body.bytes);
+            return formed ? DELIVERED : ANSWER_UNREAD;
+        }
+        free(body.bytes);
         if ((status >= 200 && status < 300) || (status >= 400 && status < 500 && refused)) return DELIVERED;
-        if (status == 408 || status == 429 || status >= 500) return AGAIN;
         return REFUSED_BY_PEER;
     }
     return AGAIN;
@@ -507,13 +633,31 @@ static void *deliver(void *context) {
         memcpy(next, w->queue + w->head, count * sizeof(*next));
         ++w->attempts;
         pthread_mutex_unlock(&w->lock);
-        int outcome = post(w, next, count);
+        Answer answer = {0};
+        int outcome = post(w, next, count, &answer);
         pthread_mutex_lock(&w->lock);
+        if (outcome == DELIVERED && answer.bytes) { /* for the owner, who then counts it delivered */
+            if (w->answer_head + w->answer_count == w->answer_capacity) {
+                if (w->answer_head) {
+                    memmove(w->answers, w->answers + w->answer_head, w->answer_count * sizeof(*w->answers));
+                    w->answer_head = 0;
+                } else {
+                    size_t capacity = w->answer_capacity ? 2 * w->answer_capacity : 16;
+                    Answer *grown = realloc(w->answers, capacity * sizeof(*grown));
+                    if (!grown) { free(answer.bytes); outcome = AGAIN; }
+                    else { w->answers = grown; w->answer_capacity = capacity; }
+                }
+            }
+            if (outcome == DELIVERED) w->answers[w->answer_head + w->answer_count++] = answer;
+        }
         if (outcome == DELIVERED) {
-            for (unsigned i = 0; i < count; ++i) free(w->queue[w->head + i].body);
+            for (unsigned i = 0; i < count; ++i) {
+                free(w->queue[w->head + i].body);
+                free(w->queue[w->head + i].sent);
+            }
             w->head += count;
             w->count -= count;
-            w->delivered = next[count - 1].seq;
+            if (!answer.bytes) w->delivered = next[count - 1].seq;
             w->deliveries += count;
             pause_ms = 0;
             pthread_mutex_unlock(&w->lock);
@@ -521,9 +665,10 @@ static void *deliver(void *context) {
             pthread_mutex_lock(&w->lock);
             continue;
         }
-        if ((outcome == REFUSED_BY_PEER || outcome == TLS_FAILED) && reported != next[0].seq) {
+        if (outcome != AGAIN && reported != next[0].seq) {
             fprintf(stderr, "{\"wire\":\"%s\",\"seq\":%llu,\"error\":\"%s\"}\n", w->source,
                     (unsigned long long)next[0].seq, outcome == TLS_FAILED ? "the receiver's certificate is not trusted; trying again" :
+                    outcome == ANSWER_UNREAD ? "its answer does not make the answer input's message; trying again" :
                     "the receiver does not take it; trying again");
             reported = next[0].seq;
         }
@@ -557,7 +702,13 @@ void pure_wire_close(PureWire *w) {
     if (w->started) pthread_join(w->thread, NULL);
     if (w->ssl) SSL_free(w->ssl);
     if (w->fd >= 0) close(w->fd);
-    for (size_t i = 0; i < w->count; ++i) free(w->queue[w->head + i].body);
+    for (size_t i = 0; i < w->count; ++i) {
+        free(w->queue[w->head + i].body);
+        free(w->queue[w->head + i].sent);
+    }
+    for (size_t i = 0; i < w->answer_count; ++i) free(w->answers[w->answer_head + i].bytes);
+    free(w->answers);
+    free(w->taken.bytes);
     free(w->queue);
     pthread_cond_destroy(&w->changed);
     pthread_mutex_destroy(&w->lock);
diff --git a/prototypes/connections/deliver.h b/prototypes/connections/deliver.h
index 13048fc..2d1da29 100644
--- a/prototypes/connections/deliver.h
+++ b/prototypes/connections/deliver.h
@@ -34,6 +34,27 @@ uint64_t pure_wire_delivered(PureWire *);
 bool pure_wire_start(PureWire *, void (*notify)(void *), void *context);
 /* Stop (a delivery under way is finished or abandoned) and free. */
 void pure_wire_close(PureWire *);
+/* Answers as events: the final answer to each send (a 2xx, or a 4xx but 408
+ * and 429) becomes a message event of input (a message without a key), in
+ * the order sent: its fields come from the answer's JSON object by name
+ * (other members are skipped), a field the answer lacks from the send's
+ * field of that name and type (as a request's id), and a u32 field named
+ * http_status is the answer's status. An answer that makes no message is
+ * logged and the send tried again every 10 s. Answers go one per request. */
+bool pure_wire_reply(PureWire *, uint32_t input, char *error, size_t size);
+/* The owner's side: the next answer in order, or false. Taking it counts
+ * its send delivered (the owner applies the answer, then trims); what out
+ * points to is the wire's until the next call. The answer is applied with
+ * pure_app_deliver from the wire's source with the send's number, so an
+ * answer that comes again after a restart is applied once. */
+typedef struct {
+    uint32_t input;
+    uint64_t seq;
+    PureSpan payload;
+    const uint32_t *placed;
+    const char *source;
+} PureWireAnswer;
+bool pure_wire_answer(PureWire *, PureWireAnswer *out);
 /* Counts for a host's report. */
 void pure_wire_counts(PureWire *, uint64_t *delivered, uint64_t *attempts);
 #endif
diff --git a/prototypes/connections/serve.c b/prototypes/connections/serve.c
index 88a9a6f..60771e1 100644
--- a/prototypes/connections/serve.c
+++ b/prototypes/connections/serve.c
@@ -150,7 +150,8 @@ typedef struct {
     /* The owner's: outputs wired to HTTP receivers, and how far each was trimmed. */
     PureWire *wires[PURE_MODULE_INPUTS];
     unsigned wire_count;
-    uint64_t trimmed[PURE_MODULE_INPUTS];
+    uint64_t trimmed[PURE_MODULE_INPUTS], answers;
+    bool held[PURE_MODULE_INPUTS]; /* an answer failed to apply: the wire's sends stay until a restart */
 } Executor;
 
 /* A request made ready: its effect, what it names, and the thread that answers it. */
@@ -1051,14 +1052,32 @@ static void after_save(Executor *ex) {
     if (ex->wire_count) pure_app_outbox(&ex->app, queue_send, &q);
 }
 
-/* What the wires delivered leaves the outbox (with the next save). */
-static void trim_delivered(Executor *ex) {
+/* The answers the wires made become events, in order; then what the wires
+ * delivered leaves the outbox (with the next save). True when an answer was
+ * applied (the caller saves). */
+static bool take_deliveries(Executor *ex) {
+    bool answered = false;
     for (unsigned i = 0; i < ex->wire_count; ++i) {
+        PureWireAnswer answer;
+        while (pure_wire_answer(ex->wires[i], &answer)) {
+            ex->app.now = wall_ms();
+            PureResult result = pure_app_deliver(&ex->app, answer.input, answer.source, answer.seq, answer.payload,
+                                                 answer.placed);
+            if (result != PURE_OK && ex->app.fault == PURE_FAULT_NONE) {
+                ex->held[i] = true; /* not applied and not refused: sent again after a restart */
+                fprintf(stderr, "{\"wire\":\"%s\",\"seq\":%" PRIu64 ",\"error\":\"its answer was not applied\"}\n",
+                        answer.source, answer.seq);
+            }
+            pure_app_finish(&ex->app);
+            answered = true;
+            ++ex->answers;
+        }
         uint64_t delivered = pure_wire_delivered(ex->wires[i]);
-        if (delivered <= ex->trimmed[i]) continue;
+        if (delivered <= ex->trimmed[i] || ex->held[i]) continue;
         pure_app_trim(&ex->app, pure_wire_output(ex->wires[i]), delivered);
         ex->trimmed[i] = delivered;
     }
+    return answered;
 }
 
 /* One save for a burst: its requests are durable with it or fail with it. */
@@ -1290,8 +1309,14 @@ static void *run_owner(void *context) {
             owner->delivered = false;
             pthread_mutex_unlock(&owner->lock);
             long due = run_timers(&owner->executor);
-            if (delivered) trim_delivered(&owner->executor);
+            bool answered = delivered && take_deliveries(&owner->executor);
             pthread_mutex_lock(&owner->lock);
+            /* Answers are saved with the next round, or now if none waits. */
+            if (answered && !owner->count && !owner->stopping) {
+                pthread_mutex_unlock(&owner->lock);
+                if (pure_app_save(&owner->executor.app) == PURE_OK) after_save(&owner->executor);
+                pthread_mutex_lock(&owner->lock);
+            }
             if (owner->count || owner->stopping) break;
             if (owner->delivered) continue;
             if (due < 0) {
@@ -1604,6 +1629,30 @@ static bool open_wires(Executor *ex, const PureModule *m, const char *const *sen
     return true;
 }
 
+/* --reply OUTPUT=INPUT: the answers to that output's sends become INPUT's events. */
+static bool wire_replies(Executor *ex, const PureModule *m, const char *const *replies, unsigned count) {
+    for (unsigned i = 0; i < count; ++i) {
+        const char *equals = strchr(replies[i], '=');
+        char output[PURE_MODULE_NAME], error[160] = "";
+        size_t length = equals ? (size_t)(equals - replies[i]) : 0;
+        int input = equals ? pure_module_input(m, equals + 1) : -1;
+        PureWire *wire = NULL;
+        if (length && length < sizeof(output)) {
+            memcpy(output, replies[i], length);
+            output[length] = 0;
+            int index = pure_module_input(m, output);
+            for (unsigned k = 0; k < ex->wire_count && index >= 0; ++k)
+                if (pure_wire_output(ex->wires[k]) == (uint32_t)index) wire = ex->wires[k];
+        }
+        if (!wire || input < 0 || !pure_wire_reply(wire, (uint32_t)input, error, sizeof(error))) {
+            fprintf(stderr, "--reply %s: %s\n", replies[i], !wire ? "takes OUTPUT=INPUT of an output given --send" :
+                                                         input < 0 ? "no such input" : error);
+            return false;
+        }
+    }
+    return true;
+}
+
 /* --threads auto: a connection thread for each CPU this process may run on,
  * less the one the owner of an app with states takes. */
 static unsigned auto_threads(bool owner) {
@@ -1617,8 +1666,8 @@ static unsigned auto_threads(bool owner) {
 int main(int argc, char **argv) {
     static PureModule module;
     static Owner owner;
-    const char *module_path = NULL, *store = NULL, *name = NULL, *sends[PURE_MODULE_INPUTS];
-    unsigned send_count = 0;
+    const char *module_path = NULL, *store = NULL, *name = NULL, *sends[PURE_MODULE_INPUTS], *replies[PURE_MODULE_INPUTS];
+    unsigned send_count = 0, reply_count = 0;
     unsigned port = 8767, threads = 1;
     size_t budget = 64u << 20, capacity = 1024;
     uint64_t timeout_ms = 10000;
@@ -1630,6 +1679,7 @@ int main(int argc, char **argv) {
         else if (!strcmp(argv[i], "--send") && send_count < PURE_MODULE_INPUTS) sends[send_count++] = argv[i + 1];
         else if (!strcmp(argv[i], "--name")) name = argv[i + 1];
         else if (!strcmp(argv[i], "--send-ca")) pure_wire_trust(argv[i + 1]);
+        else if (!strcmp(argv[i], "--reply") && reply_count < PURE_MODULE_INPUTS) replies[reply_count++] = argv[i + 1];
         else if (!strcmp(argv[i], "--port") && number(argv[i + 1], 65535, &value)) port = (unsigned)value;
         else if (!strcmp(argv[i], "--budget") && number(argv[i + 1], SIZE_MAX, &value) && value) budget = (size_t)value;
         else if (!strcmp(argv[i], "--connections") && number(argv[i + 1], 1u << 20, &value) && value) capacity = (size_t)value;
@@ -1671,7 +1721,8 @@ int main(int argc, char **argv) {
             return 1;
         owner.executor.reads_first = true;
         owner.executor.publishes = true;
-        if (!open_wires(&owner.executor, &module, sends, send_count, name)) return 2;
+        if (!open_wires(&owner.executor, &module, sends, send_count, name) ||
+            !wire_replies(&owner.executor, &module, replies, reply_count)) return 2;
         after_save(&owner.executor); /* the snapshot, and the sends stored before */
         for (unsigned i = 0; i < owner.executor.wire_count; ++i)
             if (!pure_wire_start(owner.executor.wires[i], wire_delivered, &owner)) { fputs("Cannot start a wire\n", stderr); return 1; }
@@ -1731,15 +1782,16 @@ int main(int argc, char **argv) {
     fprintf(stderr, "{\"active\":%zu,\"retained_payload_bytes\":%zu,\"accepted\":%" PRIu64 ",\"requests\":%" PRIu64
             ",\"bursts\":%" PRIu64 ",\"burst_events\":%" PRIu64 ",\"ticks\":%" PRIu64 ",\"expired\":%" PRIu64
             ",\"threads\":%u,\"owner\":%s,\"snapshots\":%" PRIu64 ",\"views_from_snapshots\":%" PRIu64
-            ",\"delivered\":%" PRIu64 ",\"delivery_attempts\":%" PRIu64 ",\"commit_groups\":%lu,\"grouped_commits\":%lu}\n",
+            ",\"delivered\":%" PRIu64 ",\"delivery_attempts\":%" PRIu64 ",\"answers\":%" PRIu64
+            ",\"commit_groups\":%lu,\"grouped_commits\":%lu}\n",
             active, retained, accepted, requests, bursts, events, ticks, expired, threads, owned ? "true" : "false",
-            owner.executor.publications, shown, delivered, attempts, groups, grouped);
+            owner.executor.publications, shown, delivered, attempts, owner.executor.answers, groups, grouped);
     free(hosts);
     free(thread);
     return 0;
 usage:
     fputs("pure-serve --module FILE.pmod --store DIRECTORY [--port 8767] [--threads 1|N|auto] [--budget BYTES] "
           "[--connections 1024] [--timeout-ms 10000] [--send OUTPUT=http[s]://host:port/path ...] [--name NAME] "
-          "[--send-ca FILE]\n", stderr);
+          "[--send-ca FILE] [--reply OUTPUT=INPUT ...]\n", stderr);
     return 2;
 }