pad/internal/collab/manager_test.go

package collab

import (
	"encoding/json"
	"errors"
	"fmt"
	"net/http"
	"net/http/httptest"
	"net/url"
	"strconv"
	"sync"
	"testing"
	"time"

	"github.com/PerpetualSoftware/pad/internal/models"
	"github.com/gorilla/websocket"
)

// fakeOpLog is an in-memory opLogStore. We don't pull in the real
// store here because the room-manager contract is narrow and the
// store dependency would drag in SQLite + migrations for what is
// fundamentally a goroutine-orchestration test.
type fakeOpLog struct {
	mu      sync.Mutex
	rows    []models.YjsUpdate
	nextID  int64
	failOn  string // if non-empty, AppendYjsUpdate returns this as an error message for any row
	appendN int

	// contentFlushedIDs simulates the items.content_flushed_op_log_id
	// watermark per item (TASK-1309 round 4). Tests populate this
	// to exercise the schema-mismatch + sweeper paths that consult
	// it. Missing key = NULL watermark = "never flushed".
	contentFlushedIDs map[string]int64

	// lastRestoreSeqs simulates the DURABLE items.last_restore_seq column
	// (BUG-2264). Tests populate it to exercise the Join stale-seed fence's
	// durable read (e.g. a fresh RoomManager on the same store = a "restart"
	// with an empty in-memory fast-path). Missing key = NULL = never restored.
	lastRestoreSeqs map[string]int64

	// failLastRestoreSeq makes ItemLastRestoreSeq return an error, to exercise
	// the Join fence's fail-closed path (a flaky DB after restart).
	failLastRestoreSeq bool

	// onListDormantHook fires once per call to
	// ListDormantOpLogItemsBefore, AFTER the listing scan but BEFORE
	// the result is returned to the caller. Used by tests that
	// simulate a row being appended in the race window between
	// "candidate query returns" and "PruneItemOpLogIfDormantBefore
	// runs under the per-item lock". The hook receives the locked
	// store so it can mutate rows directly. nil = no-op.
	//
	// Per Codex review of TASK-1309 round 2 NIT:
	// TestPruneSweepSkipsRowAddedMidSweep wasn't actually exercising
	// the conditional-DELETE path because it inserted the recent row
	// before the listing query.
	onListDormantHook func(f *fakeOpLog)
}

func (f *fakeOpLog) AppendYjsUpdate(itemID string, data []byte, schemaVersion string) (int64, error) {
	f.mu.Lock()
	defer f.mu.Unlock()
	f.appendN++
	if f.failOn != "" {
		return 0, errors.New(f.failOn)
	}
	f.nextID++
	cp := make([]byte, len(data))
	copy(cp, data)
	row := models.YjsUpdate{
		ID:            f.nextID,
		ItemID:        itemID,
		UpdateData:    cp,
		SchemaVersion: schemaVersion,
		CreatedAt:     time.Now().UTC(),
	}
	f.rows = append(f.rows, row)
	return row.ID, nil
}

func (f *fakeOpLog) LoadYjsUpdatesSince(itemID string, sinceID int64) ([]models.YjsUpdate, error) {
	f.mu.Lock()
	defer f.mu.Unlock()
	var out []models.YjsUpdate
	for _, r := range f.rows {
		if r.ItemID == itemID && r.ID > sinceID {
			out = append(out, r)
		}
	}
	return out, nil
}

// LatestYjsUpdateSchemaVersion mirrors the production store: returns
// the most recent op-log row's schema_version AND id for an item, or
// (false) when no rows exist. The fake walks rows[] in append order
// and returns the latest matching entry — same semantic as the
// store's `ORDER BY id DESC LIMIT 1`.
func (f *fakeOpLog) LatestYjsUpdateSchemaVersion(itemID string) (string, int64, bool, error) {
	f.mu.Lock()
	defer f.mu.Unlock()
	for i := len(f.rows) - 1; i >= 0; i-- {
		if f.rows[i].ItemID == itemID {
			return f.rows[i].SchemaVersion, f.rows[i].ID, true, nil
		}
	}
	return "", 0, false, nil
}

// GetItemContentFlushedOpLogID returns the per-item watermark as
// configured by the test via fakeOpLog.contentFlushedIDs (a map
// itemID → id). Missing key = NULL watermark, returned as (0, false).
func (f *fakeOpLog) GetItemContentFlushedOpLogID(itemID string) (int64, bool, error) {
	f.mu.Lock()
	defer f.mu.Unlock()
	if f.contentFlushedIDs == nil {
		return 0, false, nil
	}
	v, ok := f.contentFlushedIDs[itemID]
	return v, ok, nil
}

// PruneYjsUpdatesBefore deletes every row for itemID whose CreatedAt
// is strictly less than the cutoff. The schema-mismatch rebuild path
// passes a far-future cutoff so this becomes "delete every row".
func (f *fakeOpLog) PruneYjsUpdatesBefore(itemID string, before time.Time) (int64, error) {
	f.mu.Lock()
	defer f.mu.Unlock()
	kept := f.rows[:0]
	var pruned int64
	for _, r := range f.rows {
		if r.ItemID == itemID && r.CreatedAt.Before(before) {
			pruned++
			continue
		}
		kept = append(kept, r)
	}
	f.rows = kept
	return pruned, nil
}

// ListDormantOpLogItemsBefore returns item_ids whose entire op-log
// is older than the cutoff. Mirrors the production store's
// `GROUP BY item_id HAVING MAX(created_at) < ?`. The fake doesn't
// model the items.content_flushed_op_log_id watermark — tests that need it
// run against the real store via the server-level GC test.
func (f *fakeOpLog) ListDormantOpLogItemsBefore(before time.Time) ([]string, error) {
	f.mu.Lock()
	// Compute MAX(created_at) per item_id; collect items where MAX < cutoff.
	maxByItem := map[string]time.Time{}
	for _, r := range f.rows {
		if cur, ok := maxByItem[r.ItemID]; !ok || r.CreatedAt.After(cur) {
			maxByItem[r.ItemID] = r.CreatedAt
		}
	}
	var ids []string
	for id, mx := range maxByItem {
		if mx.Before(before) {
			ids = append(ids, id)
		}
	}
	hook := f.onListDormantHook
	f.mu.Unlock()
	// Fire the test hook (if any) AFTER the listing has been
	// computed but BEFORE returning, simulating a row appended in
	// the candidate-query→DELETE race window.
	if hook != nil {
		hook(f)
	}
	return ids, nil
}

// PruneItemOpLogIfDormantBefore atomically deletes every row for
// itemID iff no row exists with CreatedAt >= cutoff. Returns the
// count deleted.
func (f *fakeOpLog) PruneItemOpLogIfDormantBefore(itemID string, before time.Time) (int64, error) {
	f.mu.Lock()
	defer f.mu.Unlock()
	// First pass: is the item still dormant?
	for _, r := range f.rows {
		if r.ItemID == itemID && !r.CreatedAt.Before(before) {
			// A recent row exists — abort, don't delete anything.
			return 0, nil
		}
	}
	// Second pass: delete every row for this item.
	kept := f.rows[:0]
	var deleted int64
	for _, r := range f.rows {
		if r.ItemID == itemID {
			deleted++
			continue
		}
		kept = append(kept, r)
	}
	f.rows = kept
	return deleted, nil
}

// MinOpLogID + MaxOpLogID power TASK-1319's resume-cursor protocol.
func (f *fakeOpLog) MinOpLogID(itemID string) (int64, bool, error) {
	f.mu.Lock()
	defer f.mu.Unlock()
	var minID int64
	found := false
	for _, r := range f.rows {
		if r.ItemID != itemID {
			continue
		}
		if !found || r.ID < minID {
			minID = r.ID
			found = true
		}
	}
	return minID, found, nil
}

func (f *fakeOpLog) MaxOpLogID(itemID string) (int64, bool, error) {
	f.mu.Lock()
	defer f.mu.Unlock()
	var maxID int64
	found := false
	for _, r := range f.rows {
		if r.ItemID != itemID {
			continue
		}
		if !found || r.ID > maxID {
			maxID = r.ID
			found = true
		}
	}
	return maxID, found, nil
}

func (f *fakeOpLog) ItemLastRestoreSeq(itemID string) (int64, bool, error) {
	f.mu.Lock()
	defer f.mu.Unlock()
	if f.failLastRestoreSeq {
		return 0, false, errors.New("simulated durable last_restore_seq read failure")
	}
	v, ok := f.lastRestoreSeqs[itemID]
	return v, ok, nil
}

func (f *fakeOpLog) appendCount() int {
	f.mu.Lock()
	defer f.mu.Unlock()
	return f.appendN
}

func (f *fakeOpLog) rowCount() int {
	f.mu.Lock()
	defer f.mu.Unlock()
	return len(f.rows)
}

// newCollabTestServer wires up an httptest server whose only handler
// upgrades the inbound WS, registers it with the manager, and waits
// for the session to end. Tests dial it via newWSClient below.
func newCollabTestServer(t *testing.T, mgr *RoomManager) *httptest.Server {
	t.Helper()
	upgrader := websocket.Upgrader{}
	mux := http.NewServeMux()
	mux.HandleFunc("/ws/", func(w http.ResponseWriter, r *http.Request) {
		itemID := r.URL.Path[len("/ws/"):]
		conn, err := upgrader.Upgrade(w, r, nil)
		if err != nil {
			t.Errorf("upgrade: %v", err)
			return
		}
		defer conn.Close()
		// `?since=N` forwards the resume cursor into Join (TASK-1319).
		// Empty / unparseable → 0, matching the production handler.
		var since int64
		if raw := r.URL.Query().Get("since"); raw != "" {
			if v, perr := strconv.ParseInt(raw, 10, 64); perr == nil && v > 0 {
				since = v
			}
		}
		var contentSeq int64
		if raw := r.URL.Query().Get("content_seq"); raw != "" {
			if v, perr := strconv.ParseInt(raw, 10, 64); perr == nil && v > 0 {
				contentSeq = v
			}
		}
		// `?readonly=1` joins as a read-only participant (canWrite=false)
		// so tests can exercise the TASK-265 gate. Default is writable.
		canWrite := r.URL.Query().Get("readonly") != "1"
		// `?applier_bracket=1` marks the conn bracket-capable (BUG-2276 residual 2).
		bracketCapable := r.URL.Query().Get("applier_bracket") == "1"
		_ = mgr.Join(itemID, conn, since, contentSeq, canWrite, bracketCapable, nil)
	})
	return httptest.NewServer(mux)
}

