diff --git a/apps/web/tests/connection-recovery.e2e.ts b/apps/web/tests/connection-recovery.e2e.ts new file mode 100644 index 0000000000..6f1568df50 --- /dev/null +++ b/apps/web/tests/connection-recovery.e2e.ts @@ -0,0 +1,89 @@ +/** Real WebSocket loss without replacing Client plugins or navigating the page. */ +import { fileURLToPath } from 'node:url' +import { readFile } from 'node:fs/promises' +import { chromium, type WebSocketRoute } from 'playwright' +import { expect, it, onTestFailed, onTestFinished } from 'vitest' +import { launchWebScaffold, seedSession, watchConsole } from './scaffold.ts' +import { newEnglishPage, saveFailureShot, writeComposerDraft } from './support.ts' + +it.each([false, true])('retains the mounted application across WebSocket recovery with an active Session: %s', async (activeSession) => { + const scaffold = await launchWebScaffold({ + extraOverlayPath: fileURLToPath(new URL('./pin-browse-picker.overlay.yml', import.meta.url)), + }) + onTestFinished(() => scaffold.close()) + const title = 'Connection recovery session' + const draft = 'Unsent draft retained across connection recovery' + if (activeSession) { + const workspace = await scaffold.ctx.workspaceRegistry.create(scaffold.workspaceCwd) + const seed = fileURLToPath(new URL('../../../snapshots/web/seeded-history/session.v3.jsonl', import.meta.url)) + const sessionId = await seedSession(scaffold, await readFile(seed, 'utf8'), 'connection-recovery-session') + await workspace.attachSession(sessionId) + await scaffold.ctx.sessionController.rename({ sessionId, title }) + } + const browser = await chromium.launch() + onTestFinished(() => browser.close()) + const page = await newEnglishPage(browser) + const console = watchConsole(page) + onTestFailed(() => saveFailureShot(page, 'web-e2e-connection-recovery')) + const sockets: { client: WebSocketRoute; server: WebSocketRoute }[] = [] + let readyFrames = 0 + let sessionBaselines = 0 + await page.routeWebSocket('**/api/remote.mux', (client) => { + const server = client.connectToServer() + sockets.push({ client, server }) + const endpoints = new Map() + client.onMessage((message) => { + if (typeof message === 'string') { + const frame = JSON.parse(message) as { type?: string; streamId: string; endpoint: string } + if (frame.type === 'open') endpoints.set(frame.streamId, frame.endpoint) + } + server.send(message) + }) + server.onMessage((message) => { + if (typeof message === 'string') { + const frame = JSON.parse(message) as { type?: string; streamId: string; value?: { type?: string } } + if (frame.type === 'item' && frame.value?.type === 'ready') readyFrames++ + if (frame.type === 'item' && endpoints.get(frame.streamId) === 'session/follow' + && frame.value?.type === 'snapshot') sessionBaselines++ + } + client.send(message) + }) + }) + await page.goto(scaffold.authenticatedUrl, { waitUntil: 'load' }) + const composer = page.locator('[data-composer-input][contenteditable="true"]') + if (activeSession) { + await page.getByRole('treeitem').filter({ has: page.getByText(title, { exact: true }) }).click({ timeout: 20_000 }) + await writeComposerDraft(page, composer, draft) + await expect.poll(() => sessionBaselines).toBeGreaterThan(0) + } else { + await page.getByText('Into the Unknown', { exact: true }).waitFor({ timeout: 20_000 }) + } + await expect.poll(() => readyFrames).toBeGreaterThan(0) + const root = await page.locator('[data-slot="root"]').elementHandle() + expect(root).not.toBeNull() + let navigations = 0 + page.on('framenavigated', () => { navigations++ }) + + for (let attempt = 0; attempt < 3; attempt++) { + const previousReady = readyFrames + const previousBaseline = sessionBaselines + const active = sockets.at(-1)! + await Promise.all([ + active.client.close({ code: 1012, reason: 'test connection loss' }), + active.server.close({ code: 1012, reason: 'test connection loss' }), + ]) + await expect.poll(() => readyFrames, { timeout: 20_000 }).toBeGreaterThan(previousReady) + if (activeSession) { + await expect.poll(() => sessionBaselines, { timeout: 20_000 }).toBeGreaterThan(previousBaseline) + } + await page.evaluate(() => new Promise((resolve) => { + requestAnimationFrame(() => { requestAnimationFrame(() => { resolve() }) }) + })) + expect(await root!.evaluate(element => element.isConnected)).toBe(true) + expect(console.pageErrors).toEqual([]) + if (activeSession) expect(await composer.textContent()).toBe(draft) + } + + expect(navigations).toBe(0) + if (!activeSession) expect(await page.getByText('Into the Unknown', { exact: true }).isVisible()).toBe(true) +}) diff --git a/apps/web/tests/server-restart.e2e.ts b/apps/web/tests/server-restart.e2e.ts new file mode 100644 index 0000000000..c3d5565e40 --- /dev/null +++ b/apps/web/tests/server-restart.e2e.ts @@ -0,0 +1,199 @@ +/** Persisted Session recovery across two real dsh server processes on the same port. */ +import { spawn, type ChildProcess } from 'node:child_process' +import { randomUUID } from 'node:crypto' +import { createServer } from 'node:http' +import { mkdir, mkdtemp, rm, writeFile } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { fileURLToPath } from 'node:url' +import { chromium } from 'playwright' +import { expect, it, onTestFailed, onTestFinished } from 'vitest' +import { REPO_ROOT, newEnglishPage, saveFailureShot, writeComposerDraft } from './support.ts' + +class RestartableServer { + private child: ChildProcess | undefined + private closed: Promise | undefined + private output = '' + + constructor(private readonly world: string, private readonly modelUrl: string) {} + + async start(port: number): Promise { + if (this.child !== undefined) throw new Error('Server is already running') + this.output = '' + const ready = Promise.withResolvers() + const child = spawn(process.execPath, [ + join(REPO_ROOT, 'apps/cli/lib/bin.js'), '--profile', 'web', + '--patch', fileURLToPath(new URL('./pin-browse-picker.overlay.yml', import.meta.url)), + '--no-open', '--port', String(port), + ], { + cwd: this.world, + env: { + ...process.env, NODE_OPTIONS: '', + DSH_HOME: join(this.world, 'home'), DSH_AGENTS_HOME: join(this.world, 'agents'), + DSH_TELEMETRY_DISABLED: '1', DEEPSEEK_API_KEY: 'keyless-server-restart-fixture', + DEEPSEEK_BASE_URL: this.modelUrl, + }, + stdio: ['ignore', 'pipe', 'pipe'], + }) + this.child = child + const receive = (data: Buffer): void => { + this.output += data.toString() + const url = /dsh web: (http:\/\/[^\s]+)/u.exec(this.output)?.[1] + if (url !== undefined) ready.resolve(url) + } + child.stdout?.on('data', receive) + child.stderr?.on('data', receive) + child.once('error', (error) => { ready.reject(error) }) + this.closed = new Promise((resolve) => { + child.once('close', (code, signal) => { + ready.reject(new Error('Server exited before readiness: ' + JSON.stringify({ code, signal }) + '\n' + this.logs())) + resolve() + }) + }) + const timer = setTimeout(() => { ready.reject(new Error('Server readiness timed out\n' + this.logs())) }, 30_000) + try { return await ready.promise } + finally { clearTimeout(timer) } + } + + async stop(): Promise { + const child = this.child + if (child === undefined) return + let forced = false + const timer = setTimeout(() => { forced = true; child.kill('SIGKILL') }, 15_000) + try { + if (child.exitCode === null && child.signalCode === null) child.kill('SIGTERM') + await this.closed + } finally { + clearTimeout(timer) + this.child = undefined + } + if (forced) throw new Error('Server did not stop after SIGTERM\n' + this.logs()) + } + + logs(): string { return this.output.replace(/([?&]token=)[A-Za-z0-9_-]+/gu, '$1') } +} + +it('keeps an opened Session and draft when the server restarts on the same port', async () => { + const world = await mkdtemp(join(tmpdir(), 'dsh-server-restart-')) + onTestFinished(() => rm(world, { recursive: true, force: true })) + const model = createServer((request, response) => { + request.resume() + request.once('end', () => { + response.writeHead(200, { 'content-type': 'text/event-stream' }) + const events = [ + { type: 'message_start', message: { id: 'restart-fixture', model: 'mock-model', usage: { input_tokens: 3, output_tokens: 0 } } }, + { type: 'content_block_start', index: 0, content_block: { type: 'text', text: 'Persisted server restart fixture.' } }, + { type: 'content_block_stop', index: 0 }, + { type: 'message_delta', delta: { stop_reason: 'end_turn' }, usage: { output_tokens: 1 } }, + { type: 'message_stop' }, + ] + response.end(events.map(event => 'data: ' + JSON.stringify(event) + '\n\n').join('')) + }) + }) + onTestFinished(() => new Promise((resolve, reject) => { + model.close((error) => { if (error) reject(error); else resolve() }) + model.closeAllConnections() + })) + await new Promise((resolve, reject) => { + model.once('error', reject) + model.listen(0, '127.0.0.1', resolve) + }) + const modelAddress = model.address() + if (modelAddress === null || typeof modelAddress === 'string') throw new Error('Model fixture did not listen') + const server = new RestartableServer(world, `http://127.0.0.1:${String(modelAddress.port)}`) + onTestFinished(() => server.stop()) + const workspacePath = join(world, 'workspace') + await mkdir(workspacePath) + + const url = await server.start(0) + const port = Number(new URL(url).port) + const authenticated = await fetch(url, { redirect: 'manual' }) + const cookie = authenticated.headers.get('set-cookie')?.split(';', 1)[0] + if (cookie === undefined) throw new Error('Server did not issue an authentication cookie') + const rpc = async (endpoint: string, args: object): Promise => { + const response = await fetch(new URL('/api/' + endpoint, url), { + method: 'POST', headers: { 'content-type': 'application/json', cookie }, + body: JSON.stringify({ type: 'client-request', rpcId: randomUUID(), method: endpoint, payload: { args } }), + }) + const result = await response.json() as { result: { ok: boolean; value: T; error?: { message: string } } } + if (!result.result.ok) throw new Error(`${endpoint}: ${String(result.result.error?.message)}`) + return result.result.value + } + await rpc('workspace/create', { request: { path: workspacePath } }) + const scenarioId = randomUUID() + + const browser = await chromium.launch({ headless: true }) + onTestFinished(() => browser.close()) + const page = await newEnglishPage(browser) + const errors: string[] = [] + const graphs: { rev: string; entries: { id: string; rev: string }[] }[] = [] + let readyFrames = 0 + let sessionBaselines = 0 + let navigations = 0 + page.on('pageerror', (error) => { errors.push(String(error)) }) + page.on('websocket', (socket) => { + const endpoints = new Map() + socket.on('framesent', ({ payload }) => { + if (typeof payload !== 'string') return + const frame = JSON.parse(payload) as { type?: string; streamId: string; endpoint: string } + if (frame.type === 'open') endpoints.set(frame.streamId, frame.endpoint) + }) + socket.on('framereceived', ({ payload }) => { + if (typeof payload !== 'string') return + const message = JSON.parse(payload) as { type?: string; streamId: string; value?: { type?: string } } + if (message.type === 'item' && message.value?.type === 'ready') readyFrames++ + if (message.type === 'item' && endpoints.get(message.streamId) === 'session/follow' + && message.value?.type === 'snapshot') sessionBaselines++ + }) + }) + const network = await page.context().newCDPSession(page) + await network.send('Network.enable') + network.on('Network.eventSourceMessageReceived', (event) => { + const message = JSON.parse(event.data) as { type?: string; graph: typeof graphs[number] } + if (message.type === 'graph') graphs.push(message.graph) + }) + onTestFailed(async () => { + await saveFailureShot(page, 'web-server-restart-' + scenarioId) + await writeFile(join(REPO_ROOT, '.artifacts', 'server-restart-' + scenarioId + '.json'), JSON.stringify({ errors, graphs, server: server.logs() }, null, 2)) + }) + await page.goto(url, { waitUntil: 'load' }) + await page.locator('[data-slot="root"]').waitFor({ state: 'attached', timeout: 20_000 }) + const notice = page.getByRole('button', { name: 'Continue', exact: true }) + await notice.click({ timeout: 15_000 }) + const composer = page.locator('[data-composer-input][contenteditable="true"]') + await writeComposerDraft(page, composer, 'Create a completed turn for the server restart test.') + await composer.press('Enter') + const response = page.getByRole('paragraph').filter({ hasText: /^Persisted server restart fixture\.$/u }) + await response.last().waitFor({ timeout: 20_000 }) + const draft = 'Unsent draft across a server restart' + await writeComposerDraft(page, composer, draft) + await expect.poll(() => graphs.length).toBeGreaterThan(0) + await expect.poll(() => sessionBaselines).toBeGreaterThan(0) + const originalRoot = await page.locator('[data-slot="root"]').elementHandle() + expect(originalRoot).not.toBeNull() + const beforeReady = readyFrames + const beforeBaseline = sessionBaselines + const beforeGraphs = graphs.length + page.on('framenavigated', () => { navigations++ }) + + await server.stop() + const restarted = await server.start(port) + expect(new URL(restarted).origin).toBe(new URL(url).origin) + await expect.poll(() => readyFrames, { timeout: 30_000 }).toBeGreaterThan(beforeReady) + await expect.poll(() => sessionBaselines, { timeout: 30_000 }).toBeGreaterThan(beforeBaseline) + await expect.poll(() => graphs.length, { timeout: 30_000 }).toBeGreaterThan(beforeGraphs) + await page.evaluate(() => new Promise((resolve) => { + requestAnimationFrame(() => { requestAnimationFrame(() => { resolve() }) }) + })) + + expect(errors).toEqual([]) + expect(graphs.at(-1)).toEqual(graphs[beforeGraphs - 1]) + expect(await originalRoot!.evaluate(element => element.isConnected)).toBe(true) + expect(await composer.textContent()).toBe(draft) + expect(await response.isVisible()).toBe(true) + + await composer.press('Enter') + await expect.poll(() => response.count(), { timeout: 20_000 }).toBe(2) + expect(errors).toEqual([]) + expect(navigations).toBe(0) +}) diff --git a/apps/web/tsconfig.json b/apps/web/tsconfig.json index 9af71ce894..0608096055 100644 --- a/apps/web/tsconfig.json +++ b/apps/web/tsconfig.json @@ -22,6 +22,8 @@ // this client-registered project must not also hold them (one program // cannot see both sides of the cordis Context merges). "exclude": [ + "tests/connection-recovery.e2e.ts", + "tests/server-restart.e2e.ts", "tests/default-product-isolation.e2e.ts", "tests/diff-context.e2e.ts", "tests/scaffold.ts", diff --git a/packages/client/ui-renderer/tests/reconnect.client.spec.tsx b/packages/client/ui-renderer/tests/reconnect.client.spec.tsx new file mode 100644 index 0000000000..55293c494c --- /dev/null +++ b/packages/client/ui-renderer/tests/reconnect.client.spec.tsx @@ -0,0 +1,27 @@ +// @vitest-environment jsdom +import { act } from '@testing-library/react' +import { describe, expect, it, onTestFinished, vi } from 'vitest' +import { RemoteMock } from '@deepseek-ai/dsh-remote-mock' +import { TestClient, remoteDefaultResponses, webApp } from '@deepseek-ai/dsh-client-test-runtime/src/assembly/index.ts' +import type {} from '@deepseek-ai/dsh-client-ui-session/client' + +describe('assembled renderer connection recovery', () => { + it('keeps the mounted root across a transport reconnect', async () => { + const mock = RemoteMock.create().load(remoteDefaultResponses) + const client = await TestClient.start({ roster: webApp }, mock, { mount: true }) + onTestFinished(() => client.dispose()) + const root = client.container!.querySelector('[data-slot="root"]') + const generation = client.connection.generation.getSnapshot()!.id + expect(root).not.toBeNull() + + await act(async () => { + client.connection.reconnect() + await vi.waitFor(() => { + expect(client.connection.generation.getSnapshot()?.id).toBeGreaterThan(generation) + expect(client.connection.state.getSnapshot()).toBe('connected') + }) + }) + + expect(client.container!.querySelector('[data-slot="root"]')).toBe(root) + }, 60_000) +}) diff --git a/tsconfig.host.json b/tsconfig.host.json index 140e9f5894..8452b2ac30 100644 --- a/tsconfig.host.json +++ b/tsconfig.host.json @@ -10,6 +10,8 @@ "files": [], "include": [ "apps/web/product-isolation.ts", + "apps/web/tests/connection-recovery.e2e.ts", + "apps/web/tests/server-restart.e2e.ts", "apps/web/tests/default-product-isolation.e2e.ts", "apps/web/tests/diff-context.e2e.ts", "apps/web/tests/scaffold.ts",