Add GET /events SSE endpoint (Phase 2)

Implements handleSSE which streams server-sent events to connected clients.
Sets SSE headers, subscribes to the broadcaster, and writes "data: update\n\n"
after each debounced publish signal. The endpoint is unauthenticated.

Also adds Flush() to the responseWriter middleware wrapper so the http.Flusher
assertion in the handler succeeds through the middleware chain.

Built with Raymond (Agent Orchestrator)

Author: vector76

internal/server/handlers_sse.go

@@ -0,0 +1,34 @@
+package server
+
+import (
+	"fmt"
+	"net/http"
+)
+
+// handleSSE streams server-sent events to the client. It sends a minimal
+// "update" event whenever the broadcaster fires, and exits when the request
+// context is cancelled.
+func (s *Server) handleSSE(w http.ResponseWriter, r *http.Request) {
+	flusher, ok := w.(http.Flusher)
+	if !ok {
+		http.Error(w, "streaming not supported", http.StatusInternalServerError)
+		return
+	}
+
+	w.Header().Set("Content-Type", "text/event-stream")
+	w.Header().Set("Cache-Control", "no-cache")
+	w.Header().Set("Connection", "keep-alive")
+
+	ch := s.broadcaster.subscribe()
+	defer s.broadcaster.unsubscribe(ch)
+
+	for {
+		select {
+		case <-ch:
+			fmt.Fprint(w, "data: update\n\n")
+			flusher.Flush()
+		case <-r.Context().Done():
+			return
+		}
+	}
+}

internal/server/handlers_sse_test.go

@@ -0,0 +1,120 @@
+package server
+
+import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+	"time"
+)
+
+func TestSSE_StatusAndContentType(t *testing.T) {
+	srv := crudServer(t)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	req := httptest.NewRequest(http.MethodGet, "/events", nil).WithContext(ctx)
+	w := httptest.NewRecorder()
+
+	done := make(chan struct{})
+	go func() {
+		srv.Router.ServeHTTP(w, req)
+		close(done)
+	}()
+
+	// Give the handler time to write headers.
+	time.Sleep(50 * time.Millisecond)
+	cancel()
+	<-done
+
+	if w.Code != http.StatusOK {
+		t.Fatalf("expected 200, got %d", w.Code)
+	}
+	ct := w.Header().Get("Content-Type")
+	if ct != "text/event-stream" {
+		t.Errorf("expected Content-Type text/event-stream, got %q", ct)
+	}
+}
+
+func TestSSE_ReceivesUpdateAfterPublish(t *testing.T) {
+	srv := crudServer(t)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	req := httptest.NewRequest(http.MethodGet, "/events", nil).WithContext(ctx)
+	w := httptest.NewRecorder()
+
+	done := make(chan struct{})
+	go func() {
+		srv.Router.ServeHTTP(w, req)
+		close(done)
+	}()
+
+	// Wait for handler to subscribe, then publish.
+	time.Sleep(50 * time.Millisecond)
+	srv.broadcaster.publish()
+
+	// Wait for debounce (200ms) plus a little margin.
+	time.Sleep(300 * time.Millisecond)
+
+	cancel()
+	<-done
+
+	body := w.Body.String()
+	if !strings.Contains(body, "data: update\n\n") {
+		t.Errorf("expected body to contain 'data: update\\n\\n', got %q", body)
+	}
+}
+
+func TestSSE_ExitsOnContextCancel(t *testing.T) {
+	srv := crudServer(t)
+
+	ctx, cancel := context.WithCancel(context.Background())
+
+	req := httptest.NewRequest(http.MethodGet, "/events", nil).WithContext(ctx)
+	w := httptest.NewRecorder()
+
+	done := make(chan struct{})
+	go func() {
+		srv.Router.ServeHTTP(w, req)
+		close(done)
+	}()
+
+	time.Sleep(50 * time.Millisecond)
+	cancel()
+
+	select {
+	case <-done:
+		// Handler exited cleanly.
+	case <-time.After(2 * time.Second):
+		t.Fatal("handler did not exit after context cancellation")
+	}
+}
+
+func TestSSE_NoAuthRequired(t *testing.T) {
+	srv := crudServer(t)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	req := httptest.NewRequest(http.MethodGet, "/events", nil).WithContext(ctx)
+	// No Authorization header — must still succeed.
+	w := httptest.NewRecorder()
+
+	done := make(chan struct{})
+	go func() {
+		srv.Router.ServeHTTP(w, req)
+		close(done)
+	}()
+
+	time.Sleep(50 * time.Millisecond)
+	cancel()
+	<-done
+
+	if w.Code != http.StatusOK {
+		t.Fatalf("expected 200 without auth, got %d", w.Code)
+	}
+}

internal/server/server.go

@@ -67,6 +67,7 @@ func New(cfg Config, p StoreProvider) (*Server, error) {
 	srv.Router.Get("/api/v1/health", srv.handleHealth)
 	srv.Router.Get("/api/v1/version", srv.handleVersion)
 	srv.Router.Get("/api/v1/beads/status", srv.handleBeadsStatus)
+	srv.Router.Get("/events", srv.handleSSE)
 
 	// All other API routes require auth
 	srv.Router.Group(func(r chi.Router) {
@@ -168,3 +169,9 @@ func (rw *responseWriter) Write(b []byte) (int, error) {
 	}
 	return rw.ResponseWriter.Write(b)
 }
+
+func (rw *responseWriter) Flush() {
+	if f, ok := rw.ResponseWriter.(http.Flusher); ok {
+		f.Flush()
+	}
+}