packet-net · M0LTE · May 5, 2026 · May 5, 2026
diff --git a/docs/app-developers/reference.md b/docs/app-developers/reference.md
@@ -203,6 +203,9 @@ DAPPS does not impose a hard payload-size limit at the app interface - submit an
 | Read source callsign           | `dapps-source` user property                      | `sourceCallsign` field                              |
 | Read residual TTL              | `dapps-ttl` user property (absent if no TTL)      | `ttl` field (`null` if no TTL)                      |
 | Set TTL on submit              | `dapps-ttl` user property                         | `ttl` field                                         |
+| Opt into ordered delivery      | `dapps-stream` user property                      | `streamId` field                                    |
+| Set ordering gap policy        | `dapps-stream-gap-timeout` user property          | `streamGapTimeoutSeconds` field                     |
+| Read delivered stream / seq    | `dapps-stream`, `dapps-stream-seq` properties     | (not surfaced on REST inbound today)                |
 
 ## DAPPSv1 wire format (summary)
 
@@ -218,7 +221,7 @@ Then a back-and-forth of one-line commands and responses:
 
 | Command                                         | Direction        | Meaning                                              |
 |-------------------------------------------------|------------------|------------------------------------------------------|
-| `ihave id=<7hex> dst=<callsign> sz=<bytes> ttl=<seconds> [src=<callsign>] [mid=<id> frag=<n>/<m>]` | sender → receiver | "I have this message; do you want it?"             |
+| `ihave id=<7hex> dst=<callsign> sz=<bytes> ttl=<seconds> [src=<callsign>] [mid=<id> frag=<n>/<m>] [sid=<stream> sn=<seq> gt=<seconds>]` | sender → receiver | "I have this message; do you want it?"             |
 | `send`                                          | receiver → sender | "Yes, send it."                                     |
 | `?`                                             | receiver → sender | "Already have it / don't recognise this command."   |
 | `data <bytes>`                                  | sender → receiver | The payload, exactly `<sz>` bytes.                  |
@@ -228,10 +231,66 @@ Then a back-and-forth of one-line commands and responses:
 | `end`                                           | reply            | End of `peers` response.                            |
 | `rev <id>[,<id>...]`                            | either           | "Send me anything you're holding for these callsigns." |
 
