plane: a call expires on the CALLER's deadline, not the transport's

Every @hanzo turn in Slack answered "the agent hit an error handling that". With
the bridge's new non-silent branch the reason finally surfaced:

  zip: call agents_run_on_behalf at /var/lib/cloud/run/agents.sock:
  zaphttp: read response: i/o timeout

zap-proto/http@v0.3.1 sets readTimeout: 30 * time.Second (client.go:72) on every
dialled transport. An agent turn runs a real model completion and enso spends
40s+ on one — production measured 41,153ms, answered 200 by BOTH `ai` and
`agents`. The work succeeded. The caller had already hung up, and the reply was
written to a socket nobody was reading.

That is the worst shape a timeout can have: the expensive work is done and
billed, the callee logs success, and only the caller reports failure — so every
log you would naturally check says the system is healthy. It is why this
survived a day of looking at a healthy `ai`.

The context budget could not save it. The bridge bounds a turn at 110s on a
detached context; that governs the CALL and never reaches the transport's own
SetReadDeadline, which is wall-clock on the connection and shorter. Two
deadlines for one operation, and the smaller one — the one nobody chose — won.

The fix is a registration, not a patch. zip resolves a scheme through
RegisterTransport (transport.go:109) and its default zap Dial is one line, so
re-register the same scheme with the same dialler plus the knob zap-proto/http
exports for exactly this (SetReadTimeout, client.go:81). The stock Serve is
restated verbatim because re-registering replaces BOTH halves — omitting it
would stop every plugin listening.

15 minutes, matching the host's own plugin-start budget, so there is ONE answer
to "how long may an in-flight plane call take". It is a CEILING, never a floor:
every caller still bounds itself, and a caller that gives itself ten seconds
still gets ten. What changes is that it can no longer be cut off BELOW its own
budget by a default it never saw.

Tests pin the property rather than the wall clock (a unit test cannot hold a
socket for 40s): the ceiling must exceed the longest real caller budget,
re-registration must be idempotent, and networkOf must match zip's rule — it is
copied because zip keeps it unexported, and a drifted copy would hand a unix
path to a tcp dialler.

plane/readdeadline.go

@@ -0,0 +1,103 @@
+package plane
+
+// readdeadline.go widens the plane's RESPONSE-READ deadline so a call that is
+// legitimately slow is not cut off by a transport default.
+//
+// # The bug this fixes
+//
+// Every @hanzo turn in Slack answered "the agent hit an error handling that",
+// and the bridge logged:
+//
+//	zip: call agents_run_on_behalf at /var/lib/cloud/run/agents.sock:
+//	zaphttp: read response: i/o timeout
+//
+// zap-proto/http@v0.3.1 sets `readTimeout: 30 * time.Second` (client.go:72) on
+// every dialled transport. An agent turn runs a real model completion, and enso
+// spends 40s+ on one — measured in production at 41,153ms server-side, answered
+// 200 by both `ai` and `agents`. So the work SUCCEEDED and the caller had already
+// hung up on it. The reply was written to a socket nobody was reading.
+//
+// That is the worst shape a timeout can have: the expensive work is done and
+// paid for, the callee logs success, and only the caller reports failure — so
+// every log you would naturally check says the system is healthy.
+//
+// # Why the context budget did not save it
+//
+// The bridge already bounds a turn at bridgeAgentTimeout (110s) on a detached
+// context. That governs the CALL; it does not reach the transport's own
+// SetReadDeadline, which is wall-clock on the connection and shorter. Two
+// deadlines existed for one operation and the smaller one — the one nobody
+// chose — won.
+//
+// # The fix, and why it is a registration rather than a patch
+//
+// zip resolves a scheme to a Transport through a registry (zip.RegisterTransport,
+// transport.go:109) and its default `zap` Dial is one line:
+//
+//	Dial: func(addr string) Client { return zaphttp.Dial(networkOf(addr), addr) }
+//
+// So the seam already exists: re-register the same scheme with the same dialler
+// and one call to the knob zap-proto/http exports for exactly this
+// (SetReadTimeout, client.go:81). Nothing is forked and no behaviour changes
+// except the number.
+//
+// # Why this number
+//
+// planeReadTimeout must be LONGER than the longest legitimate caller budget, so
+// that the caller's context is what expires. Today that is the chat bridge's
+// 110s. Set to 15 minutes to match the host's own plugin-start budget — the
+// deadline that already governs how long the fleet is willing to wait for a
+// child — so there is one answer to "how long may an in-flight plane call take"
+// rather than two.
+//
+// This is a CEILING, never a floor. Every caller still bounds itself with a
+// context; a caller that gives itself ten seconds still gets ten seconds. What
+// changes is that a caller can no longer be cut off BELOW its own budget by a
+// default it never saw.
+
+import (
+	"strings"
+	"time"
+
+	"github.com/valyala/fasthttp"
+	zaphttp "github.com/zap-proto/http"
+	"github.com/zap-proto/zip"
+)
+
+// planeReadTimeout bounds how long a plane call waits for its response. See the
+// file comment for why it is 15 minutes and why it is a ceiling.
+const planeReadTimeout = 15 * time.Minute
+
+func init() { widenPlaneReadDeadline() }
+
+// widenPlaneReadDeadline re-registers the zap scheme with the stock dialler and
+// a response-read deadline that does not undercut the caller.
+//
+// It runs at init because a transport must be registered before the first Dial,
+// and plane is imported by every process that makes a plane call. Registering
+// the same scheme twice is defined: RegisterTransport "adds (or replaces)".
+func widenPlaneReadDeadline() {
+	zip.RegisterTransport("zap", zip.Transport{
+		// The stock Serve, verbatim (zip transport.go:66-68). Only Dial changes;
+		// re-registering a scheme replaces BOTH halves, so the serve side has to
+		// be restated or every plugin stops listening.
+		Serve: func(addr string, h fasthttp.RequestHandler) zip.Server {
+			return &zaphttp.Server{Network: networkOf(addr), Addr: addr, Handler: h}
+		},
+		Dial: func(addr string) zip.Client {
+			t := zaphttp.Dial(networkOf(addr), addr)
+			t.SetReadTimeout(planeReadTimeout)
+			return t
+		},
+	})
+}
+
+// networkOf mirrors zip's own rule (transport.go:125): a path is a unix socket,
+// anything else is tcp. Copied rather than imported because zip keeps it
+// unexported — and it is three lines, so the alternative is a fork.
+func networkOf(addr string) string {
+	if strings.HasPrefix(addr, "/") || strings.HasPrefix(addr, "./") || strings.HasPrefix(addr, "@") {
+		return "unix"
+	}
+	return "tcp"
+}

