feat: reply-on-connection — daemon --auto-answer + pilotctl --reply-on-conn (always-safe fallback)

Author: Alex Godoroja

CHANGELOG.md

@@ -7,6 +7,42 @@ project uses [Semantic Versioning](https://semver.org/).
 Detailed per-release notes are on the
 [GitHub Releases page](https://github.com/TeoSlayer/pilotprotocol/releases).
 
+## [Unreleased]
+
+### Added
+- **Reply-on-connection** — answers ride back on the connection the sender
+  opened, eliminating the dial-back for opted-in requests. Two halves, each
+  independently rollable and backwards-compatible:
+  - **Daemon `--auto-answer <url>` (SPECIALISED — service/directory agents
+    ONLY; regular nodes must not set it).** When set, a `TypeAutoAnswer` request
+    is dispatched to `<url>` (the local responder's `/dispatch`) and the reply is
+    written back on the same connection, then the connection closes after exactly
+    one request + one reply. Plain requests are untouched and still use the inbox
+    path, so enabling this never changes how the node serves ordinary senders.
+  - **`pilotctl send-message --reply-on-conn`** (env `PILOT_REPLY_ON_CONN=1`) —
+    sends the request as `TypeAutoAnswer` and reads the reply off the connection
+    into `~/.pilot/inbox/` (same format as a dial-back reply), with no `--wait`
+    and no dial-back. **Always safe — never worse than a plain send:** against an
+    `--auto-answer` agent the reply comes on the connection; against any other
+    agent it falls back to a normal dial-back (an updated agent saved it; an
+    old/stock agent acks the frame as `UNKNOWN`, which triggers an automatic
+    resend as plain `TEXT`). The on-connection benefit is gained only against
+    `--auto-answer` agents.
+- The reply lands as an ordinary inbox message — `pilotctl inbox` reads it
+  unchanged.
+
+### Notes / limitations
+- Reply-on-connection helps senders that **run a client which reads the reply**
+  (updated `pilotctl`/SDK). It does not change the ceiling for transient clients
+  with no daemon/inbox.
+- The new loop is gated on the **request type**, so it is safe to enable
+  `--auto-answer` on a live directory agent without disturbing current traffic.
+- **Sender-side fan-out is out of scope:** a single requester daemon opening
+  *many* simultaneous dials (dozens) can see sender-side `dial` failures under
+  NAT/relay pressure — a property of overloading one sender daemon, not of the
+  receiver (which dropped zero replies in testing). Regular senders issue a few
+  dials and are unaffected.
+
 ## [1.10.7] - 2026-06-06
 
 ### Fixed

cmd/daemon/main.go

@@ -6,8 +6,11 @@ import (
 	"context"
 	"flag"
 	"fmt"
+	"io"
 	"log"
 	"log/slog"
+	"net/http"
+	"net/url"
 	"os"
 	"os/signal"
 	"strconv"
@@ -35,6 +38,34 @@ import (
 
 var version = "dev"
 
+// newAutoAnswerReplyHook builds the dataexchange ReplyHook for --auto-answer: it
+// dispatches each request to the responder's HTTP endpoint and returns the body
+// as a TEXT reply. The read is bounded at the data-exchange frame limit — the
+// same effective ceiling the normal inbox→dial-back path hits — so
+// reply-on-connection delivers byte-for-byte what a dial-back would, never
+// truncating a large reply. Returns ok=false (no reply written) on transport
+// error or an empty body, which lets the sender fall back to a dial-back.
+func newAutoAnswerReplyHook(endpoint string, hc *http.Client) func(reqType uint32, payload []byte) (uint32, []byte, bool) {
+	return func(reqType uint32, payload []byte) (uint32, []byte, bool) {
+		resp, err := hc.Get(endpoint + "?message=" + url.QueryEscape(string(payload)))
+		if err != nil {
+			return 0, nil, false
+		}
+		defer resp.Body.Close()
+		// Only a 200 carries a real reply. 204 (dropped by loop-prevention) and
+		// any error status (e.g. 400 "missing message") must NOT be delivered as
+		// a reply — returning ok=false lets the sender fall back to a dial-back.
+		if resp.StatusCode != http.StatusOK {
+			return 0, nil, false
+		}
+		body, _ := io.ReadAll(io.LimitReader(resp.Body, dataexchange.MaxFrameSize))
+		if len(body) == 0 {
+			return 0, nil, false
+		}
+		return dataexchange.TypeText, body, true
+	}
+}
+
 func main() {
 	configPath := flag.String("config", "", "path to config file (JSON)")
 	registryDefault := "34.71.57.205:9000"
@@ -80,6 +111,7 @@ func main() {
 	hostname := flag.String("hostname", "", "hostname for discovery (lowercase alphanumeric + hyphens, max 63 chars)")
 	noEcho := flag.Bool("no-echo", false, "disable built-in echo service (port 7)")
 	noDataExchange := flag.Bool("no-dataexchange", false, "disable built-in data exchange service (port 1001)")
+	autoAnswer := flag.String("auto-answer", "", "SPECIALISED — directory/service agents ONLY (e.g. list-agents); regular nodes MUST leave this empty. Enables reply-on-connection: an opted-in request (TypeAutoAnswer, from `send-message --reply-on-conn`) is dispatched to this HTTP endpoint and the reply is written back on the SAME connection the sender opened, then the connection closes (exactly 1 request + 1 response — no loop). This lets a sender receive the answer in its inbox with no dial-back, even when it is unreachable. Plain requests are unaffected and still flow through the normal inbox path, so setting this never changes how this node serves ordinary senders. A regular node has no reason to hold connections open to generate replies and should never set this.")
 	dataExchangeB64 := flag.Bool("dataexchange-b64", false, "include raw base64 payload (`data_b64`) alongside `data` in inbox messages — needed only for binary payloads (e.g. zlib-compressed envelopes)")
 	noEventStream := flag.Bool("no-eventstream", false, "disable built-in event stream service (port 1002)")
 	webhookURL := flag.String("webhook", "", "HTTP(S) endpoint for event notifications (empty = disabled)")
@@ -223,9 +255,21 @@ func main() {
 	}
 
 	if !*noDataExchange {
-		if err := rt.Register(dataexchange.NewService(dataexchange.ServiceConfig{
-			IncludeBase64: *dataExchangeB64,
-		})); err != nil {
+		dxCfg := dataexchange.ServiceConfig{IncludeBase64: *dataExchangeB64}
+		if *autoAnswer != "" {
+			ru := *autoAnswer
+			// Timeout chain (must be strictly increasing so no stage cuts off
+			// the one beneath it): responder fetch (35s, the service call) <
+			// this ReplyHook HTTP client (38s, the /dispatch round trip that
+			// CONTAINS that fetch) < service autoAnswerWindow watchdog (40s) <=
+			// sender on-connection read window (45s). At 38s this client gives
+			// the responder's 35s fetch room to finish before the daemon gives up.
+			hc := &http.Client{Timeout: 38 * time.Second}
+			dxCfg.AutoAnswer = true
+			dxCfg.ReplyHook = newAutoAnswerReplyHook(ru, hc)
+			slog.Info("dataexchange auto-answer (reply-on-connection) enabled", "endpoint", ru)
+		}
+		if err := rt.Register(dataexchange.NewService(dxCfg)); err != nil {
 			log.Fatalf("register dataexchange: %v", err)
 		}
 	}

cmd/daemon/zz_autoanswer_test.go

@@ -0,0 +1,71 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package main
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/pilot-protocol/dataexchange"
+)
+
+// TestAutoAnswerReplyHook_LargeBodyNotTruncated proves the reply-on-connection
+// read is NOT capped at the old 1 MiB limit: a body well over 1 MiB is returned
+// in full, so a large directory reply is delivered byte-for-byte like a
+// dial-back would.
+func TestAutoAnswerReplyHook_LargeBodyNotTruncated(t *testing.T) {
+	const size = 3 << 20 // 3 MiB, well past the retired 1 MiB cap
+	big := strings.Repeat("x", size)
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Query().Get("message") == "" {
+			t.Error("reply hook did not forward the message query param")
+		}
+		_, _ = w.Write([]byte(big))
+	}))
+	defer srv.Close()
+
+	hook := newAutoAnswerReplyHook(srv.URL, &http.Client{Timeout: 10 * time.Second})
+	rt, body, ok := hook(dataexchange.TypeAutoAnswer, []byte("/data"))
+	if !ok {
+		t.Fatal("hook returned ok=false for a valid large body")
+	}
+	if rt != dataexchange.TypeText {
+		t.Errorf("reply type = %d, want TypeText", rt)
+	}
+	if len(body) != size {
+		t.Fatalf("reply truncated: got %d bytes, want %d", len(body), size)
+	}
+}
+
+// TestAutoAnswerReplyHook_EmptyAndError: an empty body or a transport error
+// yields ok=false (no reply on the connection → the sender falls back).
+func TestAutoAnswerReplyHook_EmptyAndError(t *testing.T) {
+	// Empty body (e.g. /dispatch 204 for a dropped message).
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		w.WriteHeader(http.StatusNoContent)
+	}))
+	hook := newAutoAnswerReplyHook(srv.URL, &http.Client{Timeout: 5 * time.Second})
+	if _, _, ok := hook(dataexchange.TypeAutoAnswer, []byte("prose")); ok {
+		t.Error("empty body should yield ok=false")
+	}
+	srv.Close()
+
+	// Transport error (server gone).
+	if _, _, ok := hook(dataexchange.TypeAutoAnswer, []byte("/data")); ok {
+		t.Error("transport error should yield ok=false")
+	}
+
+	// Non-200 with a non-empty body (e.g. 400 "missing message") must NOT be
+	// delivered as a reply.
+	errSrv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		http.Error(w, "missing message", http.StatusBadRequest)
+	}))
+	defer errSrv.Close()
+	h2 := newAutoAnswerReplyHook(errSrv.URL, &http.Client{Timeout: 5 * time.Second})
+	if _, body, ok := h2(dataexchange.TypeAutoAnswer, []byte("/data")); ok {
+		t.Errorf("a 400 response must yield ok=false, got body=%q", body)
+	}
+}