-Headers on `ihave` are forward-compatible - receivers ignore unknown ones. New optional fields (e.g. `src=` for source tracking, `mid=` + `frag=N/M` for multi-part) ride the existing `DAPPSv1>` prompt. Breaking changes bump the prompt to `DAPPSv2>`.
+Headers on `ihave` are forward-compatible - receivers ignore unknown ones. New optional fields (e.g. `src=` for source tracking, `mid=` + `frag=N/M` for multi-part, `sid=`/`sn=`/`gt=` for opt-in ordering) ride the existing `DAPPSv1>` prompt. Breaking changes bump the prompt to `DAPPSv2>`.
 
 The full wire spec lives in the [main repository README](https://github.com/M0LTE/dapps/blob/master/README.md#on-air-protocol).
 
+## Message ordering (opt-in)
+
+DAPPS doesn't order messages by default. Each submission is independent; under retries and routing reconvergence the receiver can see them in any order. For most apps this is correct: idempotent or content-addressed work doesn't care.
+
+When an app does care - chat transcripts, telemetry sequences, change-log streams - opt-in ordering is available. Setting `streamId` on a submission tags the message as part of a per-sender ordered stream; the daemon mints a monotonic sequence number and the receiving daemon delivers messages on that stream in submit order.
+
+### Opting in
+
+REST:
+
+```bash
+curl -sS -X POST http://localhost:5086/AppApi/outbound \
+  -H 'content-type: application/json' \
+  -d '{
+    "app": "chat",
+    "destCallsign": "M0LTE",
+    "payload": "aGVsbG8=",
+    "streamId": "c1",
+    "streamGapTimeoutSeconds": 600
+  }'
+```
+
+MQTT:
+
+```
+publish dapps/out/chat/M0LTE
+  user-property dapps-stream=c1
+  user-property dapps-stream-gap-timeout=600
+  payload <bytes>
+```
+
+`streamId` is sender-scoped: two senders can pick the same id without colliding because the receiver keys its cursor on `(originator-callsign, streamId)`. Pick something short (it travels on every wire frame for that stream).
+
+`streamGapTimeoutSeconds` chooses the policy when a message is missing:
+
+- **`0` (default, "strict")**: stall forever waiting for the missing seq. Later messages park until the gap fills. Use when you'd rather wait than skip.
+- **`>0` ("timeout")**: stall for that many seconds, then skip past the gap and deliver waiting messages. Use when stale data is worse than missing data.
+
+### What the receiver sees
+
+Inbound messages tagged with a stream show two extra MQTT user properties:
+
+- `dapps-stream` - the stream id the sender chose.
+- `dapps-stream-seq` - the seq within that stream, ascending.
+
+Apps that don't care can ignore them; apps that opted in can use them to detect stream id changes (the sender rotated to a fresh stream after a reset) or to assert seq monotonicity for their own bookkeeping.
+
+### Tradeoffs
+
+- **Latency cost**. One missing message stalls the whole stream until it arrives or the timeout fires. On lossy radio links this is real - opt-in is the right default.
+- **Sender resets**. The sender persists its counter to disk; a fresh install / wiped database starts back at `sn=1`. Re-using the same `streamId` after a reset will cause receivers to drop the new messages as `stream-stale` (their cursor is well past `sn=1`). Mitigate by appending a short epoch suffix to the stream id when you reset (e.g. `chat:tom.2`).
+- **End-to-end semantics**. Ordering is enforced at the receiving daemon, not at intermediate forwarders. Hops can reorder, retry, and flood freely - the trio rides the envelope verbatim.
+- **Forward compatibility**. A daemon that doesn't understand `sid`/`sn`/`gt` ignores the keys and delivers each message immediately. Apps subscribed to a partially-ordering-aware mesh see ordered delivery only between aware nodes.
+
+The dashboard's `/Streams` page surfaces both sender-side counters and receiver-side cursors plus pending row counts; a stalled stream shows up as a non-empty pending column.
+
 ## See also
 
 - [Concepts](concepts.md) - the mental model.

diff --git a/src/dapps/dapps.client/Backhaul/BackhaulMessage.cs b/src/dapps/dapps.client/Backhaul/BackhaulMessage.cs
@@ -26,7 +26,10 @@ public sealed record BackhaulMessage(
     IReadOnlyList<string>? TraversedHops = null,
     string? MasterId = null,
     int? FragmentIndex = null,
-    int? FragmentTotal = null);
+    int? FragmentTotal = null,
+    string? StreamId = null,
+    uint? StreamSeq = null,
+    uint? StreamGapTimeoutSeconds = null);
 
 // LinkSourceCallsign: the *immediate sender's* callsign, distinct from
 // Originator (the F1 end-to-end source). Carried on bearers that don't
@@ -72,3 +75,13 @@ public sealed record BackhaulMessage(
 // `mid=…` + `frag=N/M` headers); the receiver's IHaveValidator
 // rejects any mismatched-presence combination. FragmentTotal ≥ 2;
 // single-fragment messages just omit all three fields.
+//
+// StreamId / StreamSeq / StreamGapTimeoutSeconds (opt-in ordering):
+// when StreamSeq is set the message is part of a per-sender ordered
+// stream identified by StreamId. The receiver delivers messages on
+// each (sender-callsign, StreamId) cursor in monotonically-increasing
+// StreamSeq order; gaps stall until the missing seq arrives or
+// StreamGapTimeoutSeconds elapses (gt=0 = strict, never skip).
+// Wire form: `sid=`, `sn=`, `gt=` keys on the ihave line; codec flag
+// bit 9 on datagram bearers. All three are required together when
+// any one is set; intermediate forwarders preserve them verbatim.
diff --git a/src/dapps/dapps.client/Backhaul/Dappsv1SessionBackhaul.cs b/src/dapps/dapps.client/Backhaul/Dappsv1SessionBackhaul.cs
@@ -81,7 +81,10 @@ public async Task<BackhaulSendResult> SendAsync(
                     originator: message.Originator,
                     masterId: message.MasterId,
                     fragmentIndex: message.FragmentIndex,
-                    fragmentTotal: message.FragmentTotal))
+                    fragmentTotal: message.FragmentTotal,
+                    streamId: message.StreamId,
+                    streamSeq: message.StreamSeq,
+                    streamGapTimeoutSeconds: message.StreamGapTimeoutSeconds))
             {
                 return BackhaulSendResult.Fail($"offer rejected for {message.Id}");
             }
