Send messages as part of events and apply delivered ones once (078)

An output (input kind 5) is a message the app sends; SEND (72) takes
an output's field values in order in a handler, lays them out as a
message's payload and keeps it, numbered per output from 1, with the
event's changes: a failed event sent nothing, and a failed save takes
back every send since the last save with its numbers. A host sees only
saved sends (pure_app_outbox) and trims what its receiver has saved
(pure_app_trim, gone with the next save). pure_app_deliver applies a
message from a named source once: a number heard before succeeds
without applying, a refused message counts as heard, a failed save
forgets what it heard.

Sends, numbers and heard numbers are the app's links state, stored
whole in one runtime document (PLK1) with the states, in the same
commit. Outputs follow their names across module changes; a module
that drops or retypes an output with sends left is refused, and a
dropped output's number is kept for when it returns. Killing the
process at every commit stage leaves an event and its send, or a
delivery and its effect, together or neither. pure-serve answers a
post to an output or a timer with 404 no_input.

8 files changed, 742 insertions, 18 deletions

diff --git a/docs/design/008-module-format.md b/docs/design/008-module-format.md
index dd16c90..e67afa8 100644
--- a/docs/design/008-module-format.md
+++ b/docs/design/008-module-format.md
@@ -29,7 +29,7 @@ v5에는 뒤에 집계(FOLD), 걸러진 표, 마크, 그리고 **키 있는 리�
 | 32 | u32 상수 블롭 바이트 수, u16 레코드 필드, u16 0 |
 | 40 | 상태 × 16 B: u32 이름, u32 타입(0 수, 1 text, 2 커서, 3 리스트, 4 인덱스), 8 B 값: 수는 f64 초기값, text는 u32 초기 상수, 커서는 u32 대상 text 상태와 u32 플래그(1이면 마크: 그 자리에 넣은 글 앞에 남는다), 리스트는 u32 레코드와 u32 키(0이면 없음, 아니면 키 필드 + 1), 인덱스는 u32 대상 리스트 상태 |
 | | 컬렉션 × 12 B: u32 이름, u32 1(바이트), u32 1(키별) |
-| | 입력 × 16 B: u32 이름, u8 종류(1 키, 2 메시지, 3 text, 4 타이머), u8 키별 여부, u16 0, u32 키 이름(키 입력)·주기(타이머, 1 ms–24시간을 ms로) 또는 0, u16 첫 필드, u16 필드 수 |
+| | 입력 × 16 B: u32 이름, u8 종류(1 키, 2 메시지, 3 text, 4 타이머, 5 출력: 앱이 보내는 메시지, 모양은 메시지와 같고 핸들러가 없다), u8 키별 여부, u16 0, u32 키 이름(키 입력)·주기(타이머, 1 ms–24시간을 ms로) 또는 0, u16 첫 필드, u16 필드 수 |
 | | 필드 × 8 B: u32 이름(식별자, 또는 그룹 안의 필드면 점으로 이은 경로 `user.name`), u32 타입(1 u32le, 2 바이트, 3 수(binary64), 4 text). 페이로드는 접두부와 가변 필드의 바이트다. 접두부에는 선언 순서로 고정 필드의 값과 가변 필드(바이트·text)의 u32 길이가 들어가는데, 마지막 가변 필드는 길이 없이 나머지를 차지한다. 가변 필드의 바이트는 접두부 뒤에 선언 순서로 이어진다(가변 필드가 하나면 예전의 "바이트 꼬리"와 같다) |
 | | 상수 × 8 B: u32 블롭 오프셋, u32 길이 (UTF-8) |
 | | 레코드 × 8 B: u32 이름, u16 첫 레코드 필드, u16 필드 수(1–16) |
@@ -105,6 +105,7 @@ v5에는 뒤에 집계(FOLD), 걸러진 표, 마크, 그리고 **키 있는 리�
 | 69 | NOW | → n | 이벤트가 보는 지금(1970년부터 ms, UTC). 호스트가 정하고 스크립트는 아무 때나 재현한다. 핸들러 전용: 보기는 상태의 순수 함수로 남는다 |
 | 70 | FLOOR | n → n | 내림 |
 | 71 | MOD | n n → n | 나눗셈을 내림한 나머지. 결과는 나누는 수의 부호를 따른다(0도). 0으로 나누면 `divide` |
+| 72 | SEND | f… → | 출력 `인자`의 필드 값을 필드 순서대로 받아(u32·수는 n, 바이트는 b, text는 t) 메시지 페이로드(접두부와 가변 필드)로 만들고 이벤트의 일부로 보낸다. u32 필드의 값이 0–2³²−1의 정수가 아니면 `limit`. 핸들러 전용 |
 
 `f`는 레코드 필드의 타입(수 또는 text)이고, `k`는 그 리스트의 키 필드 타입이다. 수 연산 결과가 유한하지 않으면 실패한다. 실패하면 그 이벤트의 모든
 핸들러가 한 일이 취소된다.
@@ -258,6 +259,27 @@ text·리스트가 남을 때만 위치를 유지한다.
 앱이 멈추고 저장소가 닫힌 뒤에도 만들 때 그대로다. 실패한 저장 뒤의 스냅숏은 실패 전과 같다. 두 스레드가 최신
 스냅숏을 읽는 동안 앱이 계속 적용한다(ASan·UBSan·LeakSanitizer 빌드 포함).
 
+## 앱 사이의 연결
+
+출력으로 보낸 메시지(outbox)와 보낸 쪽별로 들은 번호(inbox)는 런타임이 소유하는 상태다([078](../research/078-app-connections.md)).
+
+- **보냄은 이벤트의 일부다.** `SEND`는 출력마다 1부터 매긴 번호를 붙여 메시지를 둔다. 이벤트가 실패하면 보내지 않았고,
+  저장이 실패하면 마지막 저장 뒤의 보냄과 그 번호가 모두 돌아간다. 호스트는 저장된 보냄만 본다(`pure_app_outbox`).
+  받는 쪽이 저장한 번호까지 호스트가 덜어 내면(`pure_app_trim`) 다음 저장과 함께 사라진다.
+- **받는 쪽은 한 번만 적용한다.** `pure_app_deliver(app, 입력, 보낸 쪽, 번호, 페이로드)`는 메시지 이벤트다. 보낸 쪽(1–64바이트,
+  공백 없는 인쇄 가능 ASCII)마다 마지막으로 들은 번호를 두고, 그 이하는 적용하지 않고 성공한다(`app->duplicate`). 번호는
+  이벤트의 변경과 함께 저장된다. 앱이 거절한 메시지(실패 종류가 있는 실패)도 들은 것으로 친다. 전달은 끝났고 효과는
+  거절이다. 저장이 실패하면 들은 번호도 돌아간다.
+- **저장.** `links` 상태 문서(`PLK1`)에 출력 표(이름, 필드 타입, 마지막 번호), 보낸 쪽별 번호, 덜어 내지 않은 보냄을
+  담는다. 바뀐 저장에서만 다시 쓴다. 모듈을 바꾸면 출력은 이름으로 이어진다. 보낸 것이 남은 출력을 없애거나 필드
+  타입을 바꾸는 모듈은 거절한다(`conflict`, 시작도 같다). 보낸 것이 없으면 그 번호를 기억해 두었다가 같은 이름의 출력이
+  돌아오면 이어서 매긴다. 받는 쪽이 이미 들은 번호를 다시 쓰지 않기 위해서다.
+
+검증(`test-module`): 실패한 이벤트의 보냄, 지연 저장의 보냄, 실패한 저장(보냄·덜어 냄·번호가 돌아간다), 재시작, 덜어 냄,
+중복·건너뛴 번호·다른 보낸 쪽의 전달, 거절한 전달, 저장 실패 뒤 다시 전달, 출력이 없는 모듈로의 교체와 되돌아옴, 잘못된
+`SEND`(타입, 출력이 아닌 인자, 모자란 값)와 출력의 핸들러. 커밋의 단계마다 프로세스를 죽여도 보내는 앱은 이벤트와
+보냄이, 받는 앱은 전달과 그 효과가 함께 남거나 함께 사라진다.
+
 ## HTTP 호스트 (`pure-serve`)
 
 `pure-serve --module 앱.pmod --store 경로`는 앱의 **보기와 입력이 곧 HTTP 인터페이스**다. 라우터 코드가 없다.
diff --git a/docs/research/078-app-connections.md b/docs/research/078-app-connections.md
index fbe7f34..5e89853 100644
--- a/docs/research/078-app-connections.md
+++ b/docs/research/078-app-connections.md
@@ -1,8 +1,9 @@
 # 078 — 앱 사이의 연결: 런타임이 소유하는 outbox와 정확히 한 번의 효과
 
