Real-time collaboration mechanics — Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Add an in-process pub/sub hub plus two new HTTP endpoints (GET /api/stream, POST /api/stream/focus) so two collaborators see Topic activity, agent-job state, and presence on the same Document within ~1 s.

Architecture: New internal/realtime package holds a goroutine-safe hub keyed by source_path. The SSE handler subscribes a client, writes a one-field subscribed handshake frame, activates the subscription for fanout, then streams events. Existing mutation handlers and the agent runner call Hub.Publish after their SQLite write commits — events are advisory; REST endpoints remain authoritative. No SQL migration.

Tech Stack: Go 1.x stdlib (net/http, http.Flusher, sync.RWMutex, channels, context), existing internal/auth, internal/collab, internal/agent packages. modernc.org/sqlite driver (unchanged).

Spec: ../specs/2026-05-13-realtime-collab-design.html. Decisions log: ../specs/2026-05-10-collaborative-annotations-decisions.md.

File structure

New:

• internal/realtime/hub.go — Hub, Subscriber, Event, Principal, ErrHubClosed. Goroutine-safe; ~250 LoC.
• internal/realtime/hub_test.go — unit tests for fanout, slow-consumer drop, no cross-source leak, Subscribe/Activate split, presence recompute, close semantics.
• internal/server/handler_stream.go — handleStream (SSE) and handleStreamFocus (POST). ~200 LoC.
• internal/server/handler_stream_test.go — integration tests with httptest.Server and a small SSE reader helper.
• internal/server/e2e_realtime_test.go — multi-client end-to-end: two simulated browsers, mutations propagate as expected events in the right order.

Modified:

• internal/server/server.go — new routes; Realtime added to Deps.
• internal/server/topics.go — handleCreateTopic, handleAppendTopicMessage call Realtime.Publish after the SQLite write commits.
• internal/server/handler_proposals.go — handleDiscardTopic, handleIncorporate call Realtime.Publish after the SQLite write commits.
• internal/agent/service.go — after CompleteJob, publish topic.message_appended → proposal.created → job.updated for successful incorporate jobs; just job.updated for failures.
• internal/agent/service.go constructor + cmd/wiki-browser/main.go — pass the hub into the service.
• cmd/wiki-browser/main.go — construct the hub before the server mux; close it during shutdown.

