Let links know nothing of the core: the host names inputs and shapes

A sender takes its messages' shape, and a listener asks its host which
input a stream goes to. The core gives the shape (pure_module_shape);
link/ now builds without the core's headers.

7 files changed, 137 insertions, 105 deletions

diff --git a/Makefile b/Makefile
index 427a08a..10a5f0e 100644
--- a/Makefile
+++ b/Makefile
@@ -38,6 +38,11 @@ $(BUILD)/obj/%.o: %.c $(HEADERS)
 	@mkdir -p $(dir $@)
 	$(CC) $(CPPFLAGS) $(CFLAGS) $(INCLUDES) -c $< -o $@
 
+# The links know nothing of the core: they build without its headers.
+$(BUILD)/obj/link/%.o: link/%.c $(wildcard link/*.h)
+	@mkdir -p $(dir $@)
+	$(CC) $(CPPFLAGS) $(CFLAGS) -c $< -o $@
+
 $(BUILD)/libpure.a: $(OBJECTS)
 	rm -f $@
 	$(AR) rcs $@ $^
diff --git a/core/module.c b/core/module.c
index cb4fe1a..8ac3b48 100644
--- a/core/module.c
+++ b/core/module.c
@@ -1094,6 +1094,25 @@ static int find(const char (*names)[PURE_MODULE_NAME], unsigned count, const cha
 
 int pure_module_state(const PureModule *m, const char *name) { return find(m->state_names, m->states, name); }
 int pure_module_input(const PureModule *m, const char *name) { return find(m->input_names, m->inputs, name); }
+static uint32_t fnv(uint32_t h, const void *bytes, size_t size) {
+    const unsigned char *p = bytes;
+    for (size_t i = 0; i < size; ++i) h = (h ^ p[i]) * 16777619u;
+    return h;
+}
+
+uint32_t pure_module_shape(const PureModule *m, uint32_t input) {
+    const PureInputShape *s = &m->input[input];
+    uint32_t h = fnv(2166136261u, &s->prefix, sizeof(s->prefix));
+    for (uint32_t f = 0; f < s->field_count; ++f) {
+        const PureField *field = &m->field[s->field_start + f];
+        unsigned char fixed[7] = {field->type, (unsigned char)field->rows, (unsigned char)(field->rows >> 8),
+                                  (unsigned char)field->offset, (unsigned char)(field->offset >> 8),
+                                  (unsigned char)(field->offset >> 16), (unsigned char)(field->offset >> 24)};
+        h = fnv(fnv(h, field->name, strlen(field->name) + 1), fixed, sizeof(fixed));
+    }
+    return h;
+}
+
 int pure_module_view(const PureModule *m, const char *name) {
     for (uint32_t i = 0; i < m->views; ++i)
         if (!m->view_table[i] && !strcmp(m->view_names[i], name)) return (int)i;
diff --git a/core/module.h b/core/module.h
index 3431c87..0b45ba2 100644
--- a/core/module.h
+++ b/core/module.h
@@ -211,6 +211,10 @@ int pure_module_state(const PureModule *, const char *name);
 int pure_module_input(const PureModule *, const char *name);
 int pure_module_view(const PureModule *, const char *name);
 int pure_module_column(const PureModule *, uint32_t table, const char *name);
+/* What a message of an input looks like, as 32 bits: its fields' names,
+ * types and places (FNV-1a). A sender's output and the input it sends to
+ * across nodes must agree on it (link/link.h). */
+uint32_t pure_module_shape(const PureModule *, uint32_t input);
 /* The index state a number view shows as its position (a view that is just
  * `at.position`), or -1: what a host moves with pure_app_point. */
 int pure_module_position(const PureModule *, uint32_t view);
diff --git a/link/link.c b/link/link.c
index 9c6794b..103dcf5 100644
--- a/link/link.c
+++ b/link/link.c
@@ -89,19 +89,6 @@ static uint32_t fnv(uint32_t h, const void *bytes, size_t size) {
 
 static uint32_t key_id(const unsigned char key[PURE_SEAL_KEY]) { return fnv(2166136261u, key, PURE_SEAL_KEY); }
 
-/* What a message of input looks like: its fields' names, types and places. */
-static uint32_t shape_of(const PureModule *m, uint32_t input) {
-    const PureInputShape *s = &m->input[input];
-    uint32_t h = fnv(2166136261u, &s->prefix, sizeof(s->prefix));
-    for (uint32_t f = 0; f < s->field_count; ++f) {
-        const PureField *field = &m->field[s->field_start + f];
-        unsigned char fixed[7] = {field->type, (unsigned char)field->rows, (unsigned char)(field->rows >> 8)};
-        put32(fixed + 3, field->offset);
-        h = fnv(fnv(h, field->name, strlen(field->name) + 1), fixed, sizeof(fixed));
-    }
-    return h;
-}
-
 static uint64_t random64(void) {
     uint64_t value = 0;
     while (getrandom(&value, sizeof(value), 0) != sizeof(value) || !value) {}
@@ -164,10 +151,11 @@ static size_t put_names(unsigned char *p, const char *source, const char *input)
 }
 
 /* A source and an input's names from p (size bytes); how many it takes, or 0. */
-static size_t get_names(const unsigned char *p, size_t size, char source[PURE_SOURCE + 1], char input[PURE_MODULE_NAME]) {
-    if (size < 1 || !p[0] || p[0] > PURE_SOURCE || size < 2u + p[0]) return 0;
+static size_t get_names(const unsigned char *p, size_t size, char source[PURE_LINK_NAME + 1],
+                        char input[PURE_LINK_NAME + 1]) {
+    if (size < 1 || !p[0] || p[0] > PURE_LINK_NAME || size < 2u + p[0]) return 0;
     size_t a = p[0], b = p[1 + a];
-    if (!b || b >= PURE_MODULE_NAME || size < 2 + a + b) return 0;
+    if (!b || b > PURE_LINK_NAME || size < 2 + a + b) return 0;
     memcpy(source, p + 1, a);
     source[a] = 0;
     memcpy(input, p + 2 + a, b);
@@ -249,9 +237,8 @@ typedef struct {
 } Flight;
 
 struct PureLink {
-    const PureModule *module;
-    uint32_t output, shape, own_id, peer_id;
-    char host[256], port[8], input[PURE_MODULE_NAME], source[PURE_SOURCE + 1];
+    uint32_t shape, own_id, peer_id;
+    char host[256], port[8], input[PURE_LINK_NAME + 1], source[PURE_LINK_NAME + 1];
     const PureSealKey *own;
     unsigned char own_public[PURE_SEAL_KEY], peer[PURE_SEAL_KEY], secret[32];
     bool peered;
@@ -293,23 +280,17 @@ struct PureLink {
     PureLinkCounts counts;
 };
 
-PureLink *pure_link_open(const PureModule *m, const char *output, const char *url, const char *source,
-                         const PureSealKey *own, const unsigned char own_public[PURE_SEAL_KEY], char *error,
-                         size_t size) {
-    int index = pure_module_input(m, output);
-    if (index < 0 || m->input[index].kind != PURE_KIND_OUTPUT) {
-        snprintf(error, size, "no output '%s'", output);
-        return NULL;
-    }
+PureLink *pure_link_open(const char *url, const char *source, uint32_t shape, const PureSealKey *own,
+                         const unsigned char own_public[PURE_SEAL_KEY], char *error, size_t size) {
     if (!own) {
         snprintf(error, size, "a pure:// link needs the node's --key");
         return NULL;
     }
     size_t length = strlen(source);
     const char *host = url + 7, *slash = strchr(host, '/'), *colon = slash ? memchr(host, ':', (size_t)(slash - host)) : NULL;
-    if (!length || length > PURE_SOURCE || strncmp(url, "pure://", 7) || !slash || !colon || colon == host ||
+    if (!length || length > PURE_LINK_NAME || strncmp(url, "pure://", 7) || !slash || !colon || colon == host ||
         (size_t)(colon - host) >= 256 || slash - colon - 1 < 1 || slash - colon - 1 > 5 || !slash[1] ||
-        strlen(slash + 1) >= PURE_MODULE_NAME) {
+        strlen(slash + 1) > PURE_LINK_NAME) {
         snprintf(error, size, "a link is pure://host:port/INPUT");
         return NULL;
     }
@@ -319,9 +300,7 @@ PureLink *pure_link_open(const PureModule *m, const char *output, const char *ur
         snprintf(error, size, "out of memory");
         return NULL;
     }
-    l->module = m;
-    l->output = (uint32_t)index;
-    l->shape = shape_of(m, (uint32_t)index);
+    l->shape = shape;
     memcpy(l->host, host, (size_t)(colon - host));
     memcpy(l->port, colon + 1, (size_t)(slash - colon - 1));
     strcpy(l->input, slash + 1);
@@ -339,8 +318,6 @@ void pure_link_peer(PureLink *l, const unsigned char peer[PURE_SEAL_KEY]) {
     l->peered = true;
 }
 
-uint32_t pure_link_output(const PureLink *l) { return l->output; }
-
 static Record *record_at(PureLink *l, size_t k) { return &l->records[(l->head + k) % l->capacity]; }
 
 /* The queued record numbered seq, or NULL (stored already, or not queued). */
@@ -397,17 +374,17 @@ static bool ring_grow(PureLink *l, size_t size) {
     return true;
 }
 
-bool pure_link_queue(PureLink *l, uint64_t seq, PureSpan payload) {
-    Record r = {.seq = seq, .size = (uint32_t)payload.size};
-    r.pieces = payload.size ? (uint32_t)((payload.size + PIECE - 1) / PIECE) : 1;
-    if (payload.size > RECORD_MOST || (r.pieces > 1 && !(r.more = calloc(r.pieces, sizeof(Piece))))) return false;
+bool pure_link_queue(PureLink *l, uint64_t seq, const unsigned char *bytes, size_t size) {
+    Record r = {.seq = seq, .size = (uint32_t)size};
+    r.pieces = size ? (uint32_t)((size + PIECE - 1) / PIECE) : 1;
+    if (size > RECORD_MOST || (r.pieces > 1 && !(r.more = calloc(r.pieces, sizeof(Piece))))) return false;
     pthread_mutex_lock(&l->lock);
     bool ok = seq > l->queued && (!l->count || seq == l->queued + 1);
-    size_t at = ok ? ring_place(l, payload.size) : SIZE_MAX;
-    if (ok && at == SIZE_MAX) ok = ring_grow(l, payload.size) && (at = ring_place(l, payload.size)) != SIZE_MAX;
+    size_t at = ok ? ring_place(l, size) : SIZE_MAX;
+    if (ok && at == SIZE_MAX) ok = ring_grow(l, size) && (at = ring_place(l, size)) != SIZE_MAX;
     if (ok) {
         r.at = at;
-        if (payload.size) memcpy(l->ring + at, payload.data, payload.size);
+        if (size) memcpy(l->ring + at, bytes, size);
     }
     if (ok && l->count == l->capacity) {
         size_t capacity = l->capacity ? 2 * l->capacity : 64;
@@ -568,7 +545,7 @@ static void on_ack(PureLink *l, const unsigned char *p, size_t size, uint64_t no
 }
 
 static void on_stored(PureLink *l, const unsigned char *p, size_t size, uint64_t now) {
-    char source[PURE_SOURCE + 1], input[PURE_MODULE_NAME];
+    char source[PURE_LINK_NAME + 1], input[PURE_LINK_NAME + 1];
     size_t names = get_names(p, size, source, input);
     if (!names || size < names + 13 || strcmp(source, l->source) || strcmp(input, l->input)) return;
     uint64_t heard = get64(p + names);
@@ -624,7 +601,7 @@ static void take_answer(PureLink *l, unsigned char *packet, ssize_t n, uint64_t
             on_ack(l, plain + at, size - at, now);
             at += 13 + 16u * plain[at + 12];
         } else if (type == F_STORED) {
-            char source[PURE_SOURCE + 1], input[PURE_MODULE_NAME];
+            char source[PURE_LINK_NAME + 1], input[PURE_LINK_NAME + 1];
             size_t names = get_names(plain + at, size - at, source, input);
             if (!names || at + names + 13 > size) break;
             on_stored(l, plain + at, size - at, now);
@@ -988,8 +965,8 @@ typedef struct {
 } Held;
 typedef struct Stream {
     Active *active;
-    char source[PURE_SOURCE + 1];
-    uint32_t input;
+    char source[PURE_LINK_NAME + 1], input_name[PURE_LINK_NAME + 1];
+    uint32_t input, shape; /* the host's number for the input; the messages' shape */
     uint64_t heard;              /* the last number heard (stored), once known */
     bool known, out, stored_due, gap, ready;
     /* The line: once the last heard is known, whole records from the one
@@ -1030,7 +1007,7 @@ static const uint64_t IDLE_FIRST = 30000 * MS, SWEEP_EVERY = 250 * MS, TRIM_WAIT
 enum { TRIM_AFTER = 1024 };
 
 struct PureLinkListener {
-    const PureModule *module;
+    PureLinkHost host;
     const PureSealKey *own;
     unsigned char own_public[PURE_SEAL_KEY];
     uint32_t own_id;
@@ -1062,31 +1039,25 @@ struct PureLinkListener {
     size_t ready_count, ready_capacity;
     Run **due; /* runs to answer */
     size_t due_count, due_capacity;
-    bool (*deliver)(void *, const PureLinkDelivery *);
-    void *context;
     unsigned char *batch;
     size_t batch_capacity;
     PureLinkCounts counts;
 };
 
-PureLinkListener *pure_link_listen(const PureModule *m, unsigned port, const PureSealKey *own,
-                                   const unsigned char own_public[PURE_SEAL_KEY],
-                                   bool (*deliver)(void *, const PureLinkDelivery *), void *context, char *error,
-                                   size_t size) {
+PureLinkListener *pure_link_listen(unsigned port, const PureSealKey *own, const unsigned char own_public[PURE_SEAL_KEY],
+                                   const PureLinkHost *host, char *error, size_t size) {
     PureLinkListener *s = own ? calloc(1, sizeof(*s)) : NULL;
     if (!s || pthread_mutex_init(&s->lock, NULL)) {
         free(s);
         snprintf(error, size, own ? "out of memory" : "--link-port needs the node's --key");
         return NULL;
     }
-    s->module = m;
+    s->host = *host;
     s->own = own;
     memcpy(s->own_public, own_public, PURE_SEAL_KEY);
     s->own_id = key_id(own_public);
     s->port = port;
     s->idle = IDLE_FIRST;
-    s->deliver = deliver;
-    s->context = context;
     s->fd = s->wake = -1;
     return s;
 }
@@ -1128,8 +1099,8 @@ bool pure_link_allow(PureLinkListener *s, const char *source, const unsigned cha
                      size_t size) {
     size_t length = strlen(source), p = SIZE_MAX;
     uint32_t id = key_id(key);
-    if (!length || length > PURE_SOURCE) {
-        snprintf(error, size, "a source's name is 1 to %d bytes", PURE_SOURCE);
+    if (!length || length > PURE_LINK_NAME) {
+        snprintf(error, size, "a source's name is 1 to %d bytes", PURE_LINK_NAME);
         return false;
     }
     for (size_t slot = s->index_size ? id & (s->index_size - 1) : 0; s->index_size && s->index[slot];
@@ -1431,18 +1402,15 @@ static bool allowed(const PureLinkListener *s, const Peer *peer, const char *sou
 }
 
 /* The stream of a peer's source to input, made the first time: NULL when the
- * source is not allowed from the peer, or input takes no such messages. */
+ * source is not allowed from the peer, or the host has no input that takes
+ * such messages. */
 static Stream *stream_of(PureLinkListener *s, Active *a, const char *source, const char *input, uint32_t shape) {
-    int index = pure_module_input(s->module, input);
-    if (index < 0) return NULL;
     for (unsigned k = 0; k < a->stream_count; ++k) {
         Stream *st = a->streams[k];
-        if (st->input == (uint32_t)index && !strcmp(st->source, source)) return st;
+        if (!strcmp(st->input_name, input) && !strcmp(st->source, source)) return st->shape == shape ? st : NULL;
     }
-    const PureInputShape *in = &s->module->input[index];
-    if (!allowed(s, &s->peers[a->peer], source) || in->kind != PURE_KIND_MESSAGE || in->keyed ||
-        shape_of(s->module, (uint32_t)index) != shape)
-        return NULL;
+    int index = allowed(s, &s->peers[a->peer], source) ? s->host.input(s->host.context, input, shape) : -1;
+    if (index < 0) return NULL;
     size_t capacity = a->stream_capacity;
     Stream **streams = room(a->streams, &capacity, a->stream_count, sizeof(*streams), 2), *st;
     if (!streams) return NULL;
@@ -1451,7 +1419,9 @@ static Stream *stream_of(PureLinkListener *s, Active *a, const char *source, con
     if (!(st = calloc(1, sizeof(*st)))) return NULL;
     st->active = a;
     st->input = (uint32_t)index;
+    st->shape = shape;
     snprintf(st->source, sizeof(st->source), "%s", source);
+    snprintf(st->input_name, sizeof(st->input_name), "%s", input);
     a->streams[a->stream_count++] = st;
     return st;
 }
@@ -1524,7 +1494,7 @@ static bool hand_over(PureLinkListener *s, Stream *st) {
         size_t bytes = 0;
         for (; n < st->line_count && n < BATCH; ++n) bytes += 4 + (size_t)get32(st->line + st->line_head + bytes);
         PureLinkDelivery d = {st->input, st->source, st->line_first, n, st->line + st->line_head, bytes, st};
-        if (!s->deliver(s->context, &d)) return false;
+        if (!s->host.deliver(s->host.context, &d)) return false;
         st->out = true;
         ++s->counts.deliveries;
         return true;
@@ -1548,7 +1518,7 @@ static bool hand_over(PureLinkListener *s, Stream *st) {
         at += 4 + held_at(st, k)->size;
     }
     PureLinkDelivery d = {st->input, st->source, held_at(st, 0)->seq, (unsigned)n, s->batch, bytes, st};
-    if (!s->deliver(s->context, &d)) return false;
+    if (!s->host.deliver(s->host.context, &d)) return false;
     st->out = true;
     ++s->counts.deliveries;
     return true;
@@ -1594,7 +1564,7 @@ static bool take_from(PureLinkListener *s, Peer *peer, unsigned char *packet, si
     for (size_t at = 0; at < size;) {
         unsigned type = plain[at++];
         if (type == F_STREAM) {
-            char source[PURE_SOURCE + 1], input[PURE_MODULE_NAME];
+            char source[PURE_LINK_NAME + 1], input[PURE_LINK_NAME + 1];
             size_t names = get_names(plain + at, size - at, source, input);
             if (!names || at + names + 4 > size) break;
             st = stream_of(s, a, source, input, get32(plain + at + names));
@@ -1687,9 +1657,9 @@ static void answer(PureLinkListener *s, uint64_t now) {
         for (unsigned j = 0; j < a->stream_count; ++j) {
             Stream *st = a->streams[j];
             if (!(st->stored_due || r->ack_due) || st->run != r || st->run_id != r->run || !st->known) continue;
-            if (at + 2 + PURE_SOURCE + PURE_MODULE_NAME + 14 > BODY) break;
+            if (at + strlen(st->source) + strlen(st->input_name) + 16 > BODY) break; /* type, names, 13 */
             body[at++] = F_STORED;
-            at += put_names(body + at, st->source, s->module->input_names[st->input]);
+            at += put_names(body + at, st->source, st->input_name);
             put64(body + at, st->heard);
             body[at + 8] = st->gap ? STORED_GAP : 0;
             size_t room = st->held_bytes + (st->line_size - st->line_head) < HELD_MOST ? CREDIT : 1;
diff --git a/link/link.h b/link/link.h
index 83cc645..1ee9473 100644
--- a/link/link.h
+++ b/link/link.h
@@ -1,6 +1,5 @@
 #ifndef PURE_LINK_H
 #define PURE_LINK_H
-#include "module.h"
 #include "seal.h"
 
 /* Links without connections (docs/design/011-pure-transport.md): one node's
@@ -47,19 +46,21 @@
  * Losses are detected as RFC 9002 says; congestion control is its NewReno
  * (RFC 9002 §7). */
 
+/* A source's or an input's name is 1 to PURE_LINK_NAME bytes. */
+enum { PURE_LINK_NAME = 64 };
+
 /* The sending side: one output's saved sends to url, pure://host:port/INPUT,
- * as source (a Pure receiver's name for it), from own (this node's key,
- * whose public half is own_public) to the receiver's public key, given by
- * pure_link_peer before pure_link_start. The owner queues and trims as it
- * does a wire's (serve/deliver.h): a send counts as delivered once the
- * receiver says it saved it. */
+ * as source (a Pure receiver's name for it), their messages of shape (the
+ * receiver's input must have the same: core/module.h's pure_module_shape),
+ * from own (this node's key, whose public half is own_public) to the
+ * receiver's public key, given by pure_link_peer before pure_link_start.
+ * The owner queues and trims as it does a wire's (serve/deliver.h): a send
+ * counts as delivered once the receiver says it saved it. */
 typedef struct PureLink PureLink;
-PureLink *pure_link_open(const PureModule *m, const char *output, const char *url, const char *source,
-                         const PureSealKey *own, const unsigned char own_public[PURE_SEAL_KEY], char *error,
-                         size_t size);
+PureLink *pure_link_open(const char *url, const char *source, uint32_t shape, const PureSealKey *own,
+                         const unsigned char own_public[PURE_SEAL_KEY], char *error, size_t size);
 void pure_link_peer(PureLink *, const unsigned char peer[PURE_SEAL_KEY]);
-uint32_t pure_link_output(const PureLink *);
-bool pure_link_queue(PureLink *, uint64_t seq, PureSpan payload);
+bool pure_link_queue(PureLink *, uint64_t seq, const unsigned char *bytes, size_t size);
 uint64_t pure_link_queued(PureLink *);
 /* The last number the receiver saved; any thread. */
 uint64_t pure_link_delivered(PureLink *);
@@ -69,12 +70,14 @@ void pure_link_close(PureLink *);
 
 /* The receiving side: packets on a UDP port, from the senders allowed (a
  * source name and the key it seals with, a few bytes each however many
- * there are: see pure_link_idle). Whole records of a stream, in
- * order, go to deliver (on the listener's thread) as one delivery; the host
- * applies them from the one after the source's last heard number, stops at
- * a gap, saves, and then calls pure_link_done (any thread) with the last
- * number heard. Records wait here until then, and only the saved ones are
- * acknowledged as stored. */
+ * there are: see pure_link_idle). A stream (a source to an input) goes to
+ * the host's input its name and shape find (its number, or -1: the stream
+ * is refused). Whole records of a stream, in order, go to deliver (both on
+ * the listener's thread) as one delivery; the host applies them from the
+ * one after the source's last heard number, stops at a gap, saves, and then
+ * calls pure_link_done (any thread) with the last number heard. Records
+ * wait here until then, and only the saved ones are acknowledged as
+ * stored. */
 typedef struct {
     uint32_t input;
     const char *source;
@@ -84,11 +87,14 @@ typedef struct {
     size_t size;
     void *token; /* for pure_link_done */
 } PureLinkDelivery;
+typedef struct {
+    int (*input)(void *context, const char *name, uint32_t shape);
+    bool (*deliver)(void *context, const PureLinkDelivery *);
+    void *context;
+} PureLinkHost;
 typedef struct PureLinkListener PureLinkListener;
-PureLinkListener *pure_link_listen(const PureModule *m, unsigned port, const PureSealKey *own,
-                                   const unsigned char own_public[PURE_SEAL_KEY],
-                                   bool (*deliver)(void *, const PureLinkDelivery *), void *context, char *error,
-                                   size_t size);
+PureLinkListener *pure_link_listen(unsigned port, const PureSealKey *own, const unsigned char own_public[PURE_SEAL_KEY],
+                                   const PureLinkHost *host, char *error, size_t size);
 bool pure_link_allow(PureLinkListener *, const char *source, const unsigned char key[PURE_SEAL_KEY], char *error,
                      size_t size);
 /* A sender quiet for ms (30 s at first) with nothing of it waiting lets its
diff --git a/research/experiments/019-idle-links/transport.c b/research/experiments/019-idle-links/transport.c
index 9972cd3..c4c9de1 100644
--- a/research/experiments/019-idle-links/transport.c
+++ b/research/experiments/019-idle-links/transport.c
@@ -50,6 +50,13 @@ static bool deliver(void *context, const PureLinkDelivery *d) {
 
 static void notify(void *context) { (void)context; }
 
+/* The receiver's input a stream goes to: "order", of the sender's shape. */
+static int input_of(void *context, const char *name, uint32_t shape) {
+    (void)context;
+    int index = pure_module_input(&receiver, name);
+    return index >= 0 && pure_module_shape(&receiver, (uint32_t)index) == shape ? index : -1;
+}
+
 static uint64_t now_ns(void) {
     struct timespec t;
     clock_gettime(CLOCK_MONOTONIC, &t);
@@ -131,23 +138,25 @@ int main(int argc, char **argv) {
         return 1;
     unsigned port = ntohs(any.sin_port);
     close(probe);
-    if (!(listener = pure_link_listen(&receiver, port, b, public_b, deliver, NULL, error, sizeof(error))) ||
+    if (!(listener = pure_link_listen(port, b, public_b, &(PureLinkHost){input_of, deliver, NULL}, error, sizeof(error))) ||
         !pure_link_allow(listener, "bench.ordered", public_a, error, sizeof(error)) ||
         !pure_link_serve(listener, error, sizeof(error)))
         return fprintf(stderr, "transport: %s\n", error), 1;
     char url[128];
     snprintf(url, sizeof(url), "pure://127.0.0.1:%u/order", port);
-    PureLink *link = pure_link_open(&sender, "ordered", url, "bench.ordered", a, public_a, error, sizeof(error));
-    if (!link) return fprintf(stderr, "transport: %s\n", error), 1;
-    pure_link_peer(link, public_b);
     int output = pure_module_input(&sender, "ordered");
+    PureLink *link = output < 0 ? NULL
+                                : pure_link_open(url, "bench.ordered", pure_module_shape(&sender, (uint32_t)output), a,
+                                                 public_a, error, sizeof(error));
+    if (!link) return fprintf(stderr, "transport: %s\n", output < 0 ? "no output 'ordered'" : error), 1;
+    pure_link_peer(link, public_b);
     payload = malloc(size + 256);
     payload_size = order(&sender, (uint32_t)output, size, payload);
     for (size_t i = 0; i < payload_size; ++i) payload[i] ^= (unsigned char)(i * 7); /* bytes that differ by place */
     uint64_t first = 1;
     if (!window) /* a backlog: all queued before the link starts */
         for (; first <= count; ++first)
-            if (!pure_link_queue(link, first, (PureSpan){payload, payload_size})) return fputs("transport: cannot queue\n", stderr), 1;
+            if (!pure_link_queue(link, first, payload, payload_size)) return fputs("transport: cannot queue\n", stderr), 1;
     uint64_t began = now_ns(), cpu_link = 0, cpu_listen = thread_cpu("pure-listen"), cpu_all = thread_cpu(NULL);
     if (!pure_link_start(link, notify, NULL, error, sizeof(error))) return fprintf(stderr, "transport: %s\n", error), 1;
     for (uint64_t seq = first; seq <= count; ++seq) {
@@ -155,7 +164,7 @@ int main(int argc, char **argv) {
             struct timespec pause = {0, 50000};
             nanosleep(&pause, NULL);
         }
-        if (!pure_link_queue(link, seq, (PureSpan){payload, payload_size})) return fputs("transport: cannot queue\n", stderr), 1;
+        if (!pure_link_queue(link, seq, payload, payload_size)) return fputs("transport: cannot queue\n", stderr), 1;
     }
     while (pure_link_delivered(link) < count) {
         struct timespec pause = {0, 50000};
diff --git a/serve/serve.c b/serve/serve.c
index d89fb13..5688d79 100644
--- a/serve/serve.c
+++ b/serve/serve.c
@@ -197,9 +197,11 @@ typedef struct {
     unsigned wire_count;
     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 */
-    /* The owner's: outputs linked to Pure receivers (pure://), and how far each was trimmed. */
+    /* The owner's: outputs linked to Pure receivers (pure://), which output
+     * each is, and how far each was trimmed. */
     PureLink *links[PURE_MODULE_INPUTS];
     unsigned link_count;
+    uint32_t link_output[PURE_MODULE_INPUTS];
     uint64_t link_trimmed[PURE_MODULE_INPUTS];
     struct Writer *writer; /* the thread writing the app's checkpoints, once one was due */
     uint64_t checkpoints;  /* images it wrote */
@@ -1501,7 +1503,7 @@ typedef struct { Executor *executor; Fresh *fresh; size_t count, capacity; uint6
 
 /* An output's outlet: its wires first, then its links. */
 static uint32_t outlet_output(Executor *ex, unsigned i) {
-    return i < ex->wire_count ? pure_wire_output(ex->wires[i]) : pure_link_output(ex->links[i - ex->wire_count]);
+    return i < ex->wire_count ? pure_wire_output(ex->wires[i]) : ex->link_output[i - ex->wire_count];
 }
 static uint64_t outlet_queued(Executor *ex, unsigned i) {
     return i < ex->wire_count ? pure_wire_queued(ex->wires[i]) : pure_link_queued(ex->links[i - ex->wire_count]);
@@ -1541,7 +1543,8 @@ static void after_save(Executor *ex) {
         const Fresh *f = &q.fresh[k];
         if (blocked[f->wire]) continue;
         blocked[f->wire] = f->wire < ex->wire_count ? !pure_wire_queue(ex->wires[f->wire], f->seq, f->payload)
-                                                    : !pure_link_queue(ex->links[f->wire - ex->wire_count], f->seq, f->payload);
+                                                    : !pure_link_queue(ex->links[f->wire - ex->wire_count], f->seq,
+                                                                       f->payload.data, f->payload.size);
     }
     free(q.fresh);
 }
@@ -1574,7 +1577,7 @@ static bool take_deliveries(Executor *ex) {
     for (unsigned i = 0; i < ex->link_count; ++i) {
         uint64_t stored = pure_link_delivered(ex->links[i]);
         if (stored <= ex->link_trimmed[i]) continue;
-        pure_app_trim(&ex->app, pure_link_output(ex->links[i]), stored);
+        pure_app_trim(&ex->app, ex->link_output[i], stored);
         ex->link_trimmed[i] = stored;
     }
     return answered;
@@ -2331,6 +2334,17 @@ static void stop_shard(Host *host) {
     free(host->views);
 }
 
+/* The input a link's stream goes to: one taking messages (not keyed) of the
+ * sender's shape. */
+static int link_input(void *context, const char *name, uint32_t shape) {
+    const PureModule *m = ((Owner *)context)->executor.app.module;
+    int index = pure_module_input(m, name);
+    if (index < 0 || m->input[index].kind != PURE_KIND_MESSAGE || m->input[index].keyed ||
+        pure_module_shape(m, (uint32_t)index) != shape)
+        return -1;
+    return index;
+}
+
 /* Records a link delivered, queued at the owner as a request is: applied
  * with the next burst and done (pure_link_done) once its save is over. */
 static PureLinkListener *link_listener;
@@ -2393,13 +2407,17 @@ static bool open_wires(Executor *ex, const PureModule *m, const char *const *sen
         output[length] = 0;
         int written = snprintf(source, sizeof(source), "%s.%s", name, output);
         if (!strncmp(equals + 1, "pure://", 7)) { /* a link: no connection, sealed with the node's key */
-            PureLink *link = written > 0 && written <= PURE_SOURCE
-                                 ? pure_link_open(m, output, equals + 1, source, seal_key, own_public, error, sizeof(error))
+            int index = pure_module_input(m, output);
+            if (index < 0 || m->input[index].kind != PURE_KIND_OUTPUT) snprintf(error, sizeof(error), "no output '%s'", output);
+            PureLink *link = index >= 0 && m->input[index].kind == PURE_KIND_OUTPUT && written > 0 && written <= PURE_SOURCE
+                                 ? pure_link_open(equals + 1, source, pure_module_shape(m, (uint32_t)index), seal_key,
+                                                  own_public, error, sizeof(error))
                                  : NULL;
             if (!link) {
                 fprintf(stderr, "--send %s: %s\n", sends[i], written > PURE_SOURCE ? "the name and output are longer than 64 bytes" : error);
                 return false;
             }
+            ex->link_output[ex->link_count] = (uint32_t)index;
             ex->links[ex->link_count++] = link;
             continue;
         }
@@ -2440,7 +2458,7 @@ static bool wire_seals(Executor *ex, const PureModule *m, const char *const *sea
             for (unsigned k = 0; k < ex->wire_count; ++k)
                 if (!strcmp(m->input_names[pure_wire_output(ex->wires[k])], output)) wire = ex->wires[k];
             for (unsigned k = 0; k < ex->link_count; ++k)
-                if (!strcmp(m->input_names[pure_link_output(ex->links[k])], output)) link = ex->links[k];
+                if (!strcmp(m->input_names[ex->link_output[k]], output)) link = ex->links[k];
         }
         if (link && pure_seal_peer(equals + 1, peer)) { /* the link's receiver */
             pure_link_peer(link, peer);
@@ -2657,7 +2675,8 @@ int main(int argc, char **argv) {
             fputs("--link-port needs an app with states and the node's --key\n", stderr);
             return 2;
         }
-        link_listener = pure_link_listen(&module, link_port, seal_key, own_public, link_deliver, &owner, error, sizeof(error));
+        link_listener = pure_link_listen(link_port, seal_key, own_public, &(PureLinkHost){link_input, link_deliver, &owner},
+                                         error, sizeof(error));
         bool allowed = link_listener != NULL && allow_senders(link_listener, senders_path, error, sizeof(error));
         for (unsigned i = 0; allowed && i < sender_count; ++i)
             allowed = pure_link_allow(link_listener, senders[i].name, senders[i].key, error, sizeof(error));