next-tjwater-drainage-frontend/tests/browser/agent-streaming-stability.e2e.ts

import { createServer, type Server } from "node:http";
import type { AddressInfo } from "node:net";
import { expect, test } from "playwright/test";

const STREAM_CHUNK_SIZE = 18;
const STREAM_CHUNK_DELAY_MS = 20;

let streamServer: Server;
let streamServerUrl: string;

test.beforeAll(async () => {
  streamServer = createServer((_request, response) => {
    response.writeHead(200, {
      "Access-Control-Allow-Origin": "*",
      "Cache-Control": "no-cache",
      "Content-Type": "text/event-stream",
      "X-Vercel-AI-UI-Message-Stream": "v1"
    });

    const content = createLongMarkdownResponse();
    const chunks = splitIntoCodePointChunks(content, STREAM_CHUNK_SIZE);
    const send = (chunk: Record<string, unknown>) => {
      response.write(`data: ${JSON.stringify(chunk)}\n\n`);
    };

    send({ type: "start", messageId: "assistant-stream-stability" });
    send({
      type: "data-progress",
      id: "inspect-context",
      data: {
        id: "inspect-context",
        phase: "planning",
        status: "running",
        title: "正在分析运行上下文",
        detail: "核对泵站、液位与降雨过程。",
        started_at: Date.now()
      }
    });
    send({
      type: "data-todo_update",
      id: "stream-stability-plan",
      data: {
        session_id: "stream-stability-session",
        message_id: "assistant-stream-stability",
        created_at: Date.now(),
        todos: [
          { id: "inspect-pump", content: "检查泵站运行状态", status: "in_progress" },
          { id: "inspect-level", content: "核对液位计连续性", status: "pending" },
          { id: "assess-overflow", content: "判断溢流风险", status: "pending" }
        ]
      }
    });
    send({ type: "text-start", id: "answer" });

    let index = 0;
    const timer = setInterval(() => {
      const delta = chunks[index];
      if (delta !== undefined) {
        index += 1;
        if (index === 3) {
          send({
            type: "data-progress",
            id: "assess-risk",
            data: {
              id: "assess-risk",
              phase: "analysis",
              status: "running",
              title: "正在判断溢流风险",
              detail: "结合流量突变与液位趋势形成判断。",
              started_at: Date.now()
            }
          });
        }
        send({ type: "text-delta", id: "answer", delta });
        send({
          type: "data-stream_token",
          data: {
            session_id: "stream-stability-session",
            message_id: "assistant-stream-stability",
            content: delta
          },
          transient: true
        });
        return;
      }

      clearInterval(timer);
      send({
        type: "data-progress",
        id: "assess-risk",
        data: {
          id: "assess-risk",
          phase: "analysis",
          status: "completed",
          title: "溢流风险判断完成",
          detail: "已完成风险点位排序。",
          ended_at: Date.now()
        }
      });
      send({ type: "text-end", id: "answer" });
      send({ type: "finish", finishReason: "stop" });
      response.end();
    }, STREAM_CHUNK_DELAY_MS);

    response.on("close", () => clearInterval(timer));
  });

  await new Promise<void>((resolve) => streamServer.listen(0, "127.0.0.1", resolve));
  const address = streamServer.address() as AddressInfo;
  streamServerUrl = `http://127.0.0.1:${address.port}/stream`;
});

test.afterAll(async () => {
  await new Promise<void>((resolve, reject) => {
    streamServer.close((error) => (error ? reject(error) : resolve()));
  });
});