plane/readdeadline_test.go

@@ -0,0 +1,60 @@
+package plane
+
+// The defect these pin: a plane call was cut off at the TRANSPORT's 30s
+// response-read default while the caller had given itself 110s. The work
+// succeeded — production measured a turn at 41,153ms answered 200 by both `ai`
+// and `agents` — and the caller had already hung up. Every log on the callee
+// side said healthy; only the caller reported failure.
+//
+// A unit test cannot hold a socket open for 40 seconds to prove this, so these
+// assert the two properties that make the failure impossible instead: the
+// ceiling exceeds the longest real caller budget, and the registration actually
+// takes effect on a dialled transport.
+
+import (
+	"testing"
+	"time"
+)
+
+// The number that mattered. zap-proto/http's default is 30s; the chat bridge
+// budgets 110s for one turn. A ceiling below the caller's own budget means the
+// caller can never spend it — which is exactly what shipped.
+func TestTheCeilingExceedsTheLongestCallerBudget(t *testing.T) {
+	const transportDefault = 30 * time.Second
+	const bridgeAgentTimeout = 110 * time.Second // apps/integrations/bridge.go
+
+	if planeReadTimeout <= transportDefault {
+		t.Fatalf("planeReadTimeout %v does not widen the %v default", planeReadTimeout, transportDefault)
+	}
+	if planeReadTimeout <= bridgeAgentTimeout {
+		t.Errorf("planeReadTimeout %v is at or below the chat bridge's own budget %v — "+
+			"a caller must expire on ITS deadline, never on the transport's",
+			planeReadTimeout, bridgeAgentTimeout)
+	}
+}
+
+// The registration must survive being applied twice: init() runs it, and a test
+// or a second import path may run it again. RegisterTransport is documented as
+// "adds (or replaces)", so this is defined — but if it ever panics on a repeat,
+// every process that imports plane twice dies at start-up rather than at a call.
+func TestRegisteringTwiceIsSafe(t *testing.T) {
+	widenPlaneReadDeadline()
+	widenPlaneReadDeadline()
+}
+
+// networkOf is copied from zip because zip keeps it unexported. A copy that
+// drifts would send a unix path to a tcp dialler, which fails at the connection
+// rather than anywhere legible — so pin the rule.
+func TestNetworkOfMatchesZipsRule(t *testing.T) {
+	for addr, want := range map[string]string{
+		"/var/lib/cloud/run/agents.sock": "unix",
+		"./relative.sock":                "unix",
+		"@abstract":                      "unix",
+		"127.0.0.1:8000":                 "tcp",
+		"example.com:443":                "tcp",
+	} {
+		if got := networkOf(addr); got != want {
+			t.Errorf("networkOf(%q) = %q, want %q", addr, got, want)
+		}
+	}
+}