Fix panic in runs.Replay when the consumer breaks out early

pkg/productize/runs/replay.go

@@ -22,14 +22,16 @@ func (r *Run) Replay(fromSeq uint64) iter.Seq2[events.Event, error] {
 			return
 		}
 
+		// errStopReplay means yield already returned false; the range-over-func
+		// contract forbids calling yield again, so it must not be reported.
 		if err := replayRemoteEvents(
 			context.Background(),
 			r.client,
 			r.summary.RunID,
 			fromSeq,
 			RemoteCursor{},
 			yield,
-		); err != nil {
+		); err != nil && !errors.Is(err, errStopReplay) {
 			yield(events.Event{}, err)
 		}
 	}

pkg/productize/runs/tail_test.go

@@ -43,6 +43,72 @@ func TestReplayPagesEventsInOrder(t *testing.T) {
 	}
 }
 
+func TestReplayStopsCleanlyWhenConsumerBreaksEarly(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name  string
+		pages []remoteRunEventPage
+	}{
+		{
+			name: "single page",
+			pages: []remoteRunEventPage{{
+				Events: []events.Event{
+					testEvent("run-replay-break", 1, events.EventKindRunStarted),
+					testEvent("run-replay-break", 2, events.EventKindJobStarted),
+				},
+				NextCursor: remoteCursor(time.Unix(2, 0).UTC(), 2),
+			}},
+		},
+		{
+			name: "paged reader",
+			pages: []remoteRunEventPage{
+				{
+					Events: []events.Event{
+						testEvent("run-replay-break", 1, events.EventKindRunStarted),
+					},
+					NextCursor: remoteCursor(time.Unix(1, 0).UTC(), 1),
+					HasMore:    true,
+				},
+				{
+					Events: []events.Event{
+						testEvent("run-replay-break", 2, events.EventKindJobStarted),
+					},
+					NextCursor: remoteCursor(time.Unix(2, 0).UTC(), 2),
+				},
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+
+			reader := &stubDaemonRunReader{eventPages: tt.pages}
+			run := &Run{
+				summary: RunSummary{RunID: "run-replay-break"},
+				client:  reader,
+			}
+
+			var seen []events.Event
+			for ev, err := range run.Replay(0) {
+				if err != nil {
+					t.Fatalf("Replay() unexpected error: %v", err)
+				}
+				seen = append(seen, ev)
+				break
+			}
+
+			if seqs := collectedSeqs(seen); !slices.Equal(seqs, []uint64{1}) {
+				t.Fatalf("Replay() seqs = %v, want [1]", seqs)
+			}
+			if len(reader.eventCalls) != 1 {
+				t.Fatalf("ListRunEvents calls = %d, want 1", len(reader.eventCalls))
+			}
+		})
+	}
+}
+
 func TestReplayReportsIncompatibleSchemaVersion(t *testing.T) {
 	run := &Run{
 		summary: RunSummary{RunID: "run-replay-schema"},