-2026-09-26. 설계 노트이며 구현 순서를 포함한다. 상태: 설계만 했다. [로드맵](../roadmap.md)의 단기 과제(대여 완성, 저장과
-적용의 겹치기, 읽기 독립화, JSON 확장)를 먼저 하고, 중기 1(외부로 나가는 호출)과 함께 구현한다. 외부 API 호출도 같은
-outbox 위에 선다(출력을 HTTP 끝점에 잇는 배선).
+2026-09-26. 설계 노트이며 구현 순서를 포함한다. 상태: 1단계(Pure 런타임: 출력, `SEND`, outbox·inbox와 그 저장)와
+2단계(aup `output`·`send`)를 2026-09-27에 구현했다([008 앱 사이의 연결](../design/008-module-format.md)). 호스트의 배선과
+전달(3·4단계)은 [로드맵](../roadmap.md) 중기 1(외부로 나가는 호출)과 함께 한다. 외부 API 호출도 같은 outbox 위에 선다(출력을
+HTTP 끝점에 잇는 배선).
 
 착수 기준:
 
diff --git a/prototypes/connections/module-save.c b/prototypes/connections/module-save.c
index a5eb95a..4149316 100644
--- a/prototypes/connections/module-save.c
+++ b/prototypes/connections/module-save.c
@@ -10,6 +10,12 @@
  * field {u8 0 number | 1 text, u8 name length, name}, then its items
  * (module-list.h). A text state: its UTF-8 bytes. */
 static const unsigned char values_magic[4] = {'P', 'V', 'S', '1'}, list_magic[4] = {'P', 'L', 'S', '1'};
+/* links: "PLK1"; u8 outputs, each {u8 name length, name, u8 field count, the
+ * fields' types, u64le last number} (the module's outputs, then those it
+ * dropped, without fields); u16le sources, each {u8 length, the name, u64le
+ * last number heard}; then the sent messages to the end, each {u8 output (in
+ * the table), u64le number, u32le size, payload}. */
+static const unsigned char links_magic[4] = {'P', 'L', 'K', '1'};
 static const PureVersion unsaved = {UINT64_MAX, UINT64_MAX};
 
 bool pure_save_enabled(const PureApp *app) { return app->data.state && app->data.save; }
