feat: harden control-plane protocol for reliable board communication

pkg/api/types.go

@@ -21,6 +21,22 @@ type Agent struct {
 	Multi       bool   `json:"multi"`
 }
 
+// ErrCodeUnknownSession is the [ErrorResponse.Code] for a request that names
+// a session the server does not know. Distinguishing it matters to clients
+// like the board: a 404 with this code means "the server is current but the
+// session does not exist (yet)" — keep waiting or recreate the session —
+// while a 404 without it means the route itself is missing (an older binary
+// that predates the endpoint).
+const ErrCodeUnknownSession = "unknown_session"
+
+// ErrorResponse is the body of an API error that clients may need to react
+// to programmatically. Message is human-readable; Code, when set, is a
+// stable machine-readable classifier (e.g. [ErrCodeUnknownSession]).
+type ErrorResponse struct {
+	Code    string `json:"code,omitempty"`
+	Message string `json:"message"`
+}
+
 // CreateAgentRequest represents a request to create an agent
 type CreateAgentRequest struct {
 	Prompt string `json:"prompt"`
@@ -327,6 +343,10 @@ type SessionSnapshotResponse struct {
 	// Zero when the session has no event stream (no events yet, or not
 	// attached to a control plane).
 	LastEventSeq uint64 `json:"last_event_seq"`
+	// Cost is the session's cumulative cost in US dollars, including
+	// sub-sessions and item-level costs (e.g. compaction). Clients should
+	// prefer it over summing per-message costs, which misses those.
+	Cost float64 `json:"cost"`
 }
 
 // RunAgentRequest is the body of POST /api/sessions/:id/agent/:agent[/:agent_name].

pkg/app/app.go

@@ -805,6 +805,12 @@ func (a *App) removeSubscriber(ch chan tea.Msg) {
 // scatters every message to all currently-registered subscribers. Sends are
 // non-blocking; if a subscriber's buffer is full the event is dropped for
 // that subscriber so one slow consumer cannot stall the others.
+//
+// Turn-boundary events are the exception: dropping a stream_started or
+// stream_stopped skews a consumer's turn accounting for good (the SSE replay
+// buffer never sees the event, so reconnecting cannot recover it). For those,
+// the oldest pending message — almost always a content delta, which the next
+// delta supersedes — is evicted to make room instead.
 func (a *App) startFanOut() {
 	throttled := a.throttleEvents(a.ctx(), a.events)
 	go func() {
@@ -816,13 +822,47 @@ func (a *App) startFanOut() {
 				select {
 				case ch <- msg:
 				default:
-					slog.Warn("app: subscriber buffer full, dropping event")
+					if !isTurnBoundaryEvent(msg) {
+						slog.Warn("app: subscriber buffer full, dropping event")
+						continue
+					}
+					// Evict the oldest pending message (racing the subscriber's
+					// own receive is fine: either way a slot frees up), then
+					// retry once. Still full means the subscriber is wedged;
+					// drop as before rather than block the fan-out.
+					select {
+					case <-ch:
+					default:
+					}
+					select {
+					case ch <- msg:
+					default:
+						slog.Warn("app: subscriber buffer full, dropping turn-boundary event")
+					}
 				}
 			}
 		}
 	}()
 }
 
+// isTurnBoundaryEvent reports whether msg is one of the events consumers use
+// to track turn state (running/waiting/failed/paused) and identity (title).
+// These are low-frequency and irrecoverable when lost, unlike the content
+// deltas that dominate the stream, so the fan-out prefers them on overflow.
+func isTurnBoundaryEvent(msg tea.Msg) bool {
+	switch msg.(type) {
+	case *runtime.StreamStartedEvent,
+		*runtime.StreamStoppedEvent,
+		*runtime.UserMessageEvent,
+		*runtime.ErrorEvent,
+		*runtime.PausedEvent,
+		*runtime.SessionTitleEvent:
+		return true
+	default:
+		return false
+	}
+}
+
 // Resume resumes the runtime with the given confirmation request
 func (a *App) Resume(req runtime.ResumeRequest) {
 	a.runtime.Resume(a.ctx(), req)

pkg/app/fanout_test.go

@@ -0,0 +1,86 @@
+package app
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	tea "charm.land/bubbletea/v2"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/docker/docker-agent/pkg/runtime"
+)
+
+// TestFanOut_TurnBoundaryEventEvictsPendingDelta verifies that when a
+// subscriber's buffer is full, a turn-boundary event (which SSE consumers
+// cannot recover if lost) evicts the oldest pending message instead of being
+// dropped.
+func TestFanOut_TurnBoundaryEventEvictsPendingDelta(t *testing.T) {
+	t.Parallel()
+
+	events := make(chan tea.Msg, 16)
+	app := &App{
+		ctx:              func() context.Context { return t.Context() },
+		events:           events,
+		throttleDuration: time.Millisecond,
+	}
+
+	// A one-slot subscriber makes the overflow deterministic. The subscriber
+	// never reads, standing in for a consumer that fell behind.
+	ch := make(chan tea.Msg, 1)
+	app.addSubscriber(ch)
+	app.fanoutOnce.Do(app.startFanOut)
+
+	// Fill the subscriber's buffer with a droppable event.
+	filler := runtime.NewTokenUsageEvent("sess", "root", &runtime.Usage{})
+	events <- filler
+	require.Eventually(t, func() bool { return len(ch) == 1 }, 2*time.Second, time.Millisecond)
+
+	// The turn-boundary event must displace the pending filler.
+	stopped := runtime.StreamStopped("sess", "root", "normal")
+	events <- stopped
+
+	require.Eventually(t, func() bool {
+		select {
+		case msg := <-ch:
+			return assert.ObjectsAreEqual(stopped, msg)
+		default:
+			return false
+		}
+	}, 2*time.Second, time.Millisecond, "the stream_stopped event must survive the overflow")
+}
+
+// TestFanOut_DroppableEventIsDroppedOnOverflow verifies the pre-existing
+// behavior for non-boundary events: on overflow they are dropped, never
+// evicting anything.
+func TestFanOut_DroppableEventIsDroppedOnOverflow(t *testing.T) {
+	t.Parallel()
+
+	events := make(chan tea.Msg, 16)
+	app := &App{
+		ctx:              func() context.Context { return t.Context() },
+		events:           events,
+		throttleDuration: time.Millisecond,
+	}
+
+	ch := make(chan tea.Msg, 1)
+	app.addSubscriber(ch)
+	// The witness is registered after ch, so once a message reaches it the
+	// fan-out has already made its keep-or-drop decision for ch.
+	witness := make(chan tea.Msg, 16)
+	app.addSubscriber(witness)
+	app.fanoutOnce.Do(app.startFanOut)
+
+	first := runtime.NewTokenUsageEvent("sess", "root", &runtime.Usage{})
+	events <- first
+	require.Eventually(t, func() bool { return len(ch) == 1 }, 2*time.Second, time.Millisecond)
+
+	// A second droppable event overflows and is dropped; the first stays.
+	second := runtime.NewTokenUsageEvent("sess", "other", &runtime.Usage{})
+	events <- second
+	require.Eventually(t, func() bool { return len(witness) == 2 }, 2*time.Second, time.Millisecond)
+
+	msg := <-ch
+	assert.Equal(t, first, msg)
+}

pkg/board/client.go

@@ -13,6 +13,8 @@ import (
 	"net/url"
 	"strconv"
 	"strings"
+	"sync/atomic"
+	"time"
 )
 
 // Event types the board reacts to on the session event stream. Every other
@@ -42,6 +44,17 @@ const (
 // cleanly, as opposed to "error", "canceled", "hook_blocked"...
 const reasonNormal = "normal"
 
+// streamIdleTimeout is how long StreamEvents tolerates a silent connection
+// once the server has proven it sends heartbeats (": ping" SSE comments,
+// emitted every 15s). Three missed heartbeats means the transport is hung —
+// e.g. the agent's VM was paused — not that the session is quiet, so the
+// stream is aborted and the watcher reconnects. Servers that predate
+// heartbeats never arm the watchdog, keeping long-lived idle streams working.
+var streamIdleTimeout = 45 * time.Second
+
+// errStreamIdle reports a stream aborted by the idle watchdog.
+var errStreamIdle = errors.New("event stream idle: heartbeats stopped")
+
 // event is the subset of a runtime event the board cares about.
 type event struct {
 	Type  string `json:"type"`
@@ -147,6 +160,8 @@ func (c *client) Followup(ctx context.Context, idempotencyKey, message string) e
 // false stops the stream cleanly. It returns nil on a clean stop and an
 // error when the connection fails.
 func (c *client) StreamEvents(ctx context.Context, since uint64, onEvent func(event) bool) error {
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
 	u := c.endpoint("events")
 	if since > 0 {
 		u += "?since=" + strconv.FormatUint(since, 10)
@@ -167,9 +182,37 @@ func (c *client) StreamEvents(ctx context.Context, since uint64, onEvent func(ev
 
 	scanner := bufio.NewScanner(resp.Body)
 	scanner.Buffer(make([]byte, 64*1024), 4*1024*1024)
+
+	// The idle watchdog is armed by the first heartbeat and re-armed by any
+	// subsequent line; when it fires it cancels the request, failing the read.
+	var idle atomic.Bool
+	var watchdog *time.Timer
+	defer func() {
+		if watchdog != nil {
+			watchdog.Stop()
+		}
+	}()
+	resetWatchdog := func() {
+		if watchdog != nil {
+			watchdog.Reset(streamIdleTimeout)
+		}
+	}
+
 	var seq uint64
 	for scanner.Scan() {
 		line := scanner.Text()
+		if strings.HasPrefix(line, ":") {
+			// Heartbeat comment: the server sends them, so silence now means
+			// a hung transport. Arm the watchdog on the first one.
+			if watchdog == nil {
+				watchdog = time.AfterFunc(streamIdleTimeout, func() {
+					idle.Store(true)
+					cancel()
+				})
+			}
+			continue
+		}
+		resetWatchdog()
 		if id, ok := strings.CutPrefix(line, "id:"); ok {
 			seq, _ = strconv.ParseUint(strings.TrimSpace(id), 10, 64)
 			continue
@@ -188,6 +231,9 @@ func (c *client) StreamEvents(ctx context.Context, since uint64, onEvent func(ev
 			return nil
 		}
 	}
+	if idle.Load() {
+		return errStreamIdle
+	}
 	if err := scanner.Err(); err != nil {
 		return err
 	}

pkg/board/client_test.go

@@ -0,0 +1,86 @@
+package board
+
+import (
+	"fmt"
+	"net"
+	"net/http"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// serveUnix runs an HTTP handler on a unix socket and returns the socket path.
+func serveUnix(t *testing.T, handler http.Handler) string {
+	t.Helper()
+	// Not t.TempDir(): its per-test path is long enough to overflow the
+	// ~104-byte unix sun_path limit under long test names.
+	dir, err := os.MkdirTemp("", "board-client") //nolint:forbidigo,usetesting // see above
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = os.RemoveAll(dir) })
+
+	socket := filepath.Join(dir, "cp.sock")
+	var lc net.ListenConfig
+	ln, err := lc.Listen(t.Context(), "unix", socket)
+	require.NoError(t, err)
+	srv := &http.Server{Handler: handler, ReadHeaderTimeout: time.Second}
+	go func() { _ = srv.Serve(ln) }()
+	t.Cleanup(func() { _ = srv.Close() })
+	return socket
+}
+
+// Heartbeat comments are invisible to the event callback, and once one has
+// been seen, a stream that goes silent is aborted with an error (instead of
+// blocking forever on a hung transport) so the watcher reconnects.
+func TestStreamEventsIdleWatchdogAbortsSilentStream(t *testing.T) {
+	old := streamIdleTimeout
+	streamIdleTimeout = 50 * time.Millisecond
+	t.Cleanup(func() { streamIdleTimeout = old })
+
+	socket := serveUnix(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		f := w.(http.Flusher)
+		fmt.Fprint(w, ": ping\n\n")
+		fmt.Fprint(w, "data: {\"type\":\"stream_started\"}\n\n")
+		f.Flush()
+		// Then hang without closing, like a wedged transport.
+		<-r.Context().Done()
+	}))
+
+	c := newClient(socket, "sess-1")
+	var got []event
+	err := c.StreamEvents(t.Context(), 0, func(ev event) bool {
+		got = append(got, ev)
+		return true
+	})
+	require.ErrorIs(t, err, errStreamIdle)
+	require.Len(t, got, 1, "heartbeat comments must not reach the callback")
+	assert.Equal(t, eventStreamStarted, got[0].Type)
+}
+
+// Without any heartbeat from the server (an older docker-agent), the watchdog
+// stays unarmed: a quiet stream is left alone and events keep flowing.
+func TestStreamEventsNoHeartbeatNoWatchdog(t *testing.T) {
+	old := streamIdleTimeout
+	streamIdleTimeout = 50 * time.Millisecond
+	t.Cleanup(func() { streamIdleTimeout = old })
+
+	socket := serveUnix(t, http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		f := w.(http.Flusher)
+		fmt.Fprint(w, "data: {\"type\":\"stream_started\"}\n\n")
+		f.Flush()
+		time.Sleep(120 * time.Millisecond) //nolint:forbidigo // real quiet time is the thing under test
+		fmt.Fprint(w, "data: {\"type\":\"stream_stopped\"}\n\n")
+	}))
+
+	c := newClient(socket, "sess-1")
+	var got []event
+	err := c.StreamEvents(t.Context(), 0, func(ev event) bool {
+		got = append(got, ev)
+		return len(got) < 2
+	})
+	require.NoError(t, err)
+	require.Len(t, got, 2)
+}