import assert from "node:assert/strict"; import test from "node:test"; import { appendConsultationRuntimeStep, canonicalDomainPlan, consultationModelStepTelemetry, consultationStepBudgetReceipt, createConsultationTools, createConsultationRuntimeState, publicConsultationRuntimeSteps, } from "../src/mastra/consultation-tools.ts"; import { ConsultationWorkflowError, consultationWorkflowFailureCode, } from "../src/mastra/consultation-workflow.ts"; import { agentExecutionReceiptSchema } from "../src/lib/consultation-agent-events.ts"; import { getJyotishAgent } from "../src/mastra/index.ts"; import { consultationAgentPublicEventSchema, createNdjsonParser } from "../src/lib/consultation-agent-events.ts"; import { createConsultationPlan } from "../src/lib/consultation-plan.ts"; import { collectAgentPublicEvents, ensureFinalResponseText, streamAgentResponse } from "../src/lib/stream-agent-response.ts"; const serverChart = { name: "测试", toolInput: { year: 1990, month: 1, day: 2, hour: 3, minute: 4, city: "台北", lat: 25.03, lon: 121.56, tz: 8 }, truth: { birthDate: "1990-01-02", reportedBirthTime: "03:04", activeBirthTime: null, selectedTimeKind: "reported" as const, birthTimeSource: "reported", birthTimeStatus: "reported", placeLabel: "台北", placeCodes: { countryCode: "TW", provinceCode: null, cityCode: null, districtCode: null }, placeId: null, placeType: "city", placeProvider: "profile", latitude: 25.03, longitude: 121.56, timezoneId: "Asia/Taipei", timezoneSource: "profile", timezoneOffset: 8, }, }; // Mastra validates against the model-facing schema before execute() runs, so a // test can only send what the model can send. The cast keeps the argument // checked against that shape without depending on Mastra's inferred type. type ModelConsultationToolInput = { question: string; domains?: string[] }; const modelInput = (input: ModelConsultationToolInput) => input as never; const rejectedByInputSchema = (input: { question: string; theme?: string; domains?: string[] }) => input as never; function workflow( theme = "career", options: { status?: "ready" | "degraded" | "blocked"; missingLayers?: string[]; preciseTiming?: boolean } = {}, ) { return { success: true, chart: {}, routing: { primary_theme: theme }, consumer_context: { route: theme, core_status: options.status ?? "ready", available_layers: [], missing_route_layers: options.missingLayers ?? [], hard_blockers: [], technique_truth: { status: "verified" }, answer_policy: { can_answer_direction: true, can_answer_precise_timing: options.preciseTiming ?? true }, }, }; } test("the server-selected domain stays authoritative when the model omits domains", async () => { let calls = 0; let captured: unknown; let capturedPlan: unknown; const state = createConsultationRuntimeState(); const plan = createConsultationPlan({ userIntent: "事业如何", theme: "career", consultationMode: "unverified_birth_time", modelCreditCost: 1, }); assert.throws( () => (plan.requestedDomains as unknown as string[]).push("timing"), TypeError, ); const tools = createConsultationTools({ userId: "u", sessionId: "s", requestId: "r", consultationMode: "unverified_birth_time", plan, theme: "career", serverChart, state, runWorkflow: async (input, options) => { calls += 1; captured = input; capturedPlan = options?.plan; return workflow(); }, }); const tool = tools["run-jyotish-consultation"]; // A single-value theme is deliberately absent: two mutually exclusive ways to // name a domain cost the model a step per call to discover the rule. assert.deepEqual(Object.keys((tool.inputSchema as unknown as { shape: object }).shape), ["question", "domains"]); const execute = tool.execute!; const context = { observe: { span: async (_n: string, fn: () => Promise) => fn(), log() {} } } as never; const [first, second] = await Promise.all([ execute(modelInput({ question: "尝试改成精确应期" }), context), execute(modelInput({ question: "尝试改成婚恋" }), context), ]); assert.equal(calls, 1); assert.deepEqual(first, second); assert.deepEqual(captured, { ...serverChart.toolInput, entryMode: "direct_chart", question: "事业如何", theme: "career" }); assert.strictEqual(capturedPlan, plan); assert.equal(state.consultationToolCallCount, 1); assert.equal(state.consultationToolSuccessCount, 1); assert.equal(state.workflowReceipt?.preciseTiming, "allowed"); assert.deepEqual(state.workflowReceipt?.domains, ["career"]); assert.deepEqual((first as { domains?: string[] }).domains, ["career"]); }); test("multi-domain plan canonicalizes aliases, de-duplicates, preserves order, and runs every domain", async () => { const calls: Array<{ theme: string; question: string }> = []; const state = createConsultationRuntimeState(); const tools = createConsultationTools({ userId: "u", sessionId: "s", requestId: "r", consultationMode: "verified_chart", serverChart, state, runWorkflow: async (input) => { calls.push({ theme: input.theme, question: input.question }); if (input.theme === "wealth") return workflow(input.theme, { status: "degraded", missingLayers: ["D11"] }); return workflow(input.theme); }, }); const result = await tools["run-jyotish-consultation"].execute!( modelInput({ question: "事业、财富和迁居怎么一起规划", domains: ["career", "finance", "career", "home"] }), { observe: { span: async (_n: string, fn: () => Promise) => fn(), log() {} } } as never, ) as { domains: string[]; consultations: Array<{ domain: string }> }; assert.deepEqual(calls, [ { theme: "career", question: "事业、财富和迁居怎么一起规划" }, { theme: "wealth", question: "事业、财富和迁居怎么一起规划" }, { theme: "migration", question: "事业、财富和迁居怎么一起规划" }, ]); assert.deepEqual(result.domains, ["career", "wealth", "migration"]); assert.deepEqual(result.consultations.map((item) => item.domain), ["career", "wealth", "migration"]); assert.deepEqual(state.workflowReceipt, { route: "multi-domain", status: "degraded", preciseTiming: "allowed", missingLayers: ["D11"], domains: ["career", "wealth", "migration"], }); }); test("domain plan rejects unknown and product domains before any workflow runs", async () => { for (const domain of ["unknown", "prashna", "muhurta", "rectification", "compatibility"]) { let calls = 0; const state = createConsultationRuntimeState(); const tool = createConsultationTools({ userId: "u", sessionId: "s", requestId: `r-${domain}`, consultationMode: "verified_chart", serverChart, state, runWorkflow: async () => { calls += 1; return workflow(); }, })["run-jyotish-consultation"]; await assert.rejects( tool.execute!( modelInput({ question: "测试", domains: ["career", domain] }), { observe: { span: async (_n: string, fn: () => Promise) => fn(), log() {} } } as never, ), /unsupported_consultation_domain/, ); assert.equal(calls, 0); assert.equal(state.consultationToolCompleted, false); } }); test("domain plan enforces the raw plan upper bound and one input mode", async () => { let calls = 0; const state = createConsultationRuntimeState(); const tool = createConsultationTools({ userId: "u", sessionId: "s", requestId: "r-limit", consultationMode: "verified_chart", serverChart, state, runWorkflow: async () => { calls += 1; return workflow(); }, })["run-jyotish-consultation"]; const context = { observe: { span: async (_n: string, fn: () => Promise) => fn(), log() {} } } as never; const inputSchema = tool.inputSchema as unknown as { safeParse: (value: unknown) => { success: boolean } }; assert.equal(inputSchema.safeParse({ question: "测试", domains: ["career", "career", "career", "career", "career", "career", "career"], }).success, false); assert.equal(calls, 0); // The model can only express a domain plan one way. The pair that used to be // representable, and cost a model step to be told was invalid, is now refused // by the schema before the tool body runs. assert.equal(inputSchema.safeParse({ question: "测试", domains: ["career"] }).success, true); assert.equal(inputSchema.safeParse({ question: "测试" }).success, true); assert.equal(inputSchema.safeParse({ question: "测试", theme: "career" }).success, false); assert.equal(inputSchema.safeParse({ question: "测试", domains: ["career"], theme: "career" }).success, false); const rejectedState = createConsultationRuntimeState(); const secondTool = createConsultationTools({ userId: "u", sessionId: "s", requestId: "r-modes", consultationMode: "verified_chart", serverChart, state: rejectedState, runWorkflow: async () => { calls += 1; return workflow(); }, })["run-jyotish-consultation"]; const refused = await secondTool.execute!( rejectedByInputSchema({ question: "测试", domains: ["career"], theme: "career" }), context, ) as Record; // Nothing about the run advances: no calculation, no counted attempt, and no // consultation payload the model could mistake for a result. assert.equal(calls, 0); assert.equal("domains" in refused, false); assert.equal(rejectedState.consultationToolStarted, false); assert.equal(rejectedState.consultationToolCallCount, 0); assert.deepEqual(rejectedState.steps, []); }); test("the single-value domain form stays available to callers without the model schema", () => { const plan = createConsultationPlan({ userIntent: "事业如何", theme: "career", consultationMode: "verified_chart", modelCreditCost: 1, }); // A single-value theme must never override the route-selected server domain. assert.deepEqual(canonicalDomainPlan({ theme: "timing" }, { plan, theme: "career" }), ["career"]); assert.deepEqual(canonicalDomainPlan({}, { plan, theme: "career" }), ["career"]); assert.deepEqual(canonicalDomainPlan({ theme: "marriage" }, {}), ["marriage"]); assert.deepEqual(canonicalDomainPlan({ domains: ["career", "finance"] }, {}), ["career", "wealth"]); assert.deepEqual(canonicalDomainPlan({ domains: ["timing"] }, { plan, theme: "career" }), ["timing"]); assert.throws( () => canonicalDomainPlan({ domains: ["career"], theme: "career" }, {}), /invalid_consultation_domain_plan/, ); assert.throws(() => canonicalDomainPlan({}, {}), /invalid_consultation_domain_plan/); assert.throws(() => canonicalDomainPlan({ theme: "prashna" }, {}), /unsupported_consultation_domain/); }); test("invalid model input does not poison a later valid contract retry", async () => { let calls = 0; const state = createConsultationRuntimeState(); const tool = createConsultationTools({ userId: "u", sessionId: "s", requestId: "r-retry", consultationMode: "verified_chart", serverChart, state, runWorkflow: async (input) => { calls += 1; return workflow(input.theme); }, })["run-jyotish-consultation"]; const context = { observe: { span: async (_n: string, fn: () => Promise) => fn(), log() {} } } as never; // BUG-205: an input the schema accepts but the domain registry rejects must // still be refused before the request-scoped calculation cache is written. await assert.rejects( tool.execute!(modelInput({ question: "先给出错误参数", domains: ["career", "unknown"] }), context), /unsupported_consultation_domain/, ); assert.equal(calls, 0); assert.equal(state.consultationToolCallCount, 0); assert.equal(state.consultationToolSuccessCount, 0); const result = await tool.execute!(modelInput({ question: "改用合法参数", domains: ["timing"] }), context) as { domains: string[] }; assert.equal(calls, 1); assert.deepEqual(result.domains, ["timing"]); assert.equal(state.consultationToolCallCount, 1); assert.equal(state.consultationToolSuccessCount, 1); assert.equal(state.consultationToolCompleted, true); }); test("a rejected workflow promise is cleared before a later tool call", async () => { let calls = 0; const state = createConsultationRuntimeState(); const tool = createConsultationTools({ userId: "u", sessionId: "s", requestId: "r-rejected-promise", consultationMode: "verified_chart", serverChart, state, runWorkflow: async (input) => { calls += 1; if (calls === 1) throw new Error("workflow_temporarily_failed"); return workflow(input.theme); }, })["run-jyotish-consultation"]; const context = { observe: { span: async (_n: string, fn: () => Promise) => fn(), log() {} } } as never; await assert.rejects( tool.execute!(modelInput({ question: "第一次计算", domains: ["career"] }), context), /workflow_temporarily_failed/, ); const result = await tool.execute!(modelInput({ question: "重新计算", domains: ["timing"] }), context) as { domains: string[] }; assert.equal(calls, 2); assert.deepEqual(result.domains, ["timing"]); assert.equal(state.consultationToolCallCount, 2); assert.equal(state.consultationToolSuccessCount, 1); }); test("a failed calculation records why it failed and forwards the request id", async () => { const state = createConsultationRuntimeState(); const seen: Array = []; const tool = createConsultationTools({ userId: "u", sessionId: "s", requestId: "req-correlation", consultationMode: "verified_chart", serverChart, state, runWorkflow: async (_input, options) => { seen.push(options?.requestId); throw new ConsultationWorkflowError("workflow_queue_full", "Async job queue is full"); }, })["run-jyotish-consultation"]; await assert.rejects( tool.execute!(modelInput({ question: "队列满时的表现", domains: ["career"] }), { observe: { span: async (_n: string, fn: () => Promise) => fn(), log() {} } } as never), /Async job queue is full/, ); assert.deepEqual(seen, ["req-correlation"]); const failedStep = state.steps.find((step) => step.status === "failed"); assert.equal(failedStep?.name, "run-jyotish-consultation"); assert.equal(failedStep?.failureCode, "workflow_queue_full"); }); test("workflow failure codes classify transport and contract faults", () => { assert.equal(consultationWorkflowFailureCode(new ConsultationWorkflowError("workflow_rate_limited", "x")), "workflow_rate_limited"); assert.equal(consultationWorkflowFailureCode(new DOMException("slow", "TimeoutError")), "workflow_timeout"); assert.equal(consultationWorkflowFailureCode(new DOMException("stop", "AbortError")), "workflow_aborted"); assert.equal(consultationWorkflowFailureCode(new Error("anything else")), undefined); }); test("the public receipt never carries the internal failure classification", () => { const state = createConsultationRuntimeState(); appendConsultationRuntimeStep(state, { kind: "tool", name: "run-jyotish-consultation", status: "failed", durationMs: 12, failureCode: "workflow_server_error", }); appendConsultationRuntimeStep(state, { kind: "skill", name: "jyotish-vedic-astrology", status: "completed" }); const steps = publicConsultationRuntimeSteps(state); assert.equal(steps.every((step) => !("failureCode" in step)), true); assert.equal(state.steps[0]?.failureCode, "workflow_server_error"); // A strict receipt schema would reject the internal field, so this also // guards the run from failing while building a successful response. const receipt = agentExecutionReceiptSchema.parse({ runId: "run", runtime: "mastra-agentic", skill: { name: "jyotish-vedic-astrology", loaded: true }, steps, workflow: { route: "career", status: "ready", preciseTiming: "blocked", missingLayers: [] }, }); assert.equal(receipt.steps.length, 2); assert.throws(() => agentExecutionReceiptSchema.parse({ runId: "run", runtime: "mastra-agentic", skill: { name: "jyotish-vedic-astrology", loaded: true }, steps: state.steps, workflow: { route: "career", status: "ready", preciseTiming: "blocked", missingLayers: [] }, })); }); test("the public receipt never carries the model step budget diagnostics", () => { const state = createConsultationRuntimeState(); state.modelStepCount = 8; state.modelFinishReason = "tool-calls"; appendConsultationRuntimeStep(state, { kind: "skill", name: "jyotish-vedic-astrology", status: "completed" }); assert.deepEqual(consultationModelStepTelemetry(state), { modelStepCount: 8, modelFinishReason: "tool-calls" }); assert.deepEqual( consultationModelStepTelemetry(createConsultationRuntimeState()), { modelStepCount: 0 }, ); // The client receipt schema is strict, so leaking either field would make a // successful run fail while serializing its own answer. const receipt = agentExecutionReceiptSchema.parse({ runId: "run", runtime: "mastra-agentic", skill: { name: "jyotish-vedic-astrology", loaded: true }, steps: publicConsultationRuntimeSteps(state), stepBudget: consultationStepBudgetReceipt(state), workflow: { route: "career", status: "ready", preciseTiming: "blocked", missingLayers: [] }, }); assert.doesNotMatch(JSON.stringify(receipt), /modelStepCount|modelFinishReason|tool-calls/); assert.throws(() => agentExecutionReceiptSchema.parse({ runId: "run", runtime: "mastra-agentic", skill: { name: "jyotish-vedic-astrology", loaded: true }, steps: publicConsultationRuntimeSteps(state), workflow: { route: "career", status: "ready", preciseTiming: "blocked", missingLayers: [] }, ...consultationModelStepTelemetry(state), })); }); test("personal Agent exposes the Jyotish Skill and named server tool", async () => { const state = createConsultationRuntimeState(); const agent = getJyotishAgent({ id: "personal-agent-probe", label: "Probe", description: "", creditCost: 1, isDefault: false, mode: "openai", model: "openai/gpt-5-mini", } as never, { userId: "u", sessionId: "s", requestId: "r", consultationMode: "verified_chart", serverChart, state, } as never); const skills = await agent.listSkills(); const toolNames = Object.keys(await agent.getToolsForExecution({ runId: "r" })); assert.equal(skills.some((skill) => skill.name === "jyotish-vedic-astrology"), true); assert.equal(toolNames.includes("skill"), true); assert.equal(toolNames.includes("run-jyotish-consultation"), true); assert.equal(toolNames.includes("consultationTool"), false); }); test("public stream filters private chunks and completes once", async () => { const chunks = [ { type: "reasoning-delta", payload: { text: "secret" } }, { type: "tool-call", payload: { toolCallId: "c1", toolName: "skill", args: { name: "jyotish-vedic-astrology", secret: "x" } } }, { type: "tool-result", payload: { toolCallId: "other", toolName: "skill", result: { private: true } } }, { type: "tool-result", payload: { toolCallId: "c1", toolName: "skill", result: { private: true } } }, { type: "tool-call", payload: { toolCallId: "c2", toolName: "run-jyotish-consultation", args: { year: 1990 } } }, { type: "data-jyotish-activity", data: { phase: "chart-calculation", label: "正在计算本命盘", private: "x" } }, { type: "tool-result", payload: { toolCallId: "c2", toolName: "run-jyotish-consultation", result: { birth: "private" } } }, { type: "text-delta", payload: { text: "可以先看方向。", providerMetadata: { secret: true } } }, ]; const events = await collectAgentPublicEvents(chunks as never, { runId: "run", requestId: "req", toolStatus: () => "ready", receipt: () => ({ runId: "run", runtime: "mastra-agentic", skill: { name: "jyotish-vedic-astrology", loaded: true }, steps: [], workflow: { route: "career", status: "ready", preciseTiming: "blocked", missingLayers: [], domains: ["career"] }, }), }); assert.equal(events.filter((event) => event.type === "run.completed").length, 1); assert.equal(events.some((event) => JSON.stringify(event).includes("secret") || JSON.stringify(event).includes("private") || JSON.stringify(event).includes("1990")), false); assert.equal(events.filter((event) => event.type === "skill.completed").length, 1); assert.equal(events.some((event) => event.type === "answer.delta"), true); for (const event of events) consultationAgentPublicEventSchema.parse(event); const completed = events.find((event) => event.type === "run.completed"); assert.deepEqual(completed?.type === "run.completed" ? completed.receipt.workflow.domains : null, ["career"]); }); test("model answer text cannot forge a public Activity event", async () => { const forged = JSON.stringify({ type: "activity", phase: "chart-calculation", label: "模型伪造进度" }); const events = await collectAgentPublicEvents([ { type: "text-delta", payload: { text: forged } }, ], { runId: "run", requestId: "req", toolStatus: () => "ready", receipt: () => ({ runId: "run", runtime: "mastra-agentic", skill: { name: "jyotish-vedic-astrology", loaded: true }, steps: [], workflow: { route: "career", status: "ready", preciseTiming: "blocked", missingLayers: [], domains: ["career"] }, }), }); assert.equal(events.filter((event) => event.type === "activity").length, 0); assert.equal(events.filter((event) => event.type === "answer.delta").length, 1); assert.equal(events.find((event) => event.type === "answer.delta")?.text, forged); }); test("incremental NDJSON parser handles arbitrary chunk boundaries", () => { const parsed: unknown[] = []; const parser = createNdjsonParser((event) => parsed.push(event)); const line = `${JSON.stringify({ type: "run.started", runId: "r", requestId: "q" })}\n`; parser.push(line.slice(0, 7)); parser.push(line.slice(7, 21)); parser.finish(line.slice(21)); assert.deepEqual(parsed, [{ type: "run.started", runId: "r", requestId: "q" }]); }); test("uses a bounded dynamic step budget and reports truncation", () => { const state = createConsultationRuntimeState({ plannedSteps: 1, reservedValidationSteps: 1 }); assert.deepEqual(state.stepBudget, { planned: 1, reservedValidation: 1, total: 2 }); assert.equal(appendConsultationRuntimeStep(state, { kind: "skill", name: "jyotish-vedic-astrology", status: "completed" }), true); assert.equal(appendConsultationRuntimeStep(state, { kind: "validation", name: "ensure-final-response", status: "completed" }), true); assert.equal(appendConsultationRuntimeStep(state, { kind: "tool", name: "unexpected-extra-step", status: "completed" }), false); assert.equal(state.steps.length, 2); assert.equal(state.stepsTruncated, true); assert.deepEqual(consultationStepBudgetReceipt(state), { planned: 2, used: 2, remaining: 0, truncated: true }); }); function receipt(state: ReturnType) { return { runId: "run", runtime: "mastra-agentic" as const, skill: { name: "jyotish-vedic-astrology" as const, loaded: state.jyotishSkillLoaded }, steps: state.steps, stepBudget: consultationStepBudgetReceipt(state), workflow: state.workflowReceipt ?? { route: "career", status: "ready", preciseTiming: "blocked", missingLayers: [] }, }; } test("holds answer text until the Skill and server tool contract completes", async () => { const state = createConsultationRuntimeState(); let completed = 0; async function* chunks() { yield { type: "text-delta", payload: { text: "只在合同完成后显示。" } }; yield { type: "tool-call", payload: { toolCallId: "skill-1", toolName: "skill", args: { name: "jyotish-vedic-astrology" } } }; state.jyotishSkillLoaded = true; yield { type: "tool-result", payload: { toolCallId: "skill-1", toolName: "skill", result: {} } }; yield { type: "tool-call", payload: { toolCallId: "tool-1", toolName: "run-jyotish-consultation", args: {} } }; state.consultationToolCallCount = 1; state.consultationToolSuccessCount = 1; state.consultationToolCompleted = true; state.workflowReceipt = { route: "career", status: "ready", preciseTiming: "blocked", missingLayers: [] }; yield { type: "tool-result", payload: { toolCallId: "tool-1", toolName: "run-jyotish-consultation", result: {} } }; } const response = streamAgentResponse({ runId: "run", requestId: "req", state, stream: chunks(), requireTool: true, toolStatus: () => "ready", receipt: () => receipt(state), onComplete: () => { completed += 1; }, }); const events: unknown[] = []; const parser = createNdjsonParser((event) => events.push(event)); parser.finish(await response.text()); assert.equal(completed, 1); assert.equal(events.filter((event) => (event as { type?: string }).type === "run.completed").length, 1); assert.equal(events.filter((event) => (event as { type?: string }).type === "answer.delta").length, 1); assert.equal((events.find((event) => (event as { type?: string }).type === "answer.delta") as { text?: string }).text, "只在合同完成后显示。"); }); test("a calculation that succeeds only after failed attempts still satisfies the contract", async () => { const state = createConsultationRuntimeState(); let completed = 0; async function* chunks() { yield { type: "tool-call", payload: { toolCallId: "skill-1", toolName: "skill", args: { name: "jyotish-vedic-astrology" } } }; state.jyotishSkillLoaded = true; yield { type: "tool-result", payload: { toolCallId: "skill-1", toolName: "skill", result: {} } }; // Two transient workflow failures, then one success, as observed in production. state.consultationToolCallCount = 3; state.consultationToolSuccessCount = 1; state.consultationToolCompleted = true; state.workflowReceipt = { route: "career", status: "ready", preciseTiming: "blocked", missingLayers: [] }; yield { type: "tool-result", payload: { toolCallId: "tool-3", toolName: "run-jyotish-consultation", result: {} } }; yield { type: "text-delta", payload: { text: "事业方向的判断如下。" } }; } const response = streamAgentResponse({ runId: "run", requestId: "req", state, stream: chunks(), requireTool: true, toolStatus: () => "ready", receipt: () => receipt(state), onComplete: () => { completed += 1; }, }); const events: unknown[] = []; const parser = createNdjsonParser((event) => events.push(event)); parser.finish(await response.text()); assert.equal(completed, 1); assert.equal(events.filter((event) => (event as { type?: string }).type === "run.failed").length, 0); assert.equal(events.filter((event) => (event as { type?: string }).type === "run.completed").length, 1); assert.equal((events.find((event) => (event as { type?: string }).type === "answer.delta") as { text?: string }).text, "事业方向的判断如下。"); }); test("a second successful calculation still fails the single-calculation boundary", async () => { const state = createConsultationRuntimeState(); let failed = 0; async function* chunks() { yield { type: "tool-call", payload: { toolCallId: "skill-1", toolName: "skill", args: { name: "jyotish-vedic-astrology" } } }; state.jyotishSkillLoaded = true; yield { type: "tool-result", payload: { toolCallId: "skill-1", toolName: "skill", result: {} } }; state.consultationToolCallCount = 2; state.consultationToolSuccessCount = 2; state.consultationToolCompleted = true; state.workflowReceipt = { route: "career", status: "ready", preciseTiming: "blocked", missingLayers: [] }; yield { type: "text-delta", payload: { text: "不应显示" } }; } const response = streamAgentResponse({ runId: "run", requestId: "req", state, stream: chunks(), requireTool: true, toolStatus: () => "ready", receipt: () => receipt(state), onError: () => { failed += 1; }, }); const events: unknown[] = []; const parser = createNdjsonParser((event) => events.push(event)); parser.finish(await response.text()); assert.equal(failed, 1); assert.equal(events.some((event) => (event as { type?: string }).type === "answer.delta"), false); assert.equal(events.filter((event) => (event as { type?: string }).type === "run.failed").length, 1); }); test("incomplete runtime contract fails without saving a successful answer", async () => { const state = createConsultationRuntimeState(); let completed = 0; let failed = 0; async function* chunks() { yield { type: "text-delta", payload: { text: "不能保存" } }; } const response = streamAgentResponse({ runId: "run", requestId: "req", state, stream: chunks(), requireTool: true, toolStatus: () => "blocked", receipt: () => receipt(state), onComplete: () => { completed += 1; }, onError: () => { failed += 1; }, }); const events: unknown[] = []; const parser = createNdjsonParser((event) => events.push(event)); parser.finish(await response.text()); assert.equal(completed, 0); assert.equal(failed, 1); assert.equal(events.some((event) => (event as { type?: string }).type === "answer.delta"), false); assert.equal(events.filter((event) => (event as { type?: string }).type === "run.failed").length, 1); }); test("ensures a controlled final response after a successful tool-only run", async () => { const state = createConsultationRuntimeState(); state.jyotishSkillLoaded = true; state.consultationToolCallCount = 1; state.consultationToolSuccessCount = 1; state.consultationToolCompleted = true; state.workflowReceipt = { route: "career", status: "ready", preciseTiming: "blocked", missingLayers: [] }; let completedOutput = ""; async function* chunks() { yield { type: "tool-result", payload: { toolCallId: "tool-1", toolName: "run-jyotish-consultation", result: {} } }; } const response = streamAgentResponse({ runId: "run", requestId: "req", state, stream: chunks(), requireTool: true, toolStatus: () => "ready", receipt: () => receipt(state), onComplete: (output) => { completedOutput = output; }, }); const events: unknown[] = []; const parser = createNdjsonParser((event) => events.push(event)); parser.finish(await response.text()); assert.equal(ensureFinalResponseText("", true), completedOutput); assert.match(completedOutput, /计算已完成/); assert.equal(events.filter((event) => (event as { type?: string }).type === "answer.delta").length, 1); assert.equal(events.filter((event) => (event as { type?: string }).type === "run.completed").length, 1); assert.equal(state.steps.at(-1)?.name, "ensure-final-response"); }); test("a completed run records the finish reason and the authoritative step count", async () => { const state = createConsultationRuntimeState(); state.jyotishSkillLoaded = true; state.consultationToolCallCount = 1; state.consultationToolSuccessCount = 1; state.consultationToolCompleted = true; state.workflowReceipt = { route: "career", status: "ready", preciseTiming: "blocked", missingLayers: [] }; async function* chunks() { yield { type: "step-finish", payload: { stepResult: { reason: "tool-calls" } } }; yield { type: "text-delta", payload: { text: "事业方向的判断如下。" } }; yield { type: "step-finish", payload: { stepResult: { reason: "stop" } } }; // The terminal chunk carries the runtime's own step list, which wins over // the chunks we counted. yield { type: "finish", payload: { stepResult: { reason: "stop" }, output: { usage: {}, steps: [{}, {}, {}] } } }; } const response = streamAgentResponse({ runId: "run", requestId: "req", state, stream: chunks(), requireTool: true, toolStatus: () => "ready", receipt: () => receipt(state), }); await response.text(); assert.equal(state.modelFinishReason, "stop"); assert.equal(state.modelStepCount, 3); }); test("a run that stops while still wanting tools records the exhausted step budget", async () => { const state = createConsultationRuntimeState({ plannedSteps: 8 }); state.jyotishSkillLoaded = true; state.consultationToolCallCount = 1; state.consultationToolSuccessCount = 1; state.consultationToolCompleted = true; state.workflowReceipt = { route: "career", status: "ready", preciseTiming: "blocked", missingLayers: [] }; // The production shape: the calculation succeeded, the model never wrote an // answer, and only the fallback text reached the user. Nothing in the public // event stream said the step budget ran out. async function* chunks() { yield { type: "tool-result", payload: { toolCallId: "tool-1", toolName: "run-jyotish-consultation", result: {} } }; yield { type: "step-finish", payload: { stepResult: { reason: "tool-calls" } } }; yield { type: "step-finish", payload: { stepResult: { reason: "tool-calls" } } }; yield { type: "finish", payload: { stepResult: { reason: "tool-calls" }, output: { usage: {} } } }; } let completedOutput = ""; const response = streamAgentResponse({ runId: "run", requestId: "req", state, stream: chunks(), requireTool: true, toolStatus: () => "ready", receipt: () => receipt(state), onComplete: (output) => { completedOutput = output; }, }); await response.text(); assert.equal(completedOutput, ensureFinalResponseText("", true)); assert.equal(state.modelFinishReason, "tool-calls"); assert.equal(state.modelStepCount, 2); }); test("a retry accumulates model steps and reports the latest finish reason", async () => { const state = createConsultationRuntimeState({ plannedSteps: 8 }); async function* firstAttempt() { yield { type: "step-finish", payload: { stepResult: { reason: "tool-calls" } } }; yield { type: "finish", payload: { stepResult: { reason: "tool-calls" }, output: { usage: {} } } }; } async function* retriedAttempt() { state.jyotishSkillLoaded = true; state.consultationToolCallCount = 1; state.consultationToolSuccessCount = 1; state.consultationToolCompleted = true; state.workflowReceipt = { route: "career", status: "ready", preciseTiming: "blocked", missingLayers: [] }; yield { type: "text-delta", payload: { text: "补齐后的回答。" } }; yield { type: "step-finish", payload: { stepResult: { reason: "stop" } } }; yield { type: "finish", payload: { stepResult: { reason: "unrecognized-provider-reason" }, output: { usage: {} } } }; } const response = streamAgentResponse({ runId: "run", requestId: "req", state, stream: firstAttempt(), requireTool: true, retry: async () => retriedAttempt(), toolStatus: () => "ready", receipt: () => receipt(state), }); await response.text(); assert.equal(state.modelStepCount, 2); assert.equal(state.modelFinishReason, "unknown"); }); test("persistence failure emits run.failed instead of run.completed", async () => { const state = createConsultationRuntimeState(); state.jyotishSkillLoaded = true; state.consultationToolCallCount = 1; state.consultationToolSuccessCount = 1; state.consultationToolCompleted = true; state.workflowReceipt = { route: "career", status: "ready", preciseTiming: "blocked", missingLayers: [] }; let failed = 0; async function* chunks() { yield { type: "text-delta", payload: { text: "不能在持久化失败后标记完成。" } }; } const response = streamAgentResponse({ runId: "run", requestId: "req", state, stream: chunks(), requireTool: true, toolStatus: () => "ready", receipt: () => receipt(state), onComplete: () => { throw new Error("persistence failed"); }, onError: () => { failed += 1; }, }); const events: unknown[] = []; const parser = createNdjsonParser((event) => events.push(event)); parser.finish(await response.text()); assert.equal(failed, 1); assert.equal(events.filter((event) => (event as { type?: string }).type === "run.completed").length, 0); assert.equal(events.filter((event) => (event as { type?: string }).type === "run.failed").length, 1); });