Take a delivery's fields by name, dropping none

A message from another node was taken only in its input's exact shape
(the link's STREAM shape, fields and order alike; over HTTP, every
field). So a board whose nodes are upgraded one after the other stopped
both ways until both were, and an output and its input whose fields
differ only in order were refused for good (docs/design/014).

An input now takes a message when it has every field the sender sends,
by name, of the same type or one that holds its values (a u32 in a
number, text in bytes); a field only the input has is 0 or empty. One
that sends a field the input lacks is refused: taking it would drop the
value, and with its number heard the value would never come again (two
boards apart for good). So add fields to receivers first and take them
from senders first, and nothing stops or is lost.

The link carries a sender's fields in a new frame, FIELDS (0x0d: names,
shape, u8 count, each {type, name length, name}): after a REFUSED for
another shape the sender puts it in its packets (after OFFER, before
the records, so a receiver that knows no FIELDS reads what it read
before) until the stream is taken, the first time at once. A receiver
seeing another shape waits for the packet's FIELDS before refusing, and
asks its host with them (PureLinkHost.input takes them); REFUSED says 4
when the input lacks a field. Deliveries carry the fields and the host
lays each record out anew (pure_module_takes, pure_module_recast,
pure_module_fields, over pure_send_recast's rule). A stream whose sender
starts again in another shape lets go what it held; the sender sends it
again laid out anew. A refusal already queued for a stream now says the
last reason judged, not the first. Over HTTP a delivery (X-Pure-Source)
may lack fields; an unknown member is still refused.

test-link named: a stream of another shape naming its fields is taken,
200 stored in 47 ms with every delivery naming them; an input that does
not take them refuses (4) and the sender backs off. check.py
server_fields: five outputs, three linked and two over HTTP, fields in
another order taken, an input with a field more taking messages without
it, one lacking a field refusing until it starts again with it, then
taking all. server_upgrade's new receiver declares its fields in another
order. 012 names the frame and the rules.

5 files changed, 228 insertions, 32 deletions

diff --git a/link/link-protocol.c b/link/link-protocol.c
index 4fa5aab..169705f 100644
--- a/link/link-protocol.c
+++ b/link/link-protocol.c
@@ -19,7 +19,7 @@ enum { DATAGRAM_MOST = PURE_LINK_DATAGRAM_MOST, BODY_MOST = DATAGRAM_MOST - HEAD
  * at least this much of it fits. */
 enum { PART_LEAST = 64 };
 enum { F_PADDING = 0, F_STREAM = 1, F_RECORD = 2, F_ACK = 3, F_STORED = 4, F_PING = 5, F_RECORDS = 6, F_CHALLENGE = 7,
-       F_RESPONSE = 8, F_OFFER = 9, F_ACCEPT = 10, F_LOST = 11, F_REFUSED = 12 };
+       F_RESPONSE = 8, F_OFFER = 9, F_ACCEPT = 10, F_LOST = 11, F_REFUSED = 12, F_FIELDS = 13 };
 /* STORED's flags: what came after a gap was not kept; the app is full (no
  * room for more now), and what came is kept for when it has room. */
 enum { STORED_GAP = 1, STORED_FULL = 2 };
@@ -307,6 +307,11 @@ struct PureLinkSender {
      * answer with no REFUSED. */
     unsigned refusal, tried;
     uint64_t refused_until, refused_wait, tried_pn;
+    /* The fields its records are laid out with, named (FIELDS) in each
+     * packet from a refusal for another shape until the stream is taken. */
+    unsigned char *fields;
+    size_t fields_size;
+    bool naming;
     PureLinkCounts counts;
     /* Epochs (forward secrecy). Keys come from the secret of the epoch it
      * sends in (epoch 0's is the two nodes' static secret), each made from
@@ -696,6 +701,7 @@ static void on_stored(PureLinkSender *l, const unsigned char *p, size_t size, ui
     unsigned flags = p[names + 8];
     uint32_t credit = get32(p + names + 9);
     l->refusal = l->tried = 0, l->refused_until = l->refused_wait = 0; /* the stream is taken, and stored */
+    l->naming = false;
     if (heard > l->stored) {
         while (l->count && record_at(l, 0)->seq <= heard) {
             Record *r = record_at(l, 0);
@@ -725,8 +731,12 @@ static void on_stored(PureLinkSender *l, const unsigned char *p, size_t size, ui
 static void refused(PureLinkSender *l, unsigned why, uint64_t now) {
     if (l->refusal && l->tried != TRIED) return;
     l->refusal = why ? why : PURE_LINK_NO_INPUT;
-    l->refused_wait = l->refused_wait ? least(2 * l->refused_wait, REFUSED_MOST) : REFUSED_FIRST;
-    l->refused_until = now + l->refused_wait;
+    /* Another shape, and its fields not named yet: named in a packet at once. */
+    bool naming = (why == PURE_LINK_OTHER_SHAPE || why == PURE_LINK_FIELDS_LACKED) && l->fields;
+    bool at_once = naming && !l->naming;
+    l->naming = l->naming || naming;
+    if (!at_once) l->refused_wait = l->refused_wait ? least(2 * l->refused_wait, REFUSED_MOST) : REFUSED_FIRST;
+    l->refused_until = at_once ? 0 : now + l->refused_wait;
     l->tried = 0;
     ++l->counts.refused;
     for (size_t k = 0; k < l->count; ++k) {
@@ -804,7 +814,8 @@ void pure_link_sender_take(PureLinkSender *l, const unsigned char *packet, size_
         } else if (type != F_PING)
             break; /* padding, or what this side does not know: the rest */
     }
-    if (l->refusal && l->tried == TRIED && !refusing) l->refusal = l->tried = 0, l->refused_until = 0; /* taken now */
+    if (l->refusal && l->tried == TRIED && !refusing) /* taken now */
+        l->refusal = l->tried = 0, l->refused_until = 0, l->naming = false;
 }
 
 /* A packet of what is not sent yet, in order, within the receiver's credit
@@ -831,6 +842,17 @@ static size_t fill(PureLinkSender *l, unsigned char *body, Flight *f, bool pad,
         memcpy(body + at + 5, l->offer_public, PURE_SEAL_KEY);
         at += EPOCH_FRAME;
     }
+    /* The fields, for a receiver whose input takes another shape: after the
+     * frames one that knows no FIELDS reads, before the records. */
+    size_t names = 2 + strlen(l->source) + strlen(l->input);
+    if (l->naming && at + 1 + names + 4 + l->fields_size <= most) {
+        body[at++] = F_FIELDS;
+        at += put_names(body + at, l->source, l->input);
+        put32(body + at, l->shape);
+        at += 4;
+        memcpy(body + at, l->fields, l->fields_size);
+        at += l->fields_size;
+    }
     if (l->renegotiating) { /* no record under the static keys */
         l->responding = 0;
         return at;
@@ -1196,6 +1218,15 @@ uint64_t pure_link_sender_stored(const PureLinkSender *l) { return l->stored; }
 
 size_t pure_link_sender_held(const PureLinkSender *l) { return l->held; }
 
+bool pure_link_sender_fields(PureLinkSender *l, const unsigned char *fields, size_t size) {
+    unsigned char *kept = size ? malloc(size) : NULL;
+    if (size && !kept) return false;
+    if (size) memcpy(kept, fields, size);
+    free(l->fields);
+    l->fields = kept, l->fields_size = size;
+    return true;
+}
+
 void pure_link_sender_counts(const PureLinkSender *l, PureLinkCounts *out) {
     *out = l->counts;
     out->stored = l->stored;
@@ -1213,6 +1244,7 @@ void pure_link_sender_free(PureLinkSender *l) {
     explicit_bzero(l->epoch_secret, sizeof(l->epoch_secret));
     explicit_bzero(l->before_secret, sizeof(l->before_secret));
     explicit_bzero(l->offer_private, sizeof(l->offer_private));
+    free(l->fields);
     free(l);
 }
 
@@ -1298,6 +1330,8 @@ typedef struct Stream {
     size_t held_first, held_count, held_capacity, held_bytes;
     Run *run;                    /* the run that sent to it last: where STORED goes */
     uint64_t run_id;
+    unsigned char *fields;       /* the fields its records are laid out with, when its input took them by name */
+    size_t fields_size;
 } Stream;
 struct Active {
     uint32_t peer;               /* its place among the peers */
@@ -1720,6 +1754,7 @@ static void let_go(PureLinkReceiver *s, Active *a) {
         while (a->streams[k]->held_count) drop_first(a->streams[k]);
         free(a->streams[k]->held);
         free(a->streams[k]->line);
+        free(a->streams[k]->fields);
         free(a->streams[k]);
     }
     free(a->streams);
@@ -1845,30 +1880,56 @@ static bool allowed(const PureLinkReceiver *s, const Peer *peer, const char *sou
     return false;
 }
 
+/* Whose fields a stream's records are laid out with, kept (NULL, 0: its
+ * input's). False when memory runs out. */
+static bool keep_fields(Stream *st, const unsigned char *fields, size_t size) {
+    unsigned char *kept = fields && size ? malloc(size) : NULL;
+    if (fields && size && !kept) return false;
+    if (kept) memcpy(kept, fields, size);
+    free(st->fields);
+    st->fields = kept, st->fields_size = kept ? size : 0;
+    return true;
+}
+
 /* The stream of a peer's source to input, made the first time: NULL when it
- * is refused (why: the source is not allowed from the peer, or the host has
- * no input that takes such messages), or memory runs out (why 0). */
+ * is refused (why: the source is not allowed from the peer, the host has no
+ * input that takes such messages, one of another shape unless it takes the
+ * fields named, which come with FIELDS), or memory runs out (why 0). A
+ * stream the sender now sends in another shape (it started again with
+ * another module) starts over with what it holds let go: the sender sends
+ * those again, laid out anew. */
 static Stream *stream_of(PureLinkReceiver *s, Active *a, const char *source, const char *input, uint32_t shape,
-                         unsigned *why) {
+                         const unsigned char *fields, size_t fields_size, unsigned *why) {
     *why = 0;
-    for (unsigned k = 0; k < a->stream_count; ++k) {
-        Stream *st = a->streams[k];
-        if (strcmp(st->input_name, input) || strcmp(st->source, source)) continue;
-        if (st->shape != shape) *why = PURE_LINK_OTHER_SHAPE;
-        return *why ? NULL : st;
-    }
-    int index = allowed(s, &s->peers[a->peer], source) ? s->host.input(s->host.context, input, shape)
-                                                       : -PURE_LINK_NOT_ALLOWED;
+    Stream *st = NULL;
+    for (unsigned k = 0; k < a->stream_count && !st; ++k)
+        if (!strcmp(a->streams[k]->input_name, input) && !strcmp(a->streams[k]->source, source)) st = a->streams[k];
+    if (st && st->shape == shape) return st;
+    int index = !allowed(s, &s->peers[a->peer], source) ? -PURE_LINK_NOT_ALLOWED :
+                s->host.input(s->host.context, input, shape, fields, fields ? fields_size : 0);
     if (index < 0) {
-        *why = index == -PURE_LINK_OTHER_SHAPE || index == -PURE_LINK_NOT_ALLOWED ? (unsigned)-index : PURE_LINK_NO_INPUT;
+        *why = index == -PURE_LINK_OTHER_SHAPE || index == -PURE_LINK_NOT_ALLOWED || index == -PURE_LINK_FIELDS_LACKED
+                   ? (unsigned)-index : PURE_LINK_NO_INPUT;
         return NULL;
     }
+    if (st) { /* another shape now */
+        if (st->out) { *why = PURE_LINK_OTHER_SHAPE; return NULL; } /* its last delivery first */
+        if (!keep_fields(st, fields, fields_size)) return NULL;
+        while (st->held_count) drop_first(st);
+        st->line_head = st->line_size = 0;
+        st->line_count = 0;
+        st->tail = st->known = false;
+        st->input = (uint32_t)index;
+        st->shape = shape;
+        return st;
+    }
     size_t capacity = a->stream_capacity;
-    Stream **streams = room(a->streams, &capacity, a->stream_count, sizeof(*streams), 2), *st;
+    Stream **streams = room(a->streams, &capacity, a->stream_count, sizeof(*streams), 2);
     if (!streams) return NULL;
     a->streams = streams;
     a->stream_capacity = (unsigned)capacity;
     if (!(st = calloc(1, sizeof(*st)))) return NULL;
+    if (!keep_fields(st, fields, fields_size)) { free(st); return NULL; }
     st->active = a;
     st->input = (uint32_t)index;
     st->shape = shape;
@@ -2023,7 +2084,8 @@ static bool hand_over(PureLinkReceiver *s, Stream *st) {
         size_t bytes = st->line_size - st->line_head; /* the whole line, when it is a batch or less */
         if (n > BATCH)
             for (n = 0, bytes = 0; 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};
+        PureLinkDelivery d = {st->input, st->source, st->line_first, n, st->line + st->line_head, bytes, st,
+                              st->fields, st->fields_size};
         if (!s->host.deliver(s->host.context, &d)) return false;
         st->out = true;
         st->out_count = n, st->out_bytes = bytes;
@@ -2052,7 +2114,8 @@ static bool hand_over(PureLinkReceiver *s, Stream *st) {
         memcpy(s->batch + at + 4, held_at(st, k)->bytes, held_at(st, k)->size);
         at += 4 + held_at(st, k)->size;
     }
-    PureLinkDelivery d = {st->input, st->source, held_at(st, first)->seq, (unsigned)n, s->batch, bytes, st};
+    PureLinkDelivery d = {st->input, st->source, held_at(st, first)->seq, (unsigned)n, s->batch, bytes, st,
+                          st->fields, st->fields_size};
     if (!s->host.deliver(s->host.context, &d)) return false;
     st->out = true;
     ++s->counts.deliveries;
@@ -2215,8 +2278,10 @@ static void lost(PureLinkReceiver *s, const Peer *peer, const Active *a, uint64_
  * to its run: once a turn. */
 static void refuse(PureLinkReceiver *s, Run *r, const unsigned char *names, size_t size, unsigned why) {
     for (size_t k = 0; k < s->refusal_count; ++k)
-        if (s->refusals[k].run == r && s->refusals[k].size == size + 2 && !memcmp(s->refusals[k].frame + 1, names, size))
+        if (s->refusals[k].run == r && s->refusals[k].size == size + 2 && !memcmp(s->refusals[k].frame + 1, names, size)) {
+            s->refusals[k].frame[1 + size] = (unsigned char)why; /* said once, as last judged (named fields, say) */
             return;
+        }
     void *more = room(s->refusals, &s->refusal_capacity, s->refusal_count, sizeof(*s->refusals), 4);
     if (!more) return;
     s->refusals = more;
@@ -2318,6 +2383,10 @@ static bool take_from(PureLinkReceiver *s, Peer *peer, const unsigned char *pack
     Stream *st = NULL;
     bool eliciting = false, records = false;
     const unsigned char *offer = NULL; /* taken after the frames: whether records came with it */
+    /* A stream of another shape: refused at the end unless this packet's
+     * FIELDS names fields its input takes. */
+    const unsigned char *other = NULL;
+    size_t other_names = 0;
     for (size_t at = 0; at < size;) {
         unsigned type = plain[at++];
         if (type == F_STREAM) {
@@ -2325,14 +2394,36 @@ static bool take_from(PureLinkReceiver *s, Peer *peer, const unsigned char *pack
             size_t names = get_names(plain + at, size - at, source, input);
             if (!names || at + names + 4 > size) break;
             unsigned why;
-            st = stream_of(s, a, source, input, get32(plain + at + names), &why);
+            st = stream_of(s, a, source, input, get32(plain + at + names), NULL, 0, &why);
             if (st) {
                 st->run = r;
                 st->run_id = r->run;
-            } else if (why)
+            } else if (why == PURE_LINK_OTHER_SHAPE)
+                other = plain + at, other_names = names;
+            else if (why)
                 refuse(s, r, plain + at, names, why);
             at += names + 4;
             eliciting = true;
+        } else if (type == F_FIELDS) {
+            char source[PURE_LINK_NAME + 1], input[PURE_LINK_NAME + 1];
+            size_t names = get_names(plain + at, size - at, source, input), length = 1;
+            if (!names || at + names + 5 > size) break;
+            const unsigned char *fields = plain + at + names + 4;
+            for (unsigned f = 0; f < fields[0] && length != SIZE_MAX; ++f) /* each {type, name length, name} */
+                length = at + names + 4 + length + 2 <= size ? length + 2 + fields[length + 1] : SIZE_MAX;
+            if (length == SIZE_MAX || at + names + 4 + length > size) break;
+            if (other && names == other_names && !memcmp(plain + at, other, names + 4)) {
+                unsigned why;
+                st = stream_of(s, a, source, input, get32(plain + at + names), fields, length, &why);
+                if (st) {
+                    st->run = r;
+                    st->run_id = r->run;
+                } else if (why)
+                    refuse(s, r, other, other_names, why);
+                other = NULL;
+            }
+            at += names + 4 + length;
+            eliciting = true;
         } else if (type == F_RECORD) {
             if (at + RECORD_HEAD - 1 > size) break;
             uint32_t length = get16(plain + at + 16);
@@ -2384,6 +2475,7 @@ static bool take_from(PureLinkReceiver *s, Peer *peer, const unsigned char *pack
         } else
             break; /* padding, or what this side does not know: the rest */
     }
+    if (other) refuse(s, r, other, other_names, PURE_LINK_OTHER_SHAPE); /* its fields not named */
     if (offer) offer_taken(s, a, r, get32(offer), offer + 4, e, !records, now);
     if (eliciting) {
         r->ack_due = true;
diff --git a/link/link-protocol.h b/link/link-protocol.h
index e4d3080..feff543 100644
--- a/link/link-protocol.h
+++ b/link/link-protocol.h
@@ -111,6 +111,12 @@
  *                sender sends one packet of them a wait from then (a
  *                second, twice as long each time, 64 s at most) until one
  *                is stored
+ *   0x0d FIELDS  source, input, u32 shape (as STREAM), then the fields the
+ *                stream's records are laid out with (core/module.h's
+ *                pure_module_fields): after a REFUSED for another shape,
+ *                in every packet (after OFFER, before records) until the
+ *                stream is taken. A receiver whose input has every field
+ *                takes it by name (docs/design/014-fields-by-name.md)
  *   0x00 PADDING to the end of the packet: a full packet sent in a burst is
  *                a whole datagram
  * Losses are detected as RFC 9002 says, with RACK's patience for packets
@@ -129,9 +135,10 @@
 enum { PURE_LINK_DATAGRAM = 1200, PURE_LINK_DATAGRAM_MOST = 1452, PURE_LINK_NAME = 64, PURE_LINK_ADDRESS = 48,
        PURE_LINK_RECORD_MOST = 16 << 20 };
 /* Why a receiver refuses a stream (REFUSED): no input of that name takes
- * messages, the input takes messages of another shape, or the sender's key
- * may not send as that source. */
-enum { PURE_LINK_NO_INPUT = 1, PURE_LINK_OTHER_SHAPE = 2, PURE_LINK_NOT_ALLOWED = 3 };
+ * messages, the input takes messages of another shape (send FIELDS), the
+ * sender's key may not send as that source, or the input lacks fields the
+ * stream sends (it would drop them). */
+enum { PURE_LINK_NO_INPUT = 1, PURE_LINK_OTHER_SHAPE = 2, PURE_LINK_NOT_ALLOWED = 3, PURE_LINK_FIELDS_LACKED = 4 };
 
 /* Counts for a host's report: packets and records sent, records sent again,
  * packets lost and acknowledged, packets received, deliveries handed over,
@@ -193,6 +200,9 @@ unsigned pure_link_sender_refusal(const PureLinkSender *);
 /* How often it moves on to a new epoch (ns), and after how many bytes sent
  * (an hour and a GiB at first). */
 void pure_link_sender_epochs(PureLinkSender *, uint64_t every_ns, uint64_t bytes);
+/* The fields its records are laid out with (pure_module_fields), for a
+ * receiver whose input takes another shape (FIELDS); none at first. */
+bool pure_link_sender_fields(PureLinkSender *, const unsigned char *fields, size_t size);
 /* The last number queued; the last the receiver stored. */
 uint64_t pure_link_sender_queued(const PureLinkSender *);
 uint64_t pure_link_sender_stored(const PureLinkSender *);
@@ -206,7 +216,9 @@ void pure_link_sender_free(PureLinkSender *);
  * pure_link_idle). A stream (a source to an input) goes to the host's input
  * its name and shape find: its number, or -PURE_LINK_OTHER_SHAPE when it
  * takes another shape, or -1 (PURE_LINK_NO_INPUT) when none takes messages
- * by that name; a stream refused is said to be (REFUSED). Whole
+ * by that name; asked again with the fields the sender names (FIELDS),
+ * its number when it takes them by name, else -PURE_LINK_FIELDS_LACKED. A
+ * stream refused is said to be (REFUSED). Whole
  * records of a stream, in order, go to deliver as one delivery; the host
  * applies them from the one after the source's last heard number, stops at
  * a gap, saves, and then says so with a done (pure_link_receiver_turn).
@@ -220,9 +232,13 @@ typedef struct {
     const unsigned char *records; /* count of: u32le size, bytes (the receiver's: the host's during the call) */
     size_t size;
     void *token; /* for its done */
+    /* The fields the records are laid out with when the input took them by
+     * name (the host lays each out anew for it), else NULL. */
+    const unsigned char *fields;
+    size_t fields_size;
 } PureLinkDelivery;
 typedef struct {
-    int (*input)(void *context, const char *name, uint32_t shape);
+    int (*input)(void *context, const char *name, uint32_t shape, const unsigned char *fields, size_t fields_size);
     bool (*deliver)(void *context, const PureLinkDelivery *);
     void *context;
 } PureLinkHost;
diff --git a/link/link.c b/link/link.c
index ad9376c..8f36e21 100644
--- a/link/link.c
+++ b/link/link.c
@@ -210,6 +210,13 @@ uint64_t pure_link_delivered(PureLink *l) {
     return stored;
 }
 
+bool pure_link_fields(PureLink *l, const unsigned char *fields, size_t size) {
+    pthread_mutex_lock(&l->lock);
+    bool kept = pure_link_sender_fields(l->sender, fields, size);
+    pthread_mutex_unlock(&l->lock);
+    return kept;
+}
+
 size_t pure_link_held(PureLink *l) {
     pthread_mutex_lock(&l->lock);
     size_t held = pure_link_sender_held(l->sender);
@@ -380,7 +387,8 @@ static void *run_link(void *context) {
                     "trying again\"}\n", l->host, l->port, l->input, l->source,
                     refusal == PURE_LINK_NOT_ALLOWED ? "this node's key may not send as that source"
                     : refusal == PURE_LINK_OTHER_SHAPE ? "its input takes messages of another shape"
-                                                       : "it has no input of that name taking messages");
+                    : refusal == PURE_LINK_FIELDS_LACKED ? "its input lacks fields this output sends (it would drop them)"
+                                                         : "it has no input of that name taking messages");
     }
     return NULL;
 }
diff --git a/link/link.h b/link/link.h
index 64846ce..c8d8635 100644
--- a/link/link.h
+++ b/link/link.h
@@ -18,6 +18,10 @@ typedef struct PureLink PureLink;
 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]);
+/* The fields its sends are laid out with (core/module.h's
+ * pure_module_fields): a receiver whose input takes another shape may take
+ * them by name (docs/design/014-fields-by-name.md). */
+bool pure_link_fields(PureLink *, const unsigned char *fields, size_t size);
 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. */
diff --git a/link/test-link.c b/link/test-link.c
index 66899ff..2e7509a 100644
--- a/link/test-link.c
+++ b/link/test-link.c
@@ -102,6 +102,8 @@ typedef struct {
      * handed records meanwhile, and how long after it had room again. */
     uint64_t full_at, full_ms, full_until, full_tries, full_late;
     bool later_open;     /* the app has an input "later" (as "order") */
+    bool takes;          /* its input takes the fields a sender of another shape names (NAMED) */
+    uint64_t named_deliveries; /* deliveries that came with those fields */
 } World;
 
 static void put_packet(World *w, bool to_receiver, const unsigned char *bytes, size_t size) {
@@ -146,10 +148,15 @@ static bool answer(void *context, const unsigned char *packet, size_t size, cons
     return true;
 }
 
-static int input_of(void *context, const char *name, uint32_t shape) {
+/* The fields a sender of another shape (0x5eee) names; opaque to the link. */
+static const unsigned char NAMED[] = {2, 4, 4, 'i', 't', 'e', 'm', 1, 5, 'c', 'o', 'u', 'n', 't'};
+
+static int input_of(void *context, const char *name, uint32_t shape, const unsigned char *fields, size_t size) {
     const World *w = context;
     if (strcmp(name, "order") && !(w && w->later_open && !strcmp(name, "later"))) return -PURE_LINK_NO_INPUT;
-    return shape == 0x5eed ? 3 : -PURE_LINK_OTHER_SHAPE;
+    if (shape == 0x5eed) return 3;
+    if (!fields) return -PURE_LINK_OTHER_SHAPE;
+    return w && w->takes && size == sizeof(NAMED) && !memcmp(fields, NAMED, size) ? 3 : -PURE_LINK_FIELDS_LACKED;
 }
 
 /* Applied from the one after the last heard, in order; a gap stops it. */
@@ -158,6 +165,7 @@ static bool deliver(void *context, const PureLinkDelivery *d) {
     CHECK(d->input == 3 && !strcmp(d->source, "test.ordered"), "a delivery to input %u from %s", d->input, d->source);
     if (w->done_count == sizeof(w->done) / sizeof(w->done[0])) return false; /* busy: tried again */
     ++w->deliveries;
+    w->named_deliveries += d->fields && d->fields_size == sizeof(NAMED) && !memcmp(d->fields, NAMED, sizeof(NAMED));
     if (w->full_ms && !w->full_until && w->heard >= w->full_at) w->full_until = w->now + w->full_ms * MS;
     if (w->now < w->full_until) {
         ++w->full_tries;
@@ -682,8 +690,8 @@ typedef struct {
     PureLinkDone done[64];
     size_t done_count;
 } Outputs;
-static int output_input(void *context, const char *name, uint32_t shape) {
-    (void)context;
+static int output_input(void *context, const char *name, uint32_t shape, const unsigned char *fields, size_t size) {
+    (void)context, (void)fields, (void)size;
     return shape != 0x5eed ? -1 : !strcmp(name, "one") ? 0 : !strcmp(name, "two") ? 1 : !strcmp(name, "three") ? 2 : -1;
 }
 static bool output_deliver(void *context, const PureLinkDelivery *d) {
@@ -958,6 +966,73 @@ static void refused_restarted(const char *dir) {
     free(w.packets);
 }
 
+/* A stream of another shape whose sender names its fields (FIELDS): the
+ * receiver's input takes them by name and every record is stored, each
+ * delivery naming them for the host to lay the records out anew; an input
+ * that does not take them refuses the stream (it would drop fields), and the
+ * sender tries a packet a wait as for any refusal. One that names none stays
+ * refused for another shape (refusals, above). */
+static void named(const char *dir) {
+    for (int takes = 1; takes >= 0; --takes) {
+        unsigned char public_a[PURE_SEAL_KEY], public_b[PURE_SEAL_KEY], out[PURE_LINK_DATAGRAM_MOST];
+        PureSealKey *a = key(dir, "a.key", public_a), *b = key(dir, "b.key", public_b);
+        char memo[512];
+        snprintf(memo, sizeof(memo), "%s/b.key.links", dir);
+        unlink(memo);
+        World w = {.packets = malloc(PACKETS * sizeof(Packet)), .takes = takes};
+        if (!w.packets) exit(1);
+        PureLinkReceiver *r = receiver_of(&w, b, public_b, public_a, memo, 120000);
+        PureLinkSender *s = pure_link_sender("test.ordered", "order", 0x5eee, a, public_a);
+        if (!s || !pure_link_sender_peer(s, public_b) || !pure_link_sender_fields(s, NAMED, sizeof(NAMED))) exit(1);
+        static unsigned char bytes[20200];
+        for (uint64_t seq = 1; seq <= 200; ++seq) {
+            size_t size = record_size(seq);
+            for (size_t k = 0; k < size; ++k) bytes[k] = record_byte(seq, k);
+            if (!pure_link_sender_queue(s, seq, bytes, size)) exit(1);
+        }
+        uint64_t number, packets = 0, stored_at = 0;
+        for (; w.now < 30000 * MS; w.now += MS) {
+            for (size_t size; (size = pure_link_sender_packet(s, out, false, w.now, &number)); ++packets)
+                pure_link_receiver_take(r, out, size, "sender0", 7, w.now);
+            size_t told = w.done_count;
+            memcpy(w.told, w.done, told * sizeof(*w.done));
+            w.done_count = 0;
+            pure_link_receiver_turn(r, w.told, told, w.now, answer, &w);
+            for (size_t k = 0; k < w.count;) {
+                if (w.packets[k].at > w.now) {
+                    ++k;
+                    continue;
+                }
+                pure_link_sender_take(s, w.packets[k].bytes, w.packets[k].size, w.now);
+                w.packets[k] = w.packets[--w.count];
+            }
+            pure_link_sender_tick(s, w.now);
+            if (!stored_at && pure_link_sender_stored(s) == 200) stored_at = w.now;
+        }
+        PureLinkCounts c;
+        pure_link_sender_counts(s, &c);
+        unsigned why = pure_link_sender_refusal(s);
+        printf("{\"case\":\"named\",\"takes\":%d,\"why\":%u,\"refused\":%llu,\"packets\":%llu,\"stored\":%llu,"
+               "\"stored_ms\":%llu,\"named_deliveries\":%llu}\n", takes, why, (unsigned long long)c.refused,
+               (unsigned long long)packets, (unsigned long long)pure_link_sender_stored(s),
+               (unsigned long long)(stored_at / MS), (unsigned long long)w.named_deliveries);
+        if (takes)
+            CHECK(!why && c.refused == 1 && stored_at && stored_at < 1000 * MS && w.heard == 200 &&
+                      w.named_deliveries == w.deliveries,
+                  "named and taken: why %u, refused %llu, stored at %llu ms, %llu of %llu deliveries named", why,
+                  (unsigned long long)c.refused, (unsigned long long)(stored_at / MS),
+                  (unsigned long long)w.named_deliveries, (unsigned long long)w.deliveries);
+        else /* refused at once, named at once, then at 1, 3, 7 and 15 s */
+            CHECK(why == PURE_LINK_FIELDS_LACKED && c.refused == 6 && !pure_link_sender_stored(s) && packets <= 20,
+                  "named and not taken: why %u, refused %llu, %llu packets", why, (unsigned long long)c.refused,
+                  (unsigned long long)packets);
+        pure_link_sender_free(s);
+        pure_link_receiver_free(r);
+        pure_seal_key_free(a), pure_seal_key_free(b);
+        free(w.packets);
+    }
+}
+
 /* The largest send (PURE_LINK_RECORD_MOST) goes whole, over a receiver that
  * never heard of the stream before; one byte more is not taken to send. */
 typedef struct { World *w; PureLinkSender *s; } Largest;
@@ -1089,6 +1164,7 @@ int main(void) {
     outputs(dir);
     refusals(dir);
     refused_restarted(dir);
+    named(dir);
     largest(dir);
     char command[600];
     snprintf(command, sizeof(command), "rm -rf %s", dir);