319 lines
19 KiB
TypeScript
319 lines
19 KiB
TypeScript
import { afterEach, beforeEach, expect, it } from "bun:test";
|
|
import express from "express";
|
|
import { mkdtemp, rm } from "node:fs/promises";
|
|
import type { Server } from "node:http";
|
|
import { tmpdir } from "node:os";
|
|
import { join } from "node:path";
|
|
import { buildChatRouter } from "../../src/routes/chat.js";
|
|
import { ChatSessionBridge } from "../../src/chat/sessionBridge.js";
|
|
import { SessionMetadataStore } from "../../src/sessions/metadataStore.js";
|
|
import { SessionUiStateStore } from "../../src/sessions/uiStateStore.js";
|
|
import { SessionWorkspaceStore } from "../../src/sessions/workspace-store.js";
|
|
import { removeRuntimeSessionContext } from "../../src/runtime/sessionContext.js";
|
|
import { toActorKey, toProjectKey } from "../../src/utils/fileStore.js";
|
|
import type { ResponseInput } from "../../src/chat/response-inputs.js";
|
|
import { InteractionRequestStore } from "../../src/sessions/interaction-request-store.js";
|
|
import { INTERACTION_VERSION } from "../../src/chat/interaction-request-contract.js";
|
|
|
|
const http = globalThis.fetch;
|
|
let root: string;
|
|
let server: Server;
|
|
let base: string;
|
|
let fail = false;
|
|
let prompted: string[];
|
|
let pendingInputs: ResponseInput[];
|
|
let aborts: number;
|
|
let gate: Promise<void> | undefined;
|
|
let runtimeEvents: AsyncIterable<unknown> | undefined;
|
|
let release: () => void = () => {};
|
|
let bridge: ChatSessionBridge;
|
|
let states: SessionUiStateStore;
|
|
let interactions: InteractionRequestStore;
|
|
let nativeState: "missing" | "enqueued" | "delivered";
|
|
let completedResult: string | undefined;
|
|
let enqueueResumes: boolean[];
|
|
const project = "10000000-0000-4000-8000-000000000001";
|
|
beforeEach(async () => {
|
|
fail = false; prompted = []; pendingInputs = []; aborts = 0; gate = undefined;
|
|
runtimeEvents = undefined;
|
|
nativeState = "missing"; completedResult = undefined; enqueueResumes = [];
|
|
root = await mkdtemp(join(tmpdir(), "tjwater-responses-"));
|
|
globalThis.fetch = (async (_input, init) => Response.json({
|
|
user_id: new Headers(init?.headers).get("authorization") === "Bearer owner" ? "owner" : "other",
|
|
project_id: project, network: "fixture",
|
|
})) as typeof fetch;
|
|
const workspaces = new SessionWorkspaceStore(join(root, "workspaces"));
|
|
const sessions = new SessionMetadataStore(join(root, "metadata"), workspaces);
|
|
states = new SessionUiStateStore(join(root, "ui"), workspaces);
|
|
interactions = new InteractionRequestStore(workspaces);
|
|
await sessions.initialize();
|
|
const context = { actorKey: toActorKey("product:owner"), projectKey: toProjectKey(`product:${project}`), projectId: project, userId: "owner", sessionId: "owned" };
|
|
await sessions.ensure(context);
|
|
await states.write(context, { sessionId: "owned", isTitleManuallyEdited: true, messages: [] });
|
|
const runtime = {
|
|
prompt: async (_id: string, prompt: string) => { prompted.push(prompt); if (fail) throw new Error("fixture execution failed"); },
|
|
messages: async () => [], abortSession: async () => { aborts++; }, waitForSessionIdle: async () => {},
|
|
listPendingUserInputs: async () => pendingInputs.map((input) => ({ ...input })),
|
|
cancelPendingUserInputs: async () => { pendingInputs = []; },
|
|
cancelPendingUserInput: async (_sessionId: string, id: string) => {
|
|
const item = pendingInputs.find((input) => input.id === id);
|
|
if (!item) throw Object.assign(new Error("Consumed"), { status: 409 });
|
|
pendingInputs = pendingInputs.filter((input) => input.id !== id);
|
|
return { ...item, status: "cancelled" };
|
|
},
|
|
enqueueUserInput: async (_sessionId: string, input: ResponseInput, options?: { resume?: boolean }) => { pendingInputs.push({ ...input }); enqueueResumes.push(options?.resume ?? true); },
|
|
agentMessageState: async () => nativeState,
|
|
readInteractionResult: async () => { if (completedResult === undefined) throw new Error("No completed result"); return completedResult; },
|
|
updatePendingUserInput: async (_sessionId: string, id: string, delivery: "steer" | "queue") => {
|
|
const item = pendingInputs.find((input) => input.id === id);
|
|
if (!item) throw Object.assign(new Error("Consumed"), { status: 409 });
|
|
item.delivery = delivery;
|
|
return { ...item };
|
|
},
|
|
subscribeEvents: async () => runtimeEvents ?? ({ async *[Symbol.asyncIterator]() {
|
|
if (gate) await gate;
|
|
yield { type: "message.part.updated", properties: { sessionID: "owned", part: { id: "answer", type: "text", text: "", sessionID: "owned", messageID: "native-answer" } } };
|
|
yield { type: "message.part.delta", properties: { sessionID: "owned", partID: "answer", field: "text", delta: "压力待核验" } };
|
|
for (const item of pendingInputs.splice(0)) {
|
|
yield { type: "session.inbox.delivered", properties: { sessionID: "owned", inboxID: item.id } };
|
|
}
|
|
yield { type: "session.idle", properties: { sessionID: "owned" } };
|
|
} }),
|
|
} as never;
|
|
const app = express(); app.use(express.json());
|
|
bridge = new ChatSessionBridge(runtime);
|
|
app.use("/api/v1", buildChatRouter({
|
|
runtime, sessionBridge: bridge, sessionMetadataStore: sessions, sessionUiStateStore: states, interactionRequestStore: interactions,
|
|
sessionTranscriptStore: { getRecentTurns: async () => [] } as never,
|
|
memoryStore: { buildPromptSnapshot: async () => "" } as never,
|
|
learningOrchestrator: { onTurnCompleted: async () => {} } as never,
|
|
frontendActionCoordinator: { cancelSession: () => {} } as never,
|
|
}));
|
|
server = app.listen(0, "127.0.0.1");
|
|
await new Promise<void>((resolve) => server.once("listening", resolve));
|
|
const address = server.address(); if (!address || typeof address === "string") throw new Error("missing address");
|
|
base = `http://127.0.0.1:${address.port}`;
|
|
});
|
|
afterEach(async () => {
|
|
release();
|
|
for (let i = 0; i < 100 && bridge.count(); i++) await Bun.sleep(10);
|
|
globalThis.fetch = http;
|
|
removeRuntimeSessionContext("owned");
|
|
server?.closeAllConnections();
|
|
if (server) await new Promise<void>((resolve) => server.close(() => resolve()));
|
|
await rm(root, { recursive: true, force: true });
|
|
});
|
|
const call = (body: unknown, token = "owner", path = "/api/v1/responses") => http(`${base}${path}`, {
|
|
method: "POST", headers: { Authorization: `Bearer ${token}`, "X-Project-Id": project, "Content-Type": "application/json" }, body: JSON.stringify(body), signal: AbortSignal.timeout(5000),
|
|
});
|
|
|
|
it("admits steer and queue while a host question is waiting without starting unowned inference", async () => {
|
|
const context = { actorKey: toActorKey("product:owner"), projectKey: toProjectKey(`product:${project}`), sessionId: "owned" };
|
|
const question = await interactions.create(context, { message_id: "question", turn_id: "turn", call_id: "ask", questions: [{ question_id: "scope", title: "范围", body: "核查哪里？", options: [], mode: "single", required: true, allow_custom: true }] });
|
|
for (const [i, delivery] of ["steer", "queue"].entries()) {
|
|
const result = await call({ input: `补充 ${i}`, conversation: "owned", tjwater: { delivery, input_id: `msg_${String(i).repeat(32)}` } });
|
|
expect(result.status).toBe(202);
|
|
expect(await result.json()).toMatchObject({ delivery, status: "pending" });
|
|
}
|
|
expect(pendingInputs).toHaveLength(2);
|
|
expect(enqueueResumes).toEqual([false, false]);
|
|
expect(prompted).toEqual([]);
|
|
expect((await interactions.get(context, question.interaction_id))?.status).toBe("pending");
|
|
expect((await call({ input: "foreign", conversation: "owned", tjwater: { delivery: "queue", input_id: `msg_${"f".repeat(32)}` } }, "other")).status).toBe(404);
|
|
expect(pendingInputs).toHaveLength(2);
|
|
});
|
|
|
|
for (const recovery of ["list", "retry"] as const) it(`reconciles a blocked acknowledgement through ${recovery} before admitting the next message`, async () => {
|
|
const context = { actorKey: toActorKey("product:owner"), projectKey: toProjectKey(`product:${project}`), sessionId: "owned" };
|
|
const question = await interactions.create(context, { message_id: "question", turn_id: "turn", call_id: "ask", questions: [{ question_id: "scope", title: "范围", body: "核查哪里？", options: [], mode: "single", required: true, allow_custom: true }] });
|
|
await interactions.reply(context, question.interaction_id, { version: INTERACTION_VERSION, request_id: "reply", expected_revision: 0, answers: [{ question_id: "scope", selected_option_ids: [], custom_text: "当前片区" }] });
|
|
await interactions.updateDelivery(context, question.interaction_id, { expected_revision: 0, status: "blocked", error_code: "CONTINUATION_UNCONFIRMED" });
|
|
nativeState = "delivered"; completedResult = "已完成核查。";
|
|
if (recovery === "retry") {
|
|
const replay = await call({ input: "重试继续", conversation: "owned", tjwater: { interaction_id: question.interaction_id } });
|
|
expect(replay.status).toBe(200);
|
|
expect(JSON.stringify(await replay.json())).toContain("已完成核查。");
|
|
}
|
|
const list = await http(`${base}/api/v1/session/owned/interactions`, { headers: { Authorization: "Bearer owner", "X-Project-Id": project } });
|
|
expect((await list.json()).items[0].delivery.status).toBe("delivered");
|
|
expect(await interactions.getBlocking(context)).toBeNull();
|
|
expect(prompted).toEqual([]);
|
|
const response = await call({ input: "继续检查压力", conversation: "owned" });
|
|
expect(response.status).toBe(200);
|
|
expect((await response.json()).status).toBe("completed");
|
|
expect(prompted).toHaveLength(1);
|
|
});
|
|
|
|
it("serves Responses SSE through the real HTTP route and preserves the owned conversation", async () => {
|
|
const response = await call({ input: [{ role: "user", content: [{ type: "input_text", text: "核查压力" }] }], conversation: "owned", stream: true });
|
|
const text = await response.text();
|
|
expect(response.status).toBe(200);
|
|
expect(response.headers.get("content-type")).toContain("text/event-stream");
|
|
expect(response.headers.has("x-vercel-ai-ui-message-stream")).toBeFalse();
|
|
expect(text).toContain("event: response.created");
|
|
expect(text).toContain("event: response.output_text.delta");
|
|
expect(text).toContain("event: response.completed");
|
|
expect(text).toContain('"conversation":{"id":"owned"}');
|
|
expect(prompted).toHaveLength(1);
|
|
expect(prompted[0]).toContain("核查压力");
|
|
});
|
|
|
|
it("delivers narration over HTTP while a tool is still waiting", async () => {
|
|
gate = new Promise<void>((resolve) => { release = resolve; });
|
|
runtimeEvents = { async *[Symbol.asyncIterator]() {
|
|
yield { type: "message.part.updated", properties: { sessionID: "owned", part: { id: "text", type: "text", text: "" } } };
|
|
yield { type: "message.part.delta", properties: { sessionID: "owned", partID: "text", field: "text", delta: "正在核查。" } };
|
|
yield { type: "message.part.updated", properties: { sessionID: "owned", part: {
|
|
id: "query", type: "tool", tool: "mcp_fixture_query", callID: "query", state: { status: "running", input: {}, time: { start: 1 } },
|
|
} } };
|
|
await gate;
|
|
yield { type: "session.idle", properties: { sessionID: "owned" } };
|
|
} };
|
|
const response = await call({ input: "核查压力", conversation: "owned", stream: true });
|
|
const reader = response.body!.getReader();
|
|
const decoder = new TextDecoder();
|
|
let received = "";
|
|
try {
|
|
while (!received.includes("event: response.output_text.delta")) {
|
|
const next = await reader.read();
|
|
if (next.done) break;
|
|
received += decoder.decode(next.value, { stream: true });
|
|
}
|
|
expect(received).toContain('"delta":"正在核查。"');
|
|
expect(received).not.toContain("event: response.completed");
|
|
} finally { release(); }
|
|
while (true) {
|
|
const next = await reader.read();
|
|
if (next.done) break;
|
|
received += decoder.decode(next.value, { stream: true });
|
|
}
|
|
expect(received).toContain("event: response.completed");
|
|
});
|
|
|
|
it("returns a Response object when streaming is omitted", async () => {
|
|
const response = await call({ input: "核查压力", conversation: { id: "owned" } });
|
|
expect(await response.json()).toMatchObject({ object: "response", status: "completed", conversation: { id: "owned" }, output: [{ type: "message", content: [{ type: "output_text", text: "压力待核验" }] }] });
|
|
});
|
|
|
|
it("admits steer/queue input without interrupting, edits pending delivery and persists consumed user messages", async () => {
|
|
gate = new Promise<void>((resolve) => { release = resolve; });
|
|
const initial = await call({ input: "原始问题", conversation: "owned", stream: true });
|
|
const first = `msg_${"1".repeat(32)}`;
|
|
const second = `msg_${"2".repeat(32)}`;
|
|
const baselineAborts = aborts;
|
|
for (const [id, text, delivery] of [[first, "补充压力测点", "steer"], [second, "之后检查流量", "queue"]] as const) {
|
|
const result = await call({ input: text, conversation: "owned", tjwater: { delivery, input_id: id } });
|
|
expect(result.status).toBe(202);
|
|
expect(await result.json()).toMatchObject({ id, text, delivery, status: "pending" });
|
|
}
|
|
expect((await call({ input: "不带方式", conversation: "owned" })).status).toBe(409);
|
|
expect(aborts).toBe(baselineAborts);
|
|
expect(prompted).toHaveLength(1);
|
|
const headers = { Authorization: "Bearer owner", "X-Project-Id": project, "Content-Type": "application/json" };
|
|
const changed = await http(`${base}/api/v1/session/owned/inputs/${second}`, { method: "PATCH", headers, body: JSON.stringify({ delivery: "steer" }) });
|
|
expect(changed.status).toBe(200);
|
|
expect(await changed.json()).toMatchObject({ id: second, delivery: "steer" });
|
|
const list = await http(`${base}/api/v1/session/owned/inputs`, { headers });
|
|
expect((await list.json()).items).toHaveLength(2);
|
|
const foreign = await http(`${base}/api/v1/session/owned/inputs/${first}`, { method: "PATCH", headers: { ...headers, Authorization: "Bearer other" }, body: JSON.stringify({ delivery: "queue" }) });
|
|
expect(foreign.status).toBe(404);
|
|
expect(aborts).toBe(baselineAborts);
|
|
release();
|
|
const output = await initial.text();
|
|
expect(output).toContain("event: response.completed");
|
|
expect(output).toContain("tjwater.user_input");
|
|
const history = await http(`${base}/api/v1/session/owned`, { headers });
|
|
expect(JSON.stringify((await history.json()).messages)).toContain("补充压力测点");
|
|
const consumed = await http(`${base}/api/v1/session/owned/inputs/${second}`, { method: "PATCH", headers, body: JSON.stringify({ delivery: "queue" }) });
|
|
expect(consumed.status).toBe(409);
|
|
});
|
|
it("rejects cross-user conversations, legacy requests and retired URLs before generation", async () => {
|
|
expect((await call({ input: "核查压力", conversation: "owned" }, "other")).status).toBe(404);
|
|
expect((await call({ message: "legacy", session_id: "owned" })).status).toBe(400);
|
|
expect((await call({ input: "核查压力" }, "owner", "/api/v1/agent/chat/stream")).status).toBe(404);
|
|
expect(prompted).toEqual([]);
|
|
});
|
|
|
|
it("withdraws only owned pending input without stopping generation or persisting a sent message", async () => {
|
|
gate = new Promise<void>((resolve) => { release = resolve; });
|
|
const initial = await call({ input: "原始问题", conversation: "owned", stream: true });
|
|
const id = `msg_${"3".repeat(32)}`;
|
|
await call({ input: "撤回后重新编辑", conversation: "owned", tjwater: { delivery: "queue", input_id: id } });
|
|
const headers = { Authorization: "Bearer owner", "X-Project-Id": project };
|
|
const url = `${base}/api/v1/session/owned/inputs/${id}`;
|
|
const baselineAborts = aborts;
|
|
expect((await http(url, { method: "DELETE", headers: { ...headers, Authorization: "Bearer other" } })).status).toBe(404);
|
|
expect(pendingInputs).toHaveLength(1);
|
|
const withdrawn = await http(url, { method: "DELETE", headers });
|
|
expect(withdrawn.status).toBe(200);
|
|
expect(await withdrawn.json()).toMatchObject({ id, text: "撤回后重新编辑", delivery: "queue", status: "cancelled" });
|
|
expect(pendingInputs).toHaveLength(0);
|
|
expect((await http(url, { method: "DELETE", headers })).status).toBe(409);
|
|
expect(aborts).toBe(baselineAborts);
|
|
release();
|
|
const output = await initial.text();
|
|
expect(output).toContain('"status":"cancelled"');
|
|
expect(output).toContain("event: response.completed");
|
|
const history = await http(`${base}/api/v1/session/owned`, { headers });
|
|
const messages = (await history.json()).messages as Array<{ role: string }>;
|
|
expect(JSON.stringify(messages.filter((message) => message.role === "user"))).not.toContain("撤回后重新编辑");
|
|
});
|
|
it("returns response.failed when the runtime fails", async () => {
|
|
fail = true;
|
|
const response = await call({ input: "核查压力", conversation: "owned", stream: true });
|
|
const text = await response.text();
|
|
expect(text).toContain("event: response.failed");
|
|
expect(text).not.toContain("event: response.completed");
|
|
});
|
|
|
|
it("keeps generating and saving the owned history after the browser disconnects", async () => {
|
|
gate = new Promise<void>((resolve) => { release = resolve; });
|
|
const response = await call({ input: "核查压力", conversation: "owned", stream: true });
|
|
const reader = response.body!.getReader();
|
|
await reader.read();
|
|
await reader.cancel();
|
|
release();
|
|
let loaded: { is_streaming?: boolean; messages?: unknown[] } = {};
|
|
for (let i = 0; i < 50; i++) {
|
|
await Bun.sleep(10);
|
|
const state = await http(`${base}/api/v1/session/owned`, { headers: { Authorization: "Bearer owner", "X-Project-Id": project } });
|
|
loaded = await state.json() as typeof loaded;
|
|
if (!loaded.is_streaming && JSON.stringify(loaded.messages).includes("压力待核验")) break;
|
|
}
|
|
expect(loaded.is_streaming).toBeFalse();
|
|
expect(JSON.stringify(loaded.messages)).toContain("压力待核验");
|
|
});
|
|
|
|
it("marks an explicitly aborted response incomplete and preserves scoped history", async () => {
|
|
gate = new Promise<void>((resolve) => { release = resolve; });
|
|
const response = await call({ input: "核查压力", conversation: "owned", stream: true });
|
|
const queued = await call({ input: "等待后续处理", conversation: "owned", tjwater: { delivery: "queue", input_id: `msg_${"3".repeat(32)}` } });
|
|
expect(queued.status).toBe(202);
|
|
const cancelled = await call({ session_id: "owned" }, "owner", "/api/v1/abort");
|
|
release();
|
|
expect(cancelled.ok).toBeTrue();
|
|
expect(pendingInputs).toEqual([]);
|
|
const text = await response.text();
|
|
expect(text).toContain("event: response.incomplete");
|
|
expect(text).not.toContain("event: response.completed");
|
|
});
|
|
|
|
it("continues a paged conversation from the complete saved history and preserves client message IDs", async () => {
|
|
const context = { actorKey: toActorKey("product:owner"), projectKey: toProjectKey(`product:${project}`), sessionId: "owned" };
|
|
const history = Array.from({ length: 105 }, (_, index) => ({ id: `old-${index}`, role: index % 2 ? "assistant" : "user",
|
|
parts: [{ type: "text", text: `历史证据 ${index}` }] }));
|
|
await states.write(context, { sessionId: "owned", isTitleManuallyEdited: true, messages: history });
|
|
const headers = { Authorization: "Bearer owner", "X-Project-Id": project };
|
|
const first = await http(`${base}/api/v1/session/owned?message_limit=40`, { headers });
|
|
expect((await first.json()).messages).toHaveLength(40);
|
|
const response = await call({ input: "继续核查", conversation: "owned", tjwater: { client_message_id: "client-turn-a" } });
|
|
expect(response.status).toBe(200);
|
|
await response.json();
|
|
const saved = (await states.read(context))!.messages as Array<{ id: string }>;
|
|
expect(saved.slice(0, 105)).toEqual(history);
|
|
expect(saved).toHaveLength(107);
|
|
expect(saved[105]?.id).toBe("client-turn-a");
|
|
expect((await call({ input: "重复提交", conversation: "owned", tjwater: { client_message_id: "client-turn-a" } })).status).toBe(409);
|
|
expect(prompted).toHaveLength(1);
|
|
});
|