@@ -112,7 +115,10 @@ public async Task<BackhaulSendResult> SendAsync(
                             Originator: polled.Originator,
                             MasterId: polled.MasterId,
                             FragmentIndex: polled.FragmentIndex,
-                            FragmentTotal: polled.FragmentTotal);
+                            FragmentTotal: polled.FragmentTotal,
+                            StreamId: polled.StreamId,
+                            StreamSeq: polled.StreamSeq,
+                            StreamGapTimeoutSeconds: polled.StreamGapTimeoutSeconds);
                         await opportunisticInbox.DeliverAsync(inbound, route.Callsign, ct);
                     }
                 }

diff --git a/src/dapps/dapps.client/Backhaul/Datagram/BackhaulMessageCodec.cs b/src/dapps/dapps.client/Backhaul/Datagram/BackhaulMessageCodec.cs
@@ -23,7 +23,8 @@ namespace dapps.client.Backhaul.Datagram;
 ///                          bit3=originator, bit4=link-source,
 ///                          bit5=flood-hops-remaining,
 ///                          bit6=source-route, bit7=traversed-hops,
-///                          bit8=fragment (F2 multi-part)
+///                          bit8=fragment (F2 multi-part),
+///                          bit9=stream (opt-in ordering)
 ///   [7]  id (UTF-8 ASCII, 7-char hex from DappsMessage.ComputeHash)
 ///   [8]  salt              (only when flags bit0)
 ///   [4]  ttl seconds       (only when flags bit1)
@@ -43,6 +44,10 @@ namespace dapps.client.Backhaul.Datagram;
 ///   [7]  master id         (only when flags bit8; ASCII)
 ///   [2]  fragment index    (only when flags bit8; UInt16, 1-based)
 ///   [2]  fragment total    (only when flags bit8; UInt16)
+///   [1]  stream id len     (only when flags bit9; max 255 bytes)
+///   [S]  stream id         (only when flags bit9; UTF-8)
+///   [4]  stream seq        (only when flags bit9; UInt32 LE)
+///   [4]  stream gap timeout (only when flags bit9; UInt32 LE seconds, 0=strict)
 ///   [2]  headers count     (only when flags bit2)
 ///   per header:
 ///     [2] key len, [K] key (UTF-8), [2] value len, [V] value (UTF-8)
@@ -56,7 +61,7 @@ public static class BackhaulMessageCodec
     /// <summary>Version this encoder writes AND the only version the
     /// decoder accepts. Bump on any wire-format change so a mismatched
     /// peer fails fast instead of silently misreading flag bits.</summary>
-    public const byte Version = 6;
+    public const byte Version = 7;
     public const int IdLength = 7;
 
     [Flags]
@@ -72,6 +77,7 @@ private enum Flags : ushort
         HasSourceRoute = 1 << 6,
         HasTraversedHops = 1 << 7,
         HasFragment = 1 << 8,
+        HasStream = 1 << 9,
     }
 
     public static byte[] Encode(BackhaulMessage message)
@@ -102,6 +108,28 @@ public static byte[] Encode(BackhaulMessage message)
             throw new ArgumentException("fragment index/total without master id", nameof(message));
         }
 
+        // Opt-in ordering trio: present together or absent together. The
+        // receiver enforces this in IHaveValidator too, but a relay that
+        // forwards a partial set would silently drop ordering for the
+        // downstream hop, which is worse than failing fast here.
+        var hasStream = !string.IsNullOrEmpty(message.StreamId)
+            || message.StreamSeq.HasValue
+            || message.StreamGapTimeoutSeconds.HasValue;
+        if (hasStream
+            && (string.IsNullOrEmpty(message.StreamId)
+                || !message.StreamSeq.HasValue
+                || !message.StreamGapTimeoutSeconds.HasValue))
+        {
+            throw new ArgumentException(
+                "stream id/seq/gap-timeout must all be set together (opt-in ordering) or all be absent",
+                nameof(message));
+        }
+        var streamIdBytes = hasStream ? Encoding.UTF8.GetBytes(message.StreamId!) : [];
+        if (hasStream && streamIdBytes.Length > byte.MaxValue)
+        {
+            throw new ArgumentException("stream id exceeds 255 bytes", nameof(message));
+        }
+
         var idBytes = Encoding.ASCII.GetBytes(message.Id);
         var dstBytes = Encoding.UTF8.GetBytes(message.Destination);
         var origBytes = string.IsNullOrEmpty(message.Originator)