func dialWS(t *testing.T, server *httptest.Server, itemID string) *websocket.Conn {
	t.Helper()
	return dialWSSince(t, server, itemID, 0)
}

// waitElectable polls until the room for itemID has at least n conns that pickApplier
// would actually elect — writable, anchored (replayDone), not frozen. Gates tests
// that drive ApplyExternalContent so they don't race the post-replay cursor write:
// an unanchored conn is deliberately NOT elected (BUG-2276 residual 2, P1), because
// its client buffers Y.Doc updates until it receives the initial cursor.
func waitElectable(t *testing.T, mgr *RoomManager, itemID string, n int) {
	t.Helper()
	deadline := time.Now().Add(2 * time.Second)
	for time.Now().Before(deadline) {
		mgr.mu.Lock()
		room := mgr.rooms[itemID]
		mgr.mu.Unlock()
		if room != nil {
			room.mu.Lock()
			c := 0
			for _, rc := range room.conns {
				if rc.canWrite.Load() && rc.replayDone.Load() && !rc.frozen.Load() {
					c++
				}
			}
			room.mu.Unlock()
			if c >= n {
				return
			}
		}
		time.Sleep(2 * time.Millisecond)
	}
	t.Fatalf("room %s did not reach %d electable conns within deadline", itemID, n)
}

// dialWSSince is the resume-cursor variant: appends `?since=<id>` to
// the WS URL so the test exercises the TASK-1319 force_refresh path.
func dialWSSince(t *testing.T, server *httptest.Server, itemID string, since int64) *websocket.Conn {
	t.Helper()
	u, _ := url.Parse(server.URL)
	wsURL := "ws://" + u.Host + "/ws/" + itemID
	if since > 0 {
		wsURL += "?since=" + strconv.FormatInt(since, 10)
	}
	c, _, err := websocket.DefaultDialer.Dial(wsURL, nil)
	if err != nil {
		t.Fatalf("dial: %v", err)
	}
	return c
}

// dialWSContentSeq appends `?content_seq=<seq>` so the test exercises the
// BUG-2264 stale-seed force_refresh fence.
func dialWSContentSeq(t *testing.T, server *httptest.Server, itemID string, contentSeq int64) *websocket.Conn {
	t.Helper()
	u, _ := url.Parse(server.URL)
	wsURL := "ws://" + u.Host + "/ws/" + itemID + "?content_seq=" + strconv.FormatInt(contentSeq, 10)
	c, _, err := websocket.DefaultDialer.Dial(wsURL, nil)
	if err != nil {
		t.Fatalf("dial: %v", err)
	}
	return c
}

// dialWSBracket joins as a bracket-capable participant (`?applier_bracket=1`) so
// the conn can be durably confirmed by a restore's finalization (BUG-2276 residual 2).
func dialWSBracket(t *testing.T, server *httptest.Server, itemID string) *websocket.Conn {
	t.Helper()
	u, _ := url.Parse(server.URL)
	wsURL := "ws://" + u.Host + "/ws/" + itemID + "?applier_bracket=1"
	c, _, err := websocket.DefaultDialer.Dial(wsURL, nil)
	if err != nil {
		t.Fatalf("dial bracket: %v", err)
	}
	return c
}

// dialWSReadOnly joins as a read-only participant (canWrite=false) via
// the harness's `?readonly=1` switch — for TASK-265 tests.
func dialWSReadOnly(t *testing.T, server *httptest.Server, itemID string) *websocket.Conn {
	t.Helper()
	u, _ := url.Parse(server.URL)
	wsURL := "ws://" + u.Host + "/ws/" + itemID + "?readonly=1"
	c, _, err := websocket.DefaultDialer.Dial(wsURL, nil)
	if err != nil {
		t.Fatalf("dial readonly: %v", err)
	}
	return c
}

func sendSync(t *testing.T, c *websocket.Conn, payload byte) {
	t.Helper()
	// Build a minimal y-protocol sync frame: leading 0x00 for "sync",
	// followed by an arbitrary trailing byte that distinguishes one
	// frame from another in test assertions.
	if err := c.WriteMessage(websocket.BinaryMessage, []byte{yMessageSync, payload}); err != nil {
		t.Fatalf("write sync: %v", err)
	}
}

func sendAwareness(t *testing.T, c *websocket.Conn, payload byte) {
	t.Helper()
	if err := c.WriteMessage(websocket.BinaryMessage, []byte{yMessageAwareness, payload}); err != nil {
		t.Fatalf("write awareness: %v", err)
	}
}

// readBinaryWithin reads a single binary frame within d. Returns the
// payload bytes or fails the test on timeout.
func readBinaryWithin(t *testing.T, c *websocket.Conn, d time.Duration) []byte {
	t.Helper()
	deadline := time.Now().Add(d)
	c.SetReadDeadline(deadline)
	defer c.SetReadDeadline(time.Time{})
	for {
		mt, data, err := c.ReadMessage()
		if err != nil {
			t.Fatalf("read: %v", err)
		}
		// Drain TASK-1319 op_log_cursor TextMessage frames so the
		// existing binary-frame assertions stay focused on Y.Doc
		// payload semantics.
		if mt == websocket.TextMessage {
			c.SetReadDeadline(deadline)
			continue
		}
		if mt != websocket.BinaryMessage {
			t.Fatalf("expected binary frame, got %d", mt)
		}
		return data
	}
}

// expectNoMessage checks that no NON-cursor frame arrives within d.
// Cursor TextMessage frames (TASK-1319) are drained without failing
// the assertion — they're metadata, not Y.Doc payload — and the
// caller's intent is "no peer ops leaked through."
func expectNoMessage(t *testing.T, c *websocket.Conn, d time.Duration) {
	t.Helper()
	deadline := time.Now().Add(d)
	c.SetReadDeadline(deadline)
	defer c.SetReadDeadline(time.Time{})
	for {
		mt, data, err := c.ReadMessage()
		if err != nil {
			return // timeout / close — exactly what we expected
		}
		if mt == websocket.TextMessage {
			c.SetReadDeadline(deadline)
			continue
		}
		t.Fatalf("expected no payload frame, got mt=%d data=%v", mt, data)
	}
}

// TestRoomManagerLazyCreate verifies the manager doesn't materialise
// rooms until the first Join.
func TestRoomManagerLazyCreate(t *testing.T) {
	bus := NewMemoryOpBus()
	defer bus.Close()
	mgr := NewRoomManager(&fakeOpLog{}, bus)

	if got := mgr.RoomCount(); got != 0 {
		t.Fatalf("fresh manager: want 0 rooms, got %d", got)
	}

	srv := newCollabTestServer(t, mgr)
	defer srv.Close()

	c := dialWS(t, srv, "item-a")
	defer c.Close()

	// Wait for the join to register.
	deadline := time.Now().Add(2 * time.Second)
	for time.Now().Before(deadline) {
		if mgr.RoomCount() == 1 {
			break
		}
		time.Sleep(5 * time.Millisecond)
	}
	if got := mgr.RoomCount(); got != 1 {
		t.Fatalf("after first Join: want 1 room, got %d", got)
	}
}

// TestRestoreBoundaryMonotonic covers the per-item restore boundary the
// collab-snapshot PATCH handler consults to reject stale pre-restore flushes
// (BUG-2264 Invariant B): unset reads as (0,false); SetRestoreBoundary records
// a value and keeps it monotonically non-decreasing (a later/higher restore
// wins; a stale/lower or non-positive call can never lower it); boundaries are
// isolated per item.
func TestRestoreBoundaryMonotonic(t *testing.T) {
	bus := NewMemoryOpBus()
	defer bus.Close()
	mgr := NewRoomManager(&fakeOpLog{}, bus)

	if v, ok := mgr.RestoreBoundary("item-a"); ok || v != 0 {
		t.Fatalf("unset boundary: want (0,false), got (%d,%v)", v, ok)
	}

	mgr.SetRestoreBoundary("item-a", 10)
	if v, ok := mgr.RestoreBoundary("item-a"); !ok || v != 10 {
		t.Fatalf("after set 10: want (10,true), got (%d,%v)", v, ok)
	}

	// A later restore with a higher op-log MAX advances the boundary.
	mgr.SetRestoreBoundary("item-a", 25)
	if v, _ := mgr.RestoreBoundary("item-a"); v != 25 {
		t.Fatalf("after set 25: want 25, got %d", v)
	}

	// A stale / duplicate call with a lower value can never lower the boundary.
	mgr.SetRestoreBoundary("item-a", 12)
	if v, _ := mgr.RestoreBoundary("item-a"); v != 25 {
		t.Fatalf("after stale set 12: boundary must stay 25, got %d", v)
	}

	// Non-positive values are ignored (no boundary to enforce).
	mgr.SetRestoreBoundary("item-a", 0)
	mgr.SetRestoreBoundary("item-a", -5)
	if v, _ := mgr.RestoreBoundary("item-a"); v != 25 {
		t.Fatalf("after non-positive sets: boundary must stay 25, got %d", v)
	}

	// Boundaries are per item.
	if v, ok := mgr.RestoreBoundary("item-b"); ok || v != 0 {
		t.Fatalf("unrelated item-b: want (0,false), got (%d,%v)", v, ok)
	}
}

