fix(sse): queued events go out before the gap announcement (BUG-2730, codex round 13)

Reading both handlers as state machines: the event channel and the gap
channel are two arms of one select, so with both ready Go picks at
random. Announcing first and then draining events the subscriber queued
BEFORE the hole is the wrong order twice over — the client is told its
position is untrustworthy and then immediately handed IDs that
re-establish one, below the hole; and on an ID-space change those queued
events belong to the space that was just abandoned.

The gap is now latched and the announcement waits for an empty channel.
Nothing is discarded to achieve it, and that restraint is the load-bearing
part on the watch stream: a queued one-shot PUSH cannot be recovered by
any reconcile, so dropping it to make the cursor tidy would destroy the
only copy. Draining cannot starve the announcement either — one event per
iteration, re-checked at the top, so it lands on the first iteration with
nothing queued, immediately when the channel was already empty.

Pinned by asserting the ORDER of the frames with twenty events queued
ahead of the gap. Without the barrier that fails on the first or third
event, roughly half the time per run.

internal/server/handlers_events.go

@@ -407,8 +407,36 @@ func (s *Server) handleSSE(w http.ResponseWriter, r *http.Request) {
 		return true
 	}
 
+	// gapSeen latches a gap the bus reported until the event channel is
+	// empty — see the ordering barrier at the top of the loop.
+	gapSeen := false
+
 	ctx := r.Context()
 	for {
+		// ORDERING BARRIER (BUG-2730). The event channel and the gap channel
+		// are two arms of one select, so when both are ready Go picks at
+		// random — and announcing first, then draining events the subscriber
+		// queued BEFORE the hole, is the wrong order twice over. The client
+		// is told its position is untrustworthy and then handed IDs that
+		// re-establish one, below the hole; and on an ID-space change those
+		// queued events belong to the abandoned space.
+		//
+		// So the announcement waits until the channel is empty. Nothing is
+		// discarded to achieve it, which matters most on the watch stream:
+		// a queued one-shot PUSH is not recoverable by any reconcile, so
+		// dropping it to make the cursor tidy would lose the only copy.
+		//
+		// Draining cannot starve the announcement — the loop delivers one
+		// event per iteration and re-checks, so the announcement lands on the
+		// first iteration where nothing is queued, immediately if the channel
+		// was empty when the gap arrived.
+		if gapSeen && len(ch) == 0 {
+			gapSeen = false
+			if gapAnn.observe() && !announceGap() {
+				return
+			}
+		}
+
 		select {
 		case <-ctx.Done():
 			// Client disconnected
@@ -465,12 +493,9 @@ func (s *Server) handleSSE(w http.ResponseWriter, r *http.Request) {
 			// named on BUG-2730 (replay from the buffer rather than resync;
 			// the dropped events are still in it, 1024 deep against the
 			// channel's 64).
-			if !gapAnn.observe() {
-				continue
-			}
-			if !announceGap() {
-				return
-			}
+			// Latched, not announced here: the barrier at the top of the
+			// loop decides when, so anything already queued goes out first.
+			gapSeen = true
 
 		case <-gapAnn.cool():
 			if gapAnn.flush() && !announceGap() {

internal/server/handlers_stream_gap_test.go

@@ -440,3 +440,61 @@ func assertGapBurstYieldsOneThenOne(t *testing.T, frames <-chan string, gaps cha
 	// silently discarded.
 	waitForFrameWithEvent(t, frames, "sync_required")
 }
+
+// TestQueuedEventsGoOutBeforeTheGapAnnouncement pins the ordering barrier.
+//
+// Both channels are arms of one select, so with both ready Go picks at random.
+// Announcing first and then draining pre-hole events hands the client IDs that
+// re-establish the cursor the announcement just retired, below the hole. The
+// barrier makes the announcement wait for an empty channel.
+//
+// The test raises the gap and publishes in the same breath, then asserts the
+// ORDER of the frames. Without the barrier this fails roughly half the time,
+// which -count makes reliable.
+func TestQueuedEventsGoOutBeforeTheGapAnnouncement(t *testing.T) {
+	srv := testServerWithEvents(t)
+	inner := events.New()
+	bus := &gapEventBus{EventBus: inner, gaps: make(chan struct{}, 1)}
+	srv.SetEventBus(bus)
+	ts := httptest.NewServer(srv)
+	defer ts.Close()
+
+	slug := createTestWorkspace(t, ts.URL, "Test")
+	wsID := workspaceIDForSlug(t, srv, slug)
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	frames := readRawSSEFrames(t, ctx, ts.URL+"/api/v1/events?workspace="+slug, "")
+	waitForFrameWithEvent(t, frames, "connected")
+
+	// Queue events FIRST so they are sitting in the subscriber's channel, then
+	// raise the gap. Both select arms are now ready.
+	const queued = 20
+	for range queued {
+		inner.Publish(events.Event{Type: events.ItemCreated, WorkspaceID: wsID, Collection: "tasks"})
+	}
+	bus.gaps <- struct{}{}
+
+	seenEvents := 0
+	deadline := time.After(5 * time.Second)
+	for {
+		select {
+		case f, ok := <-frames:
+			if !ok {
+				t.Fatalf("stream closed after %d/%d events and no announcement", seenEvents, queued)
+			}
+			switch {
+			case strings.Contains(f, "event: sync_required"):
+				if seenEvents != queued {
+					t.Fatalf("the gap was announced after %d of %d queued events; "+
+						"the remaining ones would re-establish a cursor below the hole", seenEvents, queued)
+				}
+				return
+			case strings.Contains(f, "event: "+events.ItemCreated):
+				seenEvents++
+			}
+		case <-deadline:
+			t.Fatalf("no announcement arrived; saw %d/%d events", seenEvents, queued)
+		}
+	}
+}

internal/server/handlers_watch_events.go

@@ -312,8 +312,36 @@ func (s *Server) handleWatchEventsStream(w http.ResponseWriter, r *http.Request)
 		return true
 	}
 
+	// gapSeen latches a gap until the notification channel is empty — see the
+	// ordering barrier at the top of the loop.
+	gapSeen := false
+
 	ctx := r.Context()
 	for {
+		// ORDERING BARRIER (BUG-2730). The event channel and the gap channel
+		// are two arms of one select, so when both are ready Go picks at
+		// random — and announcing first, then draining events the subscriber
+		// queued BEFORE the hole, is the wrong order twice over. The client
+		// is told its position is untrustworthy and then handed IDs that
+		// re-establish one, below the hole; and on an ID-space change those
+		// queued events belong to the abandoned space.
+		//
+		// So the announcement waits until the channel is empty. Nothing is
+		// discarded to achieve it, which matters most on the watch stream:
+		// a queued one-shot PUSH is not recoverable by any reconcile, so
+		// dropping it to make the cursor tidy would lose the only copy.
+		//
+		// Draining cannot starve the announcement — the loop delivers one
+		// event per iteration and re-checks, so the announcement lands on the
+		// first iteration where nothing is queued, immediately if the channel
+		// was empty when the gap arrived.
+		if gapSeen && len(ch) == 0 {
+			gapSeen = false
+			if gapAnn.observe() && !announceGap() {
+				return
+			}
+		}
+
 		select {
 		case <-ctx.Done():
 			return
@@ -355,12 +383,9 @@ func (s *Server) handleWatchEventsStream(w http.ResponseWriter, r *http.Request)
 			//
 			// Rate-limited per connection with nothing dropped — a gap inside
 			// the window is latched, not discarded. See gapAnnouncer.
-			if !gapAnn.observe() {
-				continue
-			}
-			if !announceGap() {
-				return
-			}
+			// Latched, not announced here: the barrier at the top of the
+			// loop decides when, so anything already queued goes out first.
+			gapSeen = true
 
 		case <-gapAnn.cool():
 			if gapAnn.flush() && !announceGap() {