jiang
692bbf609c
Share project JSON transport, interaction replies and report selectors. Settle tool progress consistently for direct streams, reconnects and history, preserving failure reasons and stopped state.
205 lines
12 KiB
TypeScript
205 lines
12 KiB
TypeScript
import { describe, expect, it, vi } from "vitest";
|
|
import type { ChatTransport, UIMessageChunk } from "ai";
|
|
import { AgentSessionStreamError, AgentStreamDisconnectedError, type AgentApiClient } from "@/features/agent";
|
|
import { WorkbenchSessionResources } from "./workbench-session-resources";
|
|
import { createAgentRuntime, mergeHistoryMessages, resumeAgentRuntime, runtimeRunning } from "./agent-session-runtime";
|
|
import { toAgentChatMessages, 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.each(["completed", "error", "aborted"] as const)("settles direct tool progress when the response is %s without a final progress event", async (status) => {
|
|
const resources = new WorkbenchSessionResources();
|
|
let controller!: ReadableStreamDefaultController<UIMessageChunk>;
|
|
const runtime = createAgentRuntime(resources, "project:a", {} as AgentApiClient, {
|
|
sendMessages: async ({ abortSignal }) => new ReadableStream<UIMessageChunk>({ start(stream) {
|
|
controller = stream;
|
|
abortSignal?.addEventListener("abort", () => stream.error(new DOMException("Aborted", "AbortError")), { once: true });
|
|
} }),
|
|
reconnectToStream: async () => null,
|
|
});
|
|
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: "data-progress", data: { id: "tool", phase: "tool", title: "正在处理", status: "running" } });
|
|
await vi.waitFor(() => expect(toAgentChatMessages(runtime.chat.messages, {}, {}).at(-1)?.progress?.[0].status).toBe("running"));
|
|
if (status === "aborted") await runtime.chat.stop();
|
|
else {
|
|
if (status === "error") controller.enqueue({ type: "error", errorText: "工具执行失败" });
|
|
controller.enqueue({ type: "finish", finishReason: status === "error" ? "error" : "stop" });
|
|
controller.close();
|
|
}
|
|
await request;
|
|
const answer = toAgentChatMessages(runtime.chat.messages, {}, {}).at(-1);
|
|
expect(answer?.progress?.[0].status).toBe(status === "aborted" ? "cancelled" : status);
|
|
expect(runtime.runStatus).toBe(status);
|
|
if (status === "error") expect(answer?.runError).toBe("工具执行失败");
|
|
} finally { resources.dispose(); }
|
|
});
|
|
|
|
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>;
|
|
const transport: ChatTransport<AgentUiMessage> = {
|
|
sendMessages: async () => new ReadableStream({ start(stream) { controller = stream; } }),
|
|
reconnectToStream: async () => null,
|
|
};
|
|
const client = { refreshCredential: vi.fn(async () => {}), submitFrontendActionResult: vi.fn(async () => {}) } as unknown as AgentApiClient;
|
|
const first = createAgentRuntime(resources, "project:a", client, transport);
|
|
const second = createAgentRuntime(resources, "project:a", client, transport);
|
|
resources.activeAgent = first;
|
|
first.sessionId = "server-a";
|
|
const response = first.chat.sendMessage({ text: "核对城南压力" });
|
|
await vi.waitFor(() => expect(controller).toBeDefined());
|
|
controller.enqueue({ type: "start", messageId: "answer-a" });
|
|
controller.enqueue({ type: "text-start", id: "text-a" });
|
|
controller.enqueue({ type: "text-delta", id: "text-a", delta: "第一段" });
|
|
await vi.waitFor(() => expect(first.chat.messages.at(-1)?.parts).toEqual(expect.arrayContaining([expect.objectContaining({ text: "第一段" })])));
|
|
resources.activeAgent = second;
|
|
controller.enqueue({ type: "data-progress", data: { session_id: "another-session", title: "迟到的其他会话事件" } });
|
|
expect(runtimeRunning(first)).toBe(true);
|
|
controller.enqueue({ type: "data-credential_refresh_required", data: { request_id: "refresh-a", session_id: "server-a" } });
|
|
controller.enqueue({ type: "text-delta", id: "text-a", delta: "，后台完成" });
|
|
controller.enqueue({ type: "text-end", id: "text-a" });
|
|
controller.enqueue({ type: "finish", finishReason: "stop" });
|
|
controller.close();
|
|
await response;
|
|
expect(first.chat.messages.at(-1)?.parts).toEqual(expect.arrayContaining([expect.objectContaining({ text: "第一段，后台完成" })]));
|
|
expect(second.chat.messages).toEqual([]);
|
|
expect(first.sessionId).toBe("server-a");
|
|
expect(client.refreshCredential).toHaveBeenCalledWith("server-a", "refresh-a");
|
|
resources.dispose();
|
|
});
|
|
it("merges overlapping pages and a newer authoritative tail without duplicates", () => {
|
|
expect(mergeHistoryMessages([message("a"), message("b", "old")], [message("b", "new"), message("c")]))
|
|
.toEqual([message("a"), message("b", "new"), message("c")]);
|
|
expect(mergeHistoryMessages([message("c"), message("d")], [message("a"), message("b"), message("c", "new")]))
|
|
.toEqual([message("a"), message("b"), message("c", "new"), message("d")]);
|
|
});
|
|
it("routes background browser actions to a scoped refusal and keeps old message pages on resume", async () => {
|
|
const resources = new WorkbenchSessionResources();
|
|
const submit = vi.fn(async () => {});
|
|
const client = { submitFrontendActionResult: submit,
|
|
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@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: {} });
|
|
}),
|
|
} as unknown as AgentApiClient;
|
|
const transport = {} as ChatTransport<AgentUiMessage>;
|
|
const runtime = createAgentRuntime(resources, "project:a", client, transport, {
|
|
id: "a", sessionId: "a", title: "压力复核", createdAt: 1, updatedAt: 1, purpose: "ordinary_chat",
|
|
isTitleManuallyEdited: false, messages: [message("a"), message("b", "old")],
|
|
});
|
|
resumeAgentRuntime(resources, runtime);
|
|
await vi.waitFor(() => expect(runtime.controller).toBeUndefined());
|
|
expect(toAgentChatMessages(runtime.chat.messages, {}, {})).toMatchObject([
|
|
{ id: "a", content: "a" }, { id: "b", content: "new" }, { id: "c", content: "c" }
|
|
]);
|
|
expect(submit).toHaveBeenCalledWith("a", "action-a", expect.objectContaining({
|
|
status: "failed", error: expect.objectContaining({ code: "SESSION_NOT_ACTIVE" }),
|
|
}));
|
|
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(); }
|
|
});
|