@@ -133,6 +161,7 @@ public static byte[] Encode(BackhaulMessage message)
         if (sourceRouteBytes.Length > 0) flags |= Flags.HasSourceRoute;
         if (traversedBytes.Length > 0) flags |= Flags.HasTraversedHops;
         if (hasFragment) flags |= Flags.HasFragment;
+        if (hasStream) flags |= Flags.HasStream;
 
         var size = 1 + 2 + IdLength
             + (message.Salt.HasValue ? 8 : 0)
@@ -144,6 +173,7 @@ public static byte[] Encode(BackhaulMessage message)
             + sourceRouteBytes.Length
             + traversedBytes.Length
             + (hasFragment ? IdLength + 2 + 2 : 0)
+            + (hasStream ? 1 + streamIdBytes.Length + 4 + 4 : 0)
             + headerBytes.Length
             + 4 + message.Payload.Length;
 
@@ -216,6 +246,17 @@ public static byte[] Encode(BackhaulMessage message)
             offset += 2;
         }
 
+        if (hasStream)
+        {
+            buffer[offset++] = (byte)streamIdBytes.Length;
+            streamIdBytes.CopyTo(buffer.AsSpan(offset));
+            offset += streamIdBytes.Length;
+            BinaryPrimitives.WriteUInt32LittleEndian(buffer.AsSpan(offset, 4), message.StreamSeq!.Value);
+            offset += 4;
+            BinaryPrimitives.WriteUInt32LittleEndian(buffer.AsSpan(offset, 4), message.StreamGapTimeoutSeconds!.Value);
+            offset += 4;
+        }
+
         headerBytes.CopyTo(buffer.AsSpan(offset));
         offset += headerBytes.Length;
 
@@ -313,6 +354,20 @@ public static BackhaulMessage Decode(ReadOnlySpan<byte> buffer)
             offset += 2;
         }
 
+        string? streamId = null;
+        uint? streamSeq = null;
+        uint? streamGapTimeout = null;
+        if ((flags & Flags.HasStream) != 0)
+        {
+            var sidLen = buffer[offset++];
+            streamId = Encoding.UTF8.GetString(buffer.Slice(offset, sidLen));
+            offset += sidLen;
+            streamSeq = BinaryPrimitives.ReadUInt32LittleEndian(buffer.Slice(offset, 4));
+            offset += 4;
+            streamGapTimeout = BinaryPrimitives.ReadUInt32LittleEndian(buffer.Slice(offset, 4));
+            offset += 4;
+        }
+
         IReadOnlyDictionary<string, string>? headers = null;
         if ((flags & Flags.HasHeaders) != 0)
         {
@@ -338,7 +393,7 @@ public static BackhaulMessage Decode(ReadOnlySpan<byte> buffer)
         offset += 4;
         var payload = buffer.Slice(offset, (int)payloadLen).ToArray();
 
-        return new BackhaulMessage(id, destination, salt, ttl, payload, headers, originator, linkSource, floodHops, sourceRoute, traversedHops, masterId, fragmentIndex, fragmentTotal);
+        return new BackhaulMessage(id, destination, salt, ttl, payload, headers, originator, linkSource, floodHops, sourceRoute, traversedHops, masterId, fragmentIndex, fragmentTotal, streamId, streamSeq, streamGapTimeout);
     }
 
     private static byte[] EncodeHeaders(IReadOnlyDictionary<string, string> headers)