cmd/pilotctl/main.go

@@ -3649,6 +3649,39 @@ func cmdSendMessage(args []string) {
 	}
 	msgType := flagString(flags, "type", "text")
 
+	// --reply-on-conn opts this request into reply-on-connection: send it as a
+	// TypeAutoAnswer frame and read the reply the receiver writes back on the
+	// SAME connection (into the inbox), instead of relying on a dial-back.
+	//
+	// ALWAYS SAFE — never worse than a plain send. Against an --auto-answer agent
+	// the reply rides back on this connection (so it works even when this sender
+	// is NAT'd / has no public port / is transient). Against ANY other agent it
+	// transparently falls back to a normal dial-back: an updated agent saved the
+	// request, and an old/stock agent (which acks the frame as UNKNOWN) triggers
+	// an automatic resend as plain TEXT — either way the reply is dial-backed
+	// exactly as for a normal send. Current senders that don't set it are
+	// untouched. Env: PILOT_REPLY_ON_CONN=1.
+	replyOnConn := flagBool(flags, "reply-on-conn") || os.Getenv("PILOT_REPLY_ON_CONN") != ""
+	if replyOnConn {
+		// reply-on-connection carries the request as a text query, so true binary
+		// would be lossily url-escaped — reject it rather than diverge silently
+		// from a plain --type binary send.
+		if msgType == "binary" {
+			fatalCode("invalid_argument", "--reply-on-conn does not support --type binary")
+		}
+		// --trace sends a TypeTrace frame for timing and is incompatible with
+		// reply-on-connection; trace wins, so disable reply-on-conn handling to
+		// avoid a spurious fallback resend (double inbox entry).
+		if traceTime {
+			replyOnConn = false
+		}
+		// An --auto-answer receiver closes after one request+reply, so a reused
+		// connection can't carry a second send. Dial fresh per send instead.
+		if reuseConn {
+			reuseConn = false
+		}
+	}
+
 	// Auto-handshake to peers in the embedded trusted-agents list.
 	// Best-effort: warns on stderr and continues if handshake fails.
 	maybeAutoHandshake(d, target, flagBool(flags, "no-auto-handshake"))
