|
|
@@ -0,0 +1,564 @@
|
|
|
+/**
|
|
|
+ * 验证 F073「正文真实增量流式输出」的两块：
|
|
|
+ * 第 1 部分 —— **纯逻辑**（不必联网）：增量规整器的前缀不变式 + 流状态机
|
|
|
+ * 第 2 部分 —— **协调器级假流**：假 SSE 流喂真实协调器，断言事件通道与去重
|
|
|
+ *
|
|
|
+ * 为什么规整器要这么较真：它手里攥着一个不变式——
|
|
|
+ * **任一时刻「已发出的增量拼接」=== `normalizeChatMarkdown(已输入的全部文本)`**
|
|
|
+ * 只要不变式成立，后端 `answer_end` 的「整块替换」在正常路径上就是纯尾缀追加
|
|
|
+ * （不会触发整段重建 → 不会让「思考中」卡片闪现重播）。破了它，观感立刻劣化。
|
|
|
+ *
|
|
|
+ * 怎么跑（先重新打包，必须带 --loader:.png=dataurl）：
|
|
|
+ * npx esbuild harness/tools/_entry-coordinator.ts --bundle --format=esm \
|
|
|
+ * --outfile=harness/tools/_coordinator.mjs --alias:@=./src \
|
|
|
+ * --define:import.meta.env='{}' --loader:.png=dataurl
|
|
|
+ * node harness/tools/verify-answer-stream.mjs
|
|
|
+ */
|
|
|
+
|
|
|
+import {
|
|
|
+ ApiChatCoordinator,
|
|
|
+ applyAnswerStreamAbort,
|
|
|
+ applyAnswerStreamEnd,
|
|
|
+ applyAnswerStreamEvent,
|
|
|
+ createAnswerStreamState,
|
|
|
+ createIncrementalMarkdownNormalizer,
|
|
|
+ normalizeChatMarkdown,
|
|
|
+ parseAnswerStreamEvent,
|
|
|
+ shouldKeepAnswerOverview,
|
|
|
+} from './_coordinator.mjs';
|
|
|
+
|
|
|
+let pass = 0;
|
|
|
+let fail = 0;
|
|
|
+const check = (name, cond, extra = '') => {
|
|
|
+ if (cond) {
|
|
|
+ pass++;
|
|
|
+ console.log(` ok ${name}`);
|
|
|
+ } else {
|
|
|
+ fail++;
|
|
|
+ console.log(` FAIL ${name}${extra === '' ? '' : ` → ${JSON.stringify(extra)}`}`);
|
|
|
+ }
|
|
|
+};
|
|
|
+
|
|
|
+/* ================================================================== *
|
|
|
+ * 第 1 部分：纯逻辑
|
|
|
+ * ================================================================== */
|
|
|
+console.log('【1】增量规整器：前缀不变式');
|
|
|
+
|
|
|
+/** 把一串输入分片喂给规整器，返回逐片输出 */
|
|
|
+const feed = (chunks) => {
|
|
|
+ const normalizer = createIncrementalMarkdownNormalizer();
|
|
|
+ const out = [];
|
|
|
+ for (const chunk of chunks) out.push(normalizer.push(chunk));
|
|
|
+ return { out, normalizer };
|
|
|
+};
|
|
|
+
|
|
|
+/**
|
|
|
+ * 不变式（流中）：**已发出的文本必须是「规整后全文」的前缀**，
|
|
|
+ * 且未发出的差额只能是「**给当前行准备的空行分隔符（可能 1~2 个 \n）+ 当前行的尾巴**」——
|
|
|
+ * 也就是「为了判断该行是不是列表项而短暂扣住的一小段」，行尾一到就释放。
|
|
|
+ *
|
|
|
+ * 这个性质正好保证 `answer_end` 的「整块替换」= 纯尾缀追加（不会整段重建）。
|
|
|
+ */
|
|
|
+const isPrefixWithHeldLineTail = (got, want) => {
|
|
|
+ if (!want.startsWith(got)) return false;
|
|
|
+ const held = want.slice(got.length).replace(/^\n+/, ''); // 去掉随行一起扣住的前置空行
|
|
|
+ return !held.includes('\n');
|
|
|
+};
|
|
|
+
|
|
|
+/** 不变式：每喂一片就核对一次 */
|
|
|
+const checkInvariant = (label, chunks, { finish = false } = {}) => {
|
|
|
+ const normalizer = createIncrementalMarkdownNormalizer();
|
|
|
+ let want = '';
|
|
|
+ let got = '';
|
|
|
+ let ok = true;
|
|
|
+ let at = 0;
|
|
|
+ for (let i = 0; i < chunks.length; i++) {
|
|
|
+ want += chunks[i];
|
|
|
+ got += normalizer.push(chunks[i]);
|
|
|
+ if (!isPrefixWithHeldLineTail(got, normalizeChatMarkdown(want))) {
|
|
|
+ ok = false;
|
|
|
+ at = i;
|
|
|
+ break;
|
|
|
+ }
|
|
|
+ }
|
|
|
+ if (ok && finish) {
|
|
|
+ got += normalizer.finish();
|
|
|
+ if (got !== normalizeChatMarkdown(want)) ok = false;
|
|
|
+ }
|
|
|
+ check(
|
|
|
+ `不变式：${label}`,
|
|
|
+ ok,
|
|
|
+ ok ? '' : { 第几片: at, 期望: normalizeChatMarkdown(want).slice(-40), 实际: got.slice(-40) }
|
|
|
+ );
|
|
|
+ return ok;
|
|
|
+};
|
|
|
+
|
|
|
+checkInvariant('整段一次 push', ['第一行\n第二行\n第三行']);
|
|
|
+checkInvariant('逐字 push（含换行）', [...'第一行\n第二行\n1. a\n2. b']);
|
|
|
+checkInvariant('双换行（空行）', ['段落一\n\n段落二']);
|
|
|
+checkInvariant('连续有序列表不插空行', ['1. 甲\n2. 乙\n3. 丙']);
|
|
|
+checkInvariant('连续无序列表不插空行', ['- 甲\n- 乙']);
|
|
|
+checkInvariant('列表后接说明行', ['- 甲\n- 乙\n以上就是全部']);
|
|
|
+checkInvariant('伪列表符（2倍）不误判', ['2倍\n3. 真列表']);
|
|
|
+checkInvariant('行首恰为列表前缀（- ）', ['前一行\n- ']);
|
|
|
+checkInvariant('行首数字未定型（12）', ['前一行\n12']);
|
|
|
+checkInvariant('行首数字带点（12.）', ['前一行\n12.']);
|
|
|
+checkInvariant('末尾无换行', ['段一\n段二（未完']);
|
|
|
+checkInvariant('以换行结尾', ['段一\n']);
|
|
|
+checkInvariant('空串与纯空白', ['', ' ', '\n', ' \n']);
|
|
|
+checkInvariant('逐字 push 且收尾 finish', [...'甲\n1. 乙\n丙'], { finish: true });
|
|
|
+
|
|
|
+// 逐字 push 与整段 push 的「拼接结果」必须一致（都 finish 掉扣住的尾巴后比）
|
|
|
+{
|
|
|
+ const text = '第一段\n第二段\n1. 甲\n2. 乙\n收尾说明';
|
|
|
+ const wholeFeed = feed([text]);
|
|
|
+ wholeFeed.out.push(wholeFeed.normalizer.finish());
|
|
|
+ const perCharFeed = feed([...text]);
|
|
|
+ perCharFeed.out.push(perCharFeed.normalizer.finish());
|
|
|
+ const whole = wholeFeed.out.join('');
|
|
|
+ const perChar = perCharFeed.out.join('');
|
|
|
+ check('逐字 push 与整段 push 输出一致', whole === perChar, { whole, perChar });
|
|
|
+ check('两者的拼接都等于整段规整结果', whole === normalizeChatMarkdown(text));
|
|
|
+}
|
|
|
+
|
|
|
+// 种子化伪随机模糊（≥500 轮）
|
|
|
+{
|
|
|
+ let seed = 20260920;
|
|
|
+ const rand = () => {
|
|
|
+ seed = (seed * 1103515245 + 12345) % 2147483648;
|
|
|
+ return seed / 2147483648;
|
|
|
+ };
|
|
|
+ const alphabet = ['甲', '乙', '丙', '\n', '\n\n', '- ', '* ', '1', '2', '.', ')', ' ', '。', '12', '-'];
|
|
|
+ let bad = null;
|
|
|
+ for (let round = 0; round < 500 && !bad; round++) {
|
|
|
+ const chunks = [];
|
|
|
+ const count = 1 + Math.floor(rand() * 20);
|
|
|
+ for (let i = 0; i < count; i++) chunks.push(alphabet[Math.floor(rand() * alphabet.length)]);
|
|
|
+ const normalizer = createIncrementalMarkdownNormalizer();
|
|
|
+ let want = '';
|
|
|
+ let got = '';
|
|
|
+ for (const chunk of chunks) {
|
|
|
+ want += chunk;
|
|
|
+ got += normalizer.push(chunk);
|
|
|
+ if (!isPrefixWithHeldLineTail(got, normalizeChatMarkdown(want))) {
|
|
|
+ bad = { round, chunks, want, got };
|
|
|
+ break;
|
|
|
+ }
|
|
|
+ }
|
|
|
+ if (!bad) {
|
|
|
+ got += normalizer.finish();
|
|
|
+ if (got !== normalizeChatMarkdown(want)) bad = { round, chunks, want, got, finish: true };
|
|
|
+ }
|
|
|
+ }
|
|
|
+ check('种子化模糊 500 轮（含 finish）不变式恒成立', !bad, bad);
|
|
|
+}
|
|
|
+
|
|
|
+// 扣住的部分必须「有界」：只是当前行的一小段，不能无限积压
|
|
|
+{
|
|
|
+ const normalizer = createIncrementalMarkdownNormalizer();
|
|
|
+ let shown = '';
|
|
|
+ let want = '';
|
|
|
+ for (const chunk of ['第一段\n', '第二段\n', '1', '.', ' ', '甲', '\n', '2', '.', ' ', '乙']) {
|
|
|
+ want += chunk;
|
|
|
+ shown += normalizer.push(chunk);
|
|
|
+ const held = normalizeChatMarkdown(want).slice(shown.length).replace(/^\n+/, '');
|
|
|
+ if (held.includes('\n') || held.length > 8) {
|
|
|
+ check(`扣住有界（本条 ${JSON.stringify(chunk)}）`, false, { held, shown });
|
|
|
+ break;
|
|
|
+ }
|
|
|
+ }
|
|
|
+ check('扣住的只是「可能成为列表项」的短前缀（本例 ≤8 字符且不含换行）', true);
|
|
|
+}
|
|
|
+
|
|
|
+console.log('\n【2】流式事件解析（形状不符 → null 走旧路径）');
|
|
|
+{
|
|
|
+ const start = parseAnswerStreamEvent({ stream_id: 's1', target: 'summary', sequence: 0 }, 'answer_start');
|
|
|
+ check('合法的 answer_start', start?.kind === 'start' && start.streamId === 's1' && start.target === 'summary', start);
|
|
|
+
|
|
|
+ check('缺 stream_id → null', parseAnswerStreamEvent({ target: 'summary' }, 'answer_start') === null);
|
|
|
+ check('空 stream_id → null', parseAnswerStreamEvent({ stream_id: ' ', target: 'summary' }, 'answer_delta') === null);
|
|
|
+ check('target 非法 → null', parseAnswerStreamEvent({ stream_id: 's1', target: 'other' }, 'answer_delta') === null);
|
|
|
+ check('非对象 → null', parseAnswerStreamEvent(null, 'answer_delta') === null && parseAnswerStreamEvent([1], 'answer_delta') === null);
|
|
|
+
|
|
|
+ const delta = parseAnswerStreamEvent({ stream_id: 's1', target: 'answer', sequence: '3', text: 123 }, 'answer_delta');
|
|
|
+ check('sequence 宽松（字符串也认）', delta?.sequence === 3, delta);
|
|
|
+ check('text 非字符串按空串', delta?.text === '', delta);
|
|
|
+
|
|
|
+ const end = parseAnswerStreamEvent({ stream_id: 's1', target: 'summary', text: 'x', status: '怪值' }, 'answer_end');
|
|
|
+ check('status 非法 → fallback', end?.kind === 'end' && end.status === 'fallback', end);
|
|
|
+ const endOk = parseAnswerStreamEvent({ stream_id: 's1', target: 'summary', text: 'x', status: 'completed' }, 'answer_end');
|
|
|
+ check('status=completed 保留', endOk?.status === 'completed');
|
|
|
+
|
|
|
+ const abort = parseAnswerStreamEvent({ stream_id: 's1', target: 'summary', reason: '怪值' }, 'answer_abort');
|
|
|
+ check('reason 非法 → generation_failed', abort?.kind === 'abort' && abort.reason === 'generation_failed', abort);
|
|
|
+ const abortOk = parseAnswerStreamEvent({ stream_id: 's1', target: 'summary', reason: 'not_finalized' }, 'answer_abort');
|
|
|
+ check('reason=not_finalized 保留', abortOk?.reason === 'not_finalized');
|
|
|
+
|
|
|
+ check('未知事件名 → null', parseAnswerStreamEvent({ stream_id: 's1', target: 'summary' }, 'answer_what') === null);
|
|
|
+}
|
|
|
+
|
|
|
+console.log('\n【3】流状态机');
|
|
|
+{
|
|
|
+ const run = (events) => {
|
|
|
+ let state = createAnswerStreamState();
|
|
|
+ const effects = [];
|
|
|
+ for (const ev of events) {
|
|
|
+ const r = applyAnswerStreamEvent(state, ev);
|
|
|
+ state = r.state;
|
|
|
+ effects.push(r.effect.kind);
|
|
|
+ }
|
|
|
+ return { state, effects };
|
|
|
+ };
|
|
|
+ const ev = (kind, extra = {}) => ({ kind, streamId: 's1', target: 'summary', sequence: 0, text: '', ...extra });
|
|
|
+
|
|
|
+ // 正常流
|
|
|
+ {
|
|
|
+ const { state, effects } = run([
|
|
|
+ ev('start', { sequence: 0 }),
|
|
|
+ ev('delta', { sequence: 1, text: '你好' }),
|
|
|
+ ev('delta', { sequence: 2, text: '世界' }),
|
|
|
+ ev('end', { sequence: 2, text: '你好，世界' }),
|
|
|
+ ]);
|
|
|
+ check('正常流：效果序列 start→append→append→replace', JSON.stringify(effects) === JSON.stringify(['start', 'append', 'append', 'replace']), effects);
|
|
|
+ check('正常流：end 后 text = 最终全文（整块替换，不拼接）', state.text === '你好，世界' && state.completed === true, state);
|
|
|
+ }
|
|
|
+
|
|
|
+ // 序号缺口
|
|
|
+ {
|
|
|
+ const { state, effects } = run([
|
|
|
+ ev('start'),
|
|
|
+ ev('delta', { sequence: 1, text: 'a' }),
|
|
|
+ ev('delta', { sequence: 5, text: 'b' }),
|
|
|
+ ]);
|
|
|
+ check('序号缺口：仍拼接、hasGap 置位', state.text === 'ab' && state.hasGap === true, state);
|
|
|
+ check('缺口不阻断渲染（仍是 append）', effects[2] === 'append');
|
|
|
+ }
|
|
|
+
|
|
|
+ // 重复序号
|
|
|
+ {
|
|
|
+ const { state } = run([ev('start'), ev('delta', { sequence: 1, text: 'a' }), ev('delta', { sequence: 1, text: 'XX' })]);
|
|
|
+ check('重复序号忽略（不重复拼接）', state.text === 'a', state);
|
|
|
+ }
|
|
|
+
|
|
|
+ // 过期流
|
|
|
+ {
|
|
|
+ const { state, effects } = run([
|
|
|
+ ev('start'),
|
|
|
+ ev('delta', { streamId: 'other', sequence: 1, text: 'X' }),
|
|
|
+ ev('end', { streamId: 'other', text: 'X' }),
|
|
|
+ ev('abort', { streamId: 'other' }),
|
|
|
+ ]);
|
|
|
+ check('过期 streamId 的 delta/end/abort 全部 none', effects.slice(1).every((k) => k === 'none'), effects);
|
|
|
+ check('过期事件不改状态', state.text === '' && state.completed === false, state);
|
|
|
+ }
|
|
|
+
|
|
|
+ // end 之后的 delta
|
|
|
+ {
|
|
|
+ const { state, effects } = run([ev('start'), ev('end', { text: '定稿' }), ev('delta', { sequence: 9, text: '迟到' })]);
|
|
|
+ check('end 之后的 delta 忽略', effects[2] === 'none' && state.text === '定稿', state);
|
|
|
+ }
|
|
|
+
|
|
|
+ // abort 复位 + 重新 start
|
|
|
+ {
|
|
|
+ const { state, effects } = run([
|
|
|
+ ev('start'),
|
|
|
+ ev('delta', { sequence: 1, text: '草稿' }),
|
|
|
+ ev('abort', { reason: 'generation_failed' }),
|
|
|
+ ev('start', { streamId: 's2' }),
|
|
|
+ ev('delta', { streamId: 's2', sequence: 1, text: '新正文' }),
|
|
|
+ ]);
|
|
|
+ check('abort 复位（discard 效果 + 状态清空后可再 start）', effects.includes('discard') && state.text === '新正文' && state.streamId === 's2', { effects, state });
|
|
|
+ }
|
|
|
+
|
|
|
+ // end 覆盖草稿（fallback 也替换）
|
|
|
+ {
|
|
|
+ const { state } = run([
|
|
|
+ ev('start'),
|
|
|
+ ev('delta', { sequence: 1, text: '草稿' }),
|
|
|
+ ev('end', { sequence: 1, text: '最终', status: 'fallback' }),
|
|
|
+ ]);
|
|
|
+ check('fallback 的 end 同样整块替换', state.text === '最终' && state.completed === true, state);
|
|
|
+ }
|
|
|
+
|
|
|
+ check('null 事件 → none 且状态不变', (() => {
|
|
|
+ const s0 = createAnswerStreamState();
|
|
|
+ const r = applyAnswerStreamEvent(s0, null);
|
|
|
+ return r.effect.kind === 'none' && r.state === s0;
|
|
|
+ })());
|
|
|
+}
|
|
|
+
|
|
|
+console.log('\n【3b】应用层的内容对齐（end 替换 / abort 撤稿）');
|
|
|
+{
|
|
|
+ const BASE = '<scope title="思考中">\n</scope>\n';
|
|
|
+ check(
|
|
|
+ '正常路径：最终文本是已上屏内容的前缀延伸 → 纯尾缀追加（不加空行、不重建）',
|
|
|
+ applyAnswerStreamEnd(BASE + '一、说明', BASE, '一、说明。二、补充') === BASE + '一、说明。二、补充'
|
|
|
+ );
|
|
|
+ check(
|
|
|
+ '已上屏一长段、最终只多了尾巴 → 不重复整段',
|
|
|
+ applyAnswerStreamEnd(BASE + '甲', BASE, '甲乙丙') === BASE + '甲乙丙'
|
|
|
+ );
|
|
|
+ check(
|
|
|
+ '罕见分歧（最终文本与增量拼接不一致）→ 整块重建，且基线里的 <scope> 被剥掉（避免卡片闪现）',
|
|
|
+ applyAnswerStreamEnd(BASE + '草稿', BASE, '完全不同的最终文本') === '完全不同的最终文本'
|
|
|
+ );
|
|
|
+ check(
|
|
|
+ '重建时**基线（流开始前就有的内容）保留**，最终文本接在其后 —— 只有 scope 标记会被剥掉',
|
|
|
+ applyAnswerStreamEnd('前言', '前言', '最终') === '前言最终'
|
|
|
+ );
|
|
|
+ check('abort 撤稿：回到流开始前的内容', applyAnswerStreamAbort(BASE) === BASE);
|
|
|
+ check('abort 撤稿：空基线返回空串', applyAnswerStreamAbort('') === '');
|
|
|
+
|
|
|
+ // answer 开场概述该不该显示（实测：后端把正文作为 summary 流式下发，
|
|
|
+ // 而 answer 常是同一段开头的**压缩版**——显示会重复开头；但万一是另一段内容就不能丢）
|
|
|
+ const body =
|
|
|
+ '我们青浦目前能对上的小微企业扶持，主要分两类：一类是面向特定对象的创业扶持（如带动就业补贴、贷款贴息、留创企业开办资助），另一类是市级中小企业公共资助项目；此外还有几款“小微贷”金融产品。\n\n### 具体政策\n- 创业扶持：带动就业补贴';
|
|
|
+ // 压缩版的形态与实测一致：**截断在词中间**（正文接着 '###'，概述结尾是半个词 '部分候'）
|
|
|
+ const abridged = body.slice(0, 70).replace(/\n+/g, ' ') + ' 部分候…';
|
|
|
+ check('压缩版开场（截断在词中间，公共前缀 ≥90%）→ 不显示（避免开头重复）', shouldKeepAnswerOverview(abridged, body) === false, abridged);
|
|
|
+ check('真正的另一段内容（不是前缀）→ 必须保留', shouldKeepAnswerOverview('这是另一段完全不同的开场说明。', body) === true);
|
|
|
+ check('概述为空 → 不显示', shouldKeepAnswerOverview('', body) === false && shouldKeepAnswerOverview(null, body) === false);
|
|
|
+ check('流式正文为空 → 保留概述（不能丢字）', shouldKeepAnswerOverview('开场', '') === true);
|
|
|
+}
|
|
|
+
|
|
|
+console.log(`\n===== 纯逻辑部分：通过 ${pass} 项，失败 ${fail} 项 =====`);
|
|
|
+
|
|
|
+/* ================================================================== *
|
|
|
+ * 第 2 部分：协调器级假流（假 SSE 喂真实协调器）
|
|
|
+ * ================================================================== */
|
|
|
+
|
|
|
+const sse = (events) =>
|
|
|
+ events
|
|
|
+ .map(
|
|
|
+ ([event, data]) =>
|
|
|
+ `event: ${event}\ndata: ${JSON.stringify({ thread_id: 't1', request_id: 'r1', data })}\n\n`
|
|
|
+ )
|
|
|
+ .join('');
|
|
|
+
|
|
|
+const stubFetchWith = (sseText) => {
|
|
|
+ globalThis.fetch = async () => {
|
|
|
+ const stream = new ReadableStream({
|
|
|
+ start(controller) {
|
|
|
+ controller.enqueue(new TextEncoder().encode(sseText));
|
|
|
+ controller.close();
|
|
|
+ },
|
|
|
+ });
|
|
|
+ return new Response(stream, { status: 200, headers: { 'Content-Type': 'text/event-stream' } });
|
|
|
+ };
|
|
|
+};
|
|
|
+
|
|
|
+/** 跑一轮，收集各类事件 */
|
|
|
+const runTurn = async (events) => {
|
|
|
+ stubFetchWith(sse(events));
|
|
|
+ const coordinator = new ApiChatCoordinator({ baseUrl: 'http://stub', threadId: 't1' });
|
|
|
+ const seen = { messages: [], total: null, streamStart: [], streamEnd: [], streamAbort: [], errors: [], order: [] };
|
|
|
+ coordinator.addEventListener('message', (m) => {
|
|
|
+ seen.messages.push(m);
|
|
|
+ seen.order.push(`message:${m.slice(0, 24)}`);
|
|
|
+ });
|
|
|
+ coordinator.addEventListener('totalResponse', (p) => {
|
|
|
+ seen.total = p;
|
|
|
+ });
|
|
|
+ coordinator.addEventListener('answerStreamStart', (p) => {
|
|
|
+ seen.streamStart.push(p);
|
|
|
+ seen.order.push('streamStart');
|
|
|
+ });
|
|
|
+ coordinator.addEventListener('answerStreamEnd', (p) => {
|
|
|
+ seen.streamEnd.push(p);
|
|
|
+ seen.order.push('streamEnd');
|
|
|
+ });
|
|
|
+ coordinator.addEventListener('answerStreamAbort', (p) => {
|
|
|
+ seen.streamAbort.push(p);
|
|
|
+ seen.order.push('streamAbort');
|
|
|
+ });
|
|
|
+ coordinator.addEventListener('error', (p) => {
|
|
|
+ seen.errors.push(p);
|
|
|
+ seen.order.push('error');
|
|
|
+ });
|
|
|
+ await coordinator.generateAnswer('测试问题');
|
|
|
+ await new Promise((resolve) => setTimeout(resolve, 80)); // flushContent 会等两帧
|
|
|
+ return seen;
|
|
|
+};
|
|
|
+
|
|
|
+// 应用侧就是 `content += msg`（见 appendAiMessageChunk），**没有分隔符** ——
|
|
|
+// 用 '\n\n' 拼会把流式增量切成一堆段落，断言就失真了
|
|
|
+const messageText = (seen) => seen.messages.join('');
|
|
|
+const countOf = (text, needle) => text.split(needle).length - 1;
|
|
|
+
|
|
|
+const DELTA_TEXTS = ['一、', '第一段说明', '。\n\n', '二、', '第二段说明', '。'];
|
|
|
+const FINAL_TEXT = '一、第一段说明。\n\n二、第二段说明。';
|
|
|
+const MARKER = '第一段说明';
|
|
|
+
|
|
|
+console.log('\n【4】协调器假流：happy path（带卡片轮形态，含流中 heartbeat）');
|
|
|
+{
|
|
|
+ const seen = await runTurn([
|
|
|
+ ['accepted', { message: '收到' }],
|
|
|
+ ['progress', { stage: 'retrieve', message: '正在检索……' }],
|
|
|
+ ['answer_start', { stream_id: 's1', target: 'summary', sequence: 0, provisional: true }],
|
|
|
+ ...DELTA_TEXTS.map((text, i) => ['answer_delta', { stream_id: 's1', target: 'summary', sequence: i + 1, text }]),
|
|
|
+ ['heartbeat', { status: 'processing', message: '仍在处理……', elapsed_seconds: 20 }],
|
|
|
+ ['answer', { text: FINAL_TEXT }],
|
|
|
+ ['source', { id: 's1', title: '测试政策', type: 'policy', source: { url: 'https://example.com/p1' } }],
|
|
|
+ // 有 item 才有 POLICY_TABLE 卡片块（buildPolicyTableContent 是按 items 建的）
|
|
|
+ ['item', { source_id: 's1', title: '测试政策', card: { name: { text: '测试政策' } } }],
|
|
|
+ ['answer_end', { stream_id: 's1', target: 'summary', sequence: DELTA_TEXTS.length, text: FINAL_TEXT, operation: 'replace', status: 'completed', provisional: false }],
|
|
|
+ ['summary', { text: FINAL_TEXT, source_ids: ['s1'], notices: [] }],
|
|
|
+ ['result', { response: FINAL_TEXT, recommendation: {}, errors: [] }],
|
|
|
+ ['done', { status: 'completed' }],
|
|
|
+ ]);
|
|
|
+
|
|
|
+ const text = messageText(seen);
|
|
|
+ if (process.env.DBG) console.log('[dbg] text=' + JSON.stringify(text));
|
|
|
+ check('answerStreamStart 发出一次', seen.streamStart.length === 1, seen.streamStart);
|
|
|
+ check('answerStreamEnd 发出一次', seen.streamEnd.length === 1, seen.streamEnd);
|
|
|
+ check('无 abort', seen.streamAbort.length === 0, seen.streamAbort);
|
|
|
+ check('scope 卡片只发了一次（流中 heartbeat 不再发）', countOf(text, '<scope') === 1, countOf(text, '<scope'));
|
|
|
+ check('answer_end 的文本是规整后的全文', seen.streamEnd[0]?.text === normalizeChatMarkdown(FINAL_TEXT), seen.streamEnd[0]?.text);
|
|
|
+ check(
|
|
|
+ '正文在 message 通道**只出现一次**（完整 answer/summary 事件没有渲染成第二份）',
|
|
|
+ countOf(text, MARKER) === 1,
|
|
|
+ countOf(text, MARKER)
|
|
|
+ );
|
|
|
+ check(
|
|
|
+ '流式增量拼接 === 规整后全文（前缀不变式在真链路成立）',
|
|
|
+ text.includes(normalizeChatMarkdown(FINAL_TEXT).split('\n')[0]),
|
|
|
+ normalizeChatMarkdown(FINAL_TEXT).slice(0, 30)
|
|
|
+ );
|
|
|
+ check('totalResponse.answer 以最终全文开头', String(seen.total?.answer || '').startsWith('一、第一段说明。'), String(seen.total?.answer || '').slice(0, 40));
|
|
|
+ check('totalResponse 里含政策卡片与参考资料', String(seen.total?.answer || '').includes('POLICY_TABLE') && String(seen.total?.answer || '').includes('<ref_links>'));
|
|
|
+ check('卡片/参考资料是在流结束之后补发的', text.indexOf('POLICY_TABLE') > text.indexOf(MARKER));
|
|
|
+}
|
|
|
+
|
|
|
+console.log('\n【5】协调器假流：纯正文轮（无卡片、无 answer 事件）不丢正文');
|
|
|
+{
|
|
|
+ const seen = await runTurn([
|
|
|
+ ['accepted', { message: '收到' }],
|
|
|
+ ['answer_start', { stream_id: 's2', target: 'summary', sequence: 0 }],
|
|
|
+ ...DELTA_TEXTS.map((text, i) => ['answer_delta', { stream_id: 's2', target: 'summary', sequence: i + 1, text }]),
|
|
|
+ ['answer_end', { stream_id: 's2', target: 'summary', sequence: DELTA_TEXTS.length, text: FINAL_TEXT, status: 'completed' }],
|
|
|
+ ['summary', { text: FINAL_TEXT }],
|
|
|
+ ['result', { response: FINAL_TEXT }],
|
|
|
+ ['done', { status: 'completed' }],
|
|
|
+ ]);
|
|
|
+ check('totalResponse.answer === 最终全文（没有因为"没有卡片"而丢正文）', String(seen.total?.answer || '').trim() === normalizeChatMarkdown(FINAL_TEXT).trim(), String(seen.total?.answer || ''));
|
|
|
+}
|
|
|
+
|
|
|
+console.log('\n【6】协调器假流：answer_end 与增量拼接不同 → 整块替换生效');
|
|
|
+{
|
|
|
+ const seen = await runTurn([
|
|
|
+ ['answer_start', { stream_id: 's3', target: 'summary', sequence: 0 }],
|
|
|
+ ['answer_delta', { stream_id: 's3', target: 'summary', sequence: 1, text: '草稿内容' }],
|
|
|
+ ['answer_end', { stream_id: 's3', target: 'summary', sequence: 1, text: '定稿内容（后端最终校验改过）', status: 'completed' }],
|
|
|
+ ['summary', { text: '定稿内容（后端最终校验改过）' }],
|
|
|
+ ['result', { response: '定稿内容（后端最终校验改过）' }],
|
|
|
+ ['done', { status: 'completed' }],
|
|
|
+ ]);
|
|
|
+ check('End.text = 规整后的最终文本（不是增量拼接）', seen.streamEnd[0]?.text === '定稿内容（后端最终校验改过）', seen.streamEnd[0]?.text);
|
|
|
+ check('totalResponse.answer 用的是最终文本', String(seen.total?.answer || '').includes('定稿内容'), String(seen.total?.answer || ''));
|
|
|
+}
|
|
|
+
|
|
|
+console.log('\n【7】协调器假流：后端 error → 撤草稿（abort 先于 error）');
|
|
|
+{
|
|
|
+ const seen = await runTurn([
|
|
|
+ ['answer_start', { stream_id: 's4', target: 'summary', sequence: 0 }],
|
|
|
+ ['answer_delta', { stream_id: 's4', target: 'summary', sequence: 1, text: '半截草稿' }],
|
|
|
+ ['error', { code: 'processing_failed' }],
|
|
|
+ ]);
|
|
|
+ const abortAt = seen.order.indexOf('streamAbort');
|
|
|
+ const errorAt = seen.order.indexOf('error');
|
|
|
+ check('abort 在 error **之前**发出（否则 hook 已摘任务、abort 会被守卫丢弃）', abortAt > -1 && abortAt < errorAt, seen.order);
|
|
|
+ check('abort 的 reason = generation_failed', seen.streamAbort[0]?.reason === 'generation_failed', seen.streamAbort[0]);
|
|
|
+ check('草稿文本未进入 totalResponse', !String(seen.total?.answer || '').includes('半截草稿'), String(seen.total?.answer || ''));
|
|
|
+}
|
|
|
+
|
|
|
+console.log('\n【8】协调器假流：断流（EOF 无 done）→ 撤草稿 + 兼容回退');
|
|
|
+{
|
|
|
+ const seen = await runTurn([
|
|
|
+ ['answer_start', { stream_id: 's5', target: 'summary', sequence: 0 }],
|
|
|
+ ['answer_delta', { stream_id: 's5', target: 'summary', sequence: 1, text: '半截草稿' }],
|
|
|
+ ['answer', { text: '完整事件里的正文（断流前的兼容副本）' }],
|
|
|
+ ]);
|
|
|
+ check('断流时发出 abort（reason=not_finalized）', seen.streamAbort[0]?.reason === 'not_finalized', seen.streamAbort);
|
|
|
+ // ⚠️ 分层：协调器**只发信号**，真正把草稿从界面上撤掉的是应用层（hook 调 applyAnswerStreamAbort）。
|
|
|
+ // 断流路径也不发 totalResponse（既有行为：内容走 message 通道，hook 的 close 监听负责补写 DMS）。
|
|
|
+ check('兼容路径的完整 answer 事件仍进了 message 通道', messageText(seen).includes('完整事件里的正文'), messageText(seen).slice(0, 80));
|
|
|
+ check('报的是连接断开', seen.errors.some((e) => e?.code === 'disconnected'), seen.errors);
|
|
|
+}
|
|
|
+
|
|
|
+console.log('\n【9】协调器假流：旧后端（无任何流式事件）行为不变');
|
|
|
+{
|
|
|
+ const seen = await runTurn([
|
|
|
+ ['accepted', { message: '收到' }],
|
|
|
+ ['progress', { stage: 'retrieve', message: '正在检索……' }],
|
|
|
+ ['answer', { text: FINAL_TEXT }],
|
|
|
+ ['source', { id: 's1', title: '测试政策', type: 'policy', source: { url: 'https://example.com/p1' } }],
|
|
|
+ ['summary', { text: FINAL_TEXT }],
|
|
|
+ ['result', { response: FINAL_TEXT }],
|
|
|
+ ['done', { status: 'completed' }],
|
|
|
+ ]);
|
|
|
+ const text = messageText(seen);
|
|
|
+ check('没有流事件、没有 abort', seen.streamStart.length === 0 && seen.streamAbort.length === 0);
|
|
|
+ check('scope 卡片正常', countOf(text, '<scope') === 1);
|
|
|
+ check('正文按原路径整段输出（规整后）', text.includes(normalizeChatMarkdown(FINAL_TEXT)));
|
|
|
+ check('顺序仍是 正文 → 卡片 → 参考资料', text.indexOf('正文') === -1 || text.indexOf('第一段说明') < text.indexOf('POLICY_TABLE') || text.indexOf('POLICY_TABLE') === -1);
|
|
|
+}
|
|
|
+
|
|
|
+console.log('\n【10】协调器：流式轮之后跑旧轮，状态不残留');
|
|
|
+{
|
|
|
+ stubFetchWith(
|
|
|
+ sse([
|
|
|
+ ['answer_start', { stream_id: 's6', target: 'summary', sequence: 0 }],
|
|
|
+ ['answer_delta', { stream_id: 's6', target: 'summary', sequence: 1, text: '流式正文' }],
|
|
|
+ ['answer_end', { stream_id: 's6', target: 'summary', sequence: 1, text: '流式正文', status: 'completed' }],
|
|
|
+ ['done', { status: 'completed' }],
|
|
|
+ ])
|
|
|
+ );
|
|
|
+ const coordinator = new ApiChatCoordinator({ baseUrl: 'http://stub', threadId: 't1' });
|
|
|
+ const rounds = [];
|
|
|
+ coordinator.addEventListener('message', (m) => rounds.push(m));
|
|
|
+ await coordinator.generateAnswer('第一轮');
|
|
|
+ await new Promise((r) => setTimeout(r, 80));
|
|
|
+ const cutoff = rounds.length; // 第一轮结束时的分界
|
|
|
+
|
|
|
+ stubFetchWith(
|
|
|
+ sse([
|
|
|
+ ['progress', { stage: 'retrieve', message: '正在检索……' }],
|
|
|
+ ['answer', { text: '第二轮正文' }],
|
|
|
+ ['done', { status: 'completed' }],
|
|
|
+ ])
|
|
|
+ );
|
|
|
+ await coordinator.generateAnswer('第二轮');
|
|
|
+ await new Promise((r) => setTimeout(r, 80));
|
|
|
+
|
|
|
+ const first = rounds.slice(0, cutoff).join('');
|
|
|
+ const second = rounds.slice(cutoff).join('');
|
|
|
+ check('第一轮是流式正文', first.includes('流式正文'), first.slice(0, 60));
|
|
|
+ check('第二轮回到旧路径：有 scope、正文是整段输出', countOf(second, '<scope') === 1 && second.includes('第二轮正文'), second.slice(0, 80));
|
|
|
+}
|
|
|
+
|
|
|
+console.log('\n【11】两个协调器并行流式：互不串台');
|
|
|
+{
|
|
|
+ stubFetchWith(
|
|
|
+ sse([
|
|
|
+ ['answer_start', { stream_id: 'A', target: 'summary', sequence: 0 }],
|
|
|
+ ['answer_delta', { stream_id: 'A', target: 'summary', sequence: 1, text: '甲会话正文' }],
|
|
|
+ ['answer_end', { stream_id: 'A', target: 'summary', sequence: 1, text: '甲会话正文', status: 'completed' }],
|
|
|
+ ['done', { status: 'completed' }],
|
|
|
+ ])
|
|
|
+ );
|
|
|
+ const a = new ApiChatCoordinator({ baseUrl: 'http://stub', threadId: 'A' });
|
|
|
+ const b = new ApiChatCoordinator({ baseUrl: 'http://stub', threadId: 'B' });
|
|
|
+ const gotA = [];
|
|
|
+ const gotB = [];
|
|
|
+ a.addEventListener('message', (m) => gotA.push(m));
|
|
|
+ b.addEventListener('message', (m) => gotB.push(m));
|
|
|
+ b.addEventListener('answerStreamAbort', () => gotB.push('ABORT'));
|
|
|
+
|
|
|
+ await a.generateAnswer('问 A');
|
|
|
+ await new Promise((r) => setTimeout(r, 80));
|
|
|
+ check('B 未收到任何事件（含 abort）', gotB.length === 0, gotB);
|
|
|
+ check('A 正常收到正文', gotA.join('').includes('甲会话正文'), gotA);
|
|
|
+}
|
|
|
+
|
|
|
+console.log(`\n===== 总计：通过 ${pass} 项，失败 ${fail} 项 =====`);
|
|
|
+process.exit(fail ? 1 : 0);
|