diff --git a/src/dapps/dapps.client/DappsProtocolClient.cs b/src/dapps/dapps.client/DappsProtocolClient.cs
@@ -87,7 +87,10 @@ public async Task<bool> OfferMessageAsync(
         string? originator = null,
         string? masterId = null,
         int? fragmentIndex = null,
-        int? fragmentTotal = null)
+        int? fragmentTotal = null,
+        string? streamId = null,
+        uint? streamSeq = null,
+        uint? streamGapTimeoutSeconds = null)
     {
         if (format != DappsMessage.MessageFormat.Plain)
         {
@@ -132,6 +135,21 @@ public async Task<bool> OfferMessageAsync(
         {
             sb.Append($" mid={masterId} frag={fragmentIndex}/{fragmentTotal}");
         }
+        // Opt-in ordering keys. All three travel together; the receiver's
+        // IHaveValidator rejects a partial set. Belt-and-braces sender-
+        // side: catch the malformed envelope before it reaches the wire.
+        var hasStream = !string.IsNullOrEmpty(streamId)
+            && streamSeq.HasValue && streamGapTimeoutSeconds.HasValue;
+        if (!hasStream
+            && (!string.IsNullOrEmpty(streamId) || streamSeq.HasValue || streamGapTimeoutSeconds.HasValue))
+        {
+            throw new ArgumentException(
+                "streamId, streamSeq, streamGapTimeoutSeconds must all be set together (opt-in ordering) or all be null");
+        }
+        if (hasStream)
+        {
+            sb.Append($" sid={streamId} sn={streamSeq} gt={streamGapTimeoutSeconds}");
+        }
         sb.Append('\n');
 
         await stream.WriteAsync(Encoding.UTF8.GetBytes(sb.ToString()), ct);
@@ -244,7 +262,10 @@ public sealed record PolledMessage(
         string? Originator,
         string? MasterId,
         int? FragmentIndex,
-        int? FragmentTotal);
+        int? FragmentTotal,
+        string? StreamId,
+        uint? StreamSeq,
+        uint? StreamGapTimeoutSeconds);
 
     /// <summary>
     /// Plan F3 - reverse forwarding from the client side. Send
@@ -340,7 +361,10 @@ public async IAsyncEnumerable<PolledMessage> PollAsync(
                 Originator: parsed.Originator,
                 MasterId: parsed.MasterId,
                 FragmentIndex: parsed.FragmentIndex,
-                FragmentTotal: parsed.FragmentTotal);
+                FragmentTotal: parsed.FragmentTotal,
+                StreamId: parsed.StreamId,
+                StreamSeq: parsed.StreamSeq,
+                StreamGapTimeoutSeconds: parsed.StreamGapTimeoutSeconds);
         }
     }
 
@@ -372,7 +396,7 @@ private async Task ReadExactlyAsync(byte[] buffer, CancellationToken ct)
     /// minimum fields aren't present.</summary>
     private static (bool Ok, ParsedOffer? Offer) TryParseOffer(string line)
     {
-        // line: "ihave <id> len=N fmt=p dst=… [s=…] [ttl=…] [src=…] [mid=… frag=N/M] …"
+        // line: "ihave <id> len=N fmt=p dst=… [s=…] [ttl=…] [src=…] [mid=… frag=N/M] [sid=… sn=… gt=…] …"
         var parts = line.Split(' ');
         if (parts.Length < 4 || parts[0] != "ihave") return (false, null);
         var id = parts[1];
@@ -384,6 +408,9 @@ private static (bool Ok, ParsedOffer? Offer) TryParseOffer(string line)
         string? masterId = null;
         int? fragIndex = null;
         int? fragTotal = null;
+        string? streamId = null;
+        uint? streamSeq = null;
+        uint? streamGapTimeout = null;
         for (var i = 2; i < parts.Length; i++)
         {
             var kv = parts[i];
@@ -409,15 +436,19 @@ private static (bool Ok, ParsedOffer? Offer) TryParseOffer(string line)
                         fragIndex = fn; fragTotal = fm;
                     }
                     break;
+                case "sid": streamId = value; break;
+                case "sn": if (uint.TryParse(value, out var snv)) streamSeq = snv; break;
+                case "gt": if (uint.TryParse(value, out var gtv)) streamGapTimeout = gtv; break;
             }
         }
-        if (destination is null || len is null) return (false, new ParsedOffer(id, "", 0, null, null, null, null, null, null));
-        return (true, new ParsedOffer(id, destination, len.Value, salt, ttl, originator, masterId, fragIndex, fragTotal));
+        if (destination is null || len is null) return (false, new ParsedOffer(id, "", 0, null, null, null, null, null, null, null, null, null));
+        return (true, new ParsedOffer(id, destination, len.Value, salt, ttl, originator, masterId, fragIndex, fragTotal, streamId, streamSeq, streamGapTimeout));
     }
 
     private sealed record ParsedOffer(
         string Id, string Destination, int Length, long? Salt, int? Ttl,
-        string? Originator, string? MasterId, int? FragmentIndex, int? FragmentTotal);
+        string? Originator, string? MasterId, int? FragmentIndex, int? FragmentTotal,
+        string? StreamId, uint? StreamSeq, uint? StreamGapTimeoutSeconds);
 
     /// <summary>
     /// Reads a line terminated by <c>\n</c>, <c>\r</c>, or <c>\r\n</c>.