// TestForceRefreshRoom verifies the server-initiated force-refresh (BUG-2264):
// it runs the caller's commit, prunes the ENTIRE per-item op-log, and sends a
// force_refresh control frame to every connected client before closing the
// connection — so peers discard their in-memory Y.Doc and rebuild from
// items.content.
func TestForceRefreshRoom(t *testing.T) {
	bus := NewMemoryOpBus()
	defer bus.Close()
	store := &fakeOpLog{}
	mgr := NewRoomManager(store, bus)
	defer mgr.Close()

	srv := newCollabTestServer(t, mgr)
	defer srv.Close()

	conn := dialWS(t, srv, "item-a")
	defer conn.Close()

	for i := 0; i < 200; i++ {
		if mgr.RoomCount() == 1 && bus.SubscriberCount("item-a") == 1 {
			break
		}
		time.Sleep(5 * time.Millisecond)
	}

	// Seed op-log rows to prove the prune wipes them.
	if _, err := store.AppendYjsUpdate("item-a", []byte{0x00, 0x01}, "1"); err != nil {
		t.Fatalf("seed 1: %v", err)
	}
	if _, err := store.AppendYjsUpdate("item-a", []byte{0x00, 0x02}, "1"); err != nil {
		t.Fatalf("seed 2: %v", err)
	}

	// The commit captures the pre-prune MAX and performs the op-log wipe
	// atomically (the real path runs both inside UpdateItem's tx via
	// MaxOpLogIDTx + PruneItemOpLogTx); the fake mirrors that by reading MAX,
	// pruning here, and returning (MAX, restored seq).
	committed := false
	if err := mgr.ForceRefreshRoom("item-a", func() (int64, int64, error) {
		committed = true
		maxID, _, merr := store.MaxOpLogID("item-a")
		if merr != nil {
			return 0, 0, merr
		}
		if _, err := store.PruneYjsUpdatesBefore("item-a", distantFuture); err != nil {
			return 0, 0, err
		}
		return maxID, 42, nil
	}, nil); err != nil {
		t.Fatalf("ForceRefreshRoom: %v", err)
	}
	if !committed {
		t.Fatal("commit callback did not run")
	}

	// Op-log fully pruned (by the commit).
	store.mu.Lock()
	remaining := 0
	for _, r := range store.rows {
		if r.ItemID == "item-a" {
			remaining++
		}
	}
	store.mu.Unlock()
	if remaining != 0 {
		t.Fatalf("op-log not pruned: %d rows remain", remaining)
	}

	// The restore published both fences: the op-log-id boundary (MAX+1) and the
	// content generation (the restored seq).
	if b, ok := mgr.RestoreBoundary("item-a"); !ok || b <= 0 {
		t.Fatalf("restore boundary not published: got (%d, %v)", b, ok)
	}
	if s, ok := mgr.LastRestoreSeq("item-a"); !ok || s != 42 {
		t.Fatalf("last restore seq = (%d, %v), want (42, true)", s, ok)
	}

	// The client received a force_refresh frame and then the conn was closed.
	_ = conn.SetReadDeadline(time.Now().Add(2 * time.Second))
	sawForceRefresh := false
	for {
		mt, data, err := conn.ReadMessage()
		if err != nil {
			break // conn closed by forceRefreshAll
		}
		if mt == websocket.TextMessage {
			var ctl ControlMessage
			if json.Unmarshal(data, &ctl) == nil && ctl.Type == ControlMessageForceRefresh {
				sawForceRefresh = true
			}
		}
	}
	if !sawForceRefresh {
		t.Fatal("client did not receive a force_refresh frame before the conn closed")
	}
}

// TestForceRefreshRoomReconcile covers the Postgres commit-outcome reconciliation
// (BUG-2276 residual 1): when the restore commit returns an error, the supplied
// reconcile callback disambiguates three outcomes — landed (take the success
// path), definitely rolled back (un-freeze + bail), and uncertain (stay frozen +
// bail). The SQLite/self-host path (reconcile == nil) is exercised by
// TestForceRefreshRoom above.
func TestForceRefreshRoomReconcile(t *testing.T) {
	// (a) The commit's tx DURABLY landed but its ack was lost (Postgres), so the
	// commit closure prunes the op-log then returns an error. reconcile reports
	// Landed → ForceRefreshRoom must take the SUCCESS path: publish the recovered
	// boundary + seq and reseed peers, NOT un-freeze onto a stale doc.
	t.Run("landed_takes_success_path", func(t *testing.T) {
		bus := NewMemoryOpBus()
		defer bus.Close()
		store := &fakeOpLog{}
		mgr := NewRoomManager(store, bus)
		defer mgr.Close()

		srv := newCollabTestServer(t, mgr)
		defer srv.Close()

		conn := dialWS(t, srv, "item-a")
		defer conn.Close()
		for i := 0; i < 200; i++ {
			if mgr.RoomCount() == 1 && bus.SubscriberCount("item-a") == 1 {
				break
			}
			time.Sleep(5 * time.Millisecond)
		}

		if _, err := store.AppendYjsUpdate("item-a", []byte{0x00, 0x01}, "1"); err != nil {
			t.Fatalf("seed: %v", err)
		}

		const wantBoundary = int64(8)
		const wantSeq = int64(99)
		reconciled := false
		err := mgr.ForceRefreshRoom("item-a",
			func() (int64, int64, error) {
				// The tx pruned the op-log and committed durably, then the ack was
				// lost at the commit boundary — so it surfaces here as an error.
				if _, perr := store.PruneYjsUpdatesBefore("item-a", distantFuture); perr != nil {
					return 0, 0, perr
				}
				return 0, 0, errors.New("commit: driver: bad connection")
			},
			func() (RestoreReconcileResult, error) {
				reconciled = true
				return RestoreReconcileResult{Landed: true, Boundary: wantBoundary, Seq: wantSeq}, nil
			})
		if err != nil {
			t.Fatalf("ForceRefreshRoom (landed) returned error, want success: %v", err)
		}
		if !reconciled {
			t.Fatal("reconcile callback did not run")
		}
		if b, ok := mgr.RestoreBoundary("item-a"); !ok || b != wantBoundary {
			t.Fatalf("restore boundary = (%d, %v), want (%d, true)", b, ok, wantBoundary)
		}
		if s, ok := mgr.LastRestoreSeq("item-a"); !ok || s != wantSeq {
			t.Fatalf("last restore seq = (%d, %v), want (%d, true)", s, ok, wantSeq)
		}
		// The success path ran the reseed: the peer got a force_refresh frame.
		_ = conn.SetReadDeadline(time.Now().Add(2 * time.Second))
		sawForceRefresh := false
		for {
			mt, data, rerr := conn.ReadMessage()
			if rerr != nil {
				break
			}
			if mt == websocket.TextMessage {
				var ctl ControlMessage
				if json.Unmarshal(data, &ctl) == nil && ctl.Type == ControlMessageForceRefresh {
					sawForceRefresh = true
				}
			}
		}
		if !sawForceRefresh {
			t.Fatal("landed reconcile did not reseed peers (no force_refresh frame)")
		}
	})

	// (b) The commit genuinely rolled back. reconcile reports !Landed →
	// ForceRefreshRoom un-freezes the conn and returns the commit error; no
	// boundary/seq published, peers keep their live Y.Doc.
	t.Run("rolled_back_unfreezes", func(t *testing.T) {
		bus := NewMemoryOpBus()
		defer bus.Close()
		mgr := NewRoomManager(&fakeOpLog{}, bus)
		defer mgr.Close()

		room := mgr.getOrCreate("item-a")
		rc := &roomConn{id: 1, conn: &websocket.Conn{}}
		rc.canWrite.Store(true)
		if err := room.addConn(rc); err != nil {
			t.Fatalf("addConn: %v", err)
		}
		// Drop the fake conn before Close so closeAll doesn't Close() a nil socket.
		defer func() {
			room.mu.Lock()
			delete(room.conns, rc.conn)
			room.mu.Unlock()
		}()

		commitErr := errors.New("commit: rolled back")
		err := mgr.ForceRefreshRoom("item-a",
			func() (int64, int64, error) { return 0, 0, commitErr },
			func() (RestoreReconcileResult, error) {
				return RestoreReconcileResult{Landed: false}, nil
			})
		if !errors.Is(err, commitErr) {
			t.Fatalf("ForceRefreshRoom (rolled back) err = %v, want the commit error", err)
		}
		if rc.frozen.Load() {
			t.Fatal("rolled-back reconcile must UN-freeze the conn, but frozen is still set")
		}
		if _, ok := mgr.RestoreBoundary("item-a"); ok {
			t.Fatal("rolled-back reconcile must NOT publish a restore boundary")
		}
		if _, ok := mgr.LastRestoreSeq("item-a"); ok {
			t.Fatal("rolled-back reconcile must NOT publish a last restore seq")
		}
	})

	// (c) The commit errored AND the reconcile read itself failed → outcome
	// UNCERTAIN. ForceRefreshRoom must return an error wrapping both causes, must
	// NOT publish a boundary/seq, and must PLAIN-CLOSE the peer's socket (NOT
	// force_refresh) so it reconnects and re-evaluates the durable fences fresh —
	// a frozen-but-open socket would silently drop every subsequent edit forever.
	t.Run("uncertain_plain_closes_peers", func(t *testing.T) {
		bus := NewMemoryOpBus()
		defer bus.Close()
		mgr := NewRoomManager(&fakeOpLog{}, bus)
		defer mgr.Close()

		srv := newCollabTestServer(t, mgr)
		defer srv.Close()

		conn := dialWS(t, srv, "item-a")
		defer conn.Close()
		for i := 0; i < 200; i++ {
			if mgr.RoomCount() == 1 && bus.SubscriberCount("item-a") == 1 {
				break
			}
			time.Sleep(5 * time.Millisecond)
		}

		commitErr := errors.New("commit: bad connection")
		reconcileErr := errors.New("reconcile: read failed")
		err := mgr.ForceRefreshRoom("item-a",
			func() (int64, int64, error) { return 0, 0, commitErr },
			func() (RestoreReconcileResult, error) {
				return RestoreReconcileResult{}, reconcileErr
			})
		if err == nil {
			t.Fatal("uncertain reconcile must return an error")
		}
		if !errors.Is(err, commitErr) || !errors.Is(err, reconcileErr) {
			t.Fatalf("uncertain err = %v, want it to wrap BOTH the commit and reconcile errors", err)
		}
		if _, ok := mgr.RestoreBoundary("item-a"); ok {
			t.Fatal("uncertain reconcile must NOT publish a restore boundary")
		}
		if _, ok := mgr.LastRestoreSeq("item-a"); ok {
			t.Fatal("uncertain reconcile must NOT publish a last restore seq")
		}
		// The peer's socket must be PLAIN-closed: no force_refresh frame arrives
		// before the read errors out. (force_refresh would falsely assert
		// items.content is authoritative, which is exactly what's unknown here.)
		_ = conn.SetReadDeadline(time.Now().Add(2 * time.Second))
		closed := false
		for {
			mt, data, rerr := conn.ReadMessage()
			if rerr != nil {
				closed = true
				break // socket closed by closeAllConnsPlain
			}
			if mt == websocket.TextMessage {
				var ctl ControlMessage
				if json.Unmarshal(data, &ctl) == nil && ctl.Type == ControlMessageForceRefresh {
					t.Fatal("uncertain reconcile plain-closes; it must NOT send a force_refresh frame")
				}
			}
		}
		if !closed {
			t.Fatal("uncertain reconcile must close the peer's socket")
		}
	})
}

