jesse-uxandGrok (agent loop)
cd7e0a3493
完成回答、直达校正、输出审核和开流前加载仍留在路由里，使 diff 停在 1500 行以内。 Co-Authored-By: Grok (agent loop)
248 lines
15 KiB
TypeScript
248 lines
15 KiB
TypeScript
import assert from "node:assert/strict";
|
||
import test from "node:test";
|
||
import { readFileSync } from "node:fs";
|
||
import { classifyConsultationTurn, consultationTurnSchema, smalltalkHistoryPair, type SmalltalkObservation } from "../src/lib/consultation-smalltalk.ts";
|
||
import { streamSmalltalkResponse } from "../src/lib/stream-smalltalk-response.ts";
|
||
import { consultationAgentPublicEventSchema } from "../src/lib/consultation-agent-events.ts";
|
||
import { settledChatMessageViews, streamingChatMessageView } from "../src/lib/chat-message-view.ts";
|
||
import { productConversationVoice, natalSpokenReportContract } from "../src/mastra/product-voice.ts";
|
||
import type { ResolvedLanguageModel } from "../src/mastra/model.ts";
|
||
import type { LanguageModelV2 } from "@ai-sdk/provider";
|
||
|
||
// Transport tests inject a generator; the adapter regression uses a local fake.
|
||
// Neither resolves credentials nor calls a real provider.
|
||
const model = { id: "fake-model", model: "openai/fake" } as ResolvedLanguageModel;
|
||
const source = (path: string) => readFileSync(new URL(path, import.meta.url), "utf8");
|
||
for (const question of ["你好", "在吗", "谢谢", "晚安", "哈哈", "我回来了"]) {
|
||
test(`smalltalk fake transport: ${question}`, async () => {
|
||
let calls = 0;
|
||
assert.deepEqual(await classifyConsultationTurn({ model, question, history: [], generate: async () => {
|
||
calls++; return { object: { kind: "smalltalk", reply: "在呢" } };
|
||
} }), { kind: "smalltalk", reply: "在呢" });
|
||
assert.equal(calls, 1);
|
||
});
|
||
}
|
||
for (const question of ["你好，帮我看看事业", "？", "你在说什么鬼", "今年怎么样", ""]) {
|
||
test(`consult fake transport: ${question || "blank"}`, async () => {
|
||
assert.deepEqual(await classifyConsultationTurn({ model, question, history: [], generate: async () => ({ object: { kind: "consult" } }) }), { kind: "consult" });
|
||
});
|
||
}
|
||
for (const object of [null, {}, { kind: "smalltalk", reply: "" }, { kind: "smalltalk", reply: "字".repeat(21) }, { kind: "smalltalk", reply: "你好。" }, { kind: "smalltalk", reply: "你好\n再见" }, { kind: "smalltalk", reply: "你好", extra: true }, { kind: "consult", reply: "你好" }]) {
|
||
test(`strict schema fails open: ${JSON.stringify(object)}`, async () => {
|
||
assert.equal(consultationTurnSchema.safeParse(object).success, false);
|
||
assert.deepEqual(await classifyConsultationTurn({ model, question: "test", history: [], generate: async () => ({ object }) }), { kind: "consult" });
|
||
});
|
||
}
|
||
test("default Mastra adapter uses one tool-free bounded model call and returns real usage", async () => {
|
||
const calls: Parameters<LanguageModelV2["doGenerate"]>[0][] = [];
|
||
const text = '{"kind":"smalltalk","reply":"你好"}';
|
||
const usage = { inputTokens: 15, outputTokens: 10, totalTokens: 25 };
|
||
const fake: LanguageModelV2 = {
|
||
specificationVersion: "v2", provider: "fake", modelId: "fake", supportedUrls: {},
|
||
async doGenerate(options) {
|
||
calls.push(options);
|
||
return { content: [{ type: "text", text }], finishReason: "stop", usage, warnings: [] };
|
||
},
|
||
async doStream(options) {
|
||
calls.push(options);
|
||
return { stream: new ReadableStream({ start(controller) {
|
||
controller.enqueue({ type: "stream-start", warnings: [] });
|
||
controller.enqueue({ type: "text-start", id: "1" });
|
||
controller.enqueue({ type: "text-delta", id: "1", delta: text });
|
||
controller.enqueue({ type: "text-end", id: "1" });
|
||
controller.enqueue({ type: "finish", finishReason: "stop", usage });
|
||
controller.close();
|
||
} }) };
|
||
},
|
||
};
|
||
const observations: SmalltalkObservation[] = [];
|
||
const result = await classifyConsultationTurn({
|
||
model: { ...model, model: fake }, question: "你好", history: [],
|
||
onObservation: (observation) => observations.push(observation),
|
||
});
|
||
assert.deepEqual(result, { kind: "smalltalk", reply: "你好" });
|
||
assert.equal(calls.length, 1);
|
||
assert.equal(calls[0]?.maxOutputTokens, 96);
|
||
assert.equal(calls[0]?.tools?.length ?? 0, 0);
|
||
assert.equal(observations[0]?.usage?.inputTokens, 15);
|
||
assert.equal(observations[0]?.usage?.outputTokens, 10);
|
||
});
|
||
for (const scenario of ["invalid_schema", "bad_json", "provider_error"] as const) {
|
||
test(`default Mastra adapter protects privacy and usage on ${scenario}`, async (t) => {
|
||
const sentinel = "FICTIONAL_PRIVATE_SENTINEL";
|
||
const text = scenario === "bad_json" ? sentinel : JSON.stringify({ kind: "smalltalk", reply: `${sentinel}。` });
|
||
const usage = { inputTokens: 23, outputTokens: 7, totalTokens: 30 };
|
||
const calls: Parameters<LanguageModelV2["doGenerate"]>[0][] = [];
|
||
const fake: LanguageModelV2 = {
|
||
specificationVersion: "v2", provider: "fake", modelId: "fake", supportedUrls: {},
|
||
async doGenerate(options) {
|
||
calls.push(options);
|
||
if (scenario === "provider_error") throw new Error(sentinel);
|
||
return { content: [{ type: "text", text }], finishReason: "stop", usage, warnings: [] };
|
||
},
|
||
async doStream(options) {
|
||
calls.push(options);
|
||
if (scenario === "provider_error") throw new Error(sentinel);
|
||
return { stream: new ReadableStream({ start(controller) {
|
||
controller.enqueue({ type: "stream-start", warnings: [] });
|
||
controller.enqueue({ type: "text-start", id: "1" });
|
||
controller.enqueue({ type: "text-delta", id: "1", delta: text });
|
||
controller.enqueue({ type: "text-end", id: "1" });
|
||
controller.enqueue({ type: "finish", finishReason: "stop", usage });
|
||
controller.close();
|
||
} }) };
|
||
},
|
||
};
|
||
const logs: string[] = [];
|
||
const observations: SmalltalkObservation[] = [];
|
||
let result;
|
||
try {
|
||
for (const method of ["log", "info", "warn", "error", "debug"] as const) {
|
||
t.mock.method(console, method, (...values: unknown[]) => { logs.push(JSON.stringify(values)); });
|
||
}
|
||
const stdoutWrite = process.stdout.write.bind(process.stdout);
|
||
const stderrWrite = process.stderr.write.bind(process.stderr);
|
||
t.mock.method(process.stdout, "write", ((
|
||
chunk: string | Uint8Array,
|
||
encoding?: BufferEncoding | ((error?: Error | null) => void),
|
||
callback?: (error?: Error | null) => void,
|
||
) => {
|
||
logs.push(String(chunk));
|
||
if (typeof encoding === "function") return stdoutWrite(chunk, encoding);
|
||
return stdoutWrite(chunk, encoding, callback);
|
||
}) as typeof process.stdout.write);
|
||
t.mock.method(process.stderr, "write", ((
|
||
chunk: string | Uint8Array,
|
||
encoding?: BufferEncoding | ((error?: Error | null) => void),
|
||
callback?: (error?: Error | null) => void,
|
||
) => {
|
||
logs.push(String(chunk));
|
||
if (typeof encoding === "function") return stderrWrite(chunk, encoding);
|
||
return stderrWrite(chunk, encoding, callback);
|
||
}) as typeof process.stderr.write);
|
||
result = await classifyConsultationTurn({ model: { ...model, model: fake }, question: sentinel, history: [], onObservation: (o) => observations.push(o) });
|
||
} finally { t.mock.restoreAll(); }
|
||
assert.deepEqual(result, { kind: "consult" });
|
||
assert.equal(calls.length, 1);
|
||
assert.equal(calls[0]?.maxOutputTokens, 96);
|
||
assert.equal(calls[0]?.tools?.length ?? 0, 0);
|
||
assert.equal(logs.join(" ").includes(sentinel), false, "SDK must not log private text");
|
||
assert.equal(JSON.stringify(observations).includes(sentinel), false);
|
||
if (scenario === "provider_error") {
|
||
assert.equal(observations[0]?.outcome, "provider_error");
|
||
assert.equal(observations[0]?.usage, undefined, "unknown usage must not become a fake zero");
|
||
} else {
|
||
assert.equal(observations[0]?.outcome, "invalid_output");
|
||
assert.equal(observations[0]?.usage?.inputTokens, 23);
|
||
assert.equal(observations[0]?.usage?.outputTokens, 7);
|
||
}
|
||
});
|
||
}
|
||
test("invalid JSON and provider errors fail open without leaking errors", async () => {
|
||
for (const generate of [async () => ({ text: "not json" }), async () => { throw new Error("private provider payload"); }]) {
|
||
const observations: SmalltalkObservation[] = [];
|
||
assert.deepEqual(await classifyConsultationTurn({ model, question: "test", history: [], generate, onObservation: (o) => observations.push(o) }), { kind: "consult" });
|
||
assert.equal(JSON.stringify(observations).includes("private"), false);
|
||
}
|
||
});
|
||
test("timeout is bounded even when provider ignores abort, and late usage remains observable", async () => {
|
||
let resolve!: (value: { object: unknown; usage: { inputTokens: number } }) => void;
|
||
let signal: AbortSignal | undefined;
|
||
const observations: SmalltalkObservation[] = [];
|
||
const started = Date.now();
|
||
const result = await classifyConsultationTurn({ model, question: "test", history: [], onObservation: (o) => observations.push(o), generate: async (_, s) => {
|
||
signal = s; return new Promise((done) => { resolve = done; });
|
||
} });
|
||
assert.deepEqual(result, { kind: "consult" });
|
||
assert.ok(Date.now() - started >= 2900 && Date.now() - started < 4500);
|
||
assert.equal(signal?.aborted, true);
|
||
assert.equal(observations[0]?.outcome, "timeout");
|
||
resolve({ object: { kind: "smalltalk", reply: "你好" }, usage: { inputTokens: 12 } });
|
||
await new Promise((done) => setImmediate(done));
|
||
assert.equal(observations[1]?.late, true);
|
||
assert.equal(observations[1]?.usage?.inputTokens, 12);
|
||
});
|
||
test("pre-cancelled input never calls the model or causes an unhandled rejection", async () => {
|
||
const c = new AbortController(); c.abort();
|
||
let calls = 0;
|
||
assert.deepEqual(await classifyConsultationTurn({ model, question: "test", history: [], signal: c.signal, generate: async () => { calls++; return {}; } }), { kind: "consult" });
|
||
assert.equal(calls, 0);
|
||
await new Promise((done) => setImmediate(done));
|
||
});
|
||
test("input projects only visible question, name and last complete history pair", async () => {
|
||
const history = [
|
||
{ role: "user", text: "older", requestId: "old" }, { role: "assistant", text: "old reply", requestId: "old" },
|
||
{ role: "user", text: "previous", requestId: "pair", birth: "excluded" },
|
||
{ role: "assistant", text: "visible previous reply", requestId: "pair", thinkingText: "excluded", workflowReceipt: {} },
|
||
{ role: "user", text: "incomplete" },
|
||
];
|
||
assert.deepEqual(smalltalkHistoryPair(history), [{ role: "user", text: "previous" }, { role: "assistant", text: "visible previous reply" }]);
|
||
assert.deepEqual(smalltalkHistoryPair([{ role: "user", text: "unanswered" }]), []);
|
||
assert.deepEqual(smalltalkHistoryPair([{ role: "user", text: "a", requestId: "a" }, { role: "assistant", text: "b", requestId: "b" }]), []);
|
||
await classifyConsultationTurn({ model, question: "current", name: "虚构称呼", history, generate: async (content) => {
|
||
assert.deepEqual(JSON.parse(content), { question: "current", name: "虚构称呼", history: smalltalkHistoryPair(history) });
|
||
return { object: { kind: "consult" } };
|
||
} });
|
||
const classifier = source("../src/lib/consultation-smalltalk.ts");
|
||
assert.doesNotMatch(classifier, /tools:|memory:|getJyotishAgent|skillBinding/);
|
||
});
|
||
test("smalltalk stream persists first and emits only answer plus receipt-free completion", async () => {
|
||
let persisted = false;
|
||
const response = streamSmalltalkResponse({ requestId: "fake", reply: "你好", complete: async () => { persisted = true; }, onError: async () => assert.fail() });
|
||
const events = (await response.text()).trim().split("\n").map((line) => consultationAgentPublicEventSchema.parse(JSON.parse(line)));
|
||
assert.equal(persisted, true);
|
||
assert.deepEqual(events, [{ type: "answer.delta", text: "你好" }, { type: "run.completed", responseKind: "smalltalk" }]);
|
||
assert.equal(consultationAgentPublicEventSchema.safeParse({ type: "run.completed" }).success, false);
|
||
});
|
||
test("failed free persistence never emits answer or completion", async () => {
|
||
let cancelled = 0;
|
||
const response = streamSmalltalkResponse({ requestId: "fake", reply: "你好", complete: async () => { throw new Error("failed"); }, onError: async () => { cancelled++; } });
|
||
const text = await response.text();
|
||
assert.equal(cancelled, 1);
|
||
assert.match(text, /run.failed/);
|
||
assert.doesNotMatch(text, /answer.delta|run.completed|你好/);
|
||
});
|
||
test("disconnect does not cancel the server-owned free completion", async () => {
|
||
let finish!: () => void;
|
||
let persisted = false;
|
||
const response = streamSmalltalkResponse({ requestId: "fake", reply: "你好", complete: async () => { await new Promise<void>((done) => { finish = done; }); persisted = true; }, onError: async () => assert.fail() });
|
||
await response.body!.cancel(); finish();
|
||
await new Promise((done) => setImmediate(done));
|
||
assert.equal(persisted, true);
|
||
});
|
||
test("smalltalk views never reconstruct an invented settled execution timeline", () => {
|
||
assert.equal(settledChatMessageViews([{ role: "assistant", text: "你好", responseKind: "smalltalk" }])[0]?.timeline, undefined);
|
||
assert.equal(streamingChatMessageView([{ role: "user", text: "你好" }], true, "你好", undefined, undefined, undefined, [], "smalltalk")?.responseKind, "smalltalk");
|
||
// 原值: /const quiet = smalltalk \|\| awaitingClassification/（分类回来前整行静默）
|
||
// 新值: /const quiet = smalltalk;/（只有寒暄回复静默；分类前显示排队行）
|
||
// 原因: TASK-consult-first-frame-and-pacing-20260928 D1 / BUG-1074
|
||
assert.match(source("../src/components/chat-message-row.tsx"), /const quiet = smalltalk;/);
|
||
assert.doesNotMatch(source("../src/components/chat-message-row.tsx"), /awaitingClassification/);
|
||
assert.match(source("../src/lib/home-cloud-sync.ts"), /stored.responseKind === "smalltalk"/);
|
||
});
|
||
test("route gates all special entries before a single selected-model classifier and keeps tool contracts", () => {
|
||
const route = source("../src/app/api/consult/route.ts");
|
||
assert.match(route, /parsed.data.entrypoint === undefined\s*\? await classifyConsultationTurn/);
|
||
assert.match(route, /classifyConsultationTurn\(\{\s*model: selectedModel,\s*question: visibleQuestion/);
|
||
assert.equal((route.match(/requireTool: true/g) ?? []).length, 2);
|
||
const smalltalk = route.slice(route.indexOf('if (turn.kind === "smalltalk")'), route.indexOf("const expectedTitle"));
|
||
const settlement = source("../src/lib/consult/settlement.ts");
|
||
// 原值: 寒暄分支源码含 complete_consultation_free
|
||
// 新值: 分支调用 consultSettlement.completeSmalltalk；RPC 在 settlement.ts
|
||
// 原因: 寒暄结算搬进工厂，分支仍不跑 agent
|
||
assert.match(smalltalk, /consultSettlement\.completeSmalltalk\(turn\.reply\)/);
|
||
assert.match(settlement, /complete_consultation_free/);
|
||
assert.doesNotMatch(smalltalk, /run-jyotish-|streamAgentResponse\(/);
|
||
});
|
||
test("all three agent voices share the exemption, natal domain skeleton stays mandatory", () => {
|
||
assert.match(productConversationVoice, /只是打招呼、道谢、告别时，回一句话，不要套开场形状/);
|
||
assert.doesNotMatch(natalSpokenReportContract, /Short chit-chat/);
|
||
// 原值: /The first natal career\/wealth\/marriage\/family answer of a session uses the opener-plus-skeleton above/
|
||
// 新值: 首轮用段落形状；追问轮仍按用户轮里的指令直接回答
|
||
// 原因: TASK-consult-conversational-answer-20261006 D1（BUG-1244）：首轮骨架取消，追问轮形状保留
|
||
assert.match(natalSpokenReportContract, /The first natal career\/wealth\/marriage\/family answer of a session uses the paragraph shape above/);
|
||
assert.match(natalSpokenReportContract, /a follow-up turn follows the shape instruction carried in the user turn/);
|
||
const index = source("../src/mastra/index.ts");
|
||
assert.ok((index.match(/\$\{productConversationVoice\}/g) ?? []).length >= 3);
|
||
});
|