Fix inbox byte-cap wipe; add request/reply IDs and dedupe

CHANGELOG.md

@@ -7,6 +7,48 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Fixed
+
+- The inbox byte cap no longer deletes the whole inbox. With the default
+  config (`InboxMaxBytes == 0`, meaning 256 MiB) the evictor compared the
+  inbox size against the raw value 0, so the first time the inbox passed
+  256 MiB every stored message was removed, including replies a waiting
+  `pilotctl send-message --wait` was about to read. Both eviction paths now
+  use the effective cap, evict oldest-first down to 90% of it, and never
+  evict for a message that cannot fit at all.
+- The byte cap is checked against a running total instead of listing and
+  stat'ing the whole inbox on every incoming message. A message that another
+  connection is still writing (admitted, but not yet acknowledged) is never
+  evicted and never counted twice.
+- When the byte cap is full the inbox now evicts the oldest messages to make
+  room (as `InboxMaxBytes` was documented to do) instead of rejecting the
+  new message.
+
+### Added
+
+- Optional request/reply correlation: `Frame.MessageID` and `Frame.ReplyTo`,
+  carried on the wire in a new `TypeTagged` (10) wrapper and recorded in
+  inbox JSON and `message.received` / `file.received` events as
+  `message_id` / `reply_to`. Frames without them are byte-for-byte
+  unchanged. `NewMessageID` and `ValidMessageID` helpers.
+- `Client.Send`, which waits for the ACK and delivers a tagged frame
+  untagged, with `SendResult.Tagged == false`, when the tagged form cannot
+  get through. That happens when the receiver predates `TypeTagged`: Send
+  recognises both the `ERR UNKNOWN(10) ...` answer of dataexchange v0.2.2 and
+  later and the `ACK UNKNOWN(10) <n> bytes` answer of v0.2.1 and older (every
+  stable daemon through v1.13.9), which drop the frame while acknowledging
+  it. It also happens when the header would push the frame over the max
+  frame size.
+- `MaxTaggedOverhead` (293 bytes) and `ErrTaggedFrameTooLarge`: the tagged
+  header counts toward `MaxFrameSize`, and `WriteFrame` refuses a tagged
+  frame that exceeds it, writing nothing, rather than send a frame the
+  receiver would drop the connection over.
+- Receiver-side duplicate suppression: an identical re-delivery of a stored
+  frame that carried a `MessageID` is acknowledged (`... (duplicate)`) but
+  not stored twice (`ServiceConfig.DedupeWindow`, default 10 min).
+  `ServiceConfig.DedupeContentWindow` (off by default) extends this to
+  frames without a `MessageID`.
+
 ### Security / hardening
 
 - Lower the default per-frame cap (`DefaultMaxFrameSize`) from 1 GiB to

README.md

@@ -30,11 +30,14 @@ rt.Register(dataexchange.NewService(dataexchange.ServiceConfig{}))
 
 | File | What it does |
 |---|---|
-| `dataexchange.go` | Wire format: `Frame`, `WriteFrame`, `ReadFrame`, `TraceFrame`, `TypeText/Binary/JSON/File/Trace/Governed`, `TypeName`. |
-| `client.go` | `Client` — `Dial` and send helpers. |
+| `dataexchange.go` | Wire format: `Frame`, `WriteFrame`, `ReadFrame`, `TraceFrame`, `TypeText/Binary/JSON/File/Trace/Governed/Tagged`, `TypeName`. |
+| `tagged.go` | Optional request/reply correlation on the wire: `Frame.MessageID` / `Frame.ReplyTo` (`TypeTagged`), `NewMessageID`, `ValidMessageID`. |
+| `client.go` | `Client` — `Dial`, `Send` (waits for the ACK, falls back for older receivers) and send helpers. |
 | `governed.go` | Signed decision envelope, receiver-side verifier, and enforceable transport constraints. |
 | `server.go` | `Server` — accept loop and handler dispatch. |
 | `service.go` | `*Service` — `coreapi.Service` adapter. Build tag `!no_dataexchange`. |
+| `inbox_budget.go` | Inbox caps: running byte total, oldest-first eviction to 90% of the byte cap, file-count cap. |
+| `dedupe.go` | Receiver-side suppression of identical re-deliveries. |
 | `service_disabled.go` | Stub `*Service` for `-tags no_dataexchange` builds. |
 
 ## Wire format