// TestForceRefreshRoomUncertainInvalidatesInMemoryFences is the BUG-2276
// residual-1 P1 guard: on the UNCERTAIN commit-outcome path, ForceRefreshRoom must
// CLEAR the item's in-memory restore fences (lastRestoreSeqs + restoreBoundaries)
// so a reconnecting stale-seed peer is judged against the DURABLE column — which
// reflects a landed-but-ack-lost restore — not a PRIOR restore's cached generation.
// Otherwise a cursor-0 peer seeded at the prior generation would be admitted against
// the stale cache and clobber the landed content.
func TestForceRefreshRoomUncertainInvalidatesInMemoryFences(t *testing.T) {
	bus := NewMemoryOpBus()
	defer bus.Close()
	// The DURABLE column reflects the LANDED-but-ack-lost restore R2 (seq=20). The
	// in-memory caches still hold the PRIOR restore R1's generation (seq=10 /
	// boundary=100) — the stale-cache trap this guards against.
	store := &fakeOpLog{lastRestoreSeqs: map[string]int64{"item-a": 20}}
	mgr := NewRoomManager(store, bus)
	defer mgr.Close()

	mgr.SetLastRestoreSeq("item-a", 10)   // R1 in-memory fast-path
	mgr.SetRestoreBoundary("item-a", 100) // R1 in-memory boundary
	if s, ok := mgr.LastRestoreSeq("item-a"); !ok || s != 10 {
		t.Fatalf("precondition: R1 last restore seq = (%d,%v), want (10,true)", s, ok)
	}
	if b, ok := mgr.RestoreBoundary("item-a"); !ok || b != 100 {
		t.Fatalf("precondition: R1 boundary = (%d,%v), want (100,true)", b, ok)
	}

	// R2's commit errors AND its reconcile read errors → UNCERTAIN.
	err := mgr.ForceRefreshRoom("item-a",
		func() (int64, int64, error) { return 0, 0, errors.New("commit: bad connection") },
		func() (RestoreReconcileResult, error) {
			return RestoreReconcileResult{}, errors.New("reconcile: read failed")
		})
	if err == nil {
		t.Fatal("uncertain reconcile must return an error")
	}

	// Both in-memory fences must be CLEARED so the next check reads durable.
	if s, ok := mgr.LastRestoreSeq("item-a"); ok {
		t.Fatalf("uncertain must invalidate the in-memory last-restore-seq fence; still cached (%d)", s)
	}
	if b, ok := mgr.RestoreBoundary("item-a"); ok {
		t.Fatalf("uncertain must invalidate the in-memory restore boundary; still cached (%d)", b)
	}

	// A peer seeded at R1's generation (content_seq=10) now reconnecting is judged
	// against the DURABLE column (R2=20): 10 < 20 → force_refresh. With the STALE
	// cache it would have compared 10 vs cached-10 (equal) and been admitted — the
	// clobber this fix prevents.
	srv := newCollabTestServer(t, mgr)
	defer srv.Close()
	stale := dialWSContentSeq(t, srv, "item-a", 10)
	defer stale.Close()
	if ctl := readControlWithin(t, stale, time.Second); ctl.Type != ControlMessageForceRefresh {
		t.Fatalf("stale R1-generation seed after uncertain: want %q (durable fence), got %q", ControlMessageForceRefresh, ctl.Type)
	}
}

// TestJoinDurableRestoreBoundaryFencesStaleSeedAfterRestart is the BUG-2264
// restart-durability guard (residual #1 closed): the stale-seed Join fence must
// survive a server restart. A restore in a PRIOR process stamped the DURABLE
// items.last_restore_seq column; after a restart the RoomManager is brand new,
// so its in-memory lastRestoreSeqs fast-path is empty. A cursor-0 tab that
// seeded PRE-restore reconnects — it must STILL be force_refreshed, off the
// durable column read, not the (absent) in-memory value.
func TestJoinDurableRestoreBoundaryFencesStaleSeedAfterRestart(t *testing.T) {
	bus := NewMemoryOpBus()
	defer bus.Close()
	// Durable boundary present in the store (a restore stamped seq=10), but the
	// manager is fresh — exactly the post-restart state.
	store := &fakeOpLog{lastRestoreSeqs: map[string]int64{"item-a": 10}}
	mgr := NewRoomManager(store, bus)
	defer mgr.Close()
	if _, ok := mgr.LastRestoreSeq("item-a"); ok {
		t.Fatal("a fresh manager must have an empty in-memory restore-boundary fast-path")
	}

	srv := newCollabTestServer(t, mgr)
	defer srv.Close()

	// Stale seed (content_seq=5 < durable 10): force_refreshed via the durable
	// column even though the in-memory fast-path is empty.
	stale := dialWSContentSeq(t, srv, "item-a", 5)
	defer stale.Close()
	if ctl := readControlWithin(t, stale, time.Second); ctl.Type != ControlMessageForceRefresh {
		t.Fatalf("stale seed after restart: want %q, got %q", ControlMessageForceRefresh, ctl.Type)
	}

	// Current seed (content_seq=10 == durable): joins normally (first control
	// frame is the op_log_cursor, never force_refresh).
	fresh := dialWSContentSeq(t, srv, "item-a", 10)
	defer fresh.Close()
	if ctl := readControlWithin(t, fresh, time.Second); ctl.Type == ControlMessageForceRefresh {
		t.Fatal("current-generation seed must NOT be force_refreshed after restart")
	}
}

// TestJoinFailsClosedWhenDurableBoundaryReadErrors is the BUG-2264 fail-closed
// guard (Codex xhigh): if the durable restore-boundary read errors on an
// in-memory miss (a flaky DB after a restart), the fence must NOT degrade open
// and admit a possibly-stale seed — that client could push its pre-restore Y.Doc
// and clobber the restored content once the DB recovers. It fails CLOSED via a
// RETRYABLE plain close (NOT a force_refresh, which would discard the Y.Doc and
// spin an unbounded refresh loop on a persistent error): the conn is closed
// without admission and WITHOUT a force_refresh frame, so the client reconnects
// with backoff, Y.Doc intact.
func TestJoinFailsClosedWhenDurableBoundaryReadErrors(t *testing.T) {
	bus := NewMemoryOpBus()
	defer bus.Close()
	// Empty in-memory fast-path (fresh manager) AND the durable read errors.
	store := &fakeOpLog{failLastRestoreSeq: true}
	mgr := NewRoomManager(store, bus)
	defer mgr.Close()

	srv := newCollabTestServer(t, mgr)
	defer srv.Close()

	c := dialWSContentSeq(t, srv, "item-a", 5)
	defer c.Close()

	// The conn is closed by the server WITHOUT a force_refresh frame (that would
	// discard the Y.Doc). We should see a close/read error and never a
	// force_refresh control message.
	_ = c.SetReadDeadline(time.Now().Add(2 * time.Second))
	for {
		mt, data, err := c.ReadMessage()
		if err != nil {
			break // conn closed — the expected retryable close
		}
		if mt == websocket.TextMessage {
			var ctl ControlMessage
			if json.Unmarshal(data, &ctl) == nil && ctl.Type == ControlMessageForceRefresh {
				t.Fatal("durable-read error must NOT force_refresh (discards Y.Doc); expected a retryable close")
			}
		}
	}
}

// TestRoomManagerOpLogReplay confirms a fresh peer receives every
// previously-persisted op-log row before any live frames.
func TestRoomManagerOpLogReplay(t *testing.T) {
	bus := NewMemoryOpBus()
	defer bus.Close()

	store := &fakeOpLog{}
	// Pre-seed the op-log with two rows for item-a. Use the y-sync
	// header (0x00) so the room treats them as sync frames if they
	// were ever fed back through the read path.
	if _, err := store.AppendYjsUpdate("item-a", []byte{yMessageSync, 0x01}, "1"); err != nil {
		t.Fatalf("seed 1: %v", err)
	}
	if _, err := store.AppendYjsUpdate("item-a", []byte{yMessageSync, 0x02}, "1"); err != nil {
		t.Fatalf("seed 2: %v", err)
	}

	mgr := NewRoomManager(store, bus)
	srv := newCollabTestServer(t, mgr)
	defer srv.Close()

	c := dialWS(t, srv, "item-a")
	defer c.Close()

	first := readBinaryWithin(t, c, 2*time.Second)
	second := readBinaryWithin(t, c, 2*time.Second)

	if len(first) < 2 || first[1] != 0x01 {
		t.Errorf("replay row 1 mismatch: got %v", first)
	}
	if len(second) < 2 || second[1] != 0x02 {
		t.Errorf("replay row 2 mismatch: got %v", second)
	}
}

// TestRoomManagerSyncBroadcastAndPersist drives a sync frame from
// peer A and confirms B receives it AND the op-log gained a row.
func TestRoomManagerSyncBroadcastAndPersist(t *testing.T) {
	bus := NewMemoryOpBus()
	defer bus.Close()
	store := &fakeOpLog{}
	mgr := NewRoomManager(store, bus)
	srv := newCollabTestServer(t, mgr)
	defer srv.Close()

	a := dialWS(t, srv, "item-a")
	defer a.Close()
	b := dialWS(t, srv, "item-a")
	defer b.Close()

	// Wait for both to be subscribed before publishing.
	deadline := time.Now().Add(2 * time.Second)
	for time.Now().Before(deadline) {
		if bus.SubscriberCount("item-a") == 2 {
			break
		}
		time.Sleep(5 * time.Millisecond)
	}
	if got := bus.SubscriberCount("item-a"); got != 2 {
		t.Fatalf("want 2 subscribers, got %d", got)
	}

	sendSync(t, a, 0x42)

	got := readBinaryWithin(t, b, 2*time.Second)
	if len(got) < 2 || got[0] != yMessageSync || got[1] != 0x42 {
		t.Errorf("B received unexpected frame: %v", got)
	}

	// A must NOT receive its own echo. Pause briefly to let any
	// stray broadcast land before asserting.
	expectNoMessage(t, a, 200*time.Millisecond)

	if store.appendCount() != 1 {
		t.Errorf("op-log appendCount: want 1, got %d", store.appendCount())
	}
}

// TestRoomManagerAwarenessNotPersisted is the symmetric test for
// awareness frames: broadcast to peers, never written to the op-log.
func TestRoomManagerAwarenessNotPersisted(t *testing.T) {
	bus := NewMemoryOpBus()
	defer bus.Close()
	store := &fakeOpLog{}
	mgr := NewRoomManager(store, bus)
	srv := newCollabTestServer(t, mgr)
	defer srv.Close()

	a := dialWS(t, srv, "item-a")
	defer a.Close()
	b := dialWS(t, srv, "item-a")
	defer b.Close()

	// Wait for B's subscription.
	deadline := time.Now().Add(2 * time.Second)
	for time.Now().Before(deadline) {
		if bus.SubscriberCount("item-a") == 2 {
			break
		}
		time.Sleep(5 * time.Millisecond)
	}

	sendAwareness(t, a, 0x77)

	got := readBinaryWithin(t, b, 2*time.Second)
	if len(got) < 2 || got[0] != yMessageAwareness || got[1] != 0x77 {
		t.Errorf("B received unexpected frame: %v", got)
	}

	if store.rowCount() != 0 {
		t.Errorf("op-log rowCount: want 0 (awareness must not persist), got %d", store.rowCount())
	}
}