Deliberately not changed in this plan (owned by #8):

• internal/server/static/chrome.js — client-side wiring lives with the UI integration sub-project.
• internal/server/templates/shell.html — no chrome changes.

Conventions every task assumes

• Working directory. All paths are relative to /home/volrath/code/orcha/wiki-browser/. Run go test, go vet, make build from that dir.
• Test command. go test ./internal/realtime/... ./internal/server/... ./internal/agent/... runs the relevant slice.
• Commit message style. Match recent history (git log --oneline -5):
  • wiki-browser: realtime — <thing> for hub work.
  • wiki-browser: server — <thing> for HTTP handler work.
  • wiki-browser: agent — <thing> for runner work.
  • No Co-Authored-By trailer (project convention).
• Test-first. Every task: write the failing test, run it (verify it fails for the right reason), write the minimal implementation, run the test (verify it passes), commit.
• DB seeding. newTestServer in internal/server/handler_doc_test.go already provisions a temp collab.Store with daniel@getorcha.com seeded into users. Use it.

Task 1: Realtime package skeleton — Event, Principal, ErrHubClosed

Files:

• Create: internal/realtime/hub.go
• Create: internal/realtime/hub_test.go

This task establishes the types the rest of the plan refers to. No behaviour yet — just enough surface for a compile-time check.

• Step 1: Write the failing test

// internal/realtime/hub_test.go
package realtime_test

import (
	"errors"
	"testing"

	"github.com/getorcha/wiki-browser/internal/realtime"
)

func TestErrHubClosed_isSentinel(t *testing.T) {
	if !errors.Is(realtime.ErrHubClosed, realtime.ErrHubClosed) {
		t.Fatal("ErrHubClosed must satisfy errors.Is against itself")
	}
	if realtime.ErrHubClosed.Error() == "" {
		t.Fatal("ErrHubClosed must have a non-empty message")
	}
}

func TestEvent_zeroValueCompiles(t *testing.T) {
	var e realtime.Event
	_ = e.Type
	_ = e.Payload
}

func TestPrincipal_zeroValueCompiles(t *testing.T) {
	var p realtime.Principal
	_ = p.UserID
	_ = p.DisplayName
	_ = p.SessionID
}

• Step 2: Run test to verify it fails

go test ./internal/realtime/ -run TestErrHubClosed -v

Expected: FAIL with no Go files in internal/realtime (the package doesn't exist yet).

• Step 3: Write the minimal implementation

// internal/realtime/hub.go
package realtime

import "errors"

// Event is a single SSE frame published to subscribers. Payload must marshal
// to JSON; the SSE handler encodes it inline.
type Event struct {
	Type    string
	Payload any
}

// Principal carries the identity of the SSE-stream consumer. Lives here (not
// imported from internal/auth) so the realtime package stays independent of
// auth's package shape. The SSE handler maps from auth.Principal +
// auth.SessionInfo into this struct on subscribe.
type Principal struct {
	UserID      string
	DisplayName string
	SessionID   string
}

// ErrHubClosed is returned by Hub.Subscribe after Hub.Close has been called.
var ErrHubClosed = errors.New("realtime: hub closed")

• Step 4: Run test to verify it passes

go test ./internal/realtime/ -v

Expected: PASS (three tests).

• Step 5: Commit

git add internal/realtime/
git commit -m "wiki-browser: realtime — package skeleton (Event, Principal, ErrHubClosed)"

Task 2: Hub with Subscribe / Activate / Publish and one-way fanout

The Subscribe/Activate split is load-bearing: it guarantees the SSE handler can write the subscribed handshake frame before any other event reaches the channel.

Files:

• Modify: internal/realtime/hub.go

• Modify: internal/realtime/hub_test.go

• Step 1: Write the failing test

Append to internal/realtime/hub_test.go:

import (
	// ... keep existing imports ...
	"sync"
	"time"
)

func newPrincipal(userID, session string) realtime.Principal {
	return realtime.Principal{UserID: userID, DisplayName: userID, SessionID: session}
}

func drain(t *testing.T, ch <-chan realtime.Event, want int, timeout time.Duration) []realtime.Event {
	t.Helper()
	got := make([]realtime.Event, 0, want)
	deadline := time.After(timeout)
	for len(got) < want {
		select {
		case ev, ok := <-ch:
			if !ok {
				return got
			}
			got = append(got, ev)
		case <-deadline:
			t.Fatalf("timed out waiting for %d events, got %d: %+v", want, len(got), got)
		}
	}
	return got
}

func TestHub_PublishReachesActivatedSubscriber(t *testing.T) {
	hub := realtime.NewHub()
	defer hub.Close()
	sub, err := hub.Subscribe("a.md", newPrincipal("u1", "s1"))
	if err != nil {
		t.Fatal(err)
	}
	hub.Activate(sub)
	hub.Publish("a.md", realtime.Event{Type: "topic.created", Payload: map[string]string{"topic_id": "t1"}})
	evts := drain(t, sub.Events(), 1, 100*time.Millisecond)
	if evts[0].Type != "topic.created" {
		t.Fatalf("got %q, want topic.created", evts[0].Type)
	}
}

func TestHub_PublishBeforeActivateIsDropped(t *testing.T) {
	hub := realtime.NewHub()
	defer hub.Close()
	sub, _ := hub.Subscribe("a.md", newPrincipal("u1", "s1"))
	hub.Publish("a.md", realtime.Event{Type: "topic.created"})
	// No Activate — the event must not be in the channel.
	select {
	case ev := <-sub.Events():
		t.Fatalf("unexpected event before Activate: %+v", ev)
	case <-time.After(20 * time.Millisecond):
		// good
	}
}

func TestHub_NoCrossSourceLeak(t *testing.T) {
	hub := realtime.NewHub()
	defer hub.Close()
	a, _ := hub.Subscribe("a.md", newPrincipal("u1", "s1"))
	b, _ := hub.Subscribe("b.md", newPrincipal("u2", "s2"))
	hub.Activate(a)
	hub.Activate(b)
	hub.Publish("a.md", realtime.Event{Type: "topic.created"})
	got := drain(t, a.Events(), 1, 100*time.Millisecond)
	if got[0].Type != "topic.created" {
		t.Fatalf("a got %q", got[0].Type)
	}
	select {
	case ev := <-b.Events():
		t.Fatalf("b received cross-source event: %+v", ev)
	case <-time.After(20 * time.Millisecond):
	}
}

func TestHub_FanoutToMultipleSubscribersOnSameSource(t *testing.T) {
	hub := realtime.NewHub()
	defer hub.Close()
	a, _ := hub.Subscribe("a.md", newPrincipal("u1", "s1"))
	b, _ := hub.Subscribe("a.md", newPrincipal("u2", "s2"))
	hub.Activate(a)
	hub.Activate(b)
	hub.Publish("a.md", realtime.Event{Type: "topic.created"})
	var wg sync.WaitGroup
	for _, s := range []*realtime.Subscriber{a, b} {
		wg.Add(1)
		go func(s *realtime.Subscriber) {
			defer wg.Done()
			drain(t, s.Events(), 1, 100*time.Millisecond)
		}(s)
	}
	wg.Wait()
}

func TestHub_SubscribeAfterCloseFails(t *testing.T) {
	hub := realtime.NewHub()
	hub.Close()
	if _, err := hub.Subscribe("a.md", newPrincipal("u1", "s1")); err == nil || err.Error() != realtime.ErrHubClosed.Error() {
		t.Fatalf("Subscribe after Close: err=%v, want ErrHubClosed", err)
	}
}

• Step 2: Run tests to verify they fail

go test ./internal/realtime/ -v

Expected: build failure (undefined: realtime.NewHub, realtime.Subscriber, etc.).

• Step 3: Implement the hub

Replace internal/realtime/hub.go with:

// internal/realtime/hub.go
package realtime

import (
	"errors"
	"sync"

	"github.com/google/uuid"
)

// Event is a single SSE frame published to subscribers.
type Event struct {
	Type    string
	Payload any
}

// Principal carries the identity of the SSE-stream consumer.
type Principal struct {
	UserID      string
	DisplayName string
	SessionID   string
}

// ErrHubClosed is returned by Subscribe after Close has been called.
var ErrHubClosed = errors.New("realtime: hub closed")

// subscriberBufferSize is the per-subscriber event channel capacity. See the
// "Backpressure" section of the design spec for the rationale (64 absorbs the
// 3-event burst of an Agent incorporation plus prior running transitions and
// presence churn, with ample headroom; 16 KiB per subscriber stays trivial on
// the Pi).
const subscriberBufferSize = 64

// Hub is the in-process pub/sub fanout. Goroutine-safe. Keyed by source_path.
type Hub struct {
	mu      sync.RWMutex
	closed  bool
	sources map[string]map[*Subscriber]struct{}
}

// Subscriber represents one live SSE stream. Owned by the calling handler;
// must be closed when the handler returns.
type Subscriber struct {
	id     string
	source string
	princ  Principal

	mu             sync.Mutex
	focusedTopicID string
	events         chan Event
	closed         bool
	activated      bool

	hub *Hub
}

func NewHub() *Hub {
	return &Hub{sources: map[string]map[*Subscriber]struct{}{}}
}

// Subscribe allocates a Subscriber for sourcePath but does NOT register it for
// fanout. The caller writes a `subscribed` handshake frame onto the SSE
// stream, then calls Activate. Splitting the two operations guarantees
// `subscribed` is the first frame on the wire — concurrent Publish calls on
// the same source cannot deliver into this Subscriber's channel until
// Activate runs.
func (h *Hub) Subscribe(sourcePath string, p Principal) (*Subscriber, error) {
	h.mu.RLock()
	closed := h.closed
	h.mu.RUnlock()
	if closed {
		return nil, ErrHubClosed
	}
	return &Subscriber{
		id:     uuid.NewString(),
		source: sourcePath,
		princ:  p,
		events: make(chan Event, subscriberBufferSize),
		hub:    h,
	}, nil
}

// Activate joins the subscriber to the source's fanout set.
func (h *Hub) Activate(s *Subscriber) {
	h.mu.Lock()
	defer h.mu.Unlock()
	if h.closed {
		return
	}
	s.mu.Lock()
	s.activated = true
	s.mu.Unlock()
	set := h.sources[s.source]
	if set == nil {
		set = map[*Subscriber]struct{}{}
		h.sources[s.source] = set
	}
	set[s] = struct{}{}
}

// Publish fans an event to every activated subscriber on sourcePath. Non-
// blocking: if a subscriber's buffer is full, the hub asynchronously closes
// that subscriber and the SSE handler will return + the client will reconnect
// + refetch via REST.
func (h *Hub) Publish(sourcePath string, evt Event) {
	h.mu.RLock()
	set := h.sources[sourcePath]
	subs := make([]*Subscriber, 0, len(set))
	for s := range set {
		subs = append(subs, s)
	}
	h.mu.RUnlock()
	for _, s := range subs {
		s.deliver(evt)
	}
}

// Close shuts down the hub. After Close, Subscribe returns ErrHubClosed.
// Existing subscribers' channels are closed.
func (h *Hub) Close() {
	h.mu.Lock()
	if h.closed {
		h.mu.Unlock()
		return
	}
	h.closed = true
	subs := []*Subscriber{}
	for _, set := range h.sources {
		for s := range set {
			subs = append(subs, s)
		}
	}
	h.sources = map[string]map[*Subscriber]struct{}{}
	h.mu.Unlock()
	for _, s := range subs {
		s.closeChan()
	}
}

// ID returns the server-generated subscriber_id sent in the `subscribed` event.
func (s *Subscriber) ID() string { return s.id }

// Source returns the source_path this subscription is bound to.
func (s *Subscriber) Source() string { return s.source }

// Principal returns the bound Principal (read-only).
func (s *Subscriber) Principal() Principal { return s.princ }

// Events is the read-only stream the SSE handler ranges over.
func (s *Subscriber) Events() <-chan Event { return s.events }

// Close removes the subscriber from the hub's fanout set and closes its
// channel. Idempotent. Safe to call from the SSE handler's defer.
func (s *Subscriber) Close() {
	s.hub.removeSubscriber(s)
}

func (s *Subscriber) deliver(evt Event) {
	s.mu.Lock()
	if s.closed {
		s.mu.Unlock()
		return
	}
	select {
	case s.events <- evt:
		s.mu.Unlock()
	default:
		// Buffer full → slow consumer. Close and let the SSE handler exit;
		// the client reconnects + refetches via REST.
		s.mu.Unlock()
		go s.hub.removeSubscriber(s)
	}
}

func (s *Subscriber) closeChan() {
	s.mu.Lock()
	if s.closed {
		s.mu.Unlock()
		return
	}
	s.closed = true
	close(s.events)
	s.mu.Unlock()
}

func (h *Hub) removeSubscriber(s *Subscriber) {
	h.mu.Lock()
	if set, ok := h.sources[s.source]; ok {
		delete(set, s)
		if len(set) == 0 {
			delete(h.sources, s.source)
		}
	}
	h.mu.Unlock()
	s.closeChan()
}

• Step 4: Run tests to verify they pass

go test ./internal/realtime/ -v

Expected: all tests pass.

• Step 5: Commit

git add internal/realtime/
git commit -m "wiki-browser: realtime — hub with Subscribe/Activate fanout"

Task 3: Backpressure — slow consumer dropped at buffer overflow

Files:

• Modify: internal/realtime/hub_test.go

The hub already implements drop-on-full in Task 2's deliver. This task tests the contract explicitly.

• Step 1: Write the failing test

Append to internal/realtime/hub_test.go:

func TestHub_SlowConsumerDropped(t *testing.T) {
	hub := realtime.NewHub()
	defer hub.Close()
	sub, _ := hub.Subscribe("a.md", newPrincipal("u1", "s1"))
	hub.Activate(sub)

	// Fill the buffer (64) without reading.
	for i := 0; i < 64; i++ {
		hub.Publish("a.md", realtime.Event{Type: "x"})
	}
	// One more should trigger backpressure → close.
	hub.Publish("a.md", realtime.Event{Type: "overflow"})

	// Drain whatever fit, then verify the channel closes.
	deadline := time.After(200 * time.Millisecond)
	gotClose := false
	for !gotClose {
		select {
		case _, ok := <-sub.Events():
			if !ok {
				gotClose = true
			}
		case <-deadline:
			t.Fatal("subscriber channel never closed after overflow")
		}
	}
}

func TestHub_PublishNeverBlocks(t *testing.T) {
	hub := realtime.NewHub()
	defer hub.Close()
	sub, _ := hub.Subscribe("a.md", newPrincipal("u1", "s1"))
	hub.Activate(sub)
	done := make(chan struct{})
	go func() {
		for i := 0; i < 1000; i++ {
			hub.Publish("a.md", realtime.Event{Type: "x"})
		}
		close(done)
	}()
	select {
	case <-done:
	case <-time.After(500 * time.Millisecond):
		t.Fatal("Publish blocked on a slow subscriber")
	}
}

• Step 2: Run tests

go test ./internal/realtime/ -run 'TestHub_(SlowConsumerDropped|PublishNeverBlocks)' -v

Expected: both pass (impl from Task 2 already satisfies them). If either fails, fix the implementation before continuing — backpressure is correctness, not nice-to-have.

• Step 3: Commit (tests only)

git add internal/realtime/hub_test.go
git commit -m "wiki-browser: realtime — backpressure tests"

Task 4: SetFocus and presence recompute on subscribe/activate/close

Files:

• Modify: internal/realtime/hub.go
• Modify: internal/realtime/hub_test.go

Presence is emitted as presence.updated events with payload {subscriptions: [...]} — one entry per active subscription on the source. The hub recomputes and republishes on every Activate, SetFocus, and removeSubscriber.

• Step 1: Write the failing test

Append to hub_test.go:

type presencePayload struct {
	Subscriptions []presenceEntry `json:"subscriptions"`
}

type presenceEntry struct {
	SubscriberID   string `json:"subscriber_id"`
	UserID         string `json:"user_id"`
	DisplayName    string `json:"display_name"`
	FocusedTopicID string `json:"focused_topic_id,omitempty"`
}

func presenceFrom(t *testing.T, ev realtime.Event) presencePayload {
	t.Helper()
	if ev.Type != "presence.updated" {
		t.Fatalf("not presence.updated: %s", ev.Type)
	}
	p, ok := ev.Payload.(realtime.PresencePayload)
	if !ok {
		t.Fatalf("payload type = %T, want realtime.PresencePayload", ev.Payload)
	}
	out := presencePayload{}
	for _, s := range p.Subscriptions {
		out.Subscriptions = append(out.Subscriptions, presenceEntry{
			SubscriberID: s.SubscriberID, UserID: s.UserID,
			DisplayName: s.DisplayName, FocusedTopicID: s.FocusedTopicID,
		})
	}
	return out
}

func TestHub_PresenceEmittedOnActivate(t *testing.T) {
	hub := realtime.NewHub()
	defer hub.Close()
	a, _ := hub.Subscribe("a.md", newPrincipal("daniel@x", "s1"))
	hub.Activate(a)
	evts := drain(t, a.Events(), 1, 100*time.Millisecond)
	p := presenceFrom(t, evts[0])
	if len(p.Subscriptions) != 1 || p.Subscriptions[0].UserID != "daniel@x" {
		t.Fatalf("snapshot = %+v", p)
	}
}

func TestHub_PresenceRecomputeOnSecondActivate(t *testing.T) {
	hub := realtime.NewHub()
	defer hub.Close()
	a, _ := hub.Subscribe("a.md", newPrincipal("daniel@x", "s1"))
	hub.Activate(a)
	drain(t, a.Events(), 1, 100*time.Millisecond) // initial snapshot
	b, _ := hub.Subscribe("a.md", newPrincipal("max@y", "s2"))
	hub.Activate(b)
	// Both subscribers receive an updated snapshot.
	got := drain(t, a.Events(), 1, 100*time.Millisecond)
	pA := presenceFrom(t, got[0])
	if len(pA.Subscriptions) != 2 {
		t.Fatalf("after second Activate, a saw %d subs", len(pA.Subscriptions))
	}
	got = drain(t, b.Events(), 1, 100*time.Millisecond)
	pB := presenceFrom(t, got[0])
	if len(pB.Subscriptions) != 2 {
		t.Fatalf("after second Activate, b saw %d subs", len(pB.Subscriptions))
	}
}

func TestHub_SetFocusRepublishesPresence(t *testing.T) {
	hub := realtime.NewHub()
	defer hub.Close()
	a, _ := hub.Subscribe("a.md", newPrincipal("daniel@x", "s1"))
	b, _ := hub.Subscribe("a.md", newPrincipal("max@y", "s2"))
	hub.Activate(a)
	hub.Activate(b)
	// Drain the two initial snapshots from both subscribers.
	drain(t, a.Events(), 2, 100*time.Millisecond)
	drain(t, b.Events(), 1, 100*time.Millisecond)

	a.SetFocus("topic-123")
	got := drain(t, b.Events(), 1, 100*time.Millisecond)
	p := presenceFrom(t, got[0])
	var aEntry presenceEntry
	for _, e := range p.Subscriptions {
		if e.UserID == "daniel@x" {
			aEntry = e
		}
	}
	if aEntry.FocusedTopicID != "topic-123" {
		t.Fatalf("after SetFocus, daniel's focus = %q, want topic-123", aEntry.FocusedTopicID)
	}
}

func TestHub_RemoveSubscriberRepublishesPresence(t *testing.T) {
	hub := realtime.NewHub()
	defer hub.Close()
	a, _ := hub.Subscribe("a.md", newPrincipal("daniel@x", "s1"))
	b, _ := hub.Subscribe("a.md", newPrincipal("max@y", "s2"))
	hub.Activate(a)
	hub.Activate(b)
	drain(t, a.Events(), 2, 100*time.Millisecond)
	drain(t, b.Events(), 1, 100*time.Millisecond)
	a.Close()
	got := drain(t, b.Events(), 1, 100*time.Millisecond)
	p := presenceFrom(t, got[0])
	if len(p.Subscriptions) != 1 || p.Subscriptions[0].UserID != "max@y" {
		t.Fatalf("after a.Close, b saw %+v", p.Subscriptions)
	}
}

• Step 2: Run tests

go test ./internal/realtime/ -v

Expected: build failure (undefined: realtime.PresencePayload, (*Subscriber).SetFocus).

• Step 3: Implement presence

Add to internal/realtime/hub.go:

// PresencePayload is the JSON-marshalable payload of presence.updated events.
// Each entry corresponds to one live subscription — two tabs from the same
// user produce two entries with the same user_id.
type PresencePayload struct {
	Subscriptions []PresenceEntry `json:"subscriptions"`
}

type PresenceEntry struct {
	SubscriberID   string `json:"subscriber_id"`
	UserID         string `json:"user_id"`
	DisplayName    string `json:"display_name"`
	FocusedTopicID string `json:"focused_topic_id,omitempty"`
}

Modify (*Hub).Activate to call presence recompute at the end:

func (h *Hub) Activate(s *Subscriber) {
	h.mu.Lock()
	if h.closed {
		h.mu.Unlock()
		return
	}
	s.mu.Lock()
	s.activated = true
	s.mu.Unlock()
	set := h.sources[s.source]
	if set == nil {
		set = map[*Subscriber]struct{}{}
		h.sources[s.source] = set
	}
	set[s] = struct{}{}
	h.mu.Unlock()
	h.publishPresence(s.source)
}

Modify (*Hub).removeSubscriber to publish after removal:

func (h *Hub) removeSubscriber(s *Subscriber) {
	h.mu.Lock()
	source := s.source
	if set, ok := h.sources[source]; ok {
		delete(set, s)
		if len(set) == 0 {
			delete(h.sources, source)
		}
	}
	h.mu.Unlock()
	s.closeChan()
	h.publishPresence(source)
}

Add SetFocus to Subscriber and publishPresence to Hub:

// SetFocus updates the subscription's focused Topic and triggers a presence
// republish on this source.
func (s *Subscriber) SetFocus(topicID string) {
	s.mu.Lock()
	s.focusedTopicID = topicID
	activated := s.activated
	s.mu.Unlock()
	if activated {
		s.hub.publishPresence(s.source)
	}
}

func (h *Hub) publishPresence(sourcePath string) {
	h.mu.RLock()
	set := h.sources[sourcePath]
	entries := make([]PresenceEntry, 0, len(set))
	subs := make([]*Subscriber, 0, len(set))
	for s := range set {
		s.mu.Lock()
		entries = append(entries, PresenceEntry{
			SubscriberID:   s.id,
			UserID:         s.princ.UserID,
			DisplayName:    s.princ.DisplayName,
			FocusedTopicID: s.focusedTopicID,
		})
		s.mu.Unlock()
		subs = append(subs, s)
	}
	h.mu.RUnlock()
	evt := Event{Type: "presence.updated", Payload: PresencePayload{Subscriptions: entries}}
	for _, s := range subs {
		s.deliver(evt)
	}
}

NOTE on ordering. The presence entries' order is unstable (map iteration). Tests must not rely on order — they filter by UserID. If a deterministic order is ever needed for the wire, sort by SubscriberID inside publishPresence — but v1 does not require it.

• Step 4: Run tests

go test ./internal/realtime/ -v

Expected: all pass.

• Step 5: Commit

git add internal/realtime/
git commit -m "wiki-browser: realtime — presence model (per-subscription, recomputed)"

Task 5: Wire Realtime into Deps, cmd/wiki-browser/main.go, and agent.Service

This task gets the hub instantiated and threaded through the dependency graph so subsequent tasks can use it. Smoke test: the existing test suite still passes.

Files:

• Modify: internal/server/server.go

• Modify: internal/server/handler_doc_test.go

• Modify: internal/agent/service.go

• Modify: cmd/wiki-browser/main.go

• Step 1: Write the failing test

Append to internal/server/handler_doc_test.go:

import (
	// existing imports …
	"github.com/getorcha/wiki-browser/internal/realtime"
)

func TestNewTestServer_acceptsRealtimeHub(t *testing.T) {
	// Smoke check that passing a hub through Deps compiles and the existing
	// routes still respond. This is a regression guard for the wiring change.
	hub := realtime.NewHub()
	t.Cleanup(hub.Close)
	ts, _, _ := newTestServerWithOptions(t, testServerOptions{
		SessionMiddleware: testSessionMiddleware("daniel@getorcha.com", "Daniel", "csrf"),
		Realtime:          hub,
	})
	resp, err := http.Get(ts.URL + "/")
	if err != nil {
		t.Fatal(err)
	}
	defer resp.Body.Close()
	if resp.StatusCode != 200 {
		t.Fatalf("status = %d", resp.StatusCode)
	}
}

• Step 2: Run test to verify it fails

go test ./internal/server/ -run TestNewTestServer_acceptsRealtimeHub -v

Expected: build failure (testServerOptions.Realtime undefined, Deps.Realtime undefined).

• Step 3: Add Realtime to Deps in internal/server/server.go

Modify the Deps struct (around line 14):

type Deps struct {
	Title  string
	Root   string
	Walker *walker.Walker
	Index  *index.Index
	Cache  *render.Cache
	Collab *collab.Store

	Auth              *auth.Handlers
	SessionMiddleware func(http.Handler) http.Handler
	AuthDevMode       bool

	AgentService AgentService

	// Realtime is the in-process pub/sub hub for SSE. Nil disables
	// /api/stream + /api/stream/focus with 503.
	Realtime *realtime.Hub
}

Add the internal/realtime import to the imports block.

• Step 4: Add Realtime to testServerOptions and pass it through Mux

In internal/server/handler_doc_test.go, extend testServerOptions:

type testServerOptions struct {
	SessionMiddleware func(http.Handler) http.Handler
	AgentService      server.AgentService
	Realtime          *realtime.Hub
}

Pass it into server.Mux(...):

mux := server.Mux(server.Deps{
	// … existing fields …
	Realtime: opts.Realtime,
})

• Step 5: Add the hub to agent.ServiceConfig and agent.Service

In internal/agent/service.go, extend ServiceConfig:

type ServiceConfig struct {
	// … existing fields …

	// Realtime is the hub the runner publishes job + proposal events on.
	// Nil disables event publishing (existing tests don't need it; production
	// always passes one).
	Realtime *realtime.Hub
}

Add internal/realtime import. Add a helper:

func (s *Service) publish(sourcePath string, evt realtime.Event) {
	if s.cfg.Realtime == nil {
		return
	}
	s.cfg.Realtime.Publish(sourcePath, evt)
}

(Tasks 9–11 will call this from (*Service).run.)

• Step 6: Instantiate the hub in cmd/wiki-browser/main.go

After collabStore is opened (around line 116) and before the agent service is constructed:

hub := realtime.NewHub()
defer hub.Close()

Defer-order note. run() returns through srv.Shutdown(...) in the signal-handler branch, then unwinds its deferred functions in LIFO order before returning to main. The hub must shut down after Shutdown drains in-flight SSE handlers — otherwise the handlers' defer sub.Close() would race a hub that's already torn down. Placing defer hub.Close() between collabStore open and agentSvc construction makes it later (closer to the top) than defer agentSvc.Stop(); LIFO means agentSvc.Stop() fires first (drains in-flight agent jobs that may still call Hub.Publish), then hub.Close(). That's the right order. Don't introduce a defer srv.Shutdown(...) — the existing select-branch path already calls it.

Add the import. Pass Realtime: hub into agent.NewService(agent.ServiceConfig{ ... }) and server.Mux(server.Deps{ ... }).

• Step 7: Run tests

go test ./internal/server/ ./internal/agent/ ./internal/realtime/ -v
go vet ./...

Expected: all pass, no vet warnings.

• Step 8: Build the binary as a final smoke

make build

Expected: builds dist/wiki-browser and dist/wb-agent without errors.

• Step 9: Commit

git add internal/server/server.go internal/server/handler_doc_test.go internal/agent/service.go cmd/wiki-browser/main.go
git commit -m "wiki-browser: wire realtime.Hub into Deps, agent.Service, main"

Task 6: GET /api/stream — happy path

Implement the SSE handler's happy path: validate source path, validate file exists, subscribe, write subscribed frame, activate, fan out events. Defer keepalive and session-touch handling to the next task.

Files:

• Create: internal/server/handler_stream.go

• Create: internal/server/handler_stream_test.go

• Modify: internal/server/server.go (register route)

• Step 1: Write the failing test

Create internal/server/handler_stream_test.go:

package server_test

import (
	"bufio"
	"encoding/json"
	"net/http"
	"strings"
	"testing"
	"time"

	"github.com/getorcha/wiki-browser/internal/realtime"
)

// sseEvent is one parsed SSE frame (event: + data: lines).
type sseEvent struct {
	Type string
	Data string
}

// openSSE opens the stream and returns a channel of parsed events plus a
// cancel func. The channel closes when the response body closes.
func openSSE(t *testing.T, url string) (<-chan sseEvent, func()) {
	t.Helper()
	req, err := http.NewRequest(http.MethodGet, url, nil)
	if err != nil {
		t.Fatal(err)
	}
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	if resp.StatusCode != http.StatusOK {
		body, _ := bufio.NewReader(resp.Body).ReadString('\n')
		resp.Body.Close()
		t.Fatalf("stream status = %d body=%q", resp.StatusCode, body)
	}
	out := make(chan sseEvent, 32)
	go func() {
		defer close(out)
		// Body close is owned by the returned cancel func — closing here
		// would double-Close.
		scanner := bufio.NewScanner(resp.Body)
		scanner.Buffer(make([]byte, 64<<10), 1<<20)
		var ev sseEvent
		for scanner.Scan() {
			line := scanner.Text()
			switch {
			case line == "":
				if ev.Type != "" {
					out <- ev
					ev = sseEvent{}
				}
			case strings.HasPrefix(line, "event: "):
				ev.Type = strings.TrimPrefix(line, "event: ")
			case strings.HasPrefix(line, "data: "):
				ev.Data = strings.TrimPrefix(line, "data: ")
			case strings.HasPrefix(line, ":"):
				// comment — keepalive, ignored
			}
		}
	}()
	return out, func() { _ = resp.Body.Close() }
}

func waitForEvent(t *testing.T, ch <-chan sseEvent, wantType string, timeout time.Duration) sseEvent {
	t.Helper()
	deadline := time.After(timeout)
	for {
		select {
		case ev, ok := <-ch:
			if !ok {
				t.Fatalf("stream closed waiting for %s", wantType)
			}
			if ev.Type == wantType {
				return ev
			}
		case <-deadline:
			t.Fatalf("timeout waiting for %s", wantType)
		}
	}
}

func TestStream_subscribedFrameFirstThenPresence(t *testing.T) {
	hub := realtime.NewHub()
	t.Cleanup(hub.Close)
	ts, _, _ := newTestServerWithOptions(t, testServerOptions{
		SessionMiddleware: testSessionMiddleware("daniel@getorcha.com", "Daniel", "csrf"),
		Realtime:          hub,
	})

	ch, cancel := openSSE(t, ts.URL+"/api/stream?source_path=a.md")
	defer cancel()

	// First frame must be `subscribed`.
	ev := <-ch
	if ev.Type != "subscribed" {
		t.Fatalf("first event = %q, want subscribed", ev.Type)
	}
	var sub struct {
		SubscriberID string `json:"subscriber_id"`
	}
	if err := json.Unmarshal([]byte(ev.Data), &sub); err != nil {
		t.Fatalf("decode subscribed: %v", err)
	}
	if sub.SubscriberID == "" {
		t.Fatal("subscriber_id empty")
	}

	// Followed (in some order, but soon) by a presence.updated frame.
	waitForEvent(t, ch, "presence.updated", 200*time.Millisecond)
}

func TestStream_badSourcePathReturns400(t *testing.T) {
	hub := realtime.NewHub()
	t.Cleanup(hub.Close)
	ts, _, _ := newTestServerWithOptions(t, testServerOptions{
		SessionMiddleware: testSessionMiddleware("daniel@getorcha.com", "Daniel", "csrf"),
		Realtime:          hub,
	})
	resp, err := http.Get(ts.URL + "/api/stream?source_path=../escape.md")
	if err != nil {
		t.Fatal(err)
	}
	defer resp.Body.Close()
	if resp.StatusCode != http.StatusBadRequest {
		t.Fatalf("status = %d, want 400", resp.StatusCode)
	}
}

func TestStream_missingFileReturns404(t *testing.T) {
	hub := realtime.NewHub()
	t.Cleanup(hub.Close)
	ts, _, _ := newTestServerWithOptions(t, testServerOptions{
		SessionMiddleware: testSessionMiddleware("daniel@getorcha.com", "Daniel", "csrf"),
		Realtime:          hub,
	})
	resp, err := http.Get(ts.URL + "/api/stream?source_path=does-not-exist.md")
	if err != nil {
		t.Fatal(err)
	}
	defer resp.Body.Close()
	if resp.StatusCode != http.StatusNotFound {
		t.Fatalf("status = %d, want 404", resp.StatusCode)
	}
}

func TestStream_anonymousReturns401(t *testing.T) {
	hub := realtime.NewHub()
	t.Cleanup(hub.Close)
	ts, _, _ := newTestServerWithOptions(t, testServerOptions{
		SessionMiddleware: nil,
		Realtime:          hub,
	})
	resp, err := http.Get(ts.URL + "/api/stream?source_path=a.md")
	if err != nil {
		t.Fatal(err)
	}
	defer resp.Body.Close()
	if resp.StatusCode != http.StatusUnauthorized {
		t.Fatalf("status = %d, want 401", resp.StatusCode)
	}
}

func TestStream_realtimeNilReturns503(t *testing.T) {
	ts, _, _ := newTestServerWithOptions(t, testServerOptions{
		SessionMiddleware: testSessionMiddleware("daniel@getorcha.com", "Daniel", "csrf"),
		Realtime:          nil,
	})
	resp, err := http.Get(ts.URL + "/api/stream?source_path=a.md")
	if err != nil {
		t.Fatal(err)
	}
	defer resp.Body.Close()
	if resp.StatusCode != http.StatusServiceUnavailable {
		t.Fatalf("status = %d, want 503", resp.StatusCode)
	}
}

• Step 2: Run tests to verify they fail

go test ./internal/server/ -run TestStream -v

Expected: 404 on /api/stream (route not registered).

• Step 3: Implement the SSE handler

Create internal/server/handler_stream.go:

package server

import (
	"encoding/json"
	"fmt"
	"net/http"
	"os"
	"path/filepath"

	"github.com/getorcha/wiki-browser/internal/auth"
	"github.com/getorcha/wiki-browser/internal/collab"
	"github.com/getorcha/wiki-browser/internal/realtime"
)

// handleStream implements GET /api/stream?source_path=…
//
// SSE protocol: one HTTP response, kept open. Frames are
//
//	event: <type>
//	data: <JSON payload>
//
// followed by a blank line. Keepalive uses `:keepalive` comment lines. The
// stream never emits `id:` lines — v1 has no replay log.
func (d Deps) handleStream(w http.ResponseWriter, r *http.Request) {
	if d.Realtime == nil || d.Collab == nil {
		writeJSONError(w, http.StatusServiceUnavailable, "collab_unavailable")
		return
	}
	sourcePath := r.URL.Query().Get("source_path")
	if _, err := collab.ValidateSourcePath(sourcePath); err != nil {
		writeJSONError(w, http.StatusBadRequest, "bad_source_path")
		return
	}
	abs := filepath.Join(d.Root, sourcePath)
	if info, err := os.Stat(abs); err != nil || info.IsDir() {
		writeJSONError(w, http.StatusNotFound, "unknown_source")
		return
	}

	flusher, ok := w.(http.Flusher)
	if !ok {
		writeJSONError(w, http.StatusInternalServerError, "no_flusher")
		return
	}

	principal, _ := auth.PrincipalFrom(r.Context())
	sessionInfo, _ := auth.SessionFrom(r.Context())

	sub, err := d.Realtime.Subscribe(sourcePath, realtime.Principal{
		UserID:      principal.UserID,
		DisplayName: principal.DisplayName,
		SessionID:   sessionInfo.IDHash,
	})
	if err != nil {
		writeJSONError(w, http.StatusServiceUnavailable, "hub_closed")
		return
	}
	defer sub.Close()

	w.Header().Set("Content-Type", "text/event-stream")
	w.Header().Set("Cache-Control", "no-store")
	w.Header().Set("Connection", "keep-alive")
	w.Header().Set("X-Accel-Buffering", "no")
	w.WriteHeader(http.StatusOK)

	// Handshake — `subscribed` must be the first frame on the wire. Write
	// before Activate so Publish-from-elsewhere cannot race in ahead of it.
	if err := writeSSEEvent(w, "subscribed", map[string]string{"subscriber_id": sub.ID()}); err != nil {
		return
	}
	flusher.Flush()

	d.Realtime.Activate(sub)

	ctx := r.Context()
	for {
		select {
		case <-ctx.Done():
			return
		case ev, ok := <-sub.Events():
			if !ok {
				return
			}
			if err := writeSSEEvent(w, ev.Type, ev.Payload); err != nil {
				return
			}
			flusher.Flush()
		}
	}
}

// writeSSEEvent serialises one SSE frame. `event:` + `data:` only, never `id:`
// — the spec forbids `id:` because v1 has no replay log.
func writeSSEEvent(w http.ResponseWriter, evType string, payload any) error {
	if _, err := fmt.Fprintf(w, "event: %s\n", evType); err != nil {
		return err
	}
	b, err := json.Marshal(payload)
	if err != nil {
		return fmt.Errorf("marshal event %s: %w", evType, err)
	}
	if _, err := fmt.Fprintf(w, "data: %s\n\n", string(b)); err != nil {
		return err
	}
	return nil
}

• Step 4: Register the route in internal/server/server.go

In Mux(...), after the /api/agent/jobs/{id} registration:

mux.Handle("GET /api/stream",
	d.withSession(auth.RequireCollaborator(http.HandlerFunc(d.handleStream))))

• Step 5: Run tests

go test ./internal/server/ -run TestStream -v

Expected: all pass.

• Step 6: Commit

git add internal/server/handler_stream.go internal/server/handler_stream_test.go internal/server/server.go
git commit -m "wiki-browser: server — GET /api/stream SSE handler (happy path)"

Task 7: Keepalive ticker with session touch

The handler must (a) emit a :keepalive comment every 15 s, (b) close the stream when the session is revoked / expired, (c) call TouchSession to extend the sliding expiry so SSE-only observers don't time out.

To make this testable without a 15 s wait, the keepalive interval comes from Deps (config-driven default, test-time override).

Files:

• Modify: internal/server/server.go

• Modify: internal/server/handler_stream.go

• Modify: internal/server/handler_stream_test.go

• Modify: internal/server/handler_doc_test.go

• Step 1: Write the failing test

Append to internal/server/handler_stream_test.go:

func TestStream_keepaliveCommentEmitted(t *testing.T) {
	hub := realtime.NewHub()
	t.Cleanup(hub.Close)
	ts, _, _ := newTestServerWithOptions(t, testServerOptions{
		SessionMiddleware: testSessionMiddleware("daniel@getorcha.com", "Daniel", "csrf"),
		Realtime:          hub,
		StreamKeepalive:   30 * time.Millisecond,
	})
	req, _ := http.NewRequest(http.MethodGet, ts.URL+"/api/stream?source_path=a.md", nil)
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	defer resp.Body.Close()
	scanner := bufio.NewScanner(resp.Body)
	scanner.Buffer(make([]byte, 64<<10), 1<<20)
	deadline := time.Now().Add(500 * time.Millisecond)
	sawKeepalive := false
	for scanner.Scan() && time.Now().Before(deadline) {
		if strings.HasPrefix(scanner.Text(), ":keepalive") {
			sawKeepalive = true
			break
		}
	}
	if !sawKeepalive {
		t.Fatal("no :keepalive line emitted within 500ms")
	}
}

func TestStream_doesNotEmitIDLines(t *testing.T) {
	// Reading first frame from the stream and asserting no `id:` line slipped
	// in — protects against accidental Last-Event-ID handling.
	hub := realtime.NewHub()
	t.Cleanup(hub.Close)
	ts, _, _ := newTestServerWithOptions(t, testServerOptions{
		SessionMiddleware: testSessionMiddleware("daniel@getorcha.com", "Daniel", "csrf"),
		Realtime:          hub,
		StreamKeepalive:   1 * time.Second,
	})
	req, _ := http.NewRequest(http.MethodGet, ts.URL+"/api/stream?source_path=a.md", nil)
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	defer resp.Body.Close()
	scanner := bufio.NewScanner(resp.Body)
	scanner.Buffer(make([]byte, 64<<10), 1<<20)
	deadline := time.Now().Add(200 * time.Millisecond)
	for scanner.Scan() && time.Now().Before(deadline) {
		line := scanner.Text()
		if strings.HasPrefix(line, "id:") {
			t.Fatalf("unexpected id: line: %q", line)
		}
	}
}

• Step 2: Add StreamKeepalive to testServerOptions

In internal/server/handler_doc_test.go:

type testServerOptions struct {
	SessionMiddleware func(http.Handler) http.Handler
	AgentService      server.AgentService
	Realtime          *realtime.Hub
	StreamKeepalive   time.Duration
}

Pass it into Deps:

mux := server.Mux(server.Deps{
	// … existing fields …
	StreamKeepalive: opts.StreamKeepalive,
})

Add "time" to imports.

• Step 3: Add StreamKeepalive to Deps

In internal/server/server.go, add to Deps:

// StreamKeepalive controls the SSE keepalive cadence. Zero means use the
// default (15s). Tests override to make assertions fast.
StreamKeepalive time.Duration

Add "time" to imports.

• Step 4: Run tests to verify they fail

go test ./internal/server/ -run 'TestStream_keepalive|TestStream_doesNotEmitID' -v

Expected: tests fail (handler doesn't emit keepalive).

• Step 5: Implement keepalive in the handler

Replace the loop in internal/server/handler_stream.go:

import (
	// existing imports …
	"time"
)

const defaultKeepaliveInterval = 15 * time.Second

func (d Deps) handleStream(w http.ResponseWriter, r *http.Request) {
	// … existing validation, subscribe, headers, subscribed frame, Activate …

	keepaliveInterval := d.StreamKeepalive
	if keepaliveInterval <= 0 {
		keepaliveInterval = defaultKeepaliveInterval
	}
	ticker := time.NewTicker(keepaliveInterval)
	defer ticker.Stop()

	ctx := r.Context()
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			if _, err := fmt.Fprint(w, ":keepalive\n\n"); err != nil {
				return
			}
			flusher.Flush()
		case ev, ok := <-sub.Events():
			if !ok {
				return
			}
			if err := writeSSEEvent(w, ev.Type, ev.Payload); err != nil {
				return
			}
			flusher.Flush()
		}
	}
}

• Step 6: Run tests

go test ./internal/server/ -run 'TestStream' -v

Expected: keepalive + no-id tests pass; pre-existing TestStream tests still pass.

• Step 7: Commit

git add internal/server/server.go internal/server/handler_stream.go internal/server/handler_stream_test.go internal/server/handler_doc_test.go
git commit -m "wiki-browser: server — SSE keepalive ticker"

Task 8: Keepalive ticks touch last_seen_at; revoked/expired sessions close the stream

This adds the H5 fix from the code review: a long-lived SSE connection participates in the sliding-session expiry that auth.SessionMiddleware already implements per-request.

Files:

• Modify: internal/server/server.go
• Modify: internal/server/handler_stream.go
• Modify: internal/server/handler_stream_test.go
• Modify: internal/server/handler_doc_test.go

The handler needs two things from the surrounding wiring:

1. auth.SessionStore — LookupSession, TouchSession. Already on collab.Store.
2. The session lifetime (so TouchSession can compute the new expires_at).

No clock injection in v1: tests use a short StreamKeepalive (e.g. 30 ms) and read real wall-clock time inside the handler. If a future test needs deterministic timing for the touch-throttle logic, plumb a NowFunc func() time.Time through Deps then — it's a one-field addition.

• Step 1: Write the failing test

Append to internal/server/handler_stream_test.go:

import (
	// existing imports …
	"github.com/getorcha/wiki-browser/internal/auth"
	"github.com/getorcha/wiki-browser/internal/collab"
)

// recordingSessionStore wraps a collab.Store and counts TouchSession calls.
// Per-instance state — never package-level — so each test is isolated.
type recordingSessionStore struct {
	inner *collab.Store

	mu      sync.Mutex
	touches int
}

func (r *recordingSessionStore) LookupSession(idHash string, now time.Time) (collab.Session, bool, error) {
	return r.inner.LookupSession(idHash, now)
}
func (r *recordingSessionStore) TouchSession(idHash string, seenAt, expiresAt time.Time) error {
	r.mu.Lock()
	r.touches++
	r.mu.Unlock()
	return r.inner.TouchSession(idHash, seenAt, expiresAt)
}
func (r *recordingSessionStore) RevokeSession(idHash string, revokedAt time.Time) error {
	return r.inner.RevokeSession(idHash, revokedAt)
}
func (r *recordingSessionStore) RotateSessionCSRF(idHash, csrfHash string) error {
	return r.inner.RotateSessionCSRF(idHash, csrfHash)
}
func (r *recordingSessionStore) Touches() int {
	r.mu.Lock()
	defer r.mu.Unlock()
	return r.touches
}

func TestStream_keepaliveTouchesSession(t *testing.T) {
	hub := realtime.NewHub()
	t.Cleanup(hub.Close)

	// newTestServerWithOptions constructs the recordingSessionStore inside so
	// we get it back as the third return value; this avoids any package-level
	// state that would leak across tests in the same process.
	ts, _, store, rec := newTestServerWithOptionsAndRecordingStore(t, testServerOptions{
		Realtime:            hub,
		StreamKeepalive:     30 * time.Millisecond,
		StreamTouchInterval: 0, // touch every tick
		SessionLifetime:     time.Hour,
	}, "daniel@getorcha.com")
	_ = store

	req, _ := http.NewRequest(http.MethodGet, ts.URL+"/api/stream?source_path=a.md", nil)
	req.AddCookie(&http.Cookie{Name: auth.SessionCookieName, Value: "test-session-token"})
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	defer resp.Body.Close()

	// Poll the touch counter rather than sleeping a fixed interval.
	deadline := time.Now().Add(500 * time.Millisecond)
	for time.Now().Before(deadline) {
		if rec.Touches() >= 1 {
			return
		}
		time.Sleep(20 * time.Millisecond)
	}
	t.Fatalf("TouchSession not called within 500ms (touches=%d)", rec.Touches())
}

func TestStream_revokedSessionClosesStream(t *testing.T) {
	hub := realtime.NewHub()
	t.Cleanup(hub.Close)
	ts, _, store, _ := newTestServerWithOptionsAndRecordingStore(t, testServerOptions{
		Realtime:            hub,
		StreamKeepalive:     30 * time.Millisecond,
		SessionLifetime:     time.Hour,
		StreamTouchInterval: time.Second, // don't touch every tick
	}, "daniel@getorcha.com")

	req, _ := http.NewRequest(http.MethodGet, ts.URL+"/api/stream?source_path=a.md", nil)
	req.AddCookie(&http.Cookie{Name: auth.SessionCookieName, Value: "test-session-token"})
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	defer resp.Body.Close()

	// Revoke the session in the DB.
	if err := store.RevokeSession(auth.TokenHash("test-session-token"), time.Now()); err != nil {
		t.Fatal(err)
	}

	// Read the response body in a goroutine that closes a channel on EOF.
	// bufio.Scanner blocks on Scan(); a wall-clock loop around it does
	// nothing once Scan is already blocked. The goroutine + select is the
	// only deadline-aware pattern here.
	closed := make(chan struct{})
	go func() {
		defer close(closed)
		scanner := bufio.NewScanner(resp.Body)
		scanner.Buffer(make([]byte, 64<<10), 1<<20)
		for scanner.Scan() {
		}
	}()
	select {
	case <-closed:
		// good — stream returned EOF
	case <-time.After(500 * time.Millisecond):
		t.Fatal("stream did not close within 500ms after session revocation")
	}
}