@@ -45,9 +48,126 @@ rt.Register(dataexchange.NewService(dataexchange.ServiceConfig{}))
 
 For `TypeFile` the payload is prefixed with `[2-byte name length][name bytes]`.
 For `TypeTrace` the payload is `[4-byte inner_type][8-byte sent_at_ns][inner payload]`.
+For `TypeTagged` (10) the payload is
+`[2-byte header_len][header: JSON object][4-byte inner_type][inner payload]`;
+see [Request/reply correlation](#requestreply-correlation-for-service-authors).
 
 Max frame size: 64 MiB by default (configurable at process start with
-`PILOT_DATAEXCHANGE_MAX_FRAME` within its documented safe range).
+`PILOT_DATAEXCHANGE_MAX_FRAME` within its documented safe range). The limit
+applies to the whole payload after the 8-byte header, so for `TypeTagged` it
+includes the correlation header (see below).
+
+## Inbox
+
+Text, JSON and binary messages are written to `~/.pilot/inbox/` as one JSON
+file each:
+
+```json
+{"type":"TEXT","from":"0:0000.0002.BBE4","bytes":5,"received_at":"2026-09-24T10:00:00.123456789Z",
+ "data":"hello","data_encoding":"utf8","message_id":"4f1c…","reply_to":"9b2e…"}
+```
+
+`message_id` and `reply_to` appear only when the sender supplied them (see
+below). Binary or non-UTF-8 payloads are stored as `data_b64` with
+`"data_encoding":"base64"`.
+
+Two caps bound the inbox, and both evict the **oldest** messages first:
+
+- `ServiceConfig.InboxMaxFiles` (default 10000), checked every 64 messages.
+- `ServiceConfig.InboxMaxBytes` (default 256 MiB), checked before every write
+  against a running byte total (no directory scan per message). When a new
+  message would not fit, the oldest messages are removed until the inbox plus
+  the new message is at or below 90% of the cap, so recent replies survive
+  and the next messages do not each trigger another eviction. A message
+  that another connection is still writing is never evicted. A single
+  message larger than the cap is rejected without evicting anything. A
+  negative value disables the byte cap.
+
+## Request/reply correlation (for service authors)
+
+A data-exchange message is fire-and-forget: the sender gets an ACK that the
+peer stored it, and any answer arrives later as a separate message in the
+sender's inbox. To let a requester tell *which* request an answer belongs to,
+a frame can carry two optional fields:
+
+| Field (Go) | Inbox / event key | Meaning |
+|---|---|---|
+| `Frame.MessageID` | `message_id` | An ID the sender chose for this message. Use a fresh random value per message (`NewMessageID()`, 32 hex chars). |
+| `Frame.ReplyTo` | `reply_to` | The `message_id` of the message this one answers. |
+
+IDs are 1–128 characters of `A-Z a-z 0-9 - _ . :`. They are correlation
+metadata only: they are not authenticated (a governed frame's signature does
+not cover them), so never use them for authorization.
+
+**If you run a service** (an agent that answers requests): when a request's
+inbox record (or `message.received` event) has a `message_id`, send your
+answer with `ReplyTo` set to that value. If you answer in several messages,
+set it on each. Give each answer its own `MessageID` too if you want the
+requester's daemon to drop duplicate copies of it (for example when you
+retry a send or it arrives over two network paths). Requests without a
+`message_id` come from older clients: answer them as before.
+
+```go
+c, err := dataexchange.Dial(drv, requester)
+if err != nil { return err }
+defer c.Close()
+res, err := c.Send(&dataexchange.Frame{
+    Type:      dataexchange.TypeJSON,
+    Payload:   answer,
+    MessageID: dataexchange.NewMessageID(),
+    ReplyTo:   request.MessageID, // the request's "message_id"
+})
+```
+
+**If you send requests**: generate a `MessageID`, send with `Client.Send`,
+then wait for an inbox record from that peer whose `reply_to` equals it.
+Services that predate this field answer without `reply_to`; for those, fall
+back to matching on the sender alone.
+
+**Compatibility.** Frames without either field use the original wire format,
+byte for byte. A frame with either field travels as `TypeTagged`. A receiver
+that predates it stores nothing and answers in one of two ways:
+
+- dataexchange v0.2.1 and older (every stable pilot daemon through v1.13.9)
+  has no case for an unknown frame type. It drops the frame and still
+  answers `ACK UNKNOWN(10) <n> bytes`.
+- dataexchange v0.2.2 and later answers `ERR UNKNOWN(10) save failed: ...`.
+
+`Client.Send` recognises both answers, re-sends the same frame without the
+fields on the same connection, and reports `SendResult.Tagged == false`, so a
+new sender can always talk to an old receiver and the message is stored
+exactly once. A raw `WriteFrame` of a tagged frame does not fall back:
+against a v0.2.1-or-older receiver the message is silently lost, so use
+`Client.Send`.
+
+**Size.** The tagged header adds up to `MaxTaggedOverhead` (293) bytes, and
+the whole tagged frame must fit in the max frame size. A payload within 293
+bytes of the limit may fit only untagged. `Client.Send` then sends it
+untagged (`SendResult.Tagged == false`), and a raw `WriteFrame` returns
+`ErrTaggedFrameTooLarge` without writing anything.
+
+Whenever `Tagged` is false, the reply cannot carry `reply_to`; match it on
+the sender alone. The fields can accompany text, JSON, binary, file,
+governed and trace frames. On `TypeFileStream` frames they are ignored.
+
+Wire layout of a tagged frame, for other languages:
+
+```
+[4-byte type = 10][4-byte length][payload]
+payload = [2-byte header_len, 1..1024][header_len bytes: JSON object]
+          [4-byte inner_type][inner payload, exactly as the inner type sends it]
+header  = {"message_id":"…","reply_to":"…"}   (both optional; unknown keys are ignored)
+length  = 2 + header_len + 4 + len(inner payload), at most the max frame size
+```
+
+**Duplicate deliveries.** The receiver remembers every stored frame that
+carried a `MessageID` for `ServiceConfig.DedupeWindow` (default 10 minutes).
+An identical re-delivery — same sender, `MessageID`, `ReplyTo`, type,
+filename and bytes — is acknowledged with the original ACK plus
+` (duplicate)` (`SendResult.Duplicate`) and is not stored again. Frames
+without a `MessageID` are always stored, as before, unless the operator sets
+`ServiceConfig.DedupeContentWindow` (off by default) to also drop
+byte-identical copies from the same sender inside a short window.
 
 ## Governed delivery
 