@@ -252,6 +258,154 @@ PURE_COLD PureResult pure_save_open_values(PureApp *app, const PureModule *m, do
 
 static bool same_version(PureVersion a, PureVersion b) { return a.ref == b.ref && a.bytes == b.bytes; }
 
+static unsigned char *put_le(unsigned char *at, uint64_t value, unsigned width) {
+    for (unsigned i = 0; i < width; ++i) *at++ = (unsigned char)(value >> (8 * i));
+    return at;
+}
+
+static uint64_t get_le(const unsigned char *at, unsigned width) {
+    uint64_t value = 0;
+    for (unsigned i = 0; i < width; ++i) value |= (uint64_t)at[i] << (8 * i);
+    return value;
+}
+
+/* The links as stored; NULL when memory runs out. */
+static unsigned char *encode_links(const PureApp *app, size_t *size) {
+    const PureModule *m = app->module;
+    unsigned outputs = 0;
+    uint8_t table[PURE_MODULE_INPUTS];
+    size_t total = 4 + 1 + 2;
+    for (uint32_t i = 0; i < m->inputs; ++i)
+        if (m->input[i].kind == PURE_KIND_OUTPUT) {
+            table[i] = (uint8_t)outputs++;
+            total += 2 + strlen(m->input_names[i]) + m->input[i].field_count + 8;
+        }
+    for (unsigned c = 0; c < app->carried_count; ++c) total += 2 + strlen(app->carried[c].name) + 8;
+    for (unsigned h = 0; h < app->heard_count; ++h) total += 1 + strlen(app->heard[h].source) + 8;
+    for (size_t s = 0; s < app->sent_count; ++s)
+        if (!app->sent[s].trimmed) total += 1 + 8 + 4 + app->sent[s].size;
+    unsigned char *bytes = malloc(total), *at = bytes;
+    if (!bytes) return NULL;
+    memcpy(at, links_magic, 4);
+    at += 4;
+    *at++ = (unsigned char)(outputs + app->carried_count);
+    for (uint32_t i = 0; i < m->inputs; ++i) {
+        if (m->input[i].kind != PURE_KIND_OUTPUT) continue;
+        size_t length = strlen(m->input_names[i]);
+        *at++ = (unsigned char)length;
+        memcpy(at, m->input_names[i], length);
+        at += length;
+        *at++ = (unsigned char)m->input[i].field_count;
+        for (uint32_t f = 0; f < m->input[i].field_count; ++f) *at++ = m->field[m->input[i].field_start + f].type;
+        at = put_le(at, app->last_sent[i], 8);
+    }
+    for (unsigned c = 0; c < app->carried_count; ++c) {
+        size_t length = strlen(app->carried[c].name);
+        *at++ = (unsigned char)length;
+        memcpy(at, app->carried[c].name, length);
+        at += length;
+        *at++ = 0;
+        at = put_le(at, app->carried[c].last, 8);
+    }
+    at = put_le(at, app->heard_count, 2);
+    for (unsigned h = 0; h < app->heard_count; ++h) {
+        size_t length = strlen(app->heard[h].source);
+        *at++ = (unsigned char)length;
+        memcpy(at, app->heard[h].source, length);
+        at = put_le(at + length, app->heard[h].seq, 8);
+    }
+    for (size_t s = 0; s < app->sent_count; ++s) {
+        const struct PureSent *sent = &app->sent[s];
+        if (sent->trimmed) continue;
+        *at++ = table[sent->output];
+        at = put_le(at, sent->seq, 8);
+        at = put_le(at, sent->size, 4);
+        if (sent->size) memcpy(at, sent->payload, sent->size);
+        at += sent->size;
+    }
+    *size = (size_t)(at - bytes);
+    return bytes;
+}
+
+/* The stored links of a starting app: outputs by name (an output whose
+ * fields changed or that the module lacks refuses the start while it has
+ * sends left: CONFLICT; one without sends keeps its number for later),
+ * sources and sent messages. */
+PURE_COLD PureResult pure_save_open_links(PureApp *app, const PureModule *m) {
+    PureResult result = app->stored_links ? PURE_OK : open_state(app, "", "links", &app->stored_links);
+    if (result != PURE_OK || !pure_draft_revision(app->stored_links)) return result;
+    unsigned char *bytes;
+    size_t size;
+    if ((result = read_whole(app->stored_links, &bytes, &size)) != PURE_OK) return result;
+    int output[256];
+    bool fits[256];
+    size_t at = 5;
+    unsigned outputs = size >= 5 && !memcmp(bytes, links_magic, 4) ? bytes[4] : 0;
+    result = size >= 5 && !memcmp(bytes, links_magic, 4) ? PURE_OK : PURE_STORE_ERROR;
+    for (unsigned t = 0; t < outputs && result == PURE_OK; ++t) {
+        char name[PURE_MODULE_NAME];
+        size_t length = at < size ? bytes[at] : 0;
+        if (!length || length >= PURE_MODULE_NAME || size - at < 2 + length) { result = PURE_STORE_ERROR; break; }
+        memcpy(name, bytes + at + 1, length);
+        name[length] = 0;
+        unsigned fields = bytes[at + 1 + length];
+        at += 2 + length;
+        if (size - at < fields + 8) { result = PURE_STORE_ERROR; break; }
+        int i = pure_module_input(m, name);
+        output[t] = i >= 0 && m->input[i].kind == PURE_KIND_OUTPUT ? i : -1;
+        fits[t] = output[t] >= 0 && m->input[i].field_count == fields;
+        for (unsigned f = 0; fits[t] && f < fields; ++f) fits[t] = m->field[m->input[i].field_start + f].type == bytes[at + f];
+        uint64_t last = get_le(bytes + at + fields, 8);
+        at += fields + 8;
+        if (output[t] >= 0) app->last_sent[output[t]] = last;
+        else if (last) {
+            struct PureCarried *grown = realloc(app->carried, (app->carried_count + 1) * sizeof(*grown));
+            if (!grown) { result = PURE_ALLOCATION; break; }
+            app->carried = grown;
+            memcpy(grown[app->carried_count].name, name, length + 1);
+            grown[app->carried_count++].last = last;
+        }
+    }
+    unsigned sources = result == PURE_OK && size - at >= 2 ? (unsigned)get_le(bytes + at, 2) : 0;
+    if (result == PURE_OK && size - at < 2) result = PURE_STORE_ERROR;
+    at += 2;
+    if (result == PURE_OK && sources && !(app->heard = calloc(sources, sizeof(*app->heard)))) result = PURE_ALLOCATION;
+    app->heard_capacity = sources;
+    for (unsigned h = 0; h < sources && result == PURE_OK; ++h) {
+        size_t length = at < size ? bytes[at] : 0;
+        if (!length || length > PURE_SOURCE || size - at < 1 + length + 8) { result = PURE_STORE_ERROR; break; }
+        struct PureHeard *heard = &app->heard[app->heard_count++];
+        memcpy(heard->source, bytes + at + 1, length);
+        heard->seq = heard->saved = get_le(bytes + at + 1 + length, 8);
+        at += 1 + length + 8;
+    }
+    while (result == PURE_OK && at < size) {
+        if (size - at < 13 || bytes[at] >= outputs) { result = PURE_STORE_ERROR; break; }
+        unsigned t = bytes[at];
+        uint64_t seq = get_le(bytes + at + 1, 8);
+        size_t length = (size_t)get_le(bytes + at + 9, 4);
+        at += 13;
+        if (size - at < length) { result = PURE_STORE_ERROR; break; }
+        if (!fits[t]) { result = PURE_CONFLICT; break; }
+        if (app->sent_count == app->sent_capacity) {
+            size_t capacity = app->sent_capacity ? 2 * app->sent_capacity : 16;
+            struct PureSent *grown = realloc(app->sent, capacity * sizeof(*grown));
+            if (!grown) { result = PURE_ALLOCATION; break; }
+            app->sent = grown;
+            app->sent_capacity = capacity;
+        }
+        unsigned char *payload = malloc(length ? length : 1);
+        if (!payload) { result = PURE_ALLOCATION; break; }
+        if (length) memcpy(payload, bytes + at, length);
+        app->sent[app->sent_count++] = (struct PureSent){(uint32_t)output[t], seq, payload, length, false};
+        at += length;
+    }
+    free(bytes);
+    app->sent_durable = app->sent_count;
+    memcpy(app->saved_sent, app->last_sent, sizeof(app->saved_sent));
+    return result;
+}
+
 PureResult pure_save_commit(PureApp *app, const double *values, void *const *documents, size_t document_count,
                             int64_t *revisions) {
     if (!pure_save_enabled(app)) {
@@ -262,8 +416,8 @@ PureResult pure_save_commit(PureApp *app, const double *values, void *const *doc
         return result;
     }
     const PureModule *m = app->module;
-    PureDraft *drafts[2 * PURE_MODULE_STATES + 1];
-    uint32_t states[2 * PURE_MODULE_STATES + 1];
+    PureDraft *drafts[2 * PURE_MODULE_STATES + 2];
+    uint32_t states[2 * PURE_MODULE_STATES + 2];
     size_t count = 0;
     for (uint32_t i = 0; i < m->states; ++i) {
         PureDraft *draft = m->state_type[i] == PURE_STATE_TEXT ? app->texts[i] :
@@ -292,6 +446,32 @@ PureResult pure_save_commit(PureApp *app, const double *values, void *const *doc
             drafts[count++] = app->stored_values;
         }
     }
+    /* The links, written whole when they changed. */
+    PureVersion links_before = {0, 0};
+    bool links_written = false;
+    if (app->links_changed) {
+        PureResult result = app->stored_links ? PURE_OK : open_state(app, "", "links", &app->stored_links);
+        size_t length;
+        unsigned char *encoded = result == PURE_OK ? encode_links(app, &length) : NULL;
+        if (result == PURE_OK && !encoded) result = PURE_ALLOCATION;
+        if (result == PURE_OK) {
+            links_before = pure_draft_version(app->stored_links);
+            pure_draft_hold(app->stored_links, true);
+            result = pure_draft_splice(app->stored_links, 0, pure_draft_size(app->stored_links), (PureSpan){encoded, length});
+            if (result != PURE_OK) pure_draft_hold(app->stored_links, false);
+        }
+        free(encoded);
+        if (result != PURE_OK) {
+            if (values_changed) {
+                pure_draft_restore(app->stored_values, values_before);
+                pure_draft_hold(app->stored_values, false);
+            }
+            return result;
+        }
+        links_written = true;
+        states[count] = UINT32_MAX;
+        drafts[count++] = app->stored_links;
+    }
     if (!count && !document_count) return PURE_OK;
     PureResult result = app->data.save(app->data.context, documents, document_count, drafts, count, revisions);
     for (size_t k = 0; result == PURE_OK && k < count; ++k)
@@ -300,5 +480,9 @@ PureResult pure_save_commit(PureApp *app, const double *values, void *const *doc
         if (result != PURE_OK) pure_draft_restore(app->stored_values, values_before);
         pure_draft_hold(app->stored_values, false);
     }
+    if (links_written) {
+        if (result != PURE_OK) pure_draft_restore(app->stored_links, links_before);
+        pure_draft_hold(app->stored_links, false);
+    }
     return result;
 }
diff --git a/prototypes/connections/module-save.h b/prototypes/connections/module-save.h
index ded9ba9..bc7adcc 100644
--- a/prototypes/connections/module-save.h
+++ b/prototypes/connections/module-save.h
@@ -21,6 +21,9 @@ PureResult pure_save_open_values(PureApp *, const PureModule *m, double *values,
                                  PureDraft *const *texts, PureList *const *lists);
 /* Store a list whose record changed in its draft, header and items anew. */
 PureResult pure_save_rewrite_list(const PureModule *m, uint32_t state, PureList *list, PureDraft *draft);
+/* The stored links of a starting app (module.h, PureApp); CONFLICT when an
+ * output with sends left is missing or has other fields. */
+PureResult pure_save_open_links(PureApp *, const PureModule *m);
 /* Publish the documents and every state changed since the last save
  * (values are the states' numbers); revisions are the documents'. */
 PureResult pure_save_commit(PureApp *, const double *values, void *const *documents, size_t document_count,
diff --git a/prototypes/connections/module.c b/prototypes/connections/module.c
index b1f58fe..ee72494 100644
--- a/prototypes/connections/module.c
+++ b/prototypes/connections/module.c
@@ -21,7 +21,7 @@ enum { READ = IN_VIEW | IN_HANDLER };
  * collection, a constant, a unit, the column's row, a branch target, a stack
  * depth, a fold, a relation or a keyed list. */
 enum { A_NONE, A_STATE, A_ITEM = A_STATE + PURE_STATE_INDEX + 1, A_FIELD, A_COLLECTION, A_CONSTANT, A_UNIT,
-       A_ROW, A_TARGET, A_DEPTH, A_FOLD, A_RELATION, A_KEYED };
+       A_ROW, A_TARGET, A_DEPTH, A_FOLD, A_RELATION, A_KEYED, A_OUTPUT };
 /* A fold's block is verified as a column that leaves one number (count) or
  * two (a value and whether to include it), a table's filter as a column that
  * leaves one: these stand for a view type. */
@@ -98,6 +98,7 @@ static const struct { unsigned char pops, in[3], out, where, arg, of; } signatur
     [PURE_OP_LIST_FIND] = {1, {T_KEY}, T_INDEX, READ, A_KEYED, O_ARG},
     [PURE_OP_LIST_HAS] = {1, {T_KEY}, T_NUM, READ, A_KEYED, O_ANY},
     [PURE_OP_NOW] = {0, {0}, T_NUM, IN_HANDLER, A_NONE, O_ANY},
+    [PURE_OP_SEND] = {0, {0}, 0, IN_HANDLER, A_OUTPUT, O_ANY},
     [PURE_OP_FLOOR] = NUMS(1), [PURE_OP_MOD] = NUMS(2),
 };
 #undef NUMS
@@ -109,6 +110,10 @@ static uint32_t read32(const unsigned char *p) {
     return (uint32_t)p[0] | (uint32_t)p[1] << 8 | (uint32_t)p[2] << 16 | (uint32_t)p[3] << 24;
 }
 
+static void write32(unsigned char *p, uint32_t value) {
+    for (unsigned i = 0; i < 4; ++i) p[i] = (unsigned char)(value >> (8 * i));
+}
+
 static double read_double(const unsigned char *p) {
     uint64_t bits = read32(p) | (uint64_t)read32(p + 4) << 32;
     double value;
@@ -187,6 +192,7 @@ PURE_COLD static bool names_arg(const PureModule *m, unsigned kind, uint32_t arg
     case A_FOLD: return (arg & 0xFF) <= PURE_FOLD_FIRST && state_of(m, arg >> 8 & 0xFF, PURE_STATE_LIST) && arg >> 16;
     case A_RELATION: return arg <= PURE_NOT_EQUAL;
     case A_KEYED: return state_of(m, arg, PURE_STATE_LIST) && m->state_key[arg];
+    case A_OUTPUT: return arg < m->inputs && m->input[arg].kind == PURE_KIND_OUTPUT;
     case A_NONE: case A_TARGET: case A_DEPTH: return true;
     default: return state_of(m, arg, (uint8_t)(kind - A_STATE));
     }
@@ -296,10 +302,17 @@ PURE_COLD static const char *verify_code(const PureModule *m, PureBlock block, i
             }
         }
         /* Operand types, bottom to top; record operations take their fields. */
-        unsigned char need[PURE_RECORD_FIELDS + 2];
+        unsigned char need[PURE_MODULE_FIELDS + 2];
         unsigned pops = signature[op->op].pops;
         memcpy(need, signature[op->op].in, sizeof(signature[op->op].in));
-        if (op->op == PURE_OP_LIST_APPEND || op->op == PURE_OP_LIST_INSERT) {
+        if (op->op == PURE_OP_SEND) {
+            const PureInputShape *output = &m->input[arg];
+            for (unsigned f = 0; f < output->field_count; ++f) {
+                uint8_t type = m->field[output->field_start + f].type;
+                need[f] = type == PURE_FIELD_TEXT ? T_TEXT : type == PURE_FIELD_BYTES ? T_BYTES : T_NUM;
+            }
+            pops = output->field_count;
+        } else if (op->op == PURE_OP_LIST_APPEND || op->op == PURE_OP_LIST_INSERT) {
             const PureRecord *record = &m->record[m->state_ref[arg]];
             for (unsigned f = 0; f < record->count; ++f) need[pops + f] = field_type(m, arg, f);
             pops += record->count;
@@ -516,7 +529,8 @@ PURE_COLD PureResult pure_module_load(const void *bytes, size_t size, PureModule
                 shape->period > PURE_TIMER_LONGEST) REFUSE("timer input", i);
             continue;
         }
-        if (shape->kind != PURE_KIND_MESSAGE || read32(at + 8) || shape->field_start != next_field ||
+        if ((shape->kind != PURE_KIND_MESSAGE && shape->kind != PURE_KIND_OUTPUT) || read32(at + 8) ||
+            (shape->kind == PURE_KIND_OUTPUT && shape->keyed) || shape->field_start != next_field ||
             shape->field_count > out->fields - next_field) REFUSE("message shape", i);
         /* The prefix holds the fixed fields and the variable fields' lengths
          * in order, but the last variable field's: it takes the rest. */
@@ -609,7 +623,8 @@ PURE_COLD PureResult pure_module_load(const void *bytes, size_t size, PureModule
         uint32_t event = read32(at);
         out->handler[i] = (PureHandler){event, {read32(at + 4), read32(at + 8)}};
         int input = event_input(event);
-        if (event > 2u * out->inputs || (input >= 0 && out->input[input].kind != PURE_KIND_KEY && !(event & 1)))
+        if (event > 2u * out->inputs || (input >= 0 && out->input[input].kind != PURE_KIND_KEY && !(event & 1)) ||
+            (input >= 0 && out->input[input].kind == PURE_KIND_OUTPUT))
             REFUSE("handler event", i);
     }
     for (uint32_t i = 0; i < out->code_count; ++i, at += OP_BYTES) {
@@ -838,6 +853,49 @@ static PureResult keep_bytes(PureArena *arena, const void *bytes, size_t length,
 /* Run a verified block. Types and stack bounds were proven at load. A change
  * the data refuses (a size limit, a missing item) fails the event with its
  * fault; other failures pass through. */
+typedef struct { unsigned char *at; } Copy;
+
+static bool copy_out(void *context, PureSpan span) {
+    Copy *copy = context;
+    if (span.size) memcpy(copy->at, span.data, span.size);
+    copy->at += span.size;
+    return true;
+}
+
+/* An output's message from its field values, laid out as a message's
+ * payload (size bytes), numbered and kept with the event's changes. */
+static PureResult send(PureApp *app, uint32_t output, const Slot *fields, const PureArena *arena, size_t size) {
+    const PureModule *m = app->module;
+    const PureInputShape *shape = &m->input[output];
+    if (app->sent_count == app->sent_capacity) {
+        size_t capacity = app->sent_capacity ? 2 * app->sent_capacity : 16;
+        struct PureSent *grown = realloc(app->sent, capacity * sizeof(*grown));
+        if (!grown) return PURE_ALLOCATION;
+        app->sent = grown;
+        app->sent_capacity = capacity;
+    }
+    unsigned char *payload = malloc(size ? size : 1);
+    if (!payload) return PURE_ALLOCATION;
+    Copy copy = {payload + shape->prefix};
+    for (uint32_t f = 0; f < shape->field_count; ++f) {
+        const PureField *field = &m->field[shape->field_start + f];
+        if (field->type == PURE_FIELD_U32) write32(payload + field->offset, (uint32_t)fields[f].num);
+        else if (field->type == PURE_FIELD_NUMBER) {
+            uint64_t bits;
+            memcpy(&bits, &fields[f].num, 8);
+            write32(payload + field->offset, (uint32_t)bits);
+            write32(payload + field->offset + 4, (uint32_t)(bits >> 32));
+        } else {
+            if (field->offset != PURE_FIELD_REST) write32(payload + field->offset, (uint32_t)fields[f].bytes.length);
+            PureResult result = visit_value(app, arena, fields[f].bytes, copy_out, &copy);
+            if (result != PURE_OK) { free(payload); return result; }
+        }
+    }
+    app->sent[app->sent_count++] = (struct PureSent){output, ++app->last_sent[output], payload, size, false};
+    app->links_changed = true;
+    return PURE_OK;
+}
+
 static PureResult run(const PureModule *m, double *values, PureBlock block, Event *ev, Slot *result) {
     Slot stack[PURE_MODULE_STACK];
     size_t sp = 0;
@@ -995,6 +1053,23 @@ static PureResult run(const PureModule *m, double *values, PureBlock block, Even
         }
         case PURE_OP_BYTES_FROM: case PURE_OP_INDEX_NUM: continue; /* the same bytes or position, retyped */
         case PURE_OP_LIST_COUNT: stack[sp++].num = (double)pure_list_count(app->lists[arg]); continue;
+        case PURE_OP_SEND: {
+            const PureInputShape *output = &m->input[arg];
+            size_t base = sp - output->field_count, size = output->prefix;
+            sp = base;
+            for (uint32_t f = 0; f < output->field_count; ++f) {
+                uint8_t type = m->field[output->field_start + f].type;
+                double number = stack[base + f].num;
+                if (type == PURE_FIELD_BYTES || type == PURE_FIELD_TEXT) size += (size_t)stack[base + f].bytes.length;
+                else if (type == PURE_FIELD_U32 && !(number >= 0 && number <= UINT32_MAX && number == floor(number))) {
+                    kind = PURE_FAULT_LIMIT;
+                    goto failed;
+                }
+            }
+            if (size > UINT32_MAX) { kind = PURE_FAULT_LIMIT; goto failed; }
+            if ((status = send(app, arg, stack + base, arena, size)) != PURE_OK) return status;
+            continue;
+        }
         case PURE_OP_LIST_APPEND: case PURE_OP_LIST_INSERT: {
             PureList *list = app->lists[arg];
             const PureRecord *record = &m->record[m->state_ref[arg]];
@@ -1253,6 +1328,37 @@ static void close_burst(PureApp *app, bool keep) {
     app->open = false;
 }
 
+static void drop_sent(PureApp *app, size_t count) {
+    while (app->sent_count > count) free(app->sent[--app->sent_count].payload);
+}
+
+/* After a save: what it stored is the links' saved state (trimmed sends go),
+ * or, when it failed, the links return to the last saved state. */
+static void settle_links(PureApp *app, bool saved) {
+    if (saved) {
+        size_t kept = 0;
+        for (size_t i = 0; i < app->sent_count; ++i) {
+            if (app->sent[i].trimmed) free(app->sent[i].payload);
+            else app->sent[kept++] = app->sent[i];
+        }
+        app->sent_count = app->sent_durable = kept;
+        memcpy(app->saved_sent, app->last_sent, sizeof(app->saved_sent));
+        for (unsigned i = 0; i < app->heard_count; ++i) app->heard[i].saved = app->heard[i].seq;
+    } else {
+        drop_sent(app, app->sent_durable);
+        for (size_t i = 0; i < app->sent_count; ++i) app->sent[i].trimmed = false;
+        memcpy(app->last_sent, app->saved_sent, sizeof(app->last_sent));
+        unsigned kept = 0;
+        for (unsigned i = 0; i < app->heard_count; ++i)
+            if (app->heard[i].saved) {
+                app->heard[kept] = app->heard[i];
+                app->heard[kept++].seq = app->heard[i].saved;
+            }
+        app->heard_count = kept;
+    }
+    app->links_changed = false;
+}
+
 PureResult pure_app_save(PureApp *app) {
     void *documents[PURE_APP_PENDING];
     int64_t revisions[PURE_APP_PENDING];
@@ -1260,6 +1366,7 @@ PureResult pure_app_save(PureApp *app) {
     for (unsigned i = 0; i < app->pending_count; ++i)
         if (app->pending[i].edited) documents[count++] = app->pending[i].document;
     PureResult result = pure_save_commit(app, app->values, documents, count, revisions);
+    settle_links(app, result == PURE_OK);
     close_burst(app, result == PURE_OK);
     return result;
 }
@@ -1315,6 +1422,11 @@ void pure_app_stop(PureApp *app) {
     } else
         pure_retired_free(&app->retired);
     pure_draft_close(app->stored_values);
+    drop_sent(app, 0);
+    free(app->sent);
+    free(app->heard);
+    free(app->carried);
+    pure_draft_close(app->stored_links);
     free(app->pending);
     pure_undo_free(&app->undo);
     for (unsigned i = 0; i < app->row_tables; ++i) free(app->rows[i].items);
@@ -1325,7 +1437,12 @@ void pure_app_stop(PureApp *app) {
 }
 
 /* Where one event began inside the burst: a failed event returns there. */
-typedef struct { PureVersion texts[PURE_MODULE_STATES], lists[PURE_MODULE_STATES]; size_t undo; bool opened; } Mark;
+typedef struct {
+    PureVersion texts[PURE_MODULE_STATES], lists[PURE_MODULE_STATES];
+    size_t undo, sent;
+    uint64_t last_sent[PURE_MODULE_INPUTS];
+    bool opened;
+} Mark;
 
 static void mark_event(PureApp *app, Mark *mark) {
     mark->opened = !app->open;
@@ -1335,6 +1452,8 @@ static void mark_event(PureApp *app, Mark *mark) {
         if (app->list_drafts[i]) mark->lists[i] = pure_draft_version(app->list_drafts[i]);
     }
     mark->undo = app->undo.count;
+    mark->sent = app->sent_count;
+    memcpy(mark->last_sent, app->last_sent, sizeof(mark->last_sent));
 }
 
 static void undo_event(PureApp *app, const Mark *mark) {
@@ -1343,9 +1462,30 @@ static void undo_event(PureApp *app, const Mark *mark) {
         if (app->list_drafts[i]) pure_draft_restore(app->list_drafts[i], mark->lists[i]);
     }
     pure_undo_rollback(&app->undo, mark->undo);
+    drop_sent(app, mark->sent);
+    memcpy(app->last_sent, mark->last_sent, sizeof(app->last_sent));
     if (mark->opened) close_burst(app, true); /* nothing happened: release the holds */
 }
 
+/* The last number heard from source, or NULL. */
+static struct PureHeard *heard_from(PureApp *app, const char *source) {
+    for (unsigned i = 0; i < app->heard_count; ++i)
+        if (!strcmp(app->heard[i].source, source)) return &app->heard[i];
+    return NULL;
+}
+
+/* Hear seq from source (room for a new source was made before). */
+static void hear(PureApp *app, const char *source, uint64_t seq) {
+    struct PureHeard *heard = heard_from(app, source);
+    if (!heard) {
+        heard = &app->heard[app->heard_count++];
+        *heard = (struct PureHeard){.saved = 0};
+        memcpy(heard->source, source, strlen(source) + 1);
+    }
+    heard->seq = seq;
+    app->links_changed = true;
+}
+
 static PureResult keep_reply(PureApp *app);
 
 /* One event: every handler for it runs in module order against pending
@@ -1384,6 +1524,10 @@ static PureResult dispatch(PureApp *app, uint32_t event, Event *ev) {
         memcpy(app->pending[app->pending_count++].key, ev->key, strlen(ev->key) + 1);
         app->revision += ev->modified; /* a save moves an edited document one revision */
     }
+    if (app->delivering.pending) {
+        hear(app, app->delivering.source, app->delivering.seq);
+        app->delivering.pending = false;
+    }
     return event && !app->deferred ? pure_app_save(app) : PURE_OK;
 }
 
@@ -1517,6 +1661,58 @@ PureResult pure_app_message_fields(PureApp *app, uint32_t input, const char *key
     return message(app, input, key, base, payload, placed);
 }
 
+void pure_app_outbox(const PureApp *app, PureSentVisit visit, void *context) {
+    for (size_t i = 0; i < app->sent_durable; ++i) {
+        const struct PureSent *sent = &app->sent[i];
+        if (!sent->trimmed && !visit(context, sent->output, sent->seq, (PureSpan){sent->payload, sent->size})) return;
+    }
+}
+
+void pure_app_trim(PureApp *app, uint32_t output, uint64_t through) {
+    for (size_t i = 0; i < app->sent_durable; ++i)
+        if (app->sent[i].output == output && app->sent[i].seq <= through && !app->sent[i].trimmed) {
+            app->sent[i].trimmed = true;
+            app->links_changed = true;
+        }
+}
+
+PureResult pure_app_deliver(PureApp *app, uint32_t input, const char *source, uint64_t seq, PureSpan payload) {
+    const PureModule *m = app->module;
+    size_t length = source ? strlen(source) : 0;
+    app->duplicate = false;
+    if (input >= m->inputs || m->input[input].kind != PURE_KIND_MESSAGE || m->input[input].keyed || !length ||
+        length > PURE_SOURCE || !seq) return PURE_INVALID;
+    for (size_t i = 0; i < length; ++i)
+        if (source[i] <= ' ' || source[i] > '~') return PURE_INVALID;
+    struct PureHeard *heard = heard_from(app, source);
+    if (heard && seq <= heard->seq) {
+        pure_app_finish(app);
+        app->duplicate = true;
+        return PURE_OK;
+    }
+    if (!heard && app->heard_count == app->heard_capacity) {
+        unsigned capacity = app->heard_capacity ? 2 * app->heard_capacity : 4;
+        struct PureHeard *grown = realloc(app->heard, capacity * sizeof(*grown));
+        if (!grown) return PURE_ALLOCATION;
+        app->heard = grown;
+        app->heard_capacity = capacity;
+    }
+    app->delivering.source = source;
+    app->delivering.seq = seq;
+    app->delivering.pending = true;
+    PureResult result = message(app, input, NULL, -1, payload, NULL);
+    if (!app->delivering.pending) return result;
+    /* Not applied. Refused by the app (a fault), the message was delivered
+     * all the same: its number is heard, saved like a key event's changes. */
+    app->delivering.pending = false;
+    if (app->fault == PURE_FAULT_NONE) return result;
+    Mark mark;
+    mark_event(app, &mark);
+    hear(app, source, seq);
+    PureResult saved = app->deferred ? PURE_OK : pure_app_save(app);
+    return saved == PURE_OK ? result : saved;
+}
+
 PureResult pure_app_reply(const PureApp *app, bool (*visit)(void *, PureSpan), void *context) {
     if (!app->has_reply) return PURE_OK;
     return visit_value(app, &app->arena, app->reply, visit, context);
@@ -1738,6 +1934,52 @@ PURE_COLD static PureResult keep_list(PureList **out, PureList *from, const Pure
  * again, and the next event or save publishes the module's states; a state
  * the module drops stays stored as it was. Starting an app opens its module
  * over none. */
+/* Outputs of old that module has by name and with the same fields keep their
+ * sends and numbers; one it lacks or changed refuses the module while sends
+ * are left on it (CONFLICT), and otherwise keeps its number for later. */
+PURE_COLD static PureResult remap_outputs(PureApp *app, const PureModule *old, const PureModule *module, bool apply) {
+    int map[PURE_MODULE_INPUTS];
+    for (uint32_t i = 0; i < old->inputs; ++i) {
+        map[i] = -1;
+        if (old->input[i].kind != PURE_KIND_OUTPUT) continue;
+        int to = pure_module_input(module, old->input_names[i]);
+        bool same = to >= 0 && module->input[to].kind == PURE_KIND_OUTPUT &&
+                    module->input[to].field_count == old->input[i].field_count;
+        for (uint32_t f = 0; same && f < old->input[i].field_count; ++f)
+            same = module->field[module->input[to].field_start + f].type == old->field[old->input[i].field_start + f].type;
+        map[i] = same ? to : -1;
+        for (size_t s = 0; !same && s < app->sent_count; ++s)
+            if (app->sent[s].output == i) return PURE_CONFLICT;
+    }
+    if (!apply) return PURE_OK;
+    uint64_t last[PURE_MODULE_INPUTS] = {0}, saved[PURE_MODULE_INPUTS] = {0};
+    for (uint32_t i = 0; i < old->inputs; ++i) {
+        if (old->input[i].kind != PURE_KIND_OUTPUT || !app->last_sent[i]) continue;
+        if (map[i] >= 0) {
+            last[map[i]] = app->last_sent[i];
+            saved[map[i]] = app->saved_sent[i];
+            continue;
+        }
+        struct PureCarried *grown = realloc(app->carried, (app->carried_count + 1) * sizeof(*grown));
+        if (!grown) continue; /* its numbers would start again: kept only while memory allows */
+        app->carried = grown;
+        memcpy(grown[app->carried_count].name, old->input_names[i], PURE_MODULE_NAME);
+        grown[app->carried_count++].last = app->last_sent[i];
+        app->links_changed = true;
+    }
+    for (unsigned c = 0; c < app->carried_count;) {
+        int to = pure_module_input(module, app->carried[c].name);
+        if (to < 0 || module->input[to].kind != PURE_KIND_OUTPUT || last[to]) { ++c; continue; }
+        last[to] = saved[to] = app->carried[c].last;
+        app->carried[c] = app->carried[--app->carried_count];
+        app->links_changed = true;
+    }
+    for (size_t s = 0; s < app->sent_count; ++s) app->sent[s].output = (uint32_t)map[app->sent[s].output];
+    memcpy(app->last_sent, last, sizeof(last));
+    memcpy(app->saved_sent, saved, sizeof(saved));
+    return PURE_OK;
+}
+
 PURE_COLD static PureResult open_module(PureApp *app, const PureModule *module) {
     const PureModule *old = app->module;
     bool stored = pure_save_enabled(app);
@@ -1797,6 +2039,8 @@ PURE_COLD static PureResult open_module(PureApp *app, const PureModule *module)
         values[i] = same ? app->values[kept[i]] : 0;
     }
     if (result == PURE_OK && stored) result = pure_save_open_values(app, module, values, kept, texts, lists);
+    /* Outputs keep their sends and numbers by name. */
+    if (result == PURE_OK) result = old ? remap_outputs(app, old, module, false) : stored ? pure_save_open_links(app, module) : PURE_OK;
     for (uint32_t i = 0; i < module->states; ++i) {
         uint32_t list = module->state_ref[i];
         if (module->state_type[i] == PURE_STATE_INDEX && moved[list] && values[i] < (double)pure_list_count(lists[list]))
@@ -1846,6 +2090,7 @@ PURE_COLD static PureResult open_module(PureApp *app, const PureModule *module)
     memcpy(app->lists, lists, sizeof(lists));
     memcpy(app->list_drafts, list_drafts, sizeof(list_drafts));
     memcpy(app->saved, saved, sizeof(saved));
+    if (old) remap_outputs(app, old, module, true);
     app->module = module;
     ++app->changes;
     return PURE_OK;
diff --git a/prototypes/connections/module.h b/prototypes/connections/module.h
index 430ccae..ed40c15 100644
--- a/prototypes/connections/module.h
+++ b/prototypes/connections/module.h
@@ -38,7 +38,7 @@ typedef enum {
     PURE_OP_ITEM_GET, PURE_OP_ITEM_SET, PURE_OP_INDEX_LOAD, PURE_OP_INDEX_STORE,
     PURE_OP_INDEX_MOVE, PURE_OP_INDEX_FIRST, PURE_OP_INDEX_LAST, PURE_OP_INDEX_NUM, PURE_OP_ROW,
     PURE_OP_PICK, PURE_OP_SLIDE, PURE_OP_FOLD, PURE_OP_TEXT_COMPARE, PURE_OP_LIST_FIND, PURE_OP_LIST_HAS,
-    PURE_OP_NOW, PURE_OP_FLOOR, PURE_OP_MOD,
+    PURE_OP_NOW, PURE_OP_FLOOR, PURE_OP_MOD, PURE_OP_SEND,
     PURE_OP_COUNT
 } PureOpcode;
 /* FOLD's argument: the kind (bits 0-7), the list state (8-15) and the length
@@ -51,9 +51,12 @@ typedef enum { PURE_FOLD_COUNT, PURE_FOLD_SUM, PURE_FOLD_MAX, PURE_FOLD_MIN, PUR
  * comparisons (LT LE GT GE EQ NE); texts order by their bytes. */
 typedef enum { PURE_LESS, PURE_LESS_EQUAL, PURE_GREATER, PURE_GREATER_EQUAL, PURE_EQUAL, PURE_NOT_EQUAL } PureRelation;
 
-/* A timer ticks every `period` milliseconds while a host runs the app. */
-typedef enum { PURE_KIND_KEY = 1, PURE_KIND_MESSAGE = 2, PURE_KIND_TEXT = 3, PURE_KIND_TIMER = 4 } PureInputKind;
-enum { PURE_TIMER_LONGEST = 86400000 };
+/* A timer ticks every `period` milliseconds while a host runs the app. An
+ * output is a message the app sends (SEND, in a handler; it has fields as a
+ * message has and no handler): what receives it is the host's wiring. */
+typedef enum { PURE_KIND_KEY = 1, PURE_KIND_MESSAGE = 2, PURE_KIND_TEXT = 3, PURE_KIND_TIMER = 4,
+               PURE_KIND_OUTPUT = 5 } PureInputKind;
+enum { PURE_TIMER_LONGEST = 86400000, PURE_SOURCE = 64 };
 /* A message field: a u32le or a binary64 (fixed, at its offset in the
  * payload's prefix), or bytes or text of any length. The prefix also holds
  * each variable field's u32le length but the last one's (its offset is
@@ -228,6 +231,22 @@ typedef struct {
     unsigned char *copied;
     int64_t revision;
     PureArena arena, view_arena;
+    /* Links (runtime-owned, saved with the states): messages sent on outputs
+     * and not trimmed, in the order sent (the first sent_durable are saved;
+     * a trimmed one goes at the next save), each output's last number (and
+     * as saved), those of outputs a module dropped, and the last number
+     * delivered from each source (and as saved). */
+    struct PureSent { uint32_t output; uint64_t seq; unsigned char *payload; size_t size; bool trimmed; } *sent;
+    size_t sent_count, sent_capacity, sent_durable;
+    uint64_t last_sent[PURE_MODULE_INPUTS], saved_sent[PURE_MODULE_INPUTS];
+    struct PureCarried { char name[PURE_MODULE_NAME]; uint64_t last; } *carried;
+    unsigned carried_count;
+    struct PureHeard { char source[PURE_SOURCE + 1]; uint64_t seq, saved; } *heard;
+    unsigned heard_count, heard_capacity;
+    PureDraft *stored_links;
+    bool links_changed; /* since the last save */
+    bool duplicate;     /* the last delivery was heard before and not applied */
+    struct { const char *source; uint64_t seq; bool pending; } delivering;
     struct PureSnapshot *snapshot; /* the newest snapshot made, or NULL */
     PureRetired retired;           /* list items let go since it was made */
     bool reading;                  /* a reader: its states are a snapshot's */
@@ -285,6 +304,21 @@ PureResult pure_app_cell_text(PureApp *, uint32_t column, size_t row, bool (*vis
  * start at their initial value. The module must outlive the app. */
 PureResult pure_app_replace(PureApp *, const PureModule *);
 
+/* Connections between apps (docs/research/078-app-connections.md). A host
+ * sees only saved sends, delivers each at least once, in order per output,
+ * and trims what its receiver has saved; a receiver applies each once. */
+typedef bool (*PureSentVisit)(void *context, uint32_t output, uint64_t seq, PureSpan payload);
+void pure_app_outbox(const PureApp *, PureSentVisit visit, void *context);
+/* Forget output's saved sends numbered up to through (saved with the next save). */
+void pure_app_trim(PureApp *, uint32_t output, uint64_t through);
+/* Deliver a message event from source (a sender's name, 1 to 64 printable
+ * ASCII bytes) numbered seq (from 1, growing per source). A number heard
+ * before is not applied again: it succeeds with app->duplicate set. The
+ * number counts as heard with the event's changes, and also when the app
+ * refuses the message (a fault: it was delivered, and refused); not when a
+ * save fails. */
+PureResult pure_app_deliver(PureApp *, uint32_t input, const char *source, uint64_t seq, PureSpan payload);
+
 /* Readers on other threads. Right after a save an app can make a snapshot of
  * its saved states, for evaluating views anywhere while it keeps applying
  * events: numbers are copied; text states share their stored versions and
diff --git a/prototypes/connections/serve.c b/prototypes/connections/serve.c
index 1eb5ca5..0029164 100644
--- a/prototypes/connections/serve.c
+++ b/prototypes/connections/serve.c
@@ -501,6 +501,9 @@ static const char *resolve(const Host *host, Connection *c, char *query, Route *
     if (route->kind == ROUTE_NONE || (route->kind == ROUTE_OBJECT && !host->documents)) return "route";
     if (route->kind == ROUTE_OBJECT) return NULL;
     int found = route->kind == ROUTE_VIEW ? pure_module_view(m, c->key) : pure_module_input(m, c->key);
+    /* An output is the app's to send and a timer ticks by itself: neither is posted. */
+    if (found >= 0 && route->kind == ROUTE_INPUT &&
+        (m->input[found].kind == PURE_KIND_OUTPUT || m->input[found].kind == PURE_KIND_TIMER)) found = -1;
     if (found < 0) return route->kind == ROUTE_VIEW ? "no_view" : "no_input";
     *status = 400;
     route->target = (uint32_t)found;
diff --git a/prototypes/connections/test-module.c b/prototypes/connections/test-module.c
index ae82c69..bb59174 100644
--- a/prototypes/connections/test-module.c
+++ b/prototypes/connections/test-module.c
@@ -127,7 +127,7 @@ static size_t build(unsigned char *out, Spec spec) {
         *p++ = (unsigned char)(in->kind ? in->kind : 1); *p++ = in->keyed; *p++ = 0; *p++ = 0;
         bool timer = in->kind == PURE_KIND_TIMER;
         p = put32(p, timer ? (uint32_t)in->nfields : in->kind ? 0 : offset[n++]);
-        uint32_t first = in->kind == PURE_KIND_MESSAGE ? field_start : 0; /* only messages have fields */
+        uint32_t first = in->kind == PURE_KIND_MESSAGE || in->kind == PURE_KIND_OUTPUT ? field_start : 0; /* only messages and outputs have fields */
         *p++ = (unsigned char)first; *p++ = (unsigned char)(first >> 8);
         *p++ = (unsigned char)(timer ? 0 : in->nfields); *p++ = 0;
         for (size_t f = 0; !timer && f < in->nfields; ++f) field_offsets[nf++] = offset[n++];
@@ -2539,6 +2539,237 @@ static void message_fields(void) {
     CHECK(load(groups, &module_b) == PURE_INVALID && strstr(error, "field and group"));
 }
 
+/* ---- Connections between apps (docs/research/078-app-connections.md): a
+ * send is part of its event and is seen once saved; a receiver applies each
+ * numbered message once, counts a refused one as heard, and forgets what a
+ * failed save undid. */
+enum { SEND = PURE_OP_SEND };
+typedef struct { uint32_t output; uint64_t seq; unsigned char payload[64]; size_t size; } Sent;
+typedef struct { Sent items[16]; size_t count; } Outbox;
+
+static bool collect_sent(void *context, uint32_t output, uint64_t seq, PureSpan payload) {
+    Outbox *o = context;
+    CHECK(o->count < 16 && payload.size <= 64);
+    Sent *s = &o->items[o->count++];
+    *s = (Sent){output, seq, {0}, payload.size};
+    memcpy(s->payload, payload.data, payload.size);
+    return true;
+}
+
+static Outbox outbox_of(const PureApp *app) {
+    Outbox o = {.count = 0};
+    pure_app_outbox(app, collect_sent, &o);
+    return o;
+}
+
+/* buy/fail {item: text, count: u32}: the count at 0, then the item. */
+static PureResult buy(PureApp *app, uint32_t input, const char *item, uint32_t count) {
+    unsigned char payload[64];
+    size_t length = strlen(item);
+    memcpy(payload, &count, 4);
+    memcpy(payload + 4, item, length);
+    PureResult result = pure_app_message(app, input, NULL, -1, (PureSpan){payload, 4 + length});
+    pure_app_finish(app);
+    return result;
+}
+
+/* order {item: text, count: u32, total: number}: count, total, then item. */
+static PureResult deliver(PureApp *app, const char *source, uint64_t seq, const char *item, uint32_t count) {
+    unsigned char payload[64];
+    size_t length = strlen(item);
+    double total = 0;
+    memcpy(payload, &count, 4);
+    memcpy(payload + 4, &total, 8);
+    memcpy(payload + 12, item, length);
+    PureResult result = pure_app_deliver(app, 0, source, seq, (PureSpan){payload, 12 + length});
+    pure_app_finish(app);
+    return result;
+}
+
+static void links(void) {
+    static const S sender_nums[] = {{"n", 0}};
+    static const F order_fields[] = {{"item", PURE_FIELD_TEXT}, {"count", PURE_FIELD_U32}};
+    static const F ordered_fields[] = {{"item", PURE_FIELD_TEXT}, {"count", PURE_FIELD_U32}, {"total", PURE_FIELD_NUMBER}};
+    static const I sender_inputs[] = {{"buy", NULL, PURE_KIND_MESSAGE, false, order_fields, 2},
+                                      {"fail", NULL, PURE_KIND_MESSAGE, false, order_fields, 2},
+                                      {"ordered", NULL, PURE_KIND_OUTPUT, false, ordered_fields, 3}};
+    static const O sender_code[] = {{L, 0, 0}, {C, 0, 1}, {ADD, 0, 0}, {ST, 0, 0}, {FIELD, 0, 0}, {FIELD, 1, 0},
+                                    {L, 0, 0}, {SEND, 2, 0},                                           /* buy */
+                                    {FIELD, 0, 0}, {FIELD, 1, 0}, {C, 0, 0}, {SEND, 2, 0}, {C, 0, 0}, {REQ, 0, 0}}; /* fail */
+    static const H sender_handlers[] = {{1, 0, 8}, {3, 8, 6}};
+    Spec sender = {sender_nums, 1, sender_inputs, 3, NULL, 0, sender_handlers, 2, sender_code, LEN(sender_code),
+                   NULL, 0, NULL, 0, NULL, 0, NULL, 0, NULL, 0, NULL, 0};
+    static const S receiver_nums[] = {{"stock", 0}};
+    static const T receiver_texts[] = {{"last", 1, 0}};
+    static const char *const receiver_constants[] = {""};
+    static const I receiver_inputs[] = {{"order", NULL, PURE_KIND_MESSAGE, false, ordered_fields, 3}};
+    static const O receiver_code[] = {{FIELD, 1, 0}, {L, 0, 0}, {ADD, 0, 0}, {ST, 0, 0}, {FIELD, 0, 0}, {TSET, 1, 0},
+                                      {FIELD, 1, 0}, {C, 0, 100}, {LT, 0, 0}, {REQ, 0, 0}}; /* refuses 100 or more */
+    static const H receiver_handlers[] = {{1, 0, 10}};
+    Spec receiver = {receiver_nums, 1, receiver_inputs, 1, NULL, 0, receiver_handlers, 1, receiver_code,
+                     LEN(receiver_code), NULL, 0, receiver_texts, 1, receiver_constants, 1, NULL, 0, NULL, 0, NULL, 0};
+    if (load(sender, &module_a) != PURE_OK || load(receiver, &module_b) != PURE_OK) { fprintf(stderr, "%s\n", error); abort(); }
+    CHECK(module_a.input[2].prefix == 12);
+
+    /* In memory: a send is its event's; the receiver applies it once per source. */
+    PureApp a, b;
+    CHECK(pure_app_start(&a, &module_a, (PureData){0}) == PURE_OK && pure_app_start(&b, &module_b, (PureData){0}) == PURE_OK);
+    CHECK(buy(&a, 0, "pen", 2) == PURE_OK && buy(&a, 1, "ink", 3) == PURE_LIMIT);
+    Outbox o = outbox_of(&a);
+    CHECK(o.count == 1 && o.items[0].output == 2 && o.items[0].seq == 1 && o.items[0].size == 15 && a.values[0] == 1);
+    PureSpan sent = {o.items[0].payload, o.items[0].size};
+    CHECK(pure_app_deliver(&b, 0, "orders.ordered", 1, sent) == PURE_OK && !b.duplicate && b.values[0] == 2 &&
+          !strcmp(state_text(&b, 1), "pen"));
+    pure_app_finish(&b);
+    CHECK(pure_app_deliver(&b, 0, "orders.ordered", 1, sent) == PURE_OK && b.duplicate && b.values[0] == 2);
+    CHECK(deliver(&b, "other", 1, "cup", 5) == PURE_OK && !b.duplicate && b.values[0] == 7);
+    CHECK(deliver(&b, "orders.ordered", 3, "hat", 1) == PURE_OK && b.values[0] == 8); /* numbers may skip */
+    CHECK(deliver(&b, "orders.ordered", 2, "old", 1) == PURE_OK && b.duplicate && b.values[0] == 8);
+    /* Refused (count 100): delivered all the same, never applied again. */
+    CHECK(deliver(&b, "other", 2, "big", 100) == PURE_LIMIT && b.values[0] == 7 + 1);
+    CHECK(deliver(&b, "other", 2, "big", 1) == PURE_OK && b.duplicate && b.values[0] == 8);
+    /* Bad deliveries change nothing. */
+    CHECK(deliver(&b, "", 4, "x", 1) == PURE_INVALID && deliver(&b, "a b", 4, "x", 1) == PURE_INVALID);
+    CHECK(deliver(&b, "other", 0, "x", 1) == PURE_INVALID && pure_app_deliver(&a, 2, "s", 1, sent) == PURE_INVALID);
+    CHECK(pure_app_message(&a, 2, NULL, -1, sent) == PURE_INVALID);
+    /* Trimmed, a send is gone at once for the host and from the store at the next save. */
+    pure_app_trim(&a, 2, 1);
+    CHECK(outbox_of(&a).count == 0 && pure_app_save(&a) == PURE_OK && a.sent_count == 0);
+    CHECK(buy(&a, 0, "cap", 1) == PURE_OK && outbox_of(&a).items[0].seq == 2);
+    pure_app_stop(&a);
+    pure_app_stop(&b);
+
+    /* Stored: sends and heard numbers survive restarts; a deferred burst's sends
+     * appear with its save, and a failed save takes them and their numbers back. */
+    char path_a[64], path_b[64];
+    snprintf(path_a, sizeof(path_a), "%s", new_store());
+    snprintf(path_b, sizeof(path_b), "%s", new_store());
+    CHECK(pure_app_start(&a, &module_a, pure_tree_data(path_a)) == PURE_OK);
+    CHECK(pure_app_start(&b, &module_b, pure_tree_data(path_b)) == PURE_OK);
+    CHECK(buy(&a, 0, "pen", 2) == PURE_OK && buy(&a, 0, "ink", 3) == PURE_OK);
+    pure_app_stop(&a);
+    CHECK(pure_app_start(&a, &module_a, pure_tree_data(path_a)) == PURE_OK);
+    o = outbox_of(&a);
+    CHECK(o.count == 2 && o.items[0].seq == 1 && o.items[1].seq == 2 && a.values[0] == 2);
+    CHECK(pure_app_defer(&a, true));
+    for (int i = 0; i < 3; ++i) CHECK(buy(&a, 0, "box", 1) == PURE_OK);
+    CHECK(outbox_of(&a).count == 2 && pure_app_save(&a) == PURE_OK);
+    o = outbox_of(&a);
+    CHECK(o.count == 5 && o.items[4].seq == 5);
+    CHECK(buy(&a, 0, "lost", 1) == PURE_OK);
+    pure_app_trim(&a, 2, 3);
+    fail_stage = PURE_AFTER_HEADER;
+    CHECK(pure_app_save(&a) == PURE_STORE_ERROR);
+    fail_stage = -1;
+    o = outbox_of(&a);
+    CHECK(o.count == 5 && o.items[0].seq == 1 && a.values[0] == 5);         /* the trim and the send undone */
+    CHECK(buy(&a, 0, "next", 1) == PURE_OK && pure_app_save(&a) == PURE_OK);
+    o = outbox_of(&a);
+    CHECK(o.count == 6 && o.items[5].seq == 6);                              /* numbered on from 5 */
+    pure_app_trim(&a, 2, 4);
+    CHECK(pure_app_save(&a) == PURE_OK);
+    pure_app_stop(&a);
+    CHECK(pure_app_start(&a, &module_a, pure_tree_data(path_a)) == PURE_OK);
+    o = outbox_of(&a);
+    CHECK(o.count == 2 && o.items[0].seq == 5 && o.items[1].seq == 6);
+    /* The receiver: heard numbers stay heard after a restart; a refused one too. */
+    CHECK(deliver(&b, "orders.ordered", 1, "pen", 2) == PURE_OK && deliver(&b, "x", 1, "big", 100) == PURE_LIMIT);
+    pure_app_stop(&b);
+    CHECK(pure_app_start(&b, &module_b, pure_tree_data(path_b)) == PURE_OK && b.values[0] == 2);
+    CHECK(deliver(&b, "orders.ordered", 1, "pen", 2) == PURE_OK && b.duplicate && b.values[0] == 2);
+    CHECK(deliver(&b, "x", 1, "big", 1) == PURE_OK && b.duplicate);
+    /* A save that fails forgets what it heard. */
+    CHECK(pure_app_defer(&b, true) && deliver(&b, "orders.ordered", 2, "ink", 3) == PURE_OK && b.values[0] == 5);
+    fail_stage = PURE_AFTER_HEADER;
+    CHECK(pure_app_save(&b) == PURE_STORE_ERROR && b.values[0] == 2);
+    fail_stage = -1;
+    CHECK(deliver(&b, "orders.ordered", 2, "ink", 3) == PURE_OK && !b.duplicate && b.values[0] == 5);
+    CHECK(pure_app_save(&b) == PURE_OK);
+    pure_app_stop(&b);
+
+    /* Replacing: a module without the output is refused while sends are left
+     * on it; once they are trimmed it replaces, and the output's number is
+     * kept for a module that has it again. */
+    static const I plain_inputs[] = {{"buy", NULL, PURE_KIND_MESSAGE, false, order_fields, 2}};
+    static const O plain_code[] = {{L, 0, 0}, {C, 0, 1}, {ADD, 0, 0}, {ST, 0, 0}};
+    static const H plain_handlers[] = {{1, 0, 4}};
+    Spec plain = {sender_nums, 1, plain_inputs, 1, NULL, 0, plain_handlers, 1, plain_code, LEN(plain_code),
+                  NULL, 0, NULL, 0, NULL, 0, NULL, 0, NULL, 0, NULL, 0};
+    CHECK(load(plain, &module_b) == PURE_OK);
+    CHECK(pure_app_replace(&a, &module_b) == PURE_CONFLICT && a.module == &module_a);
+    pure_app_trim(&a, 2, 6);
+    CHECK(pure_app_save(&a) == PURE_OK && pure_app_replace(&a, &module_b) == PURE_OK);
+    CHECK(buy(&a, 0, "plain", 1) == PURE_OK && pure_app_save(&a) == PURE_OK);
+    pure_app_stop(&a);
+    CHECK(pure_app_start(&a, &module_b, pure_tree_data(path_a)) == PURE_OK); /* the number kept in the store */
+    CHECK(pure_app_replace(&a, &module_a) == PURE_OK && buy(&a, 0, "back", 1) == PURE_OK && pure_app_save(&a) == PURE_OK);
+    o = outbox_of(&a);
+    CHECK(o.count == 1 && o.items[0].seq == 7);
+    /* A stored send whose output the module lacks refuses the start. */
+    pure_app_stop(&a);
+    CHECK(pure_app_start(&a, &module_b, pure_tree_data(path_a)) == PURE_CONFLICT);
+    remove_store(path_a);
+    remove_store(path_b);
+
+    /* A process that dies at each commit stage leaves an event and its send
+     * together or neither, and a delivery and its effect together or neither. */
+    snprintf(path_a, sizeof(path_a), "%s", new_store());
+    snprintf(path_b, sizeof(path_b), "%s", new_store());
+    CHECK(load(receiver, &module_b) == PURE_OK);
+    uint64_t heard = 0;
+    size_t kept[2] = {0, 0}; /* sends and deliveries a death left in place */
+    for (int round = 0; round < 2; ++round)
+    for (int stage = PURE_AFTER_HEADER; stage <= PURE_BEFORE_DIR_SYNC; ++stage) {
+        pid_t child = fork();
+        CHECK(child >= 0);
+        if (!child) {
+            PureApp doomed, taker;
+            if (pure_app_start(&doomed, &module_a, pure_tree_data(path_a)) != PURE_OK ||
+                pure_app_start(&taker, &module_b, pure_tree_data(path_b)) != PURE_OK) _exit(1);
+            crash = 1;
+            fail_stage = stage;
+            if (round) buy(&doomed, 0, "crash", 1);
+            else deliver(&taker, "orders.ordered", heard + 1, "crash", 1);
+            _exit(0);
+        }
+        int status;
+        CHECK(waitpid(child, &status, 0) == child && WIFEXITED(status) && WEXITSTATUS(status) == 77);
+        CHECK(pure_app_start(&a, &module_a, pure_tree_data(path_a)) == PURE_OK);
+        CHECK(pure_app_start(&b, &module_b, pure_tree_data(path_b)) == PURE_OK);
+        o = outbox_of(&a);
+        CHECK(o.count == (size_t)a.values[0] && (!o.count || o.items[o.count - 1].seq == o.count));
+        heard = b.heard_count ? b.heard[0].seq : 0;
+        CHECK(b.values[0] == (double)heard);
+        kept[0] = (size_t)heard;
+        kept[1] = o.count;
+        pure_app_stop(&a);
+        pure_app_stop(&b);
+    }
+    CHECK(kept[0] == 1 && kept[1] == 1); /* only a death after the head was written keeps it */
+    remove_store(path_a);
+    remove_store(path_b);
+
+    /* What an output and SEND may not be. */
+    const struct { const O *code; size_t count; const char *reason; } refused[] = {
+        {(const O[]){{FIELD, 1, 0}, {FIELD, 0, 0}, {L, 0, 0}, {SEND, 2, 0}}, 4, "operand type"},
+        {(const O[]){{FIELD, 0, 0}, {FIELD, 1, 0}, {L, 0, 0}, {SEND, 1, 0}}, 4, "no such state"},
+        {(const O[]){{FIELD, 0, 0}, {FIELD, 1, 0}, {SEND, 2, 0}}, 3, "stack underflow"},
+    };
+    for (size_t i = 0; i < LEN(refused); ++i) {
+        Spec bad = sender;
+        H handler = {1, 0, (uint32_t)refused[i].count};
+        bad.o = refused[i].code; bad.no = refused[i].count; bad.h = &handler; bad.nh = 1;
+        if (load(bad, &module_b) != PURE_INVALID || !strstr(error, refused[i].reason)) {
+            fprintf(stderr, "links case %zu: expected '%s', got '%s'\n", i, refused[i].reason, error);
+            abort();
+        }
+    }
+    H on_output[] = {{1, 0, 8}, {5, 8, 6}};
+    Spec bad = sender;
+    bad.h = on_output;
+    CHECK(load(bad, &module_b) == PURE_INVALID && strstr(error, "handler event"));
+}
+
 /* A timer's tick sees the time the host set; floor and a floored modulo
  * take it apart. now is for handlers only. */
 enum { NOW = PURE_OP_NOW, FLOOR = PURE_OP_FLOOR, MOD = PURE_OP_MOD };
@@ -2608,6 +2839,7 @@ int main(void) {
     keyed();
     keyed_mutations();
     snapshots();
+    links();
     fold_mutations();
     pointing();
     list_mutations();
@@ -2616,6 +2848,6 @@ int main(void) {
     stored_states();
     deferred();
     printf("{\"sizeof_module\":%zu,\"sizeof_app\":%zu}\n", sizeof(PureModule), sizeof(PureApp));
-    puts("{\"module_tests\":\"verified-load-atomic-events-faults-signed-zero-if-else-replace-by-name-rejections-mutations-messages-documents-typed-verification-text-cursors-utf8-views-rollback-replace-lists-indices-undo-migration-joins-tables-columns-pick-slide-stored-equivalence-crashes-frames-messages-migration-deferred-bursts-pointing-marks-folds-filters-keyed-lists-text-order-message-fields-timers-now-snapshots\"}");
+    puts("{\"module_tests\":\"verified-load-atomic-events-faults-signed-zero-if-else-replace-by-name-rejections-mutations-messages-documents-typed-verification-text-cursors-utf8-views-rollback-replace-lists-indices-undo-migration-joins-tables-columns-pick-slide-stored-equivalence-crashes-frames-messages-migration-deferred-bursts-pointing-marks-folds-filters-keyed-lists-text-order-message-fields-timers-now-snapshots-links\"}");
     return 0;
 }