feat(agent): add pressure trend demo

src/routes/chat.ts

@@ -1,4 +1,9 @@
 import { Router } from "express";
+import {
+  UI_MESSAGE_STREAM_HEADERS,
+  type UIMessage,
+  type UIMessageChunk,
+} from "ai";
 import { z } from "zod";
 
 import { getLocalAgentContext } from "../context/localContext.js";
@@ -24,6 +29,11 @@ import { type SessionRecord } from "../sessions/metadataStore.js";
 import { uiRegistryResponse } from "../uiEnvelope/registry.js";
 import type { UIEnvelopePayload } from "../uiEnvelope/types.js";
 import { toActorKey, toProjectKey } from "../utils/fileStore.js";
+import {
+  createPressureTrendDemoEnvelopePayload,
+  createPressureTrendDemoToolCall,
+  isPressureTrendDemoPrompt,
+} from "./pressureTrendDemo.js";
 import {
   buildPromptWithLearningContext,
   extractLatestFrontendTurn,
@@ -61,7 +71,9 @@ import {
 } from "./chatUiState.js";
 
 const payloadSchema = z.object({
-  message: z.string().min(1).max(10000),
+  id: z.string().max(128).optional(),
+  messages: z.array(z.unknown()).max(200).optional(),
+  message: z.string().min(1).max(10000).optional(),
   session_id: z.string().max(128).optional(),
   model: z.string().refine(isSupportedModel, {
     message: "unsupported model",
@@ -114,6 +126,73 @@ const runtimeHasConversation = async (
   );
 };
 
+const isUiMessage = (value: unknown): value is UIMessage =>
+  isObjectRecord(value) &&
+  typeof value.id === "string" &&
+  (value.role === "system" || value.role === "user" || value.role === "assistant") &&
+  Array.isArray(value.parts);
+
+const readUiMessageText = (message: UIMessage | undefined) =>
+  message?.parts
+    .flatMap((part) =>
+      isObjectRecord(part) && part.type === "text" && typeof part.text === "string"
+        ? [part.text]
+        : [],
+    )
+    .join("")
+    .trim() ?? "";
+
+const readLatestUserText = (messages: UIMessage[]) => {
+  for (let index = messages.length - 1; index >= 0; index -= 1) {
+    const message = messages[index];
+    if (message.role !== "user") {
+      continue;
+    }
+    const text = readUiMessageText(message);
+    if (text) {
+      return text;
+    }
+  }
+  return "";
+};
+
+const toUiMessages = (messages: unknown[] | undefined) =>
+  Array.isArray(messages) ? messages.filter(isUiMessage) : [];
+
+const toLegacyMessages = (messages: UIMessage[]) =>
+  messages.flatMap((message) => {
+    const content = readUiMessageText(message);
+    if (!content || (message.role !== "user" && message.role !== "assistant")) {
+      return [];
+    }
+    return [
+      {
+        id: message.id,
+        role: message.role,
+        content,
+      },
+    ];
+  });
+
+const createFallbackUserUiMessage = (content: string): UIMessage => ({
+  id: `user-${Date.now().toString(36)}`,
+  role: "user",
+  parts: [{ type: "text", text: content }],
+});
+
+const pipeUiChunk = (
+  res: {
+    write: (chunk: string) => void;
+    writableEnded?: boolean;
+    destroyed?: boolean;
+  },
+  chunk: UIMessageChunk,
+) => {
+  if (!res.writableEnded && !res.destroyed) {
+    res.write(`data: ${JSON.stringify(chunk)}\n\n`);
+  }
+};
+
 export const buildChatRouter = (
   sessionBridge: ChatSessionBridge,
   runtime: OpencodeRuntimeAdapter,
@@ -153,7 +232,18 @@ export const buildChatRouter = (
     const actorKey = toActorKey(userId);
     const projectKey = toProjectKey(projectId);
     const requestedSessionId = parsed.data.session_id?.trim();
-    const sessionId = requestedSessionId || (await runtime.createSession()).id;
+    let sessionId = requestedSessionId;
+    if (!sessionId) {
+      try {
+        sessionId = (await runtime.createSession()).id;
+      } catch (error) {
+        sessionId = sessionBridge.createClientSessionId();
+        logger.warn(
+          { err: error, sessionId },
+          "runtime unavailable while creating chat session; using local degraded session",
+        );
+      }
+    }
 
     const { record, created } = await sessionMetadataStore.ensure({
       actorKey,
@@ -518,7 +608,21 @@ export const buildChatRouter = (
       const userId = authContext.userId;
       const actorKey = toActorKey(userId);
       const projectKey = toProjectKey(projectId);
-      const requestedSessionId = parsed.data.session_id?.trim();
+      const requestUiMessages = toUiMessages(parsed.data.messages);
+      const promptText =
+        parsed.data.message?.trim() || readLatestUserText(requestUiMessages);
+      if (!promptText) {
+        res.status(400).json({
+          message: "message or messages with latest user text is required",
+        });
+        return;
+      }
+      const requestedSessionId =
+        parsed.data.session_id?.trim() ||
+        parsed.data.id?.trim() ||
+        (isPressureTrendDemoPrompt(promptText)
+          ? sessionBridge.createClientSessionId()
+          : undefined);
       const existingSessionRecord = requestedSessionId
         ? await sessionMetadataStore.get(
             { actorKey, projectId, projectKey, userId },
@@ -559,8 +663,16 @@ export const buildChatRouter = (
       const initialSessionState = await sessionUiStateStore.read(
         toSessionUiStateContext(activeSessionRecord),
       );
-      const persistedMessages = initialSessionState?.messages ?? [];
-      const baseMessages = persistedMessages;
+      const persistedUiMessages = toUiMessages(initialSessionState?.messages);
+      const requestMessages =
+        requestUiMessages.length > 0
+          ? requestUiMessages
+          : [
+              ...persistedUiMessages,
+              createFallbackUserUiMessage(promptText),
+            ];
+      let latestUiMessages = requestMessages;
+      const baseMessages = toLegacyMessages(requestMessages);
       if (activeRuns.get(activeSessionRecord.sessionId)?.status === "running") {
         res.status(409).json({
           message: "session is already streaming",
@@ -584,9 +696,9 @@ export const buildChatRouter = (
       );
 
       res.status(200);
-      res.setHeader("Content-Type", "text/event-stream; charset=utf-8");
-      res.setHeader("Cache-Control", "no-cache");
-      res.setHeader("Connection", "keep-alive");
+      for (const [header, value] of Object.entries(UI_MESSAGE_STREAM_HEADERS)) {
+        res.setHeader(header, value);
+      }
       res.setHeader("X-Accel-Buffering", "no");
       res.flushHeaders?.();
 
@@ -596,7 +708,7 @@ export const buildChatRouter = (
       sessionBridge.registerAbortController(clientSessionId, abortController);
       const initialMessages = createInitialStreamingMessages(
         baseMessages,
-        parsed.data.message,
+        promptText,
       );
       const activeRun: ActiveRun = {
         clientSessionId,
@@ -610,16 +722,110 @@ export const buildChatRouter = (
       activeRuns.set(clientSessionId, activeRun);
       lastRunStatuses.set(clientSessionId, "running");
       const sessionUiStateContext = toSessionUiStateContext(activeSessionRecord);
+      const assistantUiMessageId = `assistant-${Date.now().toString(36)}`;
+      const textPartId = "answer";
+      const assistantParts: Record<string, unknown>[] = [];
+      let responseText = "";
+      let textStarted = false;
+      let textEnded = false;
+      let uiFinished = false;
+      const syncLatestUiMessages = () => {
+        latestUiMessages = [
+          ...requestMessages,
+          {
+            id: assistantUiMessageId,
+            role: "assistant",
+            parts: assistantParts.map((part) => ({ ...part })),
+          } as UIMessage,
+        ];
+      };
+      const writeUiChunk = (chunk: UIMessageChunk) => {
+        pipeUiChunk(res, chunk);
+      };
+      const writeTextDelta = (delta: string) => {
+        if (!delta || uiFinished) {
+          return;
+        }
+        if (!textStarted) {
+          textStarted = true;
+          assistantParts.push({
+            type: "text",
+            text: "",
+            state: "streaming",
+          });
+          writeUiChunk({ type: "text-start", id: textPartId });
+        }
+        responseText += delta;
+        const textPart = assistantParts.find(
+          (part) => part.type === "text",
+        );
+        if (textPart) {
+          textPart.text = responseText;
+          textPart.state = "streaming";
+        }
+        writeUiChunk({ type: "text-delta", id: textPartId, delta });
+        syncLatestUiMessages();
+      };
+      const writeDataPart = (event: string, data: Record<string, unknown>) => {
+        if (uiFinished) {
+          return;
+        }
+        const type = `data-${event}`;
+        const id =
+          typeof data.id === "string"
+            ? data.id
+            : typeof data.request_id === "string"
+              ? data.request_id
+              : typeof data.envelope_id === "string"
+                ? data.envelope_id
+                : undefined;
+        const existingIndex =
+          id === undefined
+            ? -1
+            : assistantParts.findIndex(
+                (part) => part.type === type && part.id === id,
+              );
+        const part = id ? { type, id, data } : { type, data };
+        if (existingIndex >= 0) {
+          assistantParts[existingIndex] = part;
+        } else {
+          assistantParts.push(part);
+        }
+        writeUiChunk(part as UIMessageChunk);
+        syncLatestUiMessages();
+      };
+      const finishUiMessage = (finishReason: "stop" | "error" = "stop") => {
+        if (uiFinished) {
+          return;
+        }
+        if (!textStarted && assistantParts.length === 0) {
+          writeTextDelta("Agent 已完成处理。");
+        }
+        if (textStarted && !textEnded) {
+          textEnded = true;
+          const textPart = assistantParts.find(
+            (part) => part.type === "text",
+          );
+          if (textPart) {
+            textPart.state = "done";
+          }
+          writeUiChunk({ type: "text-end", id: textPartId });
+        }
+        syncLatestUiMessages();
+        writeUiChunk({ type: "finish", finishReason });
+        uiFinished = true;
+      };
+      writeUiChunk({ type: "start", messageId: assistantUiMessageId });
       let persistQueue = sessionUiStateStore.write(sessionUiStateContext, {
         sessionId: activeSessionRecord.sessionId,
         isTitleManuallyEdited: initialSessionState?.isTitleManuallyEdited ?? false,
-        messages: initialMessages,
+        messages: latestUiMessages,
       });
       const queueSessionUiStatePersist = () => {
         const snapshot = {
           sessionId: activeSessionRecord.sessionId,
           isTitleManuallyEdited: initialSessionState?.isTitleManuallyEdited ?? false,
-          messages: activeRun.messages,
+          messages: latestUiMessages,
         };
         persistQueue = persistQueue
           .catch((error) => {
@@ -633,9 +839,29 @@ export const buildChatRouter = (
       };
       const primarySubscriber: StreamSubscriber = {
         write: (event, data) => {
-          if (!streamClosed && !res.writableEnded && !res.destroyed) {
-            res.write(toSse(event, data));
+          if (streamClosed || res.writableEnded || res.destroyed) {
+            return;
           }
+          if (event === "token") {
+            writeTextDelta(typeof data.content === "string" ? data.content : "");
+            return;
+          }
+          if (event === "done") {
+            finishUiMessage("stop");
+            return;
+          }
+          if (event === "error") {
+            writeUiChunk({
+              type: "error",
+              errorText:
+                typeof data.message === "string"
+                  ? data.message
+                  : "Agent stream failed",
+            });
+            finishUiMessage("error");
+            return;
+          }
+          writeDataPart(event, data);
         },
         close: () => {
           if (!res.writableEnded && !res.destroyed) {
@@ -793,6 +1019,39 @@ export const buildChatRouter = (
       };
 
       try {
+        if (isPressureTrendDemoPrompt(promptText)) {
+          const toolCall = createPressureTrendDemoToolCall();
+          publish("progress", {
+            session_id: clientSessionId,
+            id: "pressure-trend-demo",
+            phase: "tool",
+            status: "running",
+            title: "生成压力趋势图",
+            detail: "已命中受控演示用例，正在通过 show_chart 生成可信 UIEnvelope。",
+            started_at: Date.now(),
+            elapsed_ms: 0,
+          });
+          publish("tool_call", {
+            session_id: clientSessionId,
+            tool: toolCall.tool,
+            params: toolCall.params,
+            reason: toolCall.reason,
+          });
+          publish(
+            "ui_envelope",
+            createPressureTrendDemoEnvelopePayload(clientSessionId, toolCall),
+          );
+          publish("token", {
+            session_id: clientSessionId,
+            content: "已生成最近压力趋势图，并通过可信 UIEnvelope 发送到前端。",
+          });
+          publish("done", {
+            session_id: clientSessionId,
+            duration_ms: 0,
+          });
+          return;
+        }
+
         const preparedMessage = await buildPromptWithLearningContext(
           memoryStore,
           requestContext.actorKey,
@@ -800,7 +1059,7 @@ export const buildChatRouter = (
           {
             recentTurns,
             persistedMessages: baseMessages,
-            message: parsed.data.message,
+            message: promptText,
             restoreConversation: shouldRestoreConversation,
           },
         );
@@ -847,7 +1106,7 @@ export const buildChatRouter = (
             sessionTitle = await generateSessionTitle(runtime, {
               sessionId: binding.sessionId,
               latestAssistantMessage: assistantText,
-              latestUserMessage: parsed.data.message,
+              latestUserMessage: promptText,
               fallbackTitle: existingSessionTitle,
             });
           }
@@ -895,10 +1154,20 @@ export const buildChatRouter = (
             progress: completeBackendProgress(message.progress),
             todos: cancelBackendTodos(message.todos),
           }));
+          if (!uiFinished && !streamClosed && !res.writableEnded && !res.destroyed) {
+            writeUiChunk({
+              type: "error",
+              errorText: "请求已中断",
+            });
+            finishUiMessage("error");
+          }
           void queueSessionUiStatePersist().catch((error) => {
             logger.warn({ err: error, sessionId: clientSessionId }, "failed to persist aborted chat stream state");
           });
         }
+        if (!uiFinished && !streamClosed && !res.writableEnded && !res.destroyed) {
+          finishUiMessage(activeRun.status === "error" ? "error" : "stop");
+        }
         await persistQueue.catch((error) => {
           logger.warn({ err: error, sessionId: clientSessionId }, "failed to persist chat stream state");
         });
@@ -934,10 +1203,14 @@ export const buildChatRouter = (
       logger.error({ err: error }, "chat stream failed");
       if (res.headersSent) {
         if (!res.writableEnded && !res.destroyed) {
-          res.write(toSse("error", {
-            message: "chat stream failed",
-            detail,
-          }));
+          pipeUiChunk(res, {
+            type: "error",
+            errorText: detail,
+          });
+          pipeUiChunk(res, {
+            type: "finish",
+            finishReason: "error",
+          });
           res.end();
         }
         return;

src/routes/pressureTrendDemo.ts

@@ -0,0 +1,66 @@
+import { createEnvelopeId } from "../uiEnvelope/ids.js";
+import { toUiEnvelopeFromToolCall } from "../uiEnvelope/fromToolCall.js";
+import type { UIEnvelopePayload } from "../uiEnvelope/types.js";
+
+export type PressureTrendDemoToolCall = {
+  tool: "show_chart";
+  params: {
+    reason: string;
+    title: string;
+    chart_type: "line";
+    x_axis_name: string;
+    y_axis_name: string;
+    x_data: string[];
+    series: Array<{
+      name: string;
+      type: "line";
+      data: number[];
+    }>;
+  };
+  reason: string;
+};
+
+export const isPressureTrendDemoPrompt = (message: string) =>
+  message.replace(/\s+/g, "").includes("展示最近压力趋势");
+
+export const createPressureTrendDemoToolCall = (): PressureTrendDemoToolCall => ({
+  tool: "show_chart",
+  reason: "展示最近 6 小时压力趋势。",
+  params: {
+    reason: "展示最近 6 小时压力趋势。",
+    title: "最近压力趋势",
+    chart_type: "line",
+    x_axis_name: "时间",
+    y_axis_name: "压力 MPa",
+    x_data: ["10:00", "11:00", "12:00", "13:00", "14:00", "15:00"],
+    series: [
+      {
+        name: "东部主干压力",
+        type: "line",
+        data: [0.42, 0.41, 0.39, 0.36, 0.35, 0.37],
+      },
+      {
+        name: "正常基线",
+        type: "line",
+        data: [0.43, 0.43, 0.42, 0.42, 0.41, 0.41],
+      },
+    ],
+  },
+});
+
+export const createPressureTrendDemoEnvelopePayload = (
+  sessionId: string,
+  toolCall: PressureTrendDemoToolCall = createPressureTrendDemoToolCall(),
+): UIEnvelopePayload => {
+  const envelope = toUiEnvelopeFromToolCall(toolCall);
+  if (!envelope) {
+    throw new Error("pressure trend demo tool call did not produce a UIEnvelope");
+  }
+
+  return {
+    session_id: sessionId,
+    envelope_id: createEnvelopeId(toolCall.tool),
+    created_at: Date.now(),
+    envelope,
+  };
+};