Fix: Enforce RequiresLater in agentop, harden the live stream, correct stale comments

cmd/agentop/apiclient/client.go

@@ -249,14 +249,15 @@ type PipelineView struct {
 // PipelinePlugin describes one plugin's position, direction, and
 // capabilities. Mirrors the server's pipelinePluginView exactly.
 type PipelinePlugin struct {
-	Name        string          `json:"name"`
-	Direction   string          `json:"direction"`
-	Position    int             `json:"position"`
-	ReadsBody   bool            `json:"readsBody"`
-	Requires    []string        `json:"requires,omitempty"`
-	RequiresAny []string        `json:"requiresAny,omitempty"`
-	Description string          `json:"description,omitempty"`
-	Config      json.RawMessage `json:"config,omitempty"`
+	Name          string          `json:"name"`
+	Direction     string          `json:"direction"`
+	Position      int             `json:"position"`
+	ReadsBody     bool            `json:"readsBody"`
+	Requires      []string        `json:"requires,omitempty"`
+	RequiresAny   []string        `json:"requiresAny,omitempty"`
+	RequiresLater []string        `json:"requiresLater,omitempty"`
+	Description   string          `json:"description,omitempty"`
+	Config        json.RawMessage `json:"config,omitempty"`
 	// OnError is the plugin's on_error policy, "" for the default (enforce): the proxy sends
 	// it only when it is not. A plugin under observe runs, but its rejections, body writes and
 	// redirects are dropped — header writes still apply — so in those respects its Config says
@@ -294,13 +295,14 @@ type PluginCatalog struct {
 // Describes a registered plugin's static type-level metadata; the
 // catalog includes plugins not currently in the active pipeline.
 type PluginCatalogEntry struct {
-	Name        string             `json:"name"`
-	Direction   string             `json:"direction,omitempty"`
-	ReadsBody   bool               `json:"readsBody,omitempty"`
-	Requires    []string           `json:"requires,omitempty"`
-	RequiresAny []string           `json:"requiresAny,omitempty"`
-	Description string             `json:"description,omitempty"`
-	Fields      []PluginFieldEntry `json:"fields,omitempty"`
+	Name          string             `json:"name"`
+	Direction     string             `json:"direction,omitempty"`
+	ReadsBody     bool               `json:"readsBody,omitempty"`
+	Requires      []string           `json:"requires,omitempty"`
+	RequiresAny   []string           `json:"requiresAny,omitempty"`
+	RequiresLater []string           `json:"requiresLater,omitempty"`
+	Description   string             `json:"description,omitempty"`
+	Fields        []PluginFieldEntry `json:"fields,omitempty"`
 }
 
 // PluginFieldEntry mirrors sessionapi.FieldSchemaEntry — per-field

cmd/agentop/apiclient/sse.go

@@ -72,13 +72,21 @@ func (c *Client) Stream(ctx context.Context, sessionFilter string) <-chan Stream
 func (c *Client) streamLoop(ctx context.Context, filter string, out chan<- StreamEvent) {
 	defer close(out)
 
+	// attempt is consecutive FAILED connects, not a lifetime counter: it
+	// resets once a connection reaches "open", so five disconnects spread
+	// over a day (laptop sleep, proxy restart) each wait 1s, not the capped
+	// 30s — and the banner's attempt number describes the current outage
+	// rather than the process's age (#1568).
 	attempt := 0
 	for {
-		err := c.streamOnce(ctx, filter, out)
+		opened, err := c.streamOnce(ctx, filter, out)
 		if ctx.Err() != nil {
 			// Caller cancelled — terminal state is the channel close on defer.
 			return
 		}
+		if opened {
+			attempt = 0
+		}
 		attempt++
 		wait := backoffSchedule(attempt)
 		sendStatus(out, StreamStatus{Phase: "reconnecting", Attempt: attempt, Wait: wait, Err: err})
@@ -92,47 +100,65 @@ func (c *Client) streamLoop(ctx context.Context, filter string, out chan<- Strea
 
 // streamOnce opens a single SSE connection and pumps events to out until
 // either the server closes, ctx is cancelled, or a transport error occurs.
-// Returns nil only if the server closed cleanly.
-func (c *Client) streamOnce(ctx context.Context, filter string, out chan<- StreamEvent) error {
+// opened reports whether the connection got as far as the "open" status —
+// the caller resets its backoff on it. Returns a nil error only on a clean
+// server close.
+func (c *Client) streamOnce(ctx context.Context, filter string, out chan<- StreamEvent) (opened bool, err error) {
 	u := c.endpoint + "/v1/events"
 	if filter != "" {
 		u += "?session=" + filter
 	}
 	req, err := http.NewRequestWithContext(ctx, "GET", u, nil)
 	if err != nil {
-		return err
+		return false, err
 	}
 	req.Header.Set("Accept", "text/event-stream")
 
 	// Reuse the shared no-timeout client so the Transport's idle-connection
 	// pool survives across reconnects.
 	resp, err := c.httpStream.Do(req)
 	if err != nil {
-		return err
+		return false, err
 	}
 	defer resp.Body.Close()
 	if resp.StatusCode != http.StatusOK {
-		return fmt.Errorf("stream: unexpected status %d", resp.StatusCode)
+		return false, fmt.Errorf("stream: unexpected status %d", resp.StatusCode)
 	}
 	sendStatus(out, StreamStatus{Phase: "open"})
 
-	return parseSSE(ctx, resp.Body, out)
+	return true, parseSSE(ctx, resp.Body, out)
 }
 
+// maxSSELine bounds how much of one SSE line parseSSE keeps. NOT a wire
+// limit: the server has no cap on event size (full inference bodies,
+// against a proxy that accepts 32 MiB request bodies), so no fixed bound
+// covers every legitimate event. A line longer than this marks its frame
+// as oversized, and the frame is skipped — the connection survives.
+const maxSSELine = 32 * 1024 * 1024
+
 // parseSSE reads SSE frames line-by-line. A frame is a run of lines ending
 // in a blank line. Lines starting with ':' are comments (our heartbeats).
 // We only care about frames with a `data:` line containing a JSON-encoded
 // SessionEvent; other line types are tolerated but ignored.
 func parseSSE(ctx context.Context, r io.Reader, out chan<- StreamEvent) error {
-	scanner := bufio.NewScanner(r)
-	// Some events (e.g. full LLM completion bodies) can be many KB. Raise the
-	// line buffer well above the default 64 KB to match the server's cap.
-	scanner.Buffer(make([]byte, 0, 16*1024), 4*1024*1024)
+	lines := newLineScan(r, maxSSELine)
 
 	var dataBuf strings.Builder
+	// oversized marks the frame in progress as unparseable — one of its
+	// lines was truncated at the cap. The frame is dropped at its blank
+	// line and the stream keeps reading; before this, a bufio.Scanner
+	// ErrTooLong killed the connection, and every event published during
+	// the reconnect gap was lost (#1568).
+	oversized := false
 
 	flush := func() {
 		defer dataBuf.Reset()
+		bad := oversized
+		oversized = false
+		if bad {
+			slog.Debug("apiclient: skipping oversized SSE frame", "bytes", dataBuf.Len())
+			return
+		}
 		if dataBuf.Len() == 0 {
 			return
 		}
@@ -156,14 +182,26 @@ func parseSSE(ctx context.Context, r io.Reader, out chan<- StreamEvent) error {
 		}
 	}
 
-	for scanner.Scan() {
-		line := scanner.Text()
+	for {
+		line, truncated, err := lines.next()
+		if err != nil {
+			// Drain trailing frame if the stream ended mid-record (rare;
+			// usual case is a clean "\n\n" before EOF).
+			flush()
+			if errors.Is(err, io.EOF) {
+				return io.EOF
+			}
+			return err
+		}
 		switch {
 		case line == "":
 			flush()
 		case strings.HasPrefix(line, ":"):
 			// Comment / heartbeat — ignore.
 		case strings.HasPrefix(line, "data:"):
+			if truncated {
+				oversized = true
+			}
 			val := strings.TrimPrefix(line, "data:")
 			val = strings.TrimPrefix(val, " ")
 			if dataBuf.Len() > 0 {
@@ -177,13 +215,62 @@ func parseSSE(ctx context.Context, r io.Reader, out chan<- StreamEvent) error {
 			return ctx.Err()
 		}
 	}
-	// Drain trailing frame if the stream ended mid-record (rare; usual case
-	// is a clean "\n\n" before EOF).
-	flush()
-	if err := scanner.Err(); err != nil && !errors.Is(err, io.EOF) {
-		return err
+}
+
+// lineScan is bufio.Scanner's line loop with one difference that matters
+// here: a line longer than max does not fail the reader. The line comes
+// back truncated (at most max bytes, truncated=true) and the reader is
+// positioned at the next line, so one huge frame costs one frame instead
+// of the connection.
+type lineScan struct {
+	br  *bufio.Reader
+	max int
+}
+
+func newLineScan(r io.Reader, max int) *lineScan {
+	return &lineScan{br: bufio.NewReaderSize(r, 16*1024), max: max}
+}
+
+// next returns the next line without its terminator. truncated reports
+// that the line exceeded the cap and was cut. A terminal error is returned
+// with whatever remained buffered; io.EOF means clean end of stream.
+func (s *lineScan) next() (line string, truncated bool, err error) {
+	var buf []byte
+	for {
+		frag, rerr := s.br.ReadSlice('\n')
+		switch rerr {
+		case nil:
+			frag = frag[:len(frag)-1] // drop the '\n'
+		case bufio.ErrBufferFull:
+			// Line longer than the read buffer: keep accumulating if within
+			// the cap, otherwise discard the overflow and remember it.
+			if len(buf)+len(frag) > s.max {
+				if len(buf) < s.max {
+					buf = append(buf, frag[:s.max-len(buf)]...)
+				}
+				truncated = true
+			} else {
+				buf = append(buf, frag...)
+			}
+			continue
+		case io.EOF:
+			if len(frag) == 0 && len(buf) == 0 {
+				return "", false, io.EOF
+			}
+			// Final line without a trailing newline.
+		default:
+			return string(buf), truncated, rerr
+		}
+		if len(buf)+len(frag) > s.max {
+			if len(buf) < s.max {
+				buf = append(buf, frag[:s.max-len(buf)]...)
+			}
+			truncated = true
+		} else {
+			buf = append(buf, frag...)
+		}
+		return string(buf), truncated, nil
 	}
-	return io.EOF
 }
 
 func sendStatus(out chan<- StreamEvent, s StreamStatus) {

cmd/agentop/apiclient/sse_test.go

@@ -3,8 +3,10 @@ package apiclient
 import (
 	"context"
 	"fmt"
+	"io"
 	"net/http"
 	"net/http/httptest"
+	"strings"
 	"sync"
 	"sync/atomic"
 	"testing"
@@ -220,3 +222,116 @@ func TestBackoffSchedule_CapsAt30s(t *testing.T) {
 		}
 	}
 }
+
+// Backoff counts consecutive FAILED connects, not a lifetime of disconnects
+// (#1568): once a connection reaches "open", the counter resets, so five
+// disconnects spread over a day each wait 1s instead of the capped 30s.
+func TestStream_BackoffResetsAfterOpen(t *testing.T) {
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "text/event-stream")
+		w.WriteHeader(200)
+		w.(http.Flusher).Flush()
+		// Every connection opens, then closes — each is a successful open
+		// followed by a disconnect.
+		<-time.After(50 * time.Millisecond)
+	}))
+	defer ts.Close()
+
+	c := New(ts.URL)
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	ch := c.Stream(ctx, "")
+
+	reconnects := 0
+	deadline := time.After(5 * time.Second)
+	for reconnects < 3 {
+		select {
+		case msg, ok := <-ch:
+			if !ok {
+				t.Fatalf("channel closed after %d reconnects", reconnects)
+			}
+			if msg.Status.Phase == "reconnecting" {
+				reconnects++
+				if msg.Status.Attempt != 1 {
+					t.Fatalf("reconnect #%d carried attempt %d, want 1 — backoff did not reset after a successful open",
+						reconnects, msg.Status.Attempt)
+				}
+			}
+		case <-deadline:
+			t.Fatalf("timeout after %d reconnects", reconnects)
+		}
+	}
+}
+
+// A frame larger than the reader's cap must cost one frame, not the
+// connection (#1568): the old bufio.Scanner died with ErrTooLong past 4 MiB
+// — a size real events reach, since the server marshals full inference
+// bodies — and everything published during the reconnect gap was lost.
+// Here the oversized frame is 5 MiB (over the old limit, under the new cap)
+// and the events on either side of it must both arrive.
+func TestStream_SkipsOversizedFrame(t *testing.T) {
+	valid := "event: session-event\ndata: {\"sessionId\":\"s1\",\"at\":\"2026-01-01T00:00:00Z\",\"direction\":\"inbound\",\"phase\":\"request\"}\n\n"
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		flusher := w.(http.Flusher)
+		w.Header().Set("Content-Type", "text/event-stream")
+		w.WriteHeader(200)
+		fmt.Fprint(w, valid)
+		flusher.Flush()
+		fmt.Fprintf(w, "data: %s\n\n", strings.Repeat("x", 5*1024*1024))
+		flusher.Flush()
+		fmt.Fprint(w, valid)
+		flusher.Flush()
+		<-r.Context().Done()
+	}))
+	defer ts.Close()
+
+	c := New(ts.URL)
+	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+	defer cancel()
+	ch := c.Stream(ctx, "")
+
+	var gotEvents int
+	deadline := time.After(3 * time.Second)
+	for gotEvents < 2 {
+		select {
+		case msg, ok := <-ch:
+			if !ok {
+				t.Fatalf("channel closed after %d events — the oversized frame killed the connection", gotEvents)
+			}
+			if msg.Event != nil {
+				gotEvents++
+			}
+		case <-deadline:
+			t.Fatalf("timeout after %d events; the oversized frame killed the connection", gotEvents)
+		}
+	}
+}
+
+// lineScan's contract, at a cap small enough to exercise directly: a line
+// within the cap passes through untouched, a longer one comes back
+// truncated with the reader positioned at the NEXT line.
+func TestLineScan_TruncatesInsteadOfFailing(t *testing.T) {
+	ls := newLineScan(strings.NewReader("short\n"+strings.Repeat("y", 64)+"\nalso short\n"), 16)
+
+	line, trunc, err := ls.next()
+	if err != nil || trunc || line != "short" {
+		t.Fatalf("first line = %q trunc=%v err=%v, want \"short\" untouched", line, trunc, err)
+	}
+	line, trunc, err = ls.next()
+	if err != nil {
+		t.Fatalf("oversized line errored: %v — the reader must survive it", err)
+	}
+	if !trunc {
+		t.Fatal("64-byte line at a 16-byte cap must report truncated")
+	}
+	if len(line) != 16 {
+		t.Fatalf("truncated line kept %d bytes, want the 16-byte cap", len(line))
+	}
+	line, trunc, err = ls.next()
+	if err != nil || trunc || line != "also short" {
+		t.Fatalf("line after the oversized one = %q trunc=%v err=%v, want \"also short\" — reader position was lost", line, trunc, err)
+	}
+	if _, _, err = ls.next(); err != io.EOF {
+		t.Fatalf("exhausted reader returned %v, want io.EOF", err)
+	}
+}