feat: expand intelligent investigation workspace

src/features/workbench/hooks/agent-session-runtime.test.ts

@@ -1,6 +1,6 @@
 import { describe, expect, it, vi } from "vitest";
 import type { ChatTransport, UIMessageChunk } from "ai";
-import type { AgentApiClient } from "@/features/agent";
+import { AgentSessionStreamError, AgentStreamDisconnectedError, type AgentApiClient } from "@/features/agent";
 import { WorkbenchSessionResources } from "./workbench-session-resources";
 import { createAgentRuntime, mergeHistoryMessages, resumeAgentRuntime, runtimeRunning } from "./agent-session-runtime";
 import type { AgentUiMessage } from "./agent-session-message-state";
@@ -8,6 +8,81 @@ import type { AgentUiMessage } from "./agent-session-message-state";
 const message = (id: string, text = id): AgentUiMessage => ({ id, role: "assistant", parts: [{ type: "text", text }] });
 
 describe("independent conversation runtimes", () => {
+  it("retries a disconnected subscription and preserves the draft, pages and reading position", async () => {
+    vi.useFakeTimers();
+    const resources = new WorkbenchSessionResources();
+    const streamSession = vi.fn(async (_id, options) => {
+      const complete = streamSession.mock.calls.length > 1;
+      options.onEvent({ type: "state", sessionId: "a", isStreaming: !complete,
+        messages: [message("b", complete ? "recovered" : "partial")], runStatus: complete ? "completed" : "running" });
+    });
+    const transport = { sendMessages: vi.fn() } as unknown as ChatTransport<AgentUiMessage>;
+    const runtime = createAgentRuntime(resources, "project:a", { streamSession } as unknown as AgentApiClient, transport, {
+      id: "a", sessionId: "a", title: "核查", createdAt: 1, updatedAt: 1, purpose: "ordinary_chat", isTitleManuallyEdited: false, messages: [message("older"), message("b")]
+    });
+    resources.activeAgent = runtime;
+    runtime.draft = "unsent";
+    runtime.readingPosition = { messageId: "older", offset: 12, atBottom: false };
+    try {
+      resumeAgentRuntime(resources, runtime);
+      await vi.advanceTimersByTimeAsync(0);
+      expect(streamSession).toHaveBeenCalledTimes(1);
+      expect(runtimeRunning(runtime)).toBe(true);
+      await vi.advanceTimersByTimeAsync(1_000);
+      expect(streamSession).toHaveBeenCalledTimes(2);
+      expect(runtime.chat.messages).toEqual([message("older"), message("b", "recovered")]);
+      expect(runtime.draft).toBe("unsent");
+      expect(runtime.readingPosition).toEqual({ messageId: "older", offset: 12, atBottom: false });
+      expect(transport.sendMessages).not.toHaveBeenCalled();
+      expect(runtime.controller).toBeUndefined();
+    } finally { resources.dispose(); vi.useRealTimers(); }
+  });
+
+  it("stops retrying after permission loss or resource disposal", async () => {
+    vi.useFakeTimers();
+    for (const forbidden of [true, false]) {
+      const resources = new WorkbenchSessionResources();
+      const streamSession = vi.fn(async () => { throw forbidden ? new AgentSessionStreamError("forbidden", 403) : new TypeError("offline"); });
+      const runtime = createAgentRuntime(resources, "project:a", { streamSession } as unknown as AgentApiClient, {} as ChatTransport<AgentUiMessage>);
+      runtime.sessionId = "a";
+      resources.activeAgent = runtime;
+      try {
+        resumeAgentRuntime(resources, runtime);
+        await vi.advanceTimersByTimeAsync(0);
+        if (forbidden) expect(runtime.controller).toBeUndefined();
+        resources.dispose();
+        await vi.advanceTimersByTimeAsync(30_000);
+        expect(streamSession).toHaveBeenCalledTimes(1);
+      } finally { resources.dispose(); }
+    }
+    vi.useRealTimers();
+  });
+
+  it("recovers an interrupted Responses stream through a GET subscription without resending the prompt", async () => {
+    const resources = new WorkbenchSessionResources();
+    let controller!: ReadableStreamDefaultController<UIMessageChunk>;
+    const sendMessages = vi.fn(async () => new ReadableStream<UIMessageChunk>({ start(value) { controller = value; } }));
+    const streamSession = vi.fn(async (_id, options) => options.onEvent({ type: "state", sessionId: "a", isStreaming: false,
+      runStatus: "completed", messages: [message("answer", "server completed")] }));
+    const runtime = createAgentRuntime(resources, "project:a", { streamSession } as unknown as AgentApiClient,
+      { sendMessages, reconnectToStream: async () => null });
+    runtime.sessionId = "a";
+    resources.activeAgent = runtime;
+    try {
+      const request = runtime.chat.sendMessage({ text: "核查压力" });
+      await vi.waitFor(() => expect(controller).toBeDefined());
+      controller.enqueue({ type: "start", messageId: "answer" });
+      controller.enqueue({ type: "text-start", id: "text" });
+      controller.enqueue({ type: "text-delta", id: "text", delta: "partial" });
+      await vi.waitFor(() => expect(runtime.chat.messages.at(-1)?.parts).toContainEqual(expect.objectContaining({ text: "partial" })));
+      controller.error(new AgentStreamDisconnectedError("connection dropped"));
+      await request;
+      await vi.waitFor(() => expect(streamSession).toHaveBeenCalledTimes(1));
+      expect(sendMessages).toHaveBeenCalledTimes(1);
+      expect(runtime.chat.messages.at(-1)).toEqual(message("answer", "server completed"));
+      expect(runtime.error).toBeUndefined();
+    } finally { resources.dispose(); }
+  });
   it("keeps the original response streaming after selecting another conversation", async () => {
     const resources = new WorkbenchSessionResources();
     let controller!: ReadableStreamDefaultController<UIMessageChunk>;
@@ -54,8 +129,7 @@ describe("independent conversation runtimes", () => {
       streamSession: vi.fn(async (_id, options) => {
         options.onEvent({ type: "state", sessionId: "a", messages: [message("b", "new"), message("c")], isStreaming: true });
         options.onEvent({ type: "frontend_action", sessionId: "a", data: {
-          version: "frontend-action@1", sessionId: "a", actionId: "action-a", toolCallId: "call-a", name: "locate_features",
-          params: {}, issuedAt: Date.now(), expiresAt: Date.now() + 15_000,
+          version: "frontend-action@2", sessionId: "a", actionId: "action-a", toolCallId: "call-a", command: { action: "locate_features", layer: "scada", feature_ids: ["D1"] }, issuedAt: Date.now(), expiresAt: Date.now() + 15_000,
         } });
         options.onEvent({ type: "done", sessionId: "a", data: {} });
       }),
@@ -74,3 +148,24 @@ describe("independent conversation runtimes", () => {
     resources.dispose();
   });
 });
+
+it.each([
+  [JSON.stringify({ code: "INTERACTION_PENDING", message: "请先回答或取消当前提问。" }), "waiting_input"],
+  ["Connection refused", "error"],
+  [JSON.stringify({ code: "OTHER_FAILURE" }), "error"]
+] as const)("keeps a pending question distinct from transport failure: %s", async (message, expected) => {
+  const resources = new WorkbenchSessionResources();
+  const transport: ChatTransport<AgentUiMessage> = {
+    sendMessages: async () => { throw new Error(message); },
+    reconnectToStream: async () => null
+  };
+  const runtime = createAgentRuntime(resources, "project:a", {} as AgentApiClient, transport);
+  resources.activeAgent = runtime;
+  const onError = vi.fn();
+  runtime.handlers = { onData: vi.fn(), onEvent: vi.fn(), onError, onFinish: vi.fn() };
+  try {
+    await runtime.chat.sendMessage({ text: "定位到公司上报地址" });
+    expect(onError).toHaveBeenCalledWith(expect.objectContaining({ message }));
+    expect(runtime.runStatus).toBe(expected);
+  } finally { resources.dispose(); }
+});