internal/api/agent.go

@@ -108,6 +108,17 @@ func (h *handlers) agentChat(c *gin.Context) {
 	send(chatEvent{Done: turn})
 }
 
+// sseWriteTimeout bounds ONE write to the stream, not the stream itself.
+//
+// It is refreshed per frame, which is the only shape that satisfies both ends:
+// the server's absolute WriteTimeout would cut a long turn (#78), while removing
+// the deadline entirely would let a client that stops reading block a write
+// forever once the socket buffer fills — pinning the run goroutine and this
+// stream's mutex with it, and taking the keep-alive down too since it needs the
+// same lock. Generous, because it is a backstop against a stuck peer and not a
+// pacing mechanism.
+const sseWriteTimeout = 30 * time.Second
+
 // eventStream serializes writes to one SSE response.
 //
 // The mutex is load-bearing, not decoration: step events are sent from the
@@ -116,6 +127,7 @@ func (h *handlers) agentChat(c *gin.Context) {
 // frames long before it crashes anything.
 type eventStream struct {
 	c  *gin.Context
+	rc *http.ResponseController
 	mu sync.Mutex
 }
 
@@ -125,29 +137,33 @@ type eventStream struct {
 // so a proxy holding the response until it looks complete can't reintroduce
 // exactly the silence streaming exists to remove.
 //
-// Clearing the write deadline is what makes a turn longer than the server's
+// Taking the write deadline off the server's absolute WriteTimeout and onto a
-// WriteTimeout possible at all (#78). That timeout is an ABSOLUTE deadline from
+// per-write one is what makes a turn longer than 30s possible at all (#78).
-// when the request header was read, not an idle timeout, so a streaming response
+// WriteTimeout is an ABSOLUTE deadline from when the request header was read,
-// has to opt out of it explicitly or it is cut mid-turn no matter how recently
+// not an idle timeout, so a streaming response is cut mid-turn however recently
 // it wrote. Without this the 4-minute runTimeout is unreachable and the
-// keep-alive below tops out at one tick — it would be pacing a connection that
+// keep-alive below tops out at one tick — pacing a connection that is destroyed
-// is destroyed underneath it.
+// underneath it.
 //
-// This failure is INVISIBLE from in here: the writes after the deadline return
+// That failure is INVISIBLE from in here: writes past the deadline return
-// err == nil and their bytes are dropped, so there is nothing to detect and
+// err == nil and their bytes are dropped, so there is nothing to detect on the
-// report. Only the client sees it, as a truncated stream it reports as a dropped
+// write path. Only the client sees it, as a truncated stream it reports as a
-// connection. That is why this is done up front rather than handled on write.
+// dropped connection. Hence a deadline set up front and refreshed per frame,
+// rather than anything checked after the fact.
 func openEventStream(c *gin.Context) *eventStream {
 	c.Header("Content-Type", "text/event-stream")
 	c.Header("Cache-Control", "no-cache")
 	c.Header("X-Accel-Buffering", "no")
-	// A failure here is worth saying out loud: it means this stream WILL be cut
+	s := &eventStream{c: c, rc: http.NewResponseController(c.Writer)}
-	// at WriteTimeout and we cannot stop it.
+	// Probe once here rather than reporting per frame: a writer that can't take
-	if err := http.NewResponseController(c.Writer).SetWriteDeadline(time.Time{}); err != nil {
+	// deadlines will fail identically on every write, and the operator needs to
-		slog.Error("api: cannot clear SSE write deadline; long turns will be truncated", "error", err)
+	// hear it once. If this fails the stream still works — it is just back to
+	// being cut at WriteTimeout, which is worth saying out loud.
+	if err := s.rc.SetWriteDeadline(time.Now().Add(sseWriteTimeout)); err != nil {
+		slog.Error("api: SSE write deadlines unavailable; long turns will be truncated at the server WriteTimeout", "error", err)
 	}
 	c.Writer.Flush()
-	return &eventStream{c: c}
+	return s
 }
 
 func (s *eventStream) send(ev chatEvent) {
@@ -162,6 +178,11 @@ func (s *eventStream) send(ev chatEvent) {
 func (s *eventStream) write(frame string) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
+	// Refresh for THIS write, so the stream as a whole is unbounded but no single
+	// write is. Error deliberately unchecked: openEventStream already reported
+	// whether deadlines work at all, and this call can only fail the same way, so
+	// checking here would log once per frame to say the same thing.
+	_ = s.rc.SetWriteDeadline(time.Now().Add(sseWriteTimeout))
 	_, _ = io.WriteString(s.c.Writer, frame)
 	s.c.Writer.Flush()
 }

internal/api/sse_deadline_test.go

@@ -22,11 +22,15 @@ import (
 // after the deadline return err == nil and their bytes are silently discarded.
 // A test that checked the return of io.WriteString would pass against the bug.
 func TestEventStreamOutlivesWriteTimeout(t *testing.T) {
-	// Long enough that the frames below straddle it, short enough to keep the
+	// tick > writeTimeout on purpose, so EVERY frame lands after the deadline has
-	// test quick. Everything else is scaled off it.
+	// already passed and the first one clears it by a full 100ms. The margin only
-	const writeTimeout = 300 * time.Millisecond
+	// ever grows: a slow or loaded CI runner delays frames further past a deadline
-	const tick = 200 * time.Millisecond
+	// that has already expired, so scheduling jitter pushes this test toward
-	const frames = 4 // last one lands at ~800ms, well past the deadline
+	// passing rather than toward flaking. Sizing it the other way — frames landing
+	// just before the deadline — would flake exactly when CI is busiest.
+	const writeTimeout = 200 * time.Millisecond
+	const tick = 300 * time.Millisecond
+	const frames = 3 // ~900ms total; the last lands 700ms past the deadline
 
 	gin.SetMode(gin.TestMode)
 	r := gin.New()