client.go

@@ -3,8 +3,10 @@
 package dataexchange
 
 import (
+	"errors"
 	"fmt"
 	"io"
+	"strings"
 	"time"
 
 	"github.com/pilot-protocol/common/decision"
@@ -26,6 +28,135 @@ func Dial(d *driver.Driver, addr protocol.Addr) (*Client, error) {
 	return &Client{conn: conn}, nil
 }
 
+// ErrRejected is wrapped by the error Client.Send returns when the receiver
+// answered with an "ERR ..." acknowledgement instead of storing the frame.
+var ErrRejected = errors.New("dataexchange: receiver rejected frame")
+
+// SendResult reports what the receiver did with a frame sent by Client.Send.
+type SendResult struct {
+	// Ack is the receiver's acknowledgement frame, normally a TypeText
+	// "ACK <TYPE> <n> bytes" (or "ERR ..." on rejection).
+	Ack *Frame
+	// Tagged is true when the frame's MessageID/ReplyTo reached the
+	// receiver. It is false when the frame carried neither, and also when
+	// Send delivered the frame without them: because the receiver predates
+	// TypeTagged, or because the tagged header would have pushed the frame
+	// over MaxFrameSize (see ErrTaggedFrameTooLarge). A requester that sees
+	// Tagged == false cannot expect a reply carrying reply_to, and should
+	// match the answer on the sender alone.
+	Tagged bool
+	// Duplicate is true when the receiver recognised the frame as an
+	// identical re-delivery of one it had already stored, and did not store
+	// it a second time.
+	Duplicate bool
+}
+
+// duplicateAckSuffix is appended to the ACK of a frame the receiver
+// suppressed as a duplicate. The ACK still starts with "ACK ", so senders
+// that only check the prefix keep treating it as success.
+const duplicateAckSuffix = " (duplicate)"
+
+// untaggedReceiverAckPrefixes are the two ways a receiver that predates
+// TypeTagged answers a tagged frame. Neither kind stores anything:
+//
+//   - dataexchange v0.1.0 through v0.2.1 (and the pseudo-versions between
+//     them) have no case for an unknown frame type. The frame matches
+//     nothing, no save is attempted, and the receiver answers
+//     "ACK UNKNOWN(10) <n> bytes": a success ACK for a frame it dropped.
+//   - dataexchange v0.2.2 and later, before TypeTagged, reject an unknown
+//     type and answer "ERR UNKNOWN(10) save failed: unsupported frame type 10".
+//     A RequireGoverned receiver answers "ERR UNKNOWN(10) save failed:
+//     unsigned legacy frame ..." instead, also without storing it.
+//
+// A receiver that knows TypeTagged unwraps the frame in ReadFrame and
+// acknowledges the inner type (which can never be TypeTagged), so its ACK
+// never starts with either prefix.
+var untaggedReceiverAckPrefixes = [...]string{
+	fmt.Sprintf("ACK UNKNOWN(%d) ", TypeTagged),
+	fmt.Sprintf("ERR UNKNOWN(%d) ", TypeTagged),
+}
+
+// receiverPredatesTagged reports whether ack is a pre-TypeTagged receiver's
+// answer to a tagged frame, which means the frame was not stored.
+func receiverPredatesTagged(ack *Frame) bool {
+	if ack.Type != TypeText {
+		return false
+	}
+	for _, prefix := range untaggedReceiverAckPrefixes {
+		if strings.HasPrefix(string(ack.Payload), prefix) {
+			return true
+		}
+	}
+	return false
+}
+
+// Send writes frame, waits for the receiver's acknowledgement and reports
+// the outcome. It is the preferred way to send a frame that carries a
+// MessageID or ReplyTo, because it delivers the frame untagged (and reports
+// SendResult.Tagged == false) in the two cases where the tagged form cannot
+// get through:
+//
+//   - The receiver predates TypeTagged. Send recognises its answer (see
+//     untaggedReceiverAckPrefixes) and re-sends the same frame without the
+//     metadata on the same connection. Such a receiver never stores the
+//     tagged copy, so the message is stored exactly once.
+//   - The tagged header would push the frame over MaxFrameSize. Send writes
+//     the frame untagged in the first place instead of sending a frame the
+//     receiver would drop the connection over.
+//
+// If the receiver answers "ERR ...", Send returns the result together with
+// an error wrapping ErrRejected. Send reads from the connection, so do not
+// use it concurrently with Recv on the same Client.
+func (c *Client) Send(frame *Frame) (*SendResult, error) {
+	return sendAndAwaitAck(c.conn, frame)
+}
+
+// withoutTags returns a copy of frame without MessageID and ReplyTo.
+func withoutTags(frame *Frame) *Frame {
+	plain := *frame
+	plain.MessageID, plain.ReplyTo = "", ""
+	return &plain
+}
+
+func sendAndAwaitAck(rw io.ReadWriter, frame *Frame) (*SendResult, error) {
+	if frame == nil {
+		return nil, fmt.Errorf("dataexchange: frame is required")
+	}
+	tagged := frame.MessageID != "" || frame.ReplyTo != ""
+	err := WriteFrame(rw, frame)
+	if tagged && errors.Is(err, ErrTaggedFrameTooLarge) {
+		// WriteFrame wrote nothing. The frame fits only without its header.
+		tagged = false
+		err = WriteFrame(rw, withoutTags(frame))
+	}
+	if err != nil {
+		return nil, err
+	}
+	ack, err := ReadFrame(rw)
+	if err != nil {
+		return nil, fmt.Errorf("dataexchange: read ack: %w", err)
+	}
+	if tagged && receiverPredatesTagged(ack) {
+		if err := WriteFrame(rw, withoutTags(frame)); err != nil {
+			return nil, err
+		}
+		if ack, err = ReadFrame(rw); err != nil {
+			return nil, fmt.Errorf("dataexchange: read ack: %w", err)
+		}
+		tagged = false
+	}
+	text := string(ack.Payload)
+	result := &SendResult{
+		Ack:       ack,
+		Tagged:    tagged,
+		Duplicate: ack.Type == TypeText && strings.HasPrefix(text, "ACK ") && strings.HasSuffix(text, duplicateAckSuffix),
+	}
+	if ack.Type == TypeText && strings.HasPrefix(text, "ERR ") {
+		return result, fmt.Errorf("%w: %s", ErrRejected, text)
+	}
+	return result, nil
+}
+
 // SendText sends a text frame.
 func (c *Client) SendText(text string) error {
 	return WriteFrame(c.conn, &Frame{Type: TypeText, Payload: []byte(text)})