test: wait on the asserted postcondition in two Windows-flaky tests (#10157)

Problem
Two tests failed the v1.38.7 release candidate's Windows CI for reasons
unrelated to the diff, each costing a 40-minute rerun:

  --- FAIL: TestRepairDrainEventuallyCompletesBeyondQueue
      repair_queue_drain_test.go:49: repair pending stuck at 2
  --- FAIL: TestTopicMigrationMarkerRescansWhenSessionFileChanges
      tabs_topic_test.go:1011: load second meta: open second.jsonl.meta:
      The process cannot access the file because it is being used by
      another process.

Root cause
Both assert on a proxy instead of the postcondition they care about.

The drain test gave the catalog 15s to reach RepairPending == 0 while
running the real filesystem repair. repairDisposition defers any transient
failure, and repairBackoff's floor is 30s -- twice the budget. One I/O
hiccup on a loaded Windows runner therefore failed the test by policy, not
by the queue-capacity invariant the test is named for. internal/sessioncatalog
has no change in the release range, so the failure was never a regression.

The migration test read second.jsonl.meta as soon as waitForCatalogTopic
returned. Catalog publication does not fence the sidecar write -- the test
already says so about the marker a few lines above -- so the migration can
still hold the handle, and Windows refuses to open a file another handle
owns.

Fix
- Options gains an unexported repairSession hook that Open installs before
  starting repairLoop. The existing testRepairSessionHook field is only
  assignable after Open returns, which races the running loop; this makes
  the injection race-free. The drain test uses it to keep the real wake path
  while removing the filesystem outcome from the assertion, and signals
  completion over a channel instead of polling a wall clock.
- The migration test waits for the meta it asserts, reusing the package's
  waitFor helper. Replacing the hand-rolled marker poll with the same helper
  keeps the file under its repolint size baseline.

Verification
- go test ./internal/sessioncatalog/ and go test -race ./internal/sessioncatalog/
  pass; the drain case runs 5x in 1.2s instead of polling for up to 15s.
- Teeth check: opening the catalog with DisableRepair so the drain can never
  progress fails the rewritten test with "repaired 0 of 8 sessions".
- cd desktop && go test . passes (110s), including the migration case 3x.
- gofmt, go vet, and go run ./tools/repolint are clean with no baseline change.

desktop/tabs_topic_test.go

@@ -988,17 +988,10 @@ func TestTopicMigrationMarkerRescansWhenSessionFileChanges(t *testing.T) {
 	// background path but not under one fsync barrier. Wait for the marker
 	// explicitly so Windows CI does not observe the topic before the stamp.
 	markerPath := filepath.Join(dir, topicMigrationMarker)
-	deadline := time.Now().Add(5 * time.Second)
-	var lastMarkerErr error
-	for {
-		if _, lastMarkerErr = os.Stat(markerPath); lastMarkerErr == nil {
-			break
-		}
-		if time.Now().After(deadline) {
-			t.Fatalf("expected migration marker after a complete pass: %v", lastMarkerErr)
-		}
-		time.Sleep(10 * time.Millisecond)
-	}
+	waitFor(t, "the migration marker after a complete pass", func() bool {
+		_, err := os.Stat(markerPath)
+		return err == nil
+	})
 
 	// A CLI-created session added after the marker invalidates the lightweight
 	// gate and gets a fresh migration pass.
@@ -1006,13 +999,12 @@ func TestTopicMigrationMarkerRescansWhenSessionFileChanges(t *testing.T) {
 	second := writeLegacySession(t, dir, "second.jsonl", "second legacy prompt", time.Now())
 	app.requestSessionCatalogReconcile(dir)
 	waitForCatalogTopic(t, app, "global", "", legacySessionTopicID(second))
-	meta, ok, err := agent.LoadBranchMeta(second)
-	if err != nil {
-		t.Fatalf("load second meta: %v", err)
-	}
-	if !ok || strings.TrimSpace(meta.TopicID) != legacySessionTopicID(second) {
-		t.Fatalf("new session after marker should be migrated, got ok=%v meta=%+v", ok, meta)
-	}
+	// Publication does not fence the sidecar write, and Windows refuses to open
+	// a .meta the migration still holds. Await the asserted postcondition.
+	waitFor(t, "second.jsonl.meta to carry the migrated topic", func() bool {
+		meta, ok, err := agent.LoadBranchMeta(second)
+		return err == nil && ok && strings.TrimSpace(meta.TopicID) == legacySessionTopicID(second)
+	})
 }
 
 func TestProjectTreeRepairsIndexedGlobalTopicsAfterMigrationMarker(t *testing.T) {

internal/sessioncatalog/catalog.go

@@ -163,6 +163,7 @@ func Open(ctx context.Context, opts Options) (*Catalog, error) {
 		}
 		c.refreshCounts(ctx)
 	}
+	c.testRepairSessionHook = opts.repairSession
 	c.workerCtx, c.workerCancel = context.WithCancel(context.Background())
 	c.workers.Add(1)
 	go c.writerLoop()

internal/sessioncatalog/repair_queue_drain_test.go

@@ -5,10 +5,18 @@ import (
 	"fmt"
 	"os"
 	"path/filepath"
+	"sync"
 	"testing"
 	"time"
+
+	"reasonix/internal/agent"
 )
 
+// The wake channel coalesces on one key and is sized by QueueCapacity, so a
+// capacity far below the pending set must not strand rows: one wake has to
+// drain every due row. Repair itself is stubbed because the filesystem path
+// defers a transient failure by repairBackoff's 30s floor, which would decide
+// this test's outcome for a reason that has nothing to do with queue capacity.
 func TestRepairDrainEventuallyCompletesBeyondQueue(t *testing.T) {
 	t.Parallel()
 	ctx := context.Background()
@@ -34,20 +42,47 @@ func TestRepairDrainEventuallyCompletesBeyondQueue(t *testing.T) {
 	if err := seed.Close(context.Background()); err != nil {
 		t.Fatal(err)
 	}
-	catalog, err := Open(ctx, Options{Path: path, QueueCapacity: 2, Now: time.Now})
+
+	var mu sync.Mutex
+	repaired := map[string]struct{}{}
+	drained := make(chan struct{})
+	catalog, err := Open(ctx, Options{
+		Path: path, QueueCapacity: 2, Now: time.Now,
+		repairSession: func(_ context.Context, session string) (agent.SessionListingRepairResult, error) {
+			mu.Lock()
+			defer mu.Unlock()
+			if _, seen := repaired[session]; !seen {
+				repaired[session] = struct{}{}
+				if len(repaired) == total {
+					close(drained)
+				}
+			}
+			return agent.SessionListingRepairResult{Status: agent.SessionListingRepairApplied, Preview: "ok", Turns: 1}, nil
+		},
+	})
 	if err != nil {
 		t.Fatal(err)
 	}
 	t.Cleanup(func() { _ = catalog.Close(context.Background()) })
-	deadline := time.Now().Add(15 * time.Second)
+
+	select {
+	case <-drained:
+	case <-time.After(30 * time.Second):
+		mu.Lock()
+		seen := len(repaired)
+		mu.Unlock()
+		t.Fatalf("repaired %d of %d sessions; a queue of 2 stranded the rest", seen, total)
+	}
+	// Every row reached repair; the batch commit that clears the pending count
+	// lands just after the last call, so wait for the count the drain owes.
+	deadline := time.Now().Add(10 * time.Second)
 	for {
-		status := catalog.Status()
-		if status.RepairPending == 0 {
+		if catalog.Status().RepairPending == 0 {
 			return
 		}
 		if time.Now().After(deadline) {
-			t.Fatalf("repair pending stuck at %d", status.RepairPending)
+			t.Fatalf("repair pending stuck at %d after every session was repaired", catalog.Status().RepairPending)
 		}
-		time.Sleep(50 * time.Millisecond)
+		time.Sleep(10 * time.Millisecond)
 	}
 }

internal/sessioncatalog/types.go

@@ -4,10 +4,12 @@
 package sessioncatalog
 
 import (
+	"context"
 	"path/filepath"
 	"strings"
 	"time"
 
+	"reasonix/internal/agent"
 	"reasonix/internal/config"
 )
 
@@ -90,6 +92,10 @@ type Options struct {
 	QueueCapacity int
 	Now           func() time.Time
 	OnRevision    func(uint64, []string, string)
+	// repairSession replaces the filesystem repair. Open installs it before
+	// starting repairLoop, so scheduler tests can drive the real wake path
+	// without racing the hook assignment.
+	repairSession func(context.Context, string) (agent.SessionListingRepairResult, error)
 }
 
 type DirectoryTarget struct {