test(kap-server): deflake three timing-sensitive tests (#2894)

* test(kap-server): poll the read-model immediate-read assertions

The 'prepares the read model at boot and serves immediate reads' test
sampled the session list / workspace session_count / paged read exactly
once after creating a session. While a mirror flush is in flight its
batch is only per-shard atomic and the pending-queue cleanup is not
linearized with reads, so a single-sample read landing inside that
window can transiently miss or double-count the new session (seen
twice on main CI as 'expected false to be true' and 'expected 0 to be
1'). Poll with vi.waitFor instead; the transient lasts at most one
in-flight flush (~100ms cadence).

* test(kap-server): retry the transcript test temp-dir teardown rm

The engine's file log writers flush synchronously on scope dispose but
their trailing async close can still create a file under the test home
after server.close() resolves, so the afterEach rm occasionally fails
with ENOTEMPTY on a loaded CI runner. Retry the rm (maxRetries: 5),
matching the existing pattern in questions.test.ts / fs.test.ts.

* test(kap-server): drain the in-flight search sync before appending

The 'serves the published generation without waiting for a blocked
background sync' test appended the delta right after a warm-up search
that had kicked a fire-and-forget background sync pass. On a starved
CI worker thread that pass can read the file after the append and
publish both documents early ('expected 2 to be 1'). settleSync before
the append makes 'no pass can index the delta' structural.

packages/kap-server/test/search/searchService.test.ts

@@ -1191,6 +1191,10 @@ describe('GlobalSearchService', () => {
       const service = track(makeService(home!, index));
       await service.reindex();
       expect((await service.search({ query: '苹果' })).items.length).toBe(1);
+      // The search above kicked a fire-and-forget background pass whose
+      // session enumeration already ran with block=false: drain it before the
+      // append, or a CI-slow pass reads the delta below and publishes it early.
+      await settleSync(service);
 
       // New bytes arrive, then the next background pass is blocked inside the
       // session enumeration. The search must return promptly with the OLD

packages/kap-server/test/sessions.test.ts

@@ -1595,24 +1595,38 @@ describe('server-v2 /api/v1/sessions (minidb read model)', () => {
     expect(status.body.code).toBe(0);
     expect(status.body.data.state).toBe('ready');
 
-    // A freshly created session lists, counts, and pages immediately — the
-    // mutation path never waited for the read model, the read path folds the
-    // mirror queue back in.
+    // A freshly created session lists, counts, and pages without waiting for
+    // the read model — the mutation path never awaited the read model, the
+    // read path folds the mirror queue back in. That fold is best-effort
+    // while a mirror flush is in flight: the flush's batch is only per-shard
+    // atomic and its pending-queue cleanup is not linearized with reads, so a
+    // read landing exactly inside that window can transiently miss (or
+    // double-count) the session — poll instead of sampling once.
     const created = await postJson<SessionWire>('/api/v1/sessions', {
       metadata: { cwd: home as string },
     });
     const id = created.body.data.id;
 
-    const listed = await getJson<PageWire>('/api/v1/sessions');
-    expect(listed.body.data.items.some((s) => s.id === id)).toBe(true);
+    await vi.waitFor(
+      async () => {
+        const listed = await getJson<PageWire>('/api/v1/sessions');
+        expect(listed.body.data.items.some((s) => s.id === id)).toBe(true);
 
-    const workspaces = await getJson<{ items: { session_count: number }[] }>('/api/v1/workspaces');
-    expect(workspaces.body.data.items[0]?.session_count).toBe(1);
+        const workspaces = await getJson<{ items: { session_count: number }[] }>(
+          '/api/v1/workspaces',
+        );
+        expect(workspaces.body.data.items[0]?.session_count).toBe(1);
 
-    const paged = await getJson<PageWire>(`/api/v1/sessions?page_size=1&before_id=${id}`);
-    expect(paged.body.data.items).toEqual([]);
-    expect(paged.body.data.has_more).toBe(false);
+        const paged = await getJson<PageWire>(`/api/v1/sessions?page_size=1&before_id=${id}`);
+        expect(paged.body.data.items).toEqual([]);
+        expect(paged.body.data.has_more).toBe(false);
+      },
+      { timeout: 10_000 },
+    );
 
+    // Archiving drains the mirror queue before responding, and the restart's
+    // boot prepare re-projects (or reuses) a fully settled generation — both
+    // of these reads are deterministic again.
     await postJson<{ archived: boolean }>(`/api/v1/sessions/${id}:archive`);
     const archivedOnly = await getJson<PageWire>('/api/v1/sessions?archived_only=true');
     expect(archivedOnly.body.data.items.map((s) => s.id)).toEqual([id]);

packages/kap-server/test/transcript.test.ts

@@ -175,7 +175,11 @@ describe('server-v2 /api/v1/sessions/{sid}/transcript', () => {
       server = undefined;
     }
     if (home !== undefined) {
-      await rm(home, { recursive: true, force: true });
+      // maxRetries: the engine's file log writers flush synchronously on scope
+      // dispose but their trailing async close can still be creating a file
+      // under home after server.close() resolves (ENOTEMPTY on a loaded CI
+      // runner) — same retry pattern as questions.test.ts / fs.test.ts.
+      await rm(home, { recursive: true, force: true, maxRetries: 5, retryDelay: 50 });
       home = undefined;
     }
   });