@@ -3687,12 +3720,16 @@ func cmdSendMessage(args []string) {
 			sentAtNs, sendErr = cl.SendTrace(innerType, []byte(data))
 		} else {
 			sendStart := time.Now()
-			switch msgType {
-			case "text":
+			switch {
+			case replyOnConn:
+				// Opt in to reply-on-connection (TypeAutoAnswer); the reply is
+				// read off this connection below and saved to the inbox.
+				sendErr = cl.SendAutoAnswer(data)
+			case msgType == "text":
 				sendErr = cl.SendText(data)
-			case "json":
+			case msgType == "json":
 				sendErr = cl.SendJSON([]byte(data))
-			case "binary":
+			case msgType == "binary":
 				sendErr = cl.SendBinary([]byte(data))
 			}
 			sentAtNs = sendStart.UnixNano()
@@ -3713,7 +3750,62 @@ func cmdSendMessage(args []string) {
 			"reused": reused,
 		}
 		if ack != nil {
-			r["ack"] = string(ack.Payload)
+			ackStr := string(ack.Payload)
+			r["ack"] = ackStr
+			// reply-on-connection delivery + GUARANTEED fallback. With
+			// --reply-on-conn we sent a TypeAutoAnswer request; decide whether a
+			// reply is (or will be) delivered, and if not, fall back so the flag
+			// is ALWAYS at least as good as a plain send:
+			//   * ACK+REPLY        → an --auto-answer agent's reply follows on
+			//                        THIS connection; read it into the inbox.
+			//   * ack names the AUTOANSWER type → an updated, non-auto-answering
+			//                        agent saved the request; it will dial back.
+			//   * anything else (an OLD/stock daemon acks UNKNOWN and does NOT
+			//                        save it; or the on-connection read failed)
+			//                        → resend as plain TEXT so ANY daemon saves
+			//                        it and the responder dial-backs the reply.
+			if replyOnConn {
+				delivered := false
+				switch replyOnConnOutcome(ackStr) {
+				case replyOnConnRead:
+					// Read window is tied to the receiver's autoAnswerWindow (40s)
+					// plus margin — NOT to --wait. --wait bounds the later inbox
+					// poll; capping the on-connection read by a small --wait would
+					// abandon a reply the receiver has already committed to send
+					// (it skipped the inbox) and force a needless resend.
+					_ = cl.SetReadDeadline(time.Now().Add(45 * time.Second))
+					reply, rerr := cl.Recv()
+					_ = cl.SetReadDeadline(time.Time{})
+					if rerr == nil && reply != nil && len(reply.Payload) > 0 {
+						if p, serr := saveReplyToInbox(target.String(), reply.Type, reply.Payload); serr == nil {
+							r["reply_bytes"] = len(reply.Payload)
+							r["reply_inbox"] = p
+							delivered = true
+						} else {
+							r["reply_error"] = serr.Error()
+						}
+					} else if rerr != nil {
+						r["reply_error"] = rerr.Error()
+					}
+				case replyOnConnDelivered:
+					// Understood by an updated daemon and saved for dial-back.
+					delivered = true
+				}
+				if !delivered {
+					// Old/stock receiver (or a lost on-connection reply): resend
+					// as a plain TEXT message — exactly the pre-feature path — so
+					// the receiver saves it and the reply is dial-backed. This is
+					// what makes --reply-on-conn never worse than a plain send.
+					if rc, derr := dataexchange.Dial(d, target); derr == nil {
+						_ = rc.SendText(data)
+						_, _ = rc.Recv() // consume ack
+						_ = rc.Close()
+						r["fallback"] = "dialback"
+					} else {
+						r["fallback_error"] = derr.Error()
+					}
+				}
+			}
 		}
 		if traceTime {
 			r["total_ms"] = float64(time.Duration(ackRecvAtNs-sentAtNs).Microseconds()) / 1000.0
@@ -5273,9 +5365,70 @@ func cmdReceived(args []string) {
 	fmt.Printf("\ntotal: %d\n", len(files))
 }
 
+// reply-on-connection sender outcomes, decided from the receiver's ack.
+const (
+	replyOnConnRead      = "read"      // --auto-answer agent: read the reply on this connection
+	replyOnConnDelivered = "delivered" // updated agent understood + saved it: a dial-back is coming
+	replyOnConnResend    = "resend"    // old/stock agent did not understand it: resend as plain TEXT
+)
+
+// replyOnConnOutcome decides what a --reply-on-conn sender must do, from the
+// receiver's ack payload. This is the core of the "always safe" guarantee: only
+// an "ACK+REPLY" ack carries a reply on the connection; only a daemon that knows
+// the AUTOANSWER type echoes it in the ack (so it understood and saved the
+// request for dial-back); everything else is an old/stock daemon that acked the
+// frame as UNKNOWN without saving it, so the sender must resend as plain TEXT.
+func replyOnConnOutcome(ackPayload string) string {
+	switch {
+	case strings.HasPrefix(ackPayload, "ACK+REPLY"):
+		return replyOnConnRead
+	case strings.Contains(ackPayload, dataexchange.TypeName(dataexchange.TypeAutoAnswer)):
+		return replyOnConnDelivered
+	default:
+		return replyOnConnResend
+	}
+}
+
+// saveReplyToInbox writes a reply-on-connection frame to ~/.pilot/inbox/ in the
+// exact shape the daemon's dataexchange service uses, so it is indistinguishable
+// from a dial-back reply to `pilotctl inbox` and any inbox tooling. Returns the
+// written path. This is how a reply-aware sender lands the answer in its own
+// inbox without ever being dialed back.
+func saveReplyToInbox(from string, frameType uint32, payload []byte) (string, error) {
+	home, err := os.UserHomeDir()
+	if err != nil {
+		return "", err
+	}
+	dir := filepath.Join(home, ".pilot", "inbox")
+	if err := os.MkdirAll(dir, 0700); err != nil {
+		return "", err
+	}
+	ts := time.Now()
+	msg := map[string]interface{}{
+		"type":        dataexchange.TypeName(frameType),
+		"from":        from,
+		"data":        string(payload),
+		"bytes":       len(payload),
+		"received_at": ts.Format(time.RFC3339Nano),
+	}
+	data, err := json.Marshal(msg)
+	if err != nil {
+		return "", err
+	}
+	// Nanosecond stamp + pid keep the filename unique without the daemon's
+	// shared atomic sequence counter.
+	filename := fmt.Sprintf("%s-%s-%06d.json", dataexchange.TypeName(frameType),
+		ts.Format("20060102-150405.000000000"), os.Getpid()%1000000)
+	destPath := filepath.Join(dir, filename)
+	if err := os.WriteFile(destPath, data, 0600); err != nil {
+		return "", err
+	}
+	return destPath, nil
+}
+
 // cmdInbox lists or clears messages received via data exchange (port 1001).
 // waitForInboxReply polls ~/.pilot/inbox/ until a JSON file arrives that is
-// newer than cutoff and (if agentHint is non-empty) has a matching "agent"
+// newer than cutoff and (if agentHint is non-empty) has a matching "from"
 // field. Returns the parsed message or an error on timeout.
 func waitForInboxReply(agentHint string, cutoff time.Time, timeout time.Duration) (map[string]interface{}, error) {
 	home, err := os.UserHomeDir()