zhifu-platform/frontend/tests/customer-sse.mjs

import assert from 'node:assert/strict';
import { mkdir, readFile, writeFile } from 'node:fs/promises';
import { execFile } from 'node:child_process';
import { promisify } from 'node:util';
import { fileURLToPath } from 'node:url';
import { chromium } from 'playwright';

const accounts = JSON.parse(await readFile(new URL('../../.work/frontend-accounts.json', import.meta.url), 'utf8'));
const started = Date.now();
const base = process.env.FRONTEND_URL || 'http://127.0.0.1:18762';
const browser = await chromium.launch({ headless: true, executablePath: process.env.PLAYWRIGHT_CHROMIUM_EXECUTABLE });
const context = await browser.newContext({ viewport: { width: 390, height: 844 } });
const page = await context.newPage();
page.setDefaultTimeout(20000);
const checks = [], errors = [], streams = [], ended = new Set();
let historyReads = 0, writes = 0, browserRequests = 0, serviceRequests = 0, customerToken, staffToken, conversationId, report;
const held = [];
page.on('pageerror', error => errors.push(error.message));
page.on('request', request => {
  if (/\/(openapi|admin)\//.test(request.url())) browserRequests++;
  if (request.url().endsWith('/events')) streams.push({ request, after: request.headers()['last-event-id'] });
  if (request.method() === 'GET' && request.url().includes('/messages?')) historyReads++;
  if (request.method() === 'POST' && request.url().endsWith('/messages')) writes++;
});
page.on('requestfailed', request => ended.add(request));
page.on('requestfinished', request => ended.add(request));
const content = page.locator('.mobile .content');
const timeout = setTimeout(() => { void browser.close(); }, 240000);

async function request(token, method, path, data, expected = 200) {
  serviceRequests++;
  const response = await context.request.fetch(`${base}${path}`, {
    method, headers: { Authorization: `Bearer ${token}`, 'Idempotency-Key': crypto.randomUUID() }, data,
  });
  assert.equal(response.status(), expected, await response.text());
  return expected === 204 ? null : response.json();
}
async function navigate(target) {
  await page.evaluate(value => { location.hash = `/customer/${value}`; }, target);
  await page.locator(`.mobile[data-page="${target}"]`).waitFor();
  await page.waitForFunction(() => !document.querySelector('.mobile .content[aria-busy="true"]'));
}
async function until(predicate, duration = 20000) {
  const deadline = Date.now() + duration;
  while (!predicate() && Date.now() < deadline) await new Promise(resolve => setTimeout(resolve, 100));
  assert(predicate(), '限定时间内没有完成预期的事件连接操作');
}
async function sendStaff(text) {
  const conversation = await request(staffToken, 'GET', `/openapi/v1/conversations/${conversationId}`);
  return request(staffToken, 'POST', `/openapi/v1/conversations/${conversationId}/messages`, { text, expected_control_version: conversation.control_version }, 201);
}

try {
  serviceRequests++;
  const staff = await context.request.post(`${base}/openapi/v1/auth/sessions`, { data: { tenant_id: accounts.tenant_id, username: 'staff01', password: accounts.password } });
  assert.equal(staff.status(), 200);
  staffToken = (await staff.json()).access_token;
  await page.goto(`${base}/#/customer/login`);
  await content.getByLabel('账户', { exact: true }).fill('customer01');
  await content.getByLabel('密码', { exact: true }).fill(accounts.password);
  await content.getByRole('checkbox').check();
  const authenticated = page.waitForResponse(response => response.request().method() === 'POST' && response.url().endsWith('/auth/sessions'));
  await content.getByRole('button', { name: '登录并继续', exact: true }).click();
  customerToken = (await (await authenticated).json()).access_token;
  await until(() => streams.length > 0);
  conversationId = new URL(streams[0].request.url()).pathname.split('/').at(-2);
  await content.getByRole('button', { name: '发送', exact: true }).waitFor();
  await page.waitForFunction(() => ![...document.querySelectorAll('.mobile button')].some(button => button.textContent === '重新连接'));
  const initial = await request(staffToken, 'GET', `/openapi/v1/conversations/${conversationId}`);
  assert.equal(initial.active_run_id, null, '本专项要求会话当前没有生成运行');
  if (initial.control_owner === 'AGENT') await request(staffToken, 'POST', `/openapi/v1/conversations/${conversationId}/takeover`, { expected_control_version: initial.control_version });
  else assert.equal(initial.assigned_staff_id, accounts.actors.staff01);
  await content.getByText('人工客服正在为你服务。', { exact: true }).waitFor();
  const initialReads = historyReads;
  const text = `SSE 人工回复 ${Date.now()}`;
  const first = await sendStaff(text);
  await content.locator('.message').filter({ hasText: text }).waitFor();
  assert.equal(await content.locator('.message').filter({ hasText: text }).count(), 1);
  assert.equal(historyReads, initialReads, '人工回复应通过 SSE 到达');
  checks.push('真实客服接管和回复通过 SSE 到达，消息只显示一次');

  const connectionCount = streams.length;
  await until(() => streams.length > connectionCount, 55000);
  assert.equal(streams.at(-1).after, String(first.sequence));
  assert.equal(historyReads, initialReads, '正常连接续传不轮询消息历史');
  checks.push('50 秒连接结束后自动携带 Last-Event-ID 续传');

  const draft = `网络恢复后保留的客户输入 ${Date.now()}`;
  await content.getByLabel('补充问题', { exact: true }).fill(draft);
  await context.setOffline(true);
  await content.getByRole('button', { name: '重新连接', exact: true }).waitFor();
  assert.equal(await content.getByRole('button', { name: '发送', exact: true }).isDisabled(), true);
  await context.setOffline(false);
  const missing = `断线期间的真实客服消息 ${Date.now()}`;
  await sendStaff(missing);
  await content.getByRole('button', { name: '重新连接', exact: true }).click();
  await content.locator('.message').filter({ hasText: missing }).waitFor();
  assert.equal(await content.getByLabel('补充问题', { exact: true }).inputValue(), draft);
  assert.equal(writes, 0, '重新连接不能自动提交客户输入');
  await content.getByRole('button', { name: '发送', exact: true }).click();
  await content.locator('.message.customer').filter({ hasText: draft }).waitFor();
  assert.equal(writes, 1);
  checks.push('真实断网禁用发送，恢复后补充读取消息并保留输入，客户手动发送一次');

  const currentStream = streams.at(-1).request;
  await navigate('profile');
  await until(() => ended.has(currentStream));
  const countAway = streams.length;
  await new Promise(resolve => setTimeout(resolve, 1000));
  assert.equal(streams.length, countAway);
  checks.push('离开咨询页面终止事件连接，其他页面不创建连接');

  const root = fileURLToPath(new URL('../../', import.meta.url));
  await promisify(execFile)(fileURLToPath(new URL('../../.venv/Scripts/python.exe', import.meta.url)), ['-m', 'zhifu.operations.conversation_retention', '--tenant', accounts.tenant_id, '--conversation', conversationId, '--before', new Date().toISOString()], { cwd: root, timeout: 20000, windowsHide: true });
  // 仅提交一次过期游标，响应及随后的状态恢复均来自真实业务服务。
  await page.route('**/events', route => route.continue({ headers: { ...route.request().headers(), 'last-event-id': '0' } }), { times: 1 });
  const expired = page.waitForResponse(response => response.url().endsWith('/events') && response.status() === 410);
  await navigate('chat');
  assert.equal((await (await expired).json()).code, 'EVENT_CURSOR_EXPIRED');
  await page.getByText('事件续传记录已超过保留期限，已重新读取历史及当前状态。', { exact: true }).waitFor();
  await content.locator('.message.customer').filter({ hasText: draft }).waitFor();
  assert.equal(await content.locator('.message.customer').filter({ hasText: draft }).count(), 1);
  await page.waitForFunction(() => ![...document.querySelectorAll('.mobile button')].some(button => button.textContent === '重新连接'));
  assert.equal(writes, 1);
  checks.push('真实保留窗口与一次过期请求游标触发 410，恢复历史和当前状态且不重复提交消息');
  await navigate('profile');
  await new Promise(resolve => setTimeout(resolve, 1000));

  // 三条真实订阅占用同一账户额度，第四条由业务服务返回 429。
  const beforeLimit = await request(customerToken, 'GET', `/openapi/v1/conversations/${conversationId}`);
  await Promise.all(Array.from({ length: 3 }, async () => {
    const controller = new AbortController(); held.push(controller);
    serviceRequests++;
    const response = await fetch(`${base}/openapi/v1/conversations/${conversationId}/events`, { headers: { Authorization: `Bearer ${customerToken}`, 'Last-Event-ID': String(beforeLimit.sequence) }, signal: controller.signal });
    assert.equal(response.status, 200);
  }));
  const limited = page.waitForResponse(response => response.url().endsWith('/events') && response.status() === 429);
  await navigate('chat');
  const retryAfter = Number((await limited).headers()['retry-after']);
  assert.ok(retryAfter > 0 && retryAfter <= 60);
  await content.getByRole('button', { name: '重新连接', exact: true }).waitFor();
  assert.equal(await content.getByRole('button', { name: '发送', exact: true }).isDisabled(), true);
  assert.equal(await content.getByRole('button', { name: '重新连接', exact: true }).isDisabled(), true);
  assert.equal(await content.getByRole('button', { name: '重新加载', exact: true }).isDisabled(), true);
  held.forEach(controller => controller.abort());
  await new Promise(resolve => setTimeout(resolve, 1000));
  await content.getByRole('button', { name: '重新连接', exact: true }).click({ timeout: (retryAfter + 5) * 1000 });
  await page.waitForFunction(() => ![...document.querySelectorAll('.mobile button')].some(button => button.textContent === '重新连接'));
  checks.push('三条真实账户订阅触发 429，释放额度后用户可以重新连接');

  const current = await request(staffToken, 'GET', `/openapi/v1/conversations/${conversationId}`);
  await request(staffToken, 'POST', `/openapi/v1/conversations/${conversationId}/release`, { expected_control_version: current.control_version });
  await content.getByText('智能助手服务中', { exact: true }).waitFor();
  await request(customerToken, 'DELETE', '/openapi/v1/auth/sessions/current', undefined, 204);
  customerToken = null;
  await page.waitForURL('**/#/customer/login');
  assert.equal(await content.locator('.message').count(), 0);
  checks.push('释放接管实时恢复助手状态，撤销真实会话后停止订阅并进入登录页');
  assert.deepEqual(errors, []);
  report = { date: new Date().toISOString(), checks, browserErrors: errors, streamConnections: streams.length, historyReads, customerWrites: writes, generatedModelCalls: 0 };
  console.log(checks.join('\n'));
} finally {
  clearTimeout(timeout);
  held.forEach(controller => controller.abort());
  await context.setOffline(false);
  if (conversationId && staffToken) {
    const current = await request(staffToken, 'GET', `/openapi/v1/conversations/${conversationId}`);
    if (current.control_owner === 'HUMAN' && current.assigned_staff_id === accounts.actors.staff01) await request(staffToken, 'POST', `/openapi/v1/conversations/${conversationId}/release`, { expected_control_version: current.control_version });
  }
  if (customerToken) await request(customerToken, 'DELETE', '/openapi/v1/auth/sessions/current', undefined, 204);
  if (staffToken) await request(staffToken, 'DELETE', '/openapi/v1/auth/sessions/current', undefined, 204);
  await browser.close();
  if (report) {
    await mkdir(new URL('../../.work/frontend-verification/', import.meta.url), { recursive: true });
    await writeFile(new URL('../../.work/frontend-verification/customer-sse.json', import.meta.url), JSON.stringify({ ...report, seconds: (Date.now() - started) / 1000, browserRequests, serviceRequests, requestCount: browserRequests + serviceRequests, durationLimitSeconds: 240 }, null, 2));
  }
}