test("stream completion and the plan card remain positionally stable", async ({ page }) => {
  await page.route("**/api/v1/agent/chat/**", async (route) => {
    const path = new URL(route.request().url()).pathname;

    if (path.endsWith("/stream")) {
      await route.continue({ url: streamServerUrl });
      return;
    }

    const body = path.endsWith("/models")
      ? {
          default_model: "test/model",
          models: [
            {
              id: "test/model",
              label: "测试模型",
              description: "流式稳定性测试",
              icon: "bolt"
            }
          ]
        }
      : path.endsWith("/sessions")
        ? { sessions: [] }
        : {};

    await route.fulfill({ json: body });
  });

  await page.goto("/", { waitUntil: "domcontentloaded" });

  await expect(
    page.getByRole("button", { name: "选择 Agent 模型" }).filter({ visible: true }).first()
  ).toContainText("测试模型");
  const prompt = page.getByPlaceholder(/输入调度问题/).filter({ visible: true }).first();
  await expect(prompt).toBeVisible();
  await prompt.fill("检查流式响应稳定性");
  const sendButton = page.getByRole("button", { name: "发送 Agent 指令" }).filter({ visible: true }).first();
  await expect(sendButton).toBeEnabled();

  await page.evaluate(() => {
    type StreamSample = {
      clientHeight: number;
      planItemTransform: string | null;
      planSectionTransform: string | null;
      planTransform: string | null;
      scrollHeight: number;
      scrollTop: number;
      time: number;
      transform: string;
    };

    const samples: StreamSample[] = [];
    const startedAt = performance.now();
    (window as typeof window & { __agentStreamSamples?: StreamSample[] }).__agentStreamSamples = samples;

    const sample = () => {
      const message = document.querySelector<HTMLElement>(".agent-panel-message");
      const wrapper = message?.parentElement?.parentElement;
      const scroll = document.querySelector<HTMLElement>(".agent-conversation-scroll");
      const plan = Array.from(document.querySelectorAll<HTMLElement>(".agent-panel-nested")).find((element) =>
        element.textContent?.includes("计划")
      );
      const planWrapper = plan?.parentElement;
      const planSection = planWrapper?.parentElement;
      const planItem = plan?.querySelector("li");

      if (message && wrapper && scroll) {
        samples.push({
          clientHeight: scroll.clientHeight,
          planItemTransform: planItem ? getComputedStyle(planItem).transform : null,
          planSectionTransform: planSection ? getComputedStyle(planSection).transform : null,
          planTransform: planWrapper ? getComputedStyle(planWrapper).transform : null,
          scrollHeight: scroll.scrollHeight,
          scrollTop: scroll.scrollTop,
          time: performance.now() - startedAt,
          transform: getComputedStyle(wrapper).transform
        });
      }

      if (performance.now() - startedAt < 8_000) {
        requestAnimationFrame(sample);
      }
    };

    requestAnimationFrame(sample);
  });

  await sendButton.click();
  await expect(page.getByText("分析完成", { exact: true }).filter({ visible: true }).first()).toBeVisible({
    timeout: 10_000
  });
  const progressToggle = page.getByRole("button", { name: /执行进度/ }).filter({ visible: true }).first();
  await expect(progressToggle).toContainText("共 2 条");
  await expect(page.getByText("溢流风险判断完成", { exact: true }).filter({ visible: true }).first()).toBeVisible();
  await page.waitForTimeout(400);

  const samples = await page.evaluate(() =>
    (window as typeof window & {
      __agentStreamSamples?: Array<{
        clientHeight: number;
        planItemTransform: string | null;
        planSectionTransform: string | null;
        planTransform: string | null;
        scrollHeight: number;
        scrollTop: number;
        time: number;
        transform: string;
      }>;
    }).__agentStreamSamples ?? []
  );

  expect(samples.length).toBeGreaterThan(5);
  const firstSampleTime = samples[0].time;
  const settledEntranceSamples = samples.filter((sample) => sample.time - firstSampleTime >= 250);
  expect(settledEntranceSamples.length).toBeGreaterThan(0);
  expect(settledEntranceSamples.every((sample) => sample.transform === "none")).toBe(true);

  const planSamples = samples.filter(
    (sample) =>
      sample.planItemTransform !== null &&
      sample.planSectionTransform !== null &&
      sample.planTransform !== null
  );
  expect(planSamples.length).toBeGreaterThan(0);
  const firstPlanSampleTime = planSamples[0].time;
  const settledPlanSamples = planSamples.filter((sample) => sample.time - firstPlanSampleTime >= 250);
  expect(settledPlanSamples.length).toBeGreaterThan(0);
  expect(
    settledPlanSamples.every(
      (sample) =>
        sample.planItemTransform === "none" &&
        sample.planSectionTransform === "none" &&
        sample.planTransform === "none"
    )
  ).toBe(true);

  const finalSample = samples.at(-1);
  expect(finalSample).toBeDefined();
  expect(
    Math.abs((finalSample!.scrollHeight - finalSample!.clientHeight) - finalSample!.scrollTop)
  ).toBeLessThanOrEqual(2);

  await progressToggle.click();
  await expect(page.getByText("正在分析运行上下文", { exact: true }).filter({ visible: true }).first()).toBeVisible();
});

function createLongMarkdownResponse() {
  let content = "## 诊断结果\n\n";
  for (let index = 1; index <= 6; index += 1) {
    content += `### 区域 ${index}\n\n`;
    content += "当前管网运行状态总体稳定，但部分区域存在水位持续升高、流量突变和监测数据缺失。需要结合实时工况核对监测点连续性。\n\n";
    content += "- 检查泵站运行状态与实时流量。\n";
    content += "- 核对液位计连续性和异常峰值。\n";
    content += "- 结合降雨过程判断溢流风险。\n\n";
  }
  return `${content}建议优先处置高风险点位，并持续跟踪后续变化。`;
}

function splitIntoCodePointChunks(content: string, chunkSize: number) {
  const codePoints = Array.from(content);
  const chunks: string[] = [];
  for (let index = 0; index < codePoints.length; index += chunkSize) {
    chunks.push(codePoints.slice(index, index + chunkSize).join(""));
  }
  return chunks;
}