test(e2e): gate the reconnect delta and give the dedup assertion a source

The post-reconnect flow(5) response was ungated, so any spec that fetched
flow 5 twice consumed it and rendered a message that never streamed. It now
serves only after a drop, behind a flag `dropAndReconnect` raises.

The no-duplicate assertion beside it had nothing that could produce a
duplicate: every id reached the page exactly once by construction. The
resubscribe now replays the id the refetch already delivered — the real
server behaviour the client dedups — followed by a sentinel that proves
the replay arrived rather than merely being awaited.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

frontend/e2e/helpers/reconnect.ts

@@ -2,12 +2,16 @@ import type { Page } from '@playwright/test';
 
 import type { MockWorld } from '../mocks/world.ts';
 
+/** Cassette entries gated on this serve only after a drop — see `dropAndReconnect`. */
+export const RECONNECTED_FLAG = 'reconnected';
+
 /**
  * Drops every live mock socket and fast-forwards the page clock past
  * graphql-ws's jittered retryWait (1–4s for the first retry), so the reconnect
  * happens deterministically instead of on a random timer.
  */
 export const dropAndReconnect = async (page: Page, world: MockWorld): Promise<void> => {
+    world.raiseFlag(RECONNECTED_FLAG);
     world.dropSockets();
     await page.clock.fastForward(5_000);
 };

frontend/e2e/mocks/cassettes/flows.ts

@@ -29,6 +29,7 @@ import {
 
 import type { Cassette } from '../cassette.ts';
 
+import { RECONNECTED_FLAG } from '../../helpers/reconnect.ts';
 import { entity, mergeCassettes } from '../cassette.ts';
 import { baseQueries, baseRest } from './base.ts';
 
@@ -118,6 +119,9 @@ const FLOW_A_TERMINAL_LOGS = [
 export const FLOW_A_INITIAL_IDS = ['101', '102', '103'];
 export const FLOW_A_STREAMED_IDS = ['104', '105'];
 export const FLOW_A_RECONNECT_ID = '106';
+export const FLOW_A_REPLAY_SENTINEL_ID = '107';
+/** Gates the resubscribe replay; the spec raises it once the reconnect has landed. */
+export const REPLAY_FLAG = 'flow-a-replay';
 export const FLOW_B_INITIAL_IDS = ['201', '202'];
 export const FLOW_B_STREAMED_IDS = ['203', '204'];
 
@@ -163,6 +167,7 @@ export const flowsCassette = (override: Cassette = {}): Cassette =>
                             FLOW_A_TERMINAL_LOGS,
                         ),
                         variables: { id: '5' },
+                        whenFlag: RECONNECTED_FLAG,
                     },
                     { data: flowQueryData(FLOW_B, messagesFor('6', FLOW_B_INITIAL_IDS)), variables: { id: '6' } },
                 ],
@@ -172,7 +177,16 @@ export const flowsCassette = (override: Cassette = {}): Cassette =>
             subscriptions: {
                 messageLogAdded: [
                     {
-                        frames: FLOW_A_STREAMED_IDS.map((id) => addedFrame(makeMessage(id, '5'), 80)),
+                        frames: [
+                            ...FLOW_A_STREAMED_IDS.map((id) => addedFrame(makeMessage(id, '5'), 80)),
+                            // Not a copy-paste: a resubscribe replays what the refetch already
+                            // delivered, and the sentinel after it proves the replay was received.
+                            {
+                                ...addedFrame(makeMessage(FLOW_A_RECONNECT_ID, '5'), 0),
+                                whenFlag: REPLAY_FLAG,
+                            },
+                            addedFrame(makeMessage(FLOW_A_REPLAY_SENTINEL_ID, '5'), 0),
+                        ],
                         variables: { flowId: '5' },
                     },
                     {

frontend/e2e/mocks/world.ts

@@ -139,6 +139,12 @@ export class MockWorld {
         return entry ? { entry, streamKey: `sub:${operationName}:${stableStringify(variables ?? {})}` } : undefined;
     }
 
+    raiseFlag(flag: string): void {
+        this.flags.add(flag);
+        this.flagWaiters.get(flag)?.forEach((resolve) => resolve());
+        this.flagWaiters.delete(flag);
+    }
+
     registerSocket(socket: MockSocket): void {
         this.sockets.add(socket);
     }
@@ -221,10 +227,4 @@ export class MockWorld {
 
         return entry;
     }
-
-    private raiseFlag(flag: string): void {
-        this.flags.add(flag);
-        this.flagWaiters.get(flag)?.forEach((resolve) => resolve());
-        this.flagWaiters.delete(flag);
-    }
 }

frontend/e2e/specs/flows/reconnect.spec.ts

@@ -5,12 +5,15 @@ import { assertNoDuplicates, extractMessageIds, MESSAGE_ID_TESTID } from '../../
 import {
     FLOW_A_INITIAL_IDS,
     FLOW_A_RECONNECT_ID,
+    FLOW_A_REPLAY_SENTINEL_ID,
     FLOW_A_STREAMED_IDS,
     flowsCassette,
+    REPLAY_FLAG,
 } from '../../mocks/cassettes/flows.ts';
 
 const BEFORE_DROP = [...FLOW_A_INITIAL_IDS, ...FLOW_A_STREAMED_IDS];
 const AFTER_RECONNECT = [...BEFORE_DROP, FLOW_A_RECONNECT_ID];
+const AFTER_REPLAY = [...AFTER_RECONNECT, FLOW_A_REPLAY_SENTINEL_ID];
 
 test.describe('flow reconnect', { tag: ['@flows', '@smoke'] }, () => {
     test.use({ cassette: flowsCassette() });
@@ -24,10 +27,14 @@ test.describe('flow reconnect', { tag: ['@flows', '@smoke'] }, () => {
 
         await expect(page.getByTestId(MESSAGE_ID_TESTID)).toHaveCount(AFTER_RECONNECT.length);
 
+        world.raiseFlag(REPLAY_FLAG);
+
+        await expect(page.getByTestId(MESSAGE_ID_TESTID)).toHaveCount(AFTER_REPLAY.length);
+
         const ids = await extractMessageIds(page.locator('body'));
 
         assertNoDuplicates(ids);
-        expect([...ids].sort()).toEqual([...AFTER_RECONNECT].sort());
+        expect([...ids].sort()).toEqual([...AFTER_REPLAY].sort());
         expectCleanPage(pageErrorLog);
     });
 });