// TestRoomManagerSerializesSyncAppends is a regression test for the
// "concurrent peers race AppendYjsUpdate" bug. With the per-room
// appendMu in place, N peers each writing M sync frames must produce
// exactly N*M op-log rows AND each peer must see every other peer's
// frames at least once. Without serialisation this could surface
// missing rows / out-of-order ids on Postgres.
func TestRoomManagerSerializesSyncAppends(t *testing.T) {
	bus := NewMemoryOpBus()
	defer bus.Close()
	store := &fakeOpLog{}
	mgr := NewRoomManager(store, bus)
	srv := newCollabTestServer(t, mgr)
	defer srv.Close()

	const peers = 4
	const writesEach = 10
	wantRows := peers * writesEach

	conns := make([]*websocket.Conn, peers)
	for i := 0; i < peers; i++ {
		conns[i] = dialWS(t, srv, "item-a")
	}
	defer func() {
		for _, c := range conns {
			c.Close()
		}
	}()

	// Wait for all subscriptions to register.
	deadline := time.Now().Add(2 * time.Second)
	for time.Now().Before(deadline) {
		if bus.SubscriberCount("item-a") == peers {
			break
		}
		time.Sleep(5 * time.Millisecond)
	}
	if got := bus.SubscriberCount("item-a"); got != peers {
		t.Fatalf("subscriber count: want %d, got %d", peers, got)
	}

	// Drain readers so the subscriber channels stay empty (otherwise
	// the slow-consumer drop kicks in and the bus loses events).
	for _, c := range conns {
		go func(c *websocket.Conn) {
			for {
				if _, _, err := c.ReadMessage(); err != nil {
					return
				}
			}
		}(c)
	}

	var wg sync.WaitGroup
	for i := 0; i < peers; i++ {
		wg.Add(1)
		go func(c *websocket.Conn, base byte) {
			defer wg.Done()
			for j := 0; j < writesEach; j++ {
				if err := c.WriteMessage(
					websocket.BinaryMessage,
					[]byte{yMessageSync, base, byte(j)},
				); err != nil {
					t.Errorf("peer %d write %d: %v", base, j, err)
					return
				}
			}
		}(conns[i], byte(i))
	}
	wg.Wait()

	// Allow the read loops on each peer's roomConn to drain the in-flight
	// frames into the op-log. Poll instead of fixed sleep so the test
	// stays fast on a quiet machine and only waits as long as needed.
	deadline = time.Now().Add(5 * time.Second)
	for time.Now().Before(deadline) {
		if store.rowCount() == wantRows {
			break
		}
		time.Sleep(10 * time.Millisecond)
	}
	if got := store.rowCount(); got != wantRows {
		t.Fatalf("op-log rowCount: want %d, got %d", wantRows, got)
	}
}

// TestRoomManagerCrossItemIsolation makes sure events for item-a do
// not leak to peers connected to item-b. Without per-item bus
// filtering this would silently scramble unrelated documents.
func TestRoomManagerCrossItemIsolation(t *testing.T) {
	bus := NewMemoryOpBus()
	defer bus.Close()
	store := &fakeOpLog{}
	mgr := NewRoomManager(store, bus)
	srv := newCollabTestServer(t, mgr)
	defer srv.Close()

	a := dialWS(t, srv, "item-a")
	defer a.Close()
	b := dialWS(t, srv, "item-b")
	defer b.Close()

	deadline := time.Now().Add(2 * time.Second)
	for time.Now().Before(deadline) {
		if bus.SubscriberCount("item-a") == 1 && bus.SubscriberCount("item-b") == 1 {
			break
		}
		time.Sleep(5 * time.Millisecond)
	}

	sendSync(t, a, 0xAA)

	expectNoMessage(t, b, 200*time.Millisecond)
}

// TestRoomManagerGraceTTLReclaim exercises the grace-timer path with
// a short TTL injected via the per-manager config. After the timer
// fires the manager's room map drops the entry, so a fresh Join
// mints a new room with no in-memory state.
func TestRoomManagerGraceTTLReclaim(t *testing.T) {
	bus := NewMemoryOpBus()
	defer bus.Close()
	store := &fakeOpLog{}
	mgr := NewRoomManagerWithConfig(store, bus, RoomManagerConfig{GraceTTL: 50 * time.Millisecond})
	srv := newCollabTestServer(t, mgr)
	defer srv.Close()

	c := dialWS(t, srv, "item-a")

	// Wait for room to register.
	deadline := time.Now().Add(2 * time.Second)
	for time.Now().Before(deadline) {
		if mgr.RoomCount() == 1 {
			break
		}
		time.Sleep(5 * time.Millisecond)
	}
	if mgr.RoomCount() != 1 {
		t.Fatalf("expected room created, got %d", mgr.RoomCount())
	}

	// Disconnect + verify reclaimed within the grace window.
	c.Close()

	deadline = time.Now().Add(1 * time.Second)
	for time.Now().Before(deadline) {
		if mgr.RoomCount() == 0 {
			break
		}
		time.Sleep(5 * time.Millisecond)
	}
	if mgr.RoomCount() != 0 {
		t.Fatalf("expected room reclaimed after grace TTL, got %d", mgr.RoomCount())
	}
}

// TestRoomManagerGraceTTLCancelledByReconnect confirms a fresh
// connection within the grace window keeps the same Room — the
// room count does NOT bounce 1→0→1, it stays 1 throughout.
func TestRoomManagerGraceTTLCancelledByReconnect(t *testing.T) {
	bus := NewMemoryOpBus()
	defer bus.Close()
	store := &fakeOpLog{}
	mgr := NewRoomManagerWithConfig(store, bus, RoomManagerConfig{GraceTTL: 200 * time.Millisecond})
	srv := newCollabTestServer(t, mgr)
	defer srv.Close()

	a := dialWS(t, srv, "item-a")

	// Wait for room registration.
	for i := 0; i < 200; i++ {
		if mgr.RoomCount() == 1 {
			break
		}
		time.Sleep(5 * time.Millisecond)
	}

	// Disconnect, reconnect well within the grace window.
	a.Close()
	time.Sleep(50 * time.Millisecond) // a fraction of grace; timer pending
	if mgr.RoomCount() != 1 {
		t.Fatalf("during grace: want room still present, got %d rooms", mgr.RoomCount())
	}

	b := dialWS(t, srv, "item-a")
	defer b.Close()

	// Wait past the original grace deadline; if the reconnect didn't
	// cancel the timer, we'd see RoomCount drop and a new room
	// allocate, which would be observable as a transient 0.
	time.Sleep(300 * time.Millisecond)
	if mgr.RoomCount() != 1 {
		t.Fatalf("after grace: want 1 room, got %d", mgr.RoomCount())
	}
}

// TestRoomManagerJoinAfterCloseFailsFast covers the post-Close race:
// http.Server.Shutdown does NOT wait for hijacked WS handlers to
// finish, so a Join can fire AFTER Close has returned. The manager
// must reject those Joins fast (errManagerClosed) instead of
// creating a fresh Room against a torn-down store.
func TestRoomManagerJoinAfterCloseFailsFast(t *testing.T) {
	bus := NewMemoryOpBus()
	defer bus.Close()
	store := &fakeOpLog{}
	mgr := NewRoomManager(store, bus)

	mgr.Close()

	// Direct Join — bypass the WS plumbing since we want to exercise
	// the closed-flag short-circuit in isolation.
	err := mgr.Join("item-a", nil, 0, 0, true, false, nil) // conn is irrelevant; we never reach the WS path
	if !errors.Is(err, errManagerClosed) {
		t.Fatalf("want errManagerClosed, got %v", err)
	}

	// Idempotent: a second Close must not panic.
	mgr.Close()
}

// TestRoomManagerCloseShutsDownAllRooms exercises Close: after the
// call, any active connections should be closed (their reads
// terminate with an error) and the manager's room map should be
// empty.
func TestRoomManagerCloseShutsDownAllRooms(t *testing.T) {
	bus := NewMemoryOpBus()
	defer bus.Close()
	store := &fakeOpLog{}
	mgr := NewRoomManager(store, bus)
	srv := newCollabTestServer(t, mgr)
	defer srv.Close()

	c := dialWS(t, srv, "item-a")
	defer c.Close()

	// Wait for room.
	for i := 0; i < 200; i++ {
		if mgr.RoomCount() == 1 {
			break
		}
		time.Sleep(5 * time.Millisecond)
	}

	mgr.Close()

	if mgr.RoomCount() != 0 {
		t.Fatalf("after Close: want 0 rooms, got %d", mgr.RoomCount())
	}
	// Read should fail because Close closed the conn from the server
	// side. Drain the post-replay op_log_cursor TextMessage
	// (TASK-1319) so we hit the close, not a normal read.
	c.SetReadDeadline(time.Now().Add(1 * time.Second))
	for {
		mt, _, err := c.ReadMessage()
		if err != nil {
			break
		}
		if mt == websocket.BinaryMessage {
			t.Errorf("unexpected payload frame after manager.Close")
			break
		}
	}
}

// TestRoomManagerSchemaMismatchPrunes is the TASK-1268 happy-path
// regression test. When a peer joins an item whose persisted op-log
// rows are stamped with a different schema version than the
// manager's current one, the room manager prunes the entire op-log
// for that item before replay so the new client doesn't replay
// (potentially incompatible) old-schema ops.
func TestRoomManagerSchemaMismatchPrunes(t *testing.T) {
	bus := NewMemoryOpBus()
	defer bus.Close()

	store := &fakeOpLog{}
	// Pre-seed two rows stamped with the OLD schema version.
	if _, err := store.AppendYjsUpdate("item-a", []byte{yMessageSync, 0x01}, "1"); err != nil {
		t.Fatalf("seed 1: %v", err)
	}
	if _, err := store.AppendYjsUpdate("item-a", []byte{yMessageSync, 0x02}, "1"); err != nil {
		t.Fatalf("seed 2: %v", err)
	}

	// Manager runs at version "2" — first new-version client to
	// arrive should trigger the prune.
	mgr := NewRoomManagerWithConfig(store, bus, RoomManagerConfig{
		SchemaVersion: "2",
	})
	srv := newCollabTestServer(t, mgr)
	defer srv.Close()

	c := dialWS(t, srv, "item-a")
	defer c.Close()

	// After the prune, the op-log replay sends NO BINARY frames.
	// The post-replay op_log_cursor TextMessage (TASK-1319) is
	// expected; everything else means the prune didn't take.
	c.SetReadDeadline(time.Now().Add(300 * time.Millisecond))
	for {
		mt, _, err := c.ReadMessage()
		if err != nil {
			break
		}
		if mt == websocket.BinaryMessage {
			t.Errorf("expected no replay binary frames after schema-mismatch prune; got one")
			break
		}
	}

	// Op-log should be empty for item-a. The store's rowCount
	// reflects the post-prune state (we pruned in maybeRebuildOnSchemaMismatch).
	if rc := store.rowCount(); rc != 0 {
		t.Errorf("after schema-mismatch prune: want 0 rows, got %d", rc)
	}
}

// TestRoomManagerSchemaCleanVersionPreservesOpLog is the negative
// case: when the persisted version matches the manager's, no prune
// runs and the replay path is undisturbed.
func TestRoomManagerSchemaCleanVersionPreservesOpLog(t *testing.T) {
	bus := NewMemoryOpBus()
	defer bus.Close()

	store := &fakeOpLog{}
	if _, err := store.AppendYjsUpdate("item-a", []byte{yMessageSync, 0x01}, "1"); err != nil {
		t.Fatalf("seed 1: %v", err)
	}

	// Manager and persisted rows agree on "1".
	mgr := NewRoomManagerWithConfig(store, bus, RoomManagerConfig{
		SchemaVersion: "1",
	})
	srv := newCollabTestServer(t, mgr)
	defer srv.Close()

	c := dialWS(t, srv, "item-a")
	defer c.Close()

	// Replay should land normally — single row.
	first := readBinaryWithin(t, c, 2*time.Second)
	if len(first) < 2 || first[1] != 0x01 {
		t.Errorf("replay row 1 mismatch: got %v", first)
	}

	if rc := store.rowCount(); rc != 1 {
		t.Errorf("on clean-version connect: op-log should be untouched (want 1 row), got %d", rc)
	}
}

// TestRoomManagerSchemaPostRebuildClean confirms that after a
// rebuild fires once, subsequent connects on the same item are
// clean: the post-prune append carries the new schema version,
// LatestYjsUpdateSchemaVersion now reads as "2", and a second peer
// connecting with version "2" does NOT re-trigger the prune.
func TestRoomManagerSchemaPostRebuildClean(t *testing.T) {
	bus := NewMemoryOpBus()
	defer bus.Close()

	store := &fakeOpLog{}
	if _, err := store.AppendYjsUpdate("item-a", []byte{yMessageSync, 0x01}, "1"); err != nil {
		t.Fatalf("seed: %v", err)
	}

	mgr := NewRoomManagerWithConfig(store, bus, RoomManagerConfig{
		SchemaVersion: "2",
	})
	srv := newCollabTestServer(t, mgr)
	defer srv.Close()

	// First connect: triggers the prune.
	c1 := dialWS(t, srv, "item-a")

	// Wait for the prune to actually land. RoomCount() == 1 isn't a
	// prune-completion barrier (getOrCreate publishes the room
	// BEFORE maybeRebuildOnSchemaMismatch runs), so poll the store
	// state directly. Per Codex review round 1 NIT.
	for i := 0; i < 200; i++ {
		if store.rowCount() == 0 {
			break
		}
		time.Sleep(5 * time.Millisecond)
	}

	if rc := store.rowCount(); rc != 0 {
		t.Errorf("after first join (mismatch): want 0 rows, got %d", rc)
	}

	// Have c1 send a sync frame so the post-prune op-log gets a row
	// stamped at the manager's "2".
	if err := c1.WriteMessage(websocket.BinaryMessage, []byte{yMessageSync, 0x55}); err != nil {
		t.Fatalf("write live frame: %v", err)
	}
	for i := 0; i < 200; i++ {
		if store.rowCount() == 1 {
			break
		}
		time.Sleep(5 * time.Millisecond)
	}
	if rc := store.rowCount(); rc != 1 {
		t.Fatalf("post-rebuild append: want 1 row, got %d", rc)
	}

	// Second connect on the same item: persisted version is now
	// "2", manager is "2", so no prune. The single row must
	// replay to the new peer untouched.
	c2 := dialWS(t, srv, "item-a")
	defer c2.Close()

	got := readBinaryWithin(t, c2, 2*time.Second)
	if len(got) < 2 || got[1] != 0x55 {
		t.Errorf("post-rebuild replay row mismatch: got %v", got)
	}

	c1.Close()
}

// TestPruneSweepRemovesDormantItems is the TASK-1309 happy-path test:
// a sweep finds items whose entire op-log is older than minAge and
// prunes them whole. Items with ANY recent activity are preserved
// completely (no prefix-pruning — Yjs causal references would break
// per the Codex P1 finding).
func TestPruneSweepRemovesDormantItems(t *testing.T) {
	bus := NewMemoryOpBus()
	defer bus.Close()

	store := &fakeOpLog{}
	mgr := NewRoomManager(store, bus)
	defer mgr.Close()

	// Item A: two old rows + one fresh row → mixed, NOT dormant.
	// Whole-log prune is unsafe here (suffix references prefix
	// structs in real Yjs). All three rows must survive.
	now := time.Now()
	store.mu.Lock()
	store.rows = append(store.rows,
		fakeYjsRow("item-a", []byte{yMessageSync, 0x01}, "1", now.Add(-48*time.Hour)),
		fakeYjsRow("item-a", []byte{yMessageSync, 0x02}, "1", now.Add(-30*time.Hour)),
		fakeYjsRow("item-a", []byte{yMessageSync, 0x03}, "1", now.Add(-1*time.Hour)),
	)
	// Item B: one fresh row → not dormant. Survives.
	store.rows = append(store.rows,
		fakeYjsRow("item-b", []byte{yMessageSync, 0x04}, "1", now.Add(-2*time.Hour)),
	)
	// Item C: two old rows, no recent activity → DORMANT. Whole
	// op-log gets pruned; cold reconnect lazy-seeds from
	// items.content (TASK-1261).
	store.rows = append(store.rows,
		fakeYjsRow("item-c", []byte{yMessageSync, 0x05}, "1", now.Add(-72*time.Hour)),
		fakeYjsRow("item-c", []byte{yMessageSync, 0x06}, "1", now.Add(-50*time.Hour)),
	)
	store.mu.Unlock()

	res, err := mgr.PruneSweep(24 * time.Hour)
	if err != nil {
		t.Fatalf("PruneSweep: %v", err)
	}

	// Only item-c is dormant (MAX(created_at) < cutoff).
	if got := res.ItemsScanned; got != 1 {
		t.Errorf("ItemsScanned: want 1 (only c is dormant), got %d", got)
	}
	if got := res.ItemsPruned; got != 1 {
		t.Errorf("ItemsPruned: want 1, got %d", got)
	}
	if got := res.RowsPruned; got != 2 {
		t.Errorf("RowsPruned: want 2 (c's whole op-log), got %d", got)
	}
	if got := res.Errors; got != 0 {
		t.Errorf("Errors: want 0, got %d", got)
	}

	// Surviving rows: item-a's three + item-b's one. item-c gone.
	store.mu.Lock()
	defer store.mu.Unlock()
	if got := len(store.rows); got != 4 {
		t.Fatalf("after prune: want 4 surviving rows (3 from item-a + 1 from item-b), got %d", got)
	}
	byItem := map[string]int{}
	for _, r := range store.rows {
		byItem[r.ItemID]++
	}
	if byItem["item-a"] != 3 {
		t.Errorf("item-a: want 3 surviving rows (whole-log preserved because non-dormant), got %d", byItem["item-a"])
	}
	if byItem["item-b"] != 1 {
		t.Errorf("item-b: want 1 surviving row, got %d", byItem["item-b"])
	}
	if byItem["item-c"] != 0 {
		t.Errorf("item-c: want 0 surviving rows (dormant, whole-log pruned), got %d", byItem["item-c"])
	}
}

// TestPruneSweepDefaultMinAge verifies that passing minAge=0 falls
// back to DefaultPruneMinAge rather than treating the cutoff as
// "now" (which would prune every row in the table).
func TestPruneSweepDefaultMinAge(t *testing.T) {
	bus := NewMemoryOpBus()
	defer bus.Close()

	store := &fakeOpLog{}
	mgr := NewRoomManager(store, bus)
	defer mgr.Close()

	// One row, 1 hour old. With DefaultPruneMinAge=24h this row
	// should be PRESERVED. If the implementation accidentally used
	// `time.Now()` as the cutoff, the row would be deleted.
	now := time.Now()
	store.mu.Lock()
	store.rows = append(store.rows,
		fakeYjsRow("item-a", []byte{yMessageSync, 0x01}, "1", now.Add(-1*time.Hour)),
	)
	store.mu.Unlock()

	res, err := mgr.PruneSweep(0)
	if err != nil {
		t.Fatalf("PruneSweep: %v", err)
	}
	if res.RowsPruned != 0 {
		t.Errorf("with default minAge, 1-hour-old row should survive; got RowsPruned=%d", res.RowsPruned)
	}
	if got := store.rowCount(); got != 1 {
		t.Errorf("rows after sweep: want 1, got %d", got)
	}
}

// TestPruneSweepEmpty handles the no-eligible-rows case cleanly.
func TestPruneSweepEmpty(t *testing.T) {
	bus := NewMemoryOpBus()
	defer bus.Close()

	store := &fakeOpLog{}
	mgr := NewRoomManager(store, bus)
	defer mgr.Close()

	res, err := mgr.PruneSweep(24 * time.Hour)
	if err != nil {
		t.Fatalf("PruneSweep: %v", err)
	}
	if res.ItemsScanned != 0 || res.RowsPruned != 0 || res.Errors != 0 {
		t.Errorf("empty store should produce zero counters, got %+v", res)
	}
}

// TestPruneSweepBailsOnClose ensures a sweep mid-loop respects a
// concurrent Close — once the manager is closed, the sweep stops
// iterating remaining items rather than continuing to touch a
// torn-down store.
func TestPruneSweepBailsOnClose(t *testing.T) {
	bus := NewMemoryOpBus()
	defer bus.Close()

	store := &fakeOpLog{}
	mgr := NewRoomManager(store, bus)

	now := time.Now()
	store.mu.Lock()
	for i := 0; i < 10; i++ {
		itemID := fmt.Sprintf("item-%d", i)
		store.rows = append(store.rows,
			fakeYjsRow(itemID, []byte{yMessageSync, byte(i)}, "1", now.Add(-48*time.Hour)),
		)
	}
	store.mu.Unlock()

	// Pre-close the manager: the sweep should bail on the first
	// per-item iteration. Close() runs synchronously here so the
	// closed flag is set before PruneSweep starts.
	mgr.Close()

	res, err := mgr.PruneSweep(24 * time.Hour)
	if err != nil {
		t.Fatalf("PruneSweep: %v", err)
	}
	// ItemsScanned reflects the initial query (which ran before
	// the closed-flag check), but no items should have been
	// pruned because the per-item loop bailed immediately.
	if res.ItemsScanned != 10 {
		t.Errorf("ItemsScanned: want 10 (query ran), got %d", res.ItemsScanned)
	}
	if res.RowsPruned != 0 {
		t.Errorf("RowsPruned after Close: want 0, got %d", res.RowsPruned)
	}
}

// TestPruneSweepSkipsActiveRoom verifies the dormant-item check
// honours an in-memory active Room — even if the database query
// returned the item as dormant, an existing Room means we shouldn't
// prune. (Active rooms can buffer awareness state, hold a grace
// timer, etc.; a fresh peer reconnecting via grace would expect to
// find the prior op-log intact.)
func TestPruneSweepSkipsActiveRoom(t *testing.T) {
	bus := NewMemoryOpBus()
	defer bus.Close()

	store := &fakeOpLog{}
	mgr := NewRoomManager(store, bus)
	defer mgr.Close()

	// Seed one dormant item.
	now := time.Now()
	store.mu.Lock()
	store.rows = append(store.rows,
		fakeYjsRow("item-a", []byte{yMessageSync, 0x01}, "1", now.Add(-48*time.Hour)),
	)
	store.mu.Unlock()

	// Create a room for item-a directly. We don't need a real WS;
	// just having the entry in m.rooms is enough to trigger the
	// active-room skip.
	mgr.mu.Lock()
	mgr.rooms["item-a"] = &Room{
		itemID: "item-a",
		store:  store,
		bus:    bus,
		conns:  make(map[*websocket.Conn]*roomConn),
	}
	mgr.mu.Unlock()

	res, err := mgr.PruneSweep(24 * time.Hour)
	if err != nil {
		t.Fatalf("PruneSweep: %v", err)
	}

	if res.ItemsPruned != 0 {
		t.Errorf("ItemsPruned: want 0 (active room blocks prune), got %d", res.ItemsPruned)
	}
	if res.ItemsSkipped != 1 {
		t.Errorf("ItemsSkipped: want 1 (item-a skipped due to active room), got %d", res.ItemsSkipped)
	}
	if got := store.rowCount(); got != 1 {
		t.Errorf("rows after sweep: want 1 (active room preserved), got %d", got)
	}
}

// TestPruneSweepSkipsRowAddedMidSweep exercises the mid-sweep race
// window: the dormancy listing returns item-a (it WAS dormant at
// query time), but a row gets appended before PruneItemOpLogIfDormantBefore
// runs under the per-item lock. The conditional DELETE in the store
// must re-check dormancy at SQL level and refuse to delete.
//
// We simulate the race via the onListDormantHook on fakeOpLog,
// which fires after the listing is computed but before
// PruneSweep iterates. The hook injects a recent row that the
// candidate-list already committed to but that the conditional
// DELETE must catch.
//
// Per Codex review of TASK-1309 round 2 NIT.
func TestPruneSweepSkipsRowAddedMidSweep(t *testing.T) {
	bus := NewMemoryOpBus()
	defer bus.Close()

	store := &fakeOpLog{}
	mgr := NewRoomManager(store, bus)
	defer mgr.Close()

	// Seed a single old row → item-a IS dormant at query time.
	now := time.Now()
	store.mu.Lock()
	store.rows = append(store.rows,
		fakeYjsRow("item-a", []byte{yMessageSync, 0x01}, "1", now.Add(-48*time.Hour)),
	)
	// Inject a recent row in the race window — listing already
	// computed, but per-item DELETE hasn't run yet.
	store.onListDormantHook = func(f *fakeOpLog) {
		f.mu.Lock()
		f.rows = append(f.rows,
			fakeYjsRow("item-a", []byte{yMessageSync, 0x02}, "1", now.Add(-1*time.Hour)),
		)
		f.mu.Unlock()
	}
	store.mu.Unlock()

	res, err := mgr.PruneSweep(24 * time.Hour)
	if err != nil {
		t.Fatalf("PruneSweep: %v", err)
	}

	// item-a was returned by the listing (1 candidate scanned),
	// but the conditional DELETE saw the newly-appended row and
	// refused. Expect 1 scanned, 0 pruned, 1 skipped.
	if res.ItemsScanned != 1 {
		t.Errorf("ItemsScanned: want 1 (listing saw dormant item), got %d", res.ItemsScanned)
	}
	if res.ItemsPruned != 0 {
		t.Errorf("ItemsPruned: want 0 (conditional DELETE refused), got %d", res.ItemsPruned)
	}
	if res.ItemsSkipped != 1 {
		t.Errorf("ItemsSkipped: want 1 (DELETE returned 0 rows), got %d", res.ItemsSkipped)
	}
	if got := store.rowCount(); got != 2 {
		t.Errorf("rows after sweep: want 2 (both preserved), got %d", got)
	}
}

// fakeYjsRow constructs a single row for the fakeOpLog seeding helpers.
// The store assigns ids in production; tests don't depend on them.
func fakeYjsRow(itemID string, data []byte, schemaVersion string, createdAt time.Time) models.YjsUpdate {
	cp := make([]byte, len(data))
	copy(cp, data)
	return models.YjsUpdate{
		ItemID:        itemID,
		UpdateData:    cp,
		SchemaVersion: schemaVersion,
		CreatedAt:     createdAt.UTC(),
	}
}

// readControlWithin reads frames off the conn within d, drains
// payload (binary) frames, and returns the FIRST TextMessage parsed
// as a ControlMessage. Tests use this to assert the post-replay
// op_log_cursor frame and force_refresh frame land cleanly.
func readControlWithin(t *testing.T, c *websocket.Conn, d time.Duration) ControlMessage {
	t.Helper()
	deadline := time.Now().Add(d)
	c.SetReadDeadline(deadline)
	defer c.SetReadDeadline(time.Time{})
	for {
		mt, data, err := c.ReadMessage()
		if err != nil {
			t.Fatalf("read control: %v", err)
		}
		if mt == websocket.TextMessage {
			var msg ControlMessage
			if err := json.Unmarshal(data, &msg); err != nil {
				t.Fatalf("unmarshal control: %v (raw=%q)", err, data)
			}
			return msg
		}
		// Skip binary replay frames; the cursor is what we want.
		c.SetReadDeadline(deadline)
	}
}

// TestRoomManagerInitialOpLogCursorAfterReplay confirms a fresh peer
// receives an op_log_cursor TextMessage after replay so its
// sessionStorage cursor is anchored without a round trip. Per TASK-1319.
func TestRoomManagerInitialOpLogCursorAfterReplay(t *testing.T) {
	bus := NewMemoryOpBus()
	defer bus.Close()
	store := &fakeOpLog{}
	if _, err := store.AppendYjsUpdate("item-a", []byte{yMessageSync, 0x01}, "1"); err != nil {
		t.Fatalf("seed: %v", err)
	}
	if _, err := store.AppendYjsUpdate("item-a", []byte{yMessageSync, 0x02}, "1"); err != nil {
		t.Fatalf("seed 2: %v", err)
	}
	mgr := NewRoomManager(store, bus)
	srv := newCollabTestServer(t, mgr)
	defer srv.Close()

	c := dialWS(t, srv, "item-a")
	defer c.Close()

	// Drain the two replay binary frames.
	_ = readBinaryWithin(t, c, time.Second)
	_ = readBinaryWithin(t, c, time.Second)

	// Now the cursor frame must arrive, anchored at the highest
	// replayed id (2 after two seeds).
	cursor := readControlWithin(t, c, time.Second)
	if cursor.Type != ControlMessageOpLogCursor {
		t.Fatalf("type: want %q, got %q", ControlMessageOpLogCursor, cursor.Type)
	}
	if cursor.OpLogID != 2 {
		t.Errorf("op_log_id: want 2, got %d", cursor.OpLogID)
	}
}

// TestRoomManagerInitialCursorEmptyOpLog confirms a fresh peer joining
// an item with no persisted ops still receives an op_log_cursor frame
// with id 0 — clears any stale sessionStorage cursor from an earlier
// life of this item id. Per TASK-1319.
func TestRoomManagerInitialCursorEmptyOpLog(t *testing.T) {
	bus := NewMemoryOpBus()
	defer bus.Close()
	mgr := NewRoomManager(&fakeOpLog{}, bus)
	srv := newCollabTestServer(t, mgr)
	defer srv.Close()

	c := dialWS(t, srv, "item-a")
	defer c.Close()

	cursor := readControlWithin(t, c, time.Second)
	if cursor.Type != ControlMessageOpLogCursor {
		t.Fatalf("type: want %q, got %q", ControlMessageOpLogCursor, cursor.Type)
	}
	if cursor.OpLogID != 0 {
		t.Errorf("op_log_id: want 0 (empty op-log), got %d", cursor.OpLogID)
	}
}

// TestRoomManagerForceRefreshOnStaleSince exercises the resume-cursor
// safety path: a client announcing `?since=N` where N is below
// MIN(item_yjs_updates.id) gets a force_refresh frame and a closed
// conn. Per TASK-1319.
func TestRoomManagerForceRefreshOnStaleSince(t *testing.T) {
	bus := NewMemoryOpBus()
	defer bus.Close()
	store := &fakeOpLog{}
	// Seed two rows then prune the older one so MIN(id) > 1.
	if _, err := store.AppendYjsUpdate("item-a", []byte{yMessageSync, 0x01}, "1"); err != nil {
		t.Fatalf("seed 1: %v", err)
	}
	if _, err := store.AppendYjsUpdate("item-a", []byte{yMessageSync, 0x02}, "1"); err != nil {
		t.Fatalf("seed 2: %v", err)
	}
	// Manually drop row id=1 to simulate a partial prune. The fake's
	// rows slice is in-place; we reuse the same field write semantics
	// as PruneYjsUpdatesBefore would.
	store.mu.Lock()
	kept := store.rows[:0]
	for _, r := range store.rows {
		if r.ID != 1 {
			kept = append(kept, r)
		}
	}
	store.rows = kept
	store.mu.Unlock()

	mgr := NewRoomManager(store, bus)
	srv := newCollabTestServer(t, mgr)
	defer srv.Close()

	// Client claims it has applied id=0 (no rows seen) but server's
	// MIN is now 2 — wait, since=0 is special-cased as "fresh client"
	// and skips the force_refresh check. Use since=1 instead — below
	// MIN=2 so the protocol fires.
	c := dialWSSince(t, srv, "item-a", 1)
	defer c.Close()

	cursor := readControlWithin(t, c, time.Second)
	if cursor.Type != ControlMessageForceRefresh {
		t.Fatalf("type: want %q, got %q", ControlMessageForceRefresh, cursor.Type)
	}
	// After the force_refresh frame the server closes; the next
	// read should error.
	c.SetReadDeadline(time.Now().Add(500 * time.Millisecond))
	if _, _, err := c.ReadMessage(); err == nil {
		t.Errorf("expected close after force_refresh; got another frame")
	}
}

// TestRoomManagerSinceAtOrAboveMinReplaysDelta confirms a client
// resuming with `?since=N` where N >= MIN gets only the rows with
// id > N (delta replay), not the entire op-log. Per TASK-1319.
func TestRoomManagerSinceAtOrAboveMinReplaysDelta(t *testing.T) {
	bus := NewMemoryOpBus()
	defer bus.Close()
	store := &fakeOpLog{}
	for i := byte(1); i <= 4; i++ {
		if _, err := store.AppendYjsUpdate("item-a", []byte{yMessageSync, i}, "1"); err != nil {
			t.Fatalf("seed %d: %v", i, err)
		}
	}
	mgr := NewRoomManager(store, bus)
	srv := newCollabTestServer(t, mgr)
	defer srv.Close()

	// since=2 → expect rows id 3 and 4 only (two binary frames),
	// then a cursor frame at id 4.
	c := dialWSSince(t, srv, "item-a", 2)
	defer c.Close()

	got1 := readBinaryWithin(t, c, time.Second)
	if len(got1) < 2 || got1[1] != 0x03 {
		t.Errorf("first replayed payload: want byte 0x03, got %v", got1)
	}
	got2 := readBinaryWithin(t, c, time.Second)
	if len(got2) < 2 || got2[1] != 0x04 {
		t.Errorf("second replayed payload: want byte 0x04, got %v", got2)
	}

	// No more binary frames — the cursor should be next.
	cursor := readControlWithin(t, c, time.Second)
	if cursor.Type != ControlMessageOpLogCursor || cursor.OpLogID != 4 {
		t.Errorf("cursor: want type=op_log_cursor id=4, got type=%q id=%d", cursor.Type, cursor.OpLogID)
	}
}

// TestRoomManagerSyncCursorBroadcast confirms that after a peer
// pushes a sync frame, the originator gets an op_log_cursor frame
// with the new id (so it can advance its session-storage cursor)
// AND every other peer also gets one piggybacked on the broadcast
// (so they stay in lockstep without a round trip).
func TestRoomManagerSyncCursorBroadcast(t *testing.T) {
	bus := NewMemoryOpBus()
	defer bus.Close()
	store := &fakeOpLog{}
	mgr := NewRoomManager(store, bus)
	srv := newCollabTestServer(t, mgr)
	defer srv.Close()

	a := dialWS(t, srv, "item-a")
	defer a.Close()
	b := dialWS(t, srv, "item-a")
	defer b.Close()

	// Drain initial cursor frames on both peers.
	_ = readControlWithin(t, a, time.Second)
	_ = readControlWithin(t, b, time.Second)

	// Wait for both subscriptions to register so the broadcast is
	// guaranteed to reach B.
	deadline := time.Now().Add(2 * time.Second)
	for time.Now().Before(deadline) {
		if bus.SubscriberCount("item-a") == 2 {
			break
		}
		time.Sleep(5 * time.Millisecond)
	}

	sendSync(t, a, 0x42)

	// A (originator) gets a cursor frame with the new id (1).
	cursorA := readControlWithin(t, a, time.Second)
	if cursorA.Type != ControlMessageOpLogCursor || cursorA.OpLogID != 1 {
		t.Errorf("originator cursor: want op_log_cursor id=1, got type=%q id=%d", cursorA.Type, cursorA.OpLogID)
	}

	// B receives the binary fan-out plus a cursor frame at id=1.
	gotB := readBinaryWithin(t, b, time.Second)
	if len(gotB) < 2 || gotB[1] != 0x42 {
		t.Errorf("peer payload: %v", gotB)
	}
	cursorB := readControlWithin(t, b, time.Second)
	if cursorB.Type != ControlMessageOpLogCursor || cursorB.OpLogID != 1 {
		t.Errorf("peer cursor: want op_log_cursor id=1, got type=%q id=%d", cursorB.Type, cursorB.OpLogID)
	}
}

// TestRoomManagerForceRefreshOnEmptyOpLogWithSince exercises the
// branch added in Codex round 2 [P1] of TASK-1319: a client
// announces `?since=N>0` against an item whose ENTIRE op-log has
// been pruned (PruneAndApply, schema rebuild, or dormant GC). MIN
// is undefined, so the original `hasMin && since < minID` predicate
// would have admitted the connection — and the client's on-open
// `Y.encodeStateAsUpdate` write would have resurrected the stale
// pre-prune document.
func TestRoomManagerForceRefreshOnEmptyOpLogWithSince(t *testing.T) {
	bus := NewMemoryOpBus()
	defer bus.Close()
	store := &fakeOpLog{}
	mgr := NewRoomManager(store, bus)
	srv := newCollabTestServer(t, mgr)
	defer srv.Close()

	// No seed rows — empty op-log, simulating post-prune state.
	c := dialWSSince(t, srv, "item-a", 5)
	defer c.Close()

	cursor := readControlWithin(t, c, time.Second)
	if cursor.Type != ControlMessageForceRefresh {
		t.Fatalf("type: want %q, got %q", ControlMessageForceRefresh, cursor.Type)
	}
}

// TestRoomManagerCursorSuppressedDuringReplay is the round 5 [P1]
// regression: a live op broadcast during the replay window MUST NOT
// generate an op_log_cursor TextMessage to a peer mid-replay. If it
// did, a client disconnecting after that cursor but before the rest
// of the replay rows would persist a cursor pointing past unreplayed
// rows.
//
// We exercise the path by:
//   - seeding 3 rows so peer B's replay has work to do,
//   - waiting for B to subscribe AND start replay,
//   - sending a sync from peer A (live op id=4) — broadcast lands in
//     B's bus channel, processed by B's writeLoop concurrently with
//     replayTo's writes.
//
// The assertion is loose-but-decisive: B sees no op_log_cursor frame
// with id == 4 BEFORE all four binary frames have arrived. After
// replay completes the post-replay cursor (id=3) and then live cursor
// (id=4) flow normally.
func TestRoomManagerCursorSuppressedDuringReplay(t *testing.T) {
	bus := NewMemoryOpBus()
	defer bus.Close()
	store := &fakeOpLog{}
	for i := byte(1); i <= 3; i++ {
		if _, err := store.AppendYjsUpdate("item-a", []byte{yMessageSync, i}, "1"); err != nil {
			t.Fatalf("seed %d: %v", i, err)
		}
	}
	mgr := NewRoomManager(store, bus)
	srv := newCollabTestServer(t, mgr)
	defer srv.Close()

	a := dialWS(t, srv, "item-a")
	defer a.Close()
	// Drain A's replay (3 binary) + post-replay cursor frame.
	_ = readBinaryWithin(t, a, time.Second)
	_ = readBinaryWithin(t, a, time.Second)
	_ = readBinaryWithin(t, a, time.Second)
	_ = readControlWithin(t, a, time.Second)

	b := dialWS(t, srv, "item-a")
	defer b.Close()

	// Push a live op from A while B is still in setup.
	sendSync(t, a, 0xFF)

	// B is racing replay + the live op fan-out. Read frames in
	// order until we've collected all 4 expected binary frames
	// (3 replayed + 1 live) and the post-replay + live cursor
	// frames. If a cursor frame for id=4 arrives BEFORE all
	// binary frames are accounted for, this is the regression.
	//
	// We count DISTINCT op payloads rather than raw frames because
	// runConn (manager.go:497-514) deliberately starts the writer
	// before replayTo so live broadcasts during a long replay aren't
	// lost; the documented trade-off is that an op appended during
	// the replay window can be delivered twice — once via the live
	// bus subscription, once via replayTo sweeping it out of the op
	// log — which Yjs update application tolerates idempotently.
	// That's a legitimate outcome of the race, not a bug: don't
	// "fix" this back to a raw frame count. Dedup by exact byte
	// content is safe here because every one of the 4 expected ops
	// has a distinguishable payload (seeded ops carry 0x01/0x02/0x03,
	// the live op carries 0xFF), so a genuinely dropped op still
	// caps distinctSeen at 3, not 4 — only a true duplicate is
	// tolerated.
	seen := map[string]int{}
	distinctSeen := 0
	deadline := time.Now().Add(2 * time.Second)
	for {
		b.SetReadDeadline(deadline)
		mt, data, err := b.ReadMessage()
		if err != nil {
			break
		}
		if mt == websocket.BinaryMessage {
			key := string(data)
			if seen[key] == 0 {
				distinctSeen++
			}
			seen[key]++
			continue
		}
		if mt != websocket.TextMessage {
			continue
		}
		var msg ControlMessage
		if jerr := json.Unmarshal(data, &msg); jerr != nil {
			continue
		}
		if msg.Type != ControlMessageOpLogCursor {
			continue
		}
		// A live cursor with id=4 must only arrive AFTER all 4
		// distinct binary ops are accounted for; otherwise
		// replay-mid cursor advancement leaked through.
		if msg.OpLogID == 4 && distinctSeen < 4 {
			t.Errorf("cursor id=4 arrived after only %d/4 distinct binary ops; replay-mid cursor leaked", distinctSeen)
		}
	}
	if distinctSeen != 4 {
		t.Errorf("distinct binary ops seen: want 4, got %d", distinctSeen)
	}
}