streams: expose SetReadDeadline so idle teardown actually engages

events.go

@@ -32,11 +32,30 @@ func (b daemonEventBus) Subscribe(pattern string) (<-chan coreapi.Event, func())
 	go func() {
 		defer close(out)
 		for ev := range src {
-			out <- coreapi.Event{
+			// Non-blocking, drop-on-full. The underlying bus deliberately
+			// drops rather than blocking its publishers; forwarding with a
+			// blocking send silently converted those semantics and turned
+			// this adapter into a goroutine leak.
+			//
+			// If a plugin consumer stopped draining `out` — its handler
+			// loop exited, panicked past a recover, or simply ran slower
+			// than the publisher for cap(src) events — this goroutine
+			// parked on the send forever, retaining itself, the buffered
+			// contents of src, and every payload map they referenced.
+			// cancel() does not rescue it: cancel closes src, which does
+			// nothing for a goroutine already blocked sending to out.
+			//
+			// Dropping matches what the bus would have done anyway once
+			// the buffer filled, so a slow consumer now loses events
+			// instead of leaking a goroutine for the process lifetime.
+			select {
+			case out <- coreapi.Event{
 				Topic:   ev.Topic,
 				NodeID:  ev.NodeID,
 				Time:    ev.Time,
 				Payload: ev.Payload,
+			}:
+			default:
 			}
 		}
 	}()

streams.go

@@ -5,6 +5,7 @@ package runtime
 import (
 	"context"
 	"errors"
+	"time"
 
 	"github.com/pilot-protocol/common/coreapi"
 	"github.com/pilot-protocol/common/daemonapi"
@@ -72,6 +73,27 @@ func (s *streamAdapter) Read(p []byte) (int, error)  { return s.rw.Read(p) }
 func (s *streamAdapter) Write(p []byte) (int, error) { return s.rw.Write(p) }
 func (s *streamAdapter) Close() error                { return s.rw.Close() }
 
+// SetReadDeadline forwards to the underlying ConnReadWriter when it
+// supports deadlines.
+//
+// This method existing at all is the point. Frame readers (notably
+// dataexchange) gate their slowloris/idle teardown on a
+// `conn.(interface{ SetReadDeadline(time.Time) error })` assertion.
+// streamAdapter did not implement it, so that assertion failed for every
+// in-daemon plugin connection and the configured idle timeout was never
+// applied — a peer that opened a stream and stalled held its handler
+// goroutine and the whole underlying connection until the process died.
+//
+// Reported as nil (no-op success) rather than an error when the transport
+// cannot do deadlines, so callers that treat a failure as fatal are not
+// broken by transports that legitimately lack the capability.
+func (s *streamAdapter) SetReadDeadline(t time.Time) error {
+	if d, ok := s.rw.(interface{ SetReadDeadline(time.Time) error }); ok {
+		return d.SetReadDeadline(t)
+	}
+	return nil
+}
+
 func (s *streamAdapter) LocalAddr() coreapi.Addr  { return s.conn.Info().LocalAddr }
 func (s *streamAdapter) LocalPort() uint16        { return s.conn.Info().LocalPort }
 func (s *streamAdapter) RemoteAddr() coreapi.Addr { return s.conn.Info().RemoteAddr }
@@ -83,4 +105,12 @@ var (
 	_ coreapi.Streams  = daemonStreams{}
 	_ coreapi.Listener = (*daemonListener)(nil)
 	_ coreapi.Stream   = (*streamAdapter)(nil)
+
+	// Frame readers (dataexchange) gate their idle/slowloris teardown on
+	// this exact optional interface. When streamAdapter stopped satisfying
+	// it, the assertion silently fell through to "no deadline" and stalled
+	// peers pinned a goroutine plus a full Connection forever — which is
+	// what OOM-killed the 431-agent service fleet on 2026-07-28. Assert it
+	// at compile time so the capability cannot be dropped again silently.
+	_ interface{ SetReadDeadline(time.Time) error } = (*streamAdapter)(nil)
 )