From 15b2668baf7f99e8a322e72d53d84c2841e73696 Mon Sep 17 00:00:00 2001 From: David Gageot Date: Thu, 9 Jul 2026 20:16:38 +0200 Subject: [PATCH 1/2] fix: reliability fixes for board event streaming and rendering Reset watchdog on every SSE line, not just data frames; add keep-alives disable on the Unix transport; drain heartbeat goroutine before return; truncate both splitTitle lines; defer mutex unlock in SetSamplingWithToolsHandler. Assisted-By: claude-opus-4-5 --- pkg/board/client.go | 6 +++--- pkg/board/client_test.go | 25 +++++++++++++++++++++++++ pkg/board/tui/tui_test.go | 5 +++++ pkg/board/tui/view.go | 2 +- pkg/server/server.go | 7 ++++++- pkg/tools/mcp/session_client.go | 2 +- 6 files changed, 41 insertions(+), 6 deletions(-) diff --git a/pkg/board/client.go b/pkg/board/client.go index f5aba4dda4..345b5775d9 100644 --- a/pkg/board/client.go +++ b/pkg/board/client.go @@ -87,6 +87,7 @@ type client struct { // unix socket and targets the given session id. func newClient(socket, session string) *client { transport := &http.Transport{ + DisableKeepAlives: true, DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) { var d net.Dialer return d.DialContext(ctx, "unix", socket) @@ -149,8 +150,7 @@ func (c *client) Followup(ctx context.Context, idempotencyKey, message string) e if resp.StatusCode != http.StatusAccepted { return fmt.Errorf("followup: %s", resp.Status) } - // Drain so the connection can be reused; the body only reports whether - // the delivery was a duplicate, which the board does not care about. + // Drain the small response body before closing it. _, _ = io.Copy(io.Discard, resp.Body) return nil } @@ -201,6 +201,7 @@ func (c *client) StreamEvents(ctx context.Context, since uint64, onEvent func(ev var seq uint64 for scanner.Scan() { line := scanner.Text() + resetWatchdog() if strings.HasPrefix(line, ":") { // Heartbeat comment: the server sends them, so silence now means // a hung transport. Arm the watchdog on the first one. @@ -212,7 +213,6 @@ func (c *client) StreamEvents(ctx context.Context, since uint64, onEvent func(ev } continue } - resetWatchdog() if id, ok := strings.CutPrefix(line, "id:"); ok { seq, _ = strconv.ParseUint(strings.TrimSpace(id), 10, 64) continue diff --git a/pkg/board/client_test.go b/pkg/board/client_test.go index 6c4dfe34b6..f1c1ef873f 100644 --- a/pkg/board/client_test.go +++ b/pkg/board/client_test.go @@ -60,6 +60,31 @@ func TestStreamEventsIdleWatchdogAbortsSilentStream(t *testing.T) { assert.Equal(t, eventStreamStarted, got[0].Type) } +func TestStreamEventsHeartbeatLinesResetWatchdog(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) + for range 4 { + fmt.Fprint(w, ": ping\n") + f.Flush() + time.Sleep(20 * time.Millisecond) //nolint:forbidigo // heartbeat cadence is under test + } + fmt.Fprint(w, "data: {\"type\":\"stream_stopped\"}\n\n") + f.Flush() + <-r.Context().Done() + })) + + c := newClient(socket, "sess-1") + err := c.StreamEvents(t.Context(), 0, func(ev event) bool { + assert.Equal(t, eventStreamStopped, ev.Type) + return false + }) + require.NoError(t, err) +} + // 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) { diff --git a/pkg/board/tui/tui_test.go b/pkg/board/tui/tui_test.go index 5b6453dcd0..1e558b0de1 100644 --- a/pkg/board/tui/tui_test.go +++ b/pkg/board/tui/tui_test.go @@ -6,6 +6,7 @@ import ( "testing" tea "charm.land/bubbletea/v2" + "charm.land/lipgloss/v2" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -480,6 +481,10 @@ func TestSplitTitle(t *testing.T) { l1, l2 = splitTitle("A somewhat longer card title", 10) assert.Equal(t, "A somewhat", l1) assert.NotEmpty(t, l2) + + l1, l2 = splitTitle(strings.Repeat("x", 30)+" tail", 10) + assert.LessOrEqual(t, lipgloss.Width(l1), 10) + assert.LessOrEqual(t, lipgloss.Width(l2), 10) } func TestColorizeDiffKeepsLineCount(t *testing.T) { diff --git a/pkg/board/tui/view.go b/pkg/board/tui/view.go index eaa9588e92..31ccae91ac 100644 --- a/pkg/board/tui/view.go +++ b/pkg/board/tui/view.go @@ -464,7 +464,7 @@ func splitTitle(title string, width int) (string, string) { candidate = line1 + " " + word } if line1 != "" && lipgloss.Width(candidate) > width { - return line1, toolcommon.TruncateText(strings.Join(words[i:], " "), width) + return toolcommon.TruncateText(line1, width), toolcommon.TruncateText(strings.Join(words[i:], " "), width) } line1 = candidate } diff --git a/pkg/server/server.go b/pkg/server/server.go index aea0d5a9e6..6f1c7e3b74 100644 --- a/pkg/server/server.go +++ b/pkg/server/server.go @@ -624,8 +624,13 @@ func (s *Server) sessionEvents(c echo.Context) error { var writeMu sync.Mutex heartbeatCtx, stopHeartbeat := context.WithCancel(c.Request().Context()) - defer stopHeartbeat() + heartbeatDone := make(chan struct{}) + defer func() { + stopHeartbeat() + <-heartbeatDone + }() go func() { + defer close(heartbeatDone) ticker := time.NewTicker(s.heartbeatInterval) defer ticker.Stop() for { diff --git a/pkg/tools/mcp/session_client.go b/pkg/tools/mcp/session_client.go index d02274f4e3..34021f2752 100644 --- a/pkg/tools/mcp/session_client.go +++ b/pkg/tools/mcp/session_client.go @@ -343,8 +343,8 @@ func (c *sessionClient) handleSamplingWithToolsRequest(ctx context.Context, req // requests carrying a tools array from the MCP server. func (c *sessionClient) SetSamplingWithToolsHandler(handler tools.SamplingWithToolsHandler) { c.mu.Lock() + defer c.mu.Unlock() c.samplingWithToolsHandler = handler - c.mu.Unlock() } // applySamplingHandlerOpts wires the SDK CreateMessage* callback into opts. From d7c61f40f952c949bb945bd524fe80748f5819ea Mon Sep 17 00:00:00 2001 From: David Gageot Date: Thu, 9 Jul 2026 20:44:07 +0200 Subject: [PATCH 2/2] test: increase timing margins for SSE watchdog heartbeat test Assisted-By: docker-agent --- pkg/board/client_test.go | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/pkg/board/client_test.go b/pkg/board/client_test.go index f1c1ef873f..766614b546 100644 --- a/pkg/board/client_test.go +++ b/pkg/board/client_test.go @@ -62,15 +62,15 @@ func TestStreamEventsIdleWatchdogAbortsSilentStream(t *testing.T) { func TestStreamEventsHeartbeatLinesResetWatchdog(t *testing.T) { old := streamIdleTimeout - streamIdleTimeout = 50 * time.Millisecond + streamIdleTimeout = 250 * time.Millisecond t.Cleanup(func() { streamIdleTimeout = old }) socket := serveUnix(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { f := w.(http.Flusher) - for range 4 { + for range 6 { fmt.Fprint(w, ": ping\n") f.Flush() - time.Sleep(20 * time.Millisecond) //nolint:forbidigo // heartbeat cadence is under test + time.Sleep(50 * time.Millisecond) //nolint:forbidigo // heartbeat cadence is under test } fmt.Fprint(w, "data: {\"type\":\"stream_stopped\"}\n\n") f.Flush()