源码
packages/desktop/test/user-questions.test.ts
import { Agent, type AgentTool } from "@zpi-dev/agent";
import { createAssistantMessageEventStream, emptyAssistant } from "@zpi-dev/ai";
import { UserQuestions } from "@zpi-dev/builtin-plugins/questions";
import { Type } from "typebox";
import { expect, it, vi } from "vitest";
import { chunk, deferred, done, fakeModel, send } from "../../../tests/fake-server.ts";
import { projectEvent } from "../src/main/projection.ts";
import { SessionHost } from "../src/main/session-host.ts";
import { cleanup, fixture } from "./helpers/host-fixture.ts";
const params = {
questions: [
{
header: "方案",
question: "使用哪种方案？",
multiSelect: false,
options: [
{ label: "A", description: "第一种" },
{ label: "B", description: "第二种" },
],
},
],
};
it.each(["__proto__", "constructor", "toString"])(
"preserves answers keyed by %s as plain data",
async (question) => {
const broker = new UserQuestions(() => {});
const execute = broker.tool("s").execute as unknown as (
...args: unknown[]
) => Promise<{ details: { answers: Record<string, string> } }>;
const pending = execute("q", { questions: [{ ...params.questions[0], question }] }, undefined);
broker.respond("s", "q", { answers: Object.fromEntries([[question, "A"]]) });
const { details } = await pending;
expect(Object.hasOwn(details.answers, question)).toBe(true);
expect(details.answers[question]).toBe("A");
const event = projectEvent(
{
type: "tool_execution_end",
toolCallId: "q",
toolName: "AskUserQuestion",
result: { content: [{ type: "text", text: "plain result" }], details },
isError: false,
},
"m",
);
expect(event?.type).toBe("tool_end");
if (event?.type === "tool_end") {
expect(Object.hasOwn(event.questionAnswers?.answers ?? {}, question)).toBe(true);
expect(event.questionAnswers?.answers[question]).toBe("A");
}
},
);
it("projects bounded question answers without forwarding malformed values or arbitrary tool details", () => {
const event = projectEvent(
{
type: "tool_execution_end",
toolCallId: "q",
toolName: "AskUserQuestion",
result: {
content: [{ type: "text", text: "result" }],
details: {
answers: { good: "A", bad: { toString: 1 }, long: "x".repeat(16001) },
arbitrary: "private",
},
},
isError: false,
},
"m",
);
expect(event).toMatchObject({ questionAnswers: { answers: { good: "A" } } });
expect(JSON.stringify(event)).not.toContain("private");
});
it("waits for an explicit answer, rejects stale replies, and returns skipped questions without inventing a preference", async () => {
const changes: unknown[] = [];
const broker = new UserQuestions((_, request) => changes.push(request));
const tool = broker.tool("s");
const execute = tool.execute as (...args: unknown[]) => Promise<{ content: { text: string }[] }>;
const pending = execute("q", params, undefined);
expect(broker.get("s")?.id).toBe("q");
expect(() => broker.respond("s", "old", { answers: {} })).toThrow("提问已结束");
expect(() => broker.respond("s", "q", { answers: { unknown: "A" } })).toThrow("无效回答");
broker.respond("s", "q", { answers: {} });
expect((await pending).content[0].text).toContain("did not provide answers");
expect(broker.get("s")).toBeNull();
expect(changes).toHaveLength(2);
});
it("cancels a pending question when the run is stopped", async () => {
const broker = new UserQuestions(() => {});
const controller = new AbortController();
const execute = broker.tool("s").execute as (...args: unknown[]) => Promise<unknown>;
const pending = execute("q", params, controller.signal);
controller.abort();
await expect(pending).rejects.toThrow("cancelled");
expect(broker.get("s")).toBeNull();
});
it("shows authoritative countdown state after a minute and continues unanswered questions at five minutes", async () => {
vi.useFakeTimers();
vi.setSystemTime(0);
try {
const broker = new UserQuestions(() => {});
const execute = broker.tool("s").execute as (
...args: unknown[]
) => Promise<{ content: { text: string }[] }>;
const pending = execute("q", params, undefined);
expect(broker.get("s")?.autoResolution).toEqual({
phase: "hiddenGrace",
startedAt: 0,
visibleAt: 60_000,
deadlineAt: 300_000,
});
await vi.advanceTimersByTimeAsync(59_000);
expect(broker.get("s")?.autoResolution?.phase).toBe("hiddenGrace");
await vi.advanceTimersByTimeAsync(1000);
expect(broker.get("s")?.autoResolution?.phase).toBe("visibleCountdown");
await vi.advanceTimersByTimeAsync(120_000);
expect(broker.get("s")?.autoResolution?.phase).toBe("visibleCountdown");
await vi.advanceTimersByTimeAsync(120_000);
expect((await pending).content[0].text).toContain("within five minutes");
expect(broker.get("s")).toBeNull();
expect(vi.getTimerCount()).toBe(0);
} finally {
vi.useRealTimers();
}
});
it.each([0, 60_000])("permanently stops question timing after user interaction at %d ms", async (delay) => {
vi.useFakeTimers();
try {
const broker = new UserQuestions(() => {});
const execute = broker.tool("s").execute as (...args: unknown[]) => Promise<unknown>;
const pending = execute("q", params, undefined);
await vi.advanceTimersByTimeAsync(delay);
broker.snooze("s", "old");
expect(broker.get("s")?.autoResolution?.phase).not.toBe("snoozed");
broker.snooze("s", "q");
broker.snooze("s", "q");
expect(broker.get("s")?.autoResolution?.phase).toBe("snoozed");
expect(vi.getTimerCount()).toBe(0);
await vi.advanceTimersByTimeAsync(300_000);
expect(broker.get("s")?.id).toBe("q");
broker.respond("s", "q", { answers: {} });
await pending;
} finally {
vi.useRealTimers();
}
});
it("snoozing one question does not disable automatic continuation for other questions", async () => {
vi.useFakeTimers();
try {
const broker = new UserQuestions(() => {});
const execute = (id: string) =>
(broker.tool(id).execute as (...args: unknown[]) => Promise<unknown>)("q", params, undefined);
const first = execute("first");
broker.snooze("first", "q");
expect(broker.get("first")?.autoResolution?.phase).toBe("snoozed");
const second = execute("second");
expect(broker.get("second")?.autoResolution?.phase).toBe("hiddenGrace");
const third = execute("third");
await vi.advanceTimersByTimeAsync(300_000);
await Promise.all([second, third]);
expect(broker.get("first")).not.toBeNull();
expect(broker.get("second")).toBeNull();
broker.respond("first", "q", { answers: {} });
await first;
expect(vi.getTimerCount()).toBe(0);
} finally {
vi.useRealTimers();
}
});
it("snapshots the plugin for new sessions, resumes with answers, and restores the choice after restart", async () => {
const f = await fixture((_, response, index) => {
if (index === 0) {
send(
response,
chunk({
tool_calls: [
{
index: 0,
id: "ask",
type: "function",
function: { name: "AskUserQuestion", arguments: JSON.stringify(params) },
},
],
}),
);
done(response, "tool_calls");
} else {
send(response, chunk({ content: "done" }));
done(response);
}
});
f.settings.setPluginEnabled("zpi.questions", true);
const session = await f.host.createSession(f.a.id);
await f.host.startRun({ sessionId: session.id, text: "choose" });
const running = f.host.activeRuns.get(session.id)?.done;
await expect.poll(() => f.host.getSessionSnapshot(session.id).view.question?.id).toBe("ask");
f.settings.setPluginEnabled("zpi.questions", false);
f.host.plugins.questions.respond(session.id, "ask", { answers: { "使用哪种方案？": "B" } });
await running;
expect(JSON.stringify(f.server.requests[1].messages)).toContain("B");
expect(f.host.getSessionSnapshot(session.id).view.question).toBeNull();
const questionBlock = () =>
f.host
.getSessionSnapshot(session.id)
.view.runs[0].orderedBlocks.find((block) => block.type === "tool" && block.name === "AskUserQuestion");
expect(questionBlock()).toMatchObject({ questionAnswers: { answers: { "使用哪种方案？": "B" } } });
const next = await f.host.createSession(f.a.id);
await f.host.startRun({ sessionId: next.id, text: "new" });
await f.host.activeRuns.get(next.id)?.done;
expect(JSON.stringify(f.server.requests.at(-1)?.tools)).not.toContain("AskUserQuestion");
await f.host.close();
const restored = new SessionHost(f.dir, f.settings, f.dir, undefined, undefined);
await restored.init();
cleanup.push(() => restored.close());
expect(restored.getSessionSnapshot(session.id).view.runs[0].orderedBlocks).toContainEqual(
expect.objectContaining({ questionAnswers: { answers: { "使用哪种方案？": "B" } } }),
);
await restored.startRun({ sessionId: session.id, text: "continue" });
await restored.activeRuns.get(session.id)?.done;
expect(JSON.stringify(f.server.requests.at(-1)?.tools)).toContain("AskUserQuestion");
});
it("shows a same-batch question while an earlier long-running tool is still executing", async () => {
const broker = new UserQuestions(() => {});
const started = deferred();
const release = deferred();
const model = fakeModel("http://localhost");
let requests = 0;
const agent = new Agent({
initialState: {
model,
tools: [
{
name: "longTask",
label: "longTask",
description: "longTask",
parameters: Type.Object({}),
execute: async () => {
started.resolve();
await release.promise;
return { content: [{ type: "text", text: "done" }], details: undefined };
},
},
broker.tool("s") as unknown as AgentTool,
],
},
streamFn: () => {
const stream = createAssistantMessageEventStream();
const message = emptyAssistant(model);
message.content =
requests++ === 0
? [
{ type: "toolCall", id: "long", name: "longTask", arguments: {} },
{ type: "toolCall", id: "q", name: "AskUserQuestion", arguments: params },
]
: [{ type: "text", text: "finished" }];
message.stopReason = requests === 1 ? "toolUse" : "stop";
stream.push({ type: "start", partial: message });
stream.end(message);
return stream;
},
});
const run = agent.prompt("choose while working");
try {
await started.promise;
await expect.poll(() => broker.get("s")?.id).toBe("q");
broker.respond("s", "q", { answers: { "使用哪种方案？": "A" } });
release.resolve();
await run;
expect(requests).toBe(2);
} finally {
release.resolve();
agent.abort();
await run;
}
});
it("queues simultaneous questions and gives each head a full answer window", async () => {
vi.useFakeTimers();
vi.setSystemTime(0);
try {
const changes: (string | null)[] = [];
const broker = new UserQuestions((_, request) => changes.push(request?.id ?? null));
const execute = broker.tool("s").execute as (...args: unknown[]) => Promise<unknown>;
const first = execute("first", params, undefined);
const second = execute("second", params, undefined);
expect(changes).toEqual(["first"]);
expect(() => broker.respond("s", "second", { answers: {} })).toThrow("提问已结束");
await vi.advanceTimersByTimeAsync(300_000);
await first;
expect(broker.get("s")).toMatchObject({
id: "second",
autoResolution: { startedAt: 300_000, deadlineAt: 600_000 },
});
expect(changes).toEqual(["first", "first", "second"]);
await vi.advanceTimersByTimeAsync(299_999);
expect(broker.get("s")?.id).toBe("second");
broker.respond("s", "second", { answers: {} });
await second;
expect(broker.get("s")).toBeNull();
expect(vi.getTimerCount()).toBe(0);
} finally {
vi.useRealTimers();
}
});
it("removes cancelled queued questions without disturbing the visible question", async () => {
const changes: (string | null)[] = [];
const broker = new UserQuestions((_, request) => changes.push(request?.id ?? null));
const execute = broker.tool("s").execute as (...args: unknown[]) => Promise<unknown>;
const controller = new AbortController();
const first = execute("first", params, undefined);
const second = execute("second", params, controller.signal);
const third = execute("third", params, undefined);
controller.abort();
await expect(second).rejects.toThrow("cancelled");
expect(broker.get("s")?.id).toBe("first");
expect(changes).toEqual(["first"]);
broker.respond("s", "first", { answers: {} });
await first;
expect(broker.get("s")?.id).toBe("third");
expect(() => broker.respond("s", "first", { answers: {} })).toThrow("提问已结束");
broker.respond("s", "third", { answers: {} });
await third;
expect(changes).toEqual(["first", "third", null]);
});
it("cancels a run's entire question queue without displaying already-aborted questions", async () => {
const changes: (string | null)[] = [];
const broker = new UserQuestions((_, request) => changes.push(request?.id ?? null));
const execute = broker.tool("s").execute as (...args: unknown[]) => Promise<unknown>;
const controller = new AbortController();
const results = Promise.allSettled([
execute("first", params, controller.signal),
execute("second", params, controller.signal),
execute("third", params, controller.signal),
]);
controller.abort();
expect((await results).every((result) => result.status === "rejected")).toBe(true);
expect(changes).toEqual(["first", null]);
expect(broker.get("s")).toBeNull();
});
it("advances after cancelling the head and keeps snooze scoped to the active question", async () => {
vi.useFakeTimers();
vi.setSystemTime(0);
try {
const broker = new UserQuestions(() => {});
const execute = broker.tool("s").execute as (...args: unknown[]) => Promise<unknown>;
const controller = new AbortController();
const first = execute("first", params, controller.signal);
const second = execute("second", params, undefined);
broker.snooze("s", "second");
expect(broker.get("s")?.autoResolution?.phase).toBe("hiddenGrace");
broker.snooze("s", "first");
await vi.advanceTimersByTimeAsync(400_000);
expect(broker.get("s")?.id).toBe("first");
controller.abort();
await expect(first).rejects.toThrow("cancelled");
expect(broker.get("s")).toMatchObject({
id: "second",
autoResolution: { phase: "hiddenGrace", startedAt: 400_000, deadlineAt: 700_000 },
});
await vi.advanceTimersByTimeAsync(300_000);
await second;
expect(broker.get("s")).toBeNull();
expect(vi.getTimerCount()).toBe(0);
} finally {
vi.useRealTimers();
}
});