From 8169bd85f7d4534a80cc8a227148cdd039415ae4 Mon Sep 17 00:00:00 2001 From: Jesse_Chen Date: Mon, 17 Aug 2026 12:44:19 +0800 Subject: [PATCH] feat(consult): classify consultation workflow failures for diagnosis Every workflow fault except abort and timeout collapsed into the single calculation_failed code, and the upstream message was discarded, so a failing run left no evidence of whether the API rejected the call or returned a payload that missed the response contract. Classify failures into a closed vocabulary carried on ConsultationWorkflowError and record it as the failureCode of the runtime step. The observability tool call schema gains one controlled optional field; upstream error text stays out of logs, as that contract requires. Forward request_id to the API so a run can be aligned with its access log. Build public receipts from an explicit allowlist. The internal failure code must not reach the client contract, whose step schema is strict and would otherwise reject a successful run. Co-authored-by: Cursor --- docs/BUG_HISTORY.md | 4 +- frontend/src/app/api/consult/route.ts | 6 +- frontend/src/lib/agent-observability.ts | 2 + frontend/src/mastra/consultation-tools.ts | 27 +++++++- frontend/src/mastra/consultation-workflow.ts | 69 ++++++++++++++++++- .../consultation-agentic-runtime.test.ts | 64 +++++++++++++++++ 6 files changed, 165 insertions(+), 7 deletions(-) diff --git a/docs/BUG_HISTORY.md b/docs/BUG_HISTORY.md index fe03100d..8016a903 100644 --- a/docs/BUG_HISTORY.md +++ b/docs/BUG_HISTORY.md @@ -3601,4 +3601,6 @@ ### 待跟进 -前两次 `calculation_failed` 的服务端原因尚未定位，需要 Python API 日志确认（工作流超时为 90s，两次失败均在 20s 内，可排除超时）。本条修复只保证瞬时失败可恢复，不替代对失败本身的排查。 +前两次 `calculation_failed` 的服务端原因尚未定位（工作流超时为 90s，两次失败均在 20s 内，可排除超时）。本条修复只保证瞬时失败可恢复，不替代对失败本身的排查。 + +排查所需的可观测性已随本批补齐：`calculation_failed` 是 `safeToolError()` 的兜底码，除中止与超时外的一切失败都会被压成它，而上游真实错误文本属于 provider payload，按 `agent-observability.ts` 的封闭契约不得进入日志。因此改为按封闭机器码分类：`runConsultationWorkflow` 抛出带 `code` 的 `ConsultationWorkflowError`，按 HTTP 状态区分 `workflow_rate_limited`(429)、`workflow_bad_request`(400)、`workflow_queue_full`(503)、`workflow_server_error`(5xx) 等，并单独标识 `workflow_contract_invalid`（HTTP 通过但响应未过 `consultationWorkflowResponseSchema`，此种情况 Python 侧日志显示成功，仅凭访问日志无法发现）。该码记入运行步骤的 `failureCode`，经 `agentObservabilityToolCallSchema` 的新增受控可选字段进入观测日志。同时把 `request_id` 透传给 Python API，用于与其访问日志交叉对齐；此前两侧无任何关联标识，只能靠时间戳猜测。公开回执改由 `publicConsultationRuntimeSteps()` 按白名单构建，内部 `failureCode` 不出现在对外契约中——`executionStepSchema` 是 strict，若直接透出会让成功运行在解析回执时报错。 diff --git a/frontend/src/app/api/consult/route.ts b/frontend/src/app/api/consult/route.ts index f2533363..d27923aa 100644 --- a/frontend/src/app/api/consult/route.ts +++ b/frontend/src/app/api/consult/route.ts @@ -39,6 +39,7 @@ import { consultationStepBudgetReceipt, createConsultationRuntimeHooks, createConsultationRuntimeState, + publicConsultationRuntimeSteps, } from "@/mastra/consultation-tools"; import { applyBirthTimeModeToWorkflowContext, @@ -549,6 +550,7 @@ export async function POST(request: Request) { name: step.name, durationMs: Math.max(0, Math.trunc(step.durationMs ?? 0)), status: step.status, + ...(step.failureCode ? { failureCode: step.failureCode } : {}), })), contractPhases: [ { @@ -652,7 +654,7 @@ export async function POST(request: Request) { runId: requestId, runtime: "mastra-agentic", skill: { name: "jyotish-vedic-astrology", loaded: state.jyotishSkillLoaded }, - steps: state.steps, + steps: publicConsultationRuntimeSteps(state), stepBudget: consultationStepBudgetReceipt(state), workflow: workflowReceipt, techniqueTruth: generalDailyContext ? "public-panchanga-only" : "not-applicable", @@ -716,7 +718,7 @@ export async function POST(request: Request) { runId: requestId, runtime: "mastra-agentic", skill: { name: "jyotish-vedic-astrology", loaded: state.jyotishSkillLoaded }, - steps: state.steps, + steps: publicConsultationRuntimeSteps(state), stepBudget: consultationStepBudgetReceipt(state), workflow: state.workflowReceipt ?? workflowReceipt, techniqueTruth: state.techniqueTruth ?? "unknown", diff --git a/frontend/src/lib/agent-observability.ts b/frontend/src/lib/agent-observability.ts index 421e97e8..f880cebb 100644 --- a/frontend/src/lib/agent-observability.ts +++ b/frontend/src/lib/agent-observability.ts @@ -72,6 +72,8 @@ export const agentObservabilityToolCallSchema = z.object({ name: machineCodeSchema, durationMs: durationMsSchema, status: z.enum(agentObservabilityStepStatuses), + // Closed classification of why the call failed. Never upstream error text. + failureCode: machineCodeSchema.optional(), }).strict().readonly(); export const agentObservabilityContractPhaseSchema = z.object({ diff --git a/frontend/src/mastra/consultation-tools.ts b/frontend/src/mastra/consultation-tools.ts index fe2d9572..7a8741bb 100644 --- a/frontend/src/mastra/consultation-tools.ts +++ b/frontend/src/mastra/consultation-tools.ts @@ -10,6 +10,7 @@ import { createConsultationPlan, type ConsultationPlan } from "../lib/consultati import type { WorkflowReceipt } from "../lib/consultation-agent-events.ts"; import { consultationInputSchema, + consultationWorkflowFailureCode, consultationWorkflowReceipt, runConsultationWorkflow, toAgentConsultationContext, @@ -40,6 +41,7 @@ export type ConsultationRuntimeStep = { name: string; status: "completed" | "failed"; durationMs?: number; + failureCode?: string; }; export type ConsultationRuntimeState = { @@ -73,6 +75,21 @@ export function createConsultationRuntimeState(options: { plannedSteps?: number; }; } +/** + * Public receipts carry only the fields the client contract allows. Building + * the list from an explicit allowlist keeps internal diagnostics, such as the + * workflow failure classification, from reaching the response. + */ +export function publicConsultationRuntimeSteps(state: ConsultationRuntimeState) { + return state.steps.map((step) => ({ + sequence: step.sequence, + kind: step.kind, + name: step.name, + status: step.status, + ...(step.durationMs === undefined ? {} : { durationMs: step.durationMs }), + })); +} + export function appendConsultationRuntimeStep(state: ConsultationRuntimeState, step: Omit) { if (state.steps.length >= state.stepBudget.total) { state.stepsTruncated = true; @@ -226,6 +243,7 @@ export function createConsultationTools(ctx: ConsultationAgentContext) { foreground: true, signal: context.abortSignal ?? ctx.abortSignal, plan: domainPlan, + requestId: ctx.requestId, }); const guarded = applyBirthTimeModeToWorkflowContext(workflow, ctx.consultationMode); const agentContext = toAgentConsultationContext(guarded); @@ -249,7 +267,14 @@ export function createConsultationTools(ctx: ConsultationAgentContext) { return toModelDomainPlanContext(executions); } catch (error) { ctx.state.consultationToolDurationMs = Date.now() - startedAt; - appendConsultationRuntimeStep(ctx.state, { kind: "tool", name: "run-jyotish-consultation", status: "failed", durationMs: ctx.state.consultationToolDurationMs }); + const failureCode = consultationWorkflowFailureCode(error); + appendConsultationRuntimeStep(ctx.state, { + kind: "tool", + name: "run-jyotish-consultation", + status: "failed", + durationMs: ctx.state.consultationToolDurationMs, + ...(failureCode ? { failureCode } : {}), + }); throw error; } })(); diff --git a/frontend/src/mastra/consultation-workflow.ts b/frontend/src/mastra/consultation-workflow.ts index bc617784..81a1d08c 100644 --- a/frontend/src/mastra/consultation-workflow.ts +++ b/frontend/src/mastra/consultation-workflow.ts @@ -45,9 +45,60 @@ function record(value: unknown): JsonRecord { const apiBase = process.env.JYOTISH_API_BASE ?? "http://127.0.0.1:5200"; +/** + * Closed vocabulary for why a consultation workflow call failed. + * + * The upstream error text is a provider payload and must not reach logs, so + * the failure is classified here into a machine code that observability can + * record without carrying free-form content. + */ +export const consultationWorkflowFailureCodes = [ + "workflow_bad_request", + "workflow_forbidden", + "workflow_unsupported_media_type", + "workflow_rate_limited", + "workflow_queue_full", + "workflow_server_error", + "workflow_http_error", + "workflow_empty_response", + "workflow_contract_invalid", + "workflow_timeout", + "workflow_aborted", + "workflow_unreachable", +] as const; + +export type ConsultationWorkflowFailureCode = typeof consultationWorkflowFailureCodes[number]; + +export class ConsultationWorkflowError extends Error { + readonly code: ConsultationWorkflowFailureCode; + + constructor(code: ConsultationWorkflowFailureCode, message: string) { + super(message); + this.name = "ConsultationWorkflowError"; + this.code = code; + } +} + +function httpFailureCode(status: number): ConsultationWorkflowFailureCode { + if (status === 400) return "workflow_bad_request"; + if (status === 403) return "workflow_forbidden"; + if (status === 415) return "workflow_unsupported_media_type"; + if (status === 429) return "workflow_rate_limited"; + if (status === 503) return "workflow_queue_full"; + if (status >= 500) return "workflow_server_error"; + return "workflow_http_error"; +} + +export function consultationWorkflowFailureCode(error: unknown): ConsultationWorkflowFailureCode | undefined { + if (error instanceof ConsultationWorkflowError) return error.code; + if (error instanceof DOMException && error.name === "TimeoutError") return "workflow_timeout"; + if (error instanceof DOMException && error.name === "AbortError") return "workflow_aborted"; + return undefined; +} + export async function runConsultationWorkflow( input: ConsultationInput, - options?: { foreground?: boolean; signal?: AbortSignal; plan?: ConsultationPlan }, + options?: { foreground?: boolean; signal?: AbortSignal; plan?: ConsultationPlan; requestId?: string }, ) { const { entryMode, question, theme, ...workflowInput } = input; const plan = options?.plan ?? createConsultationPlan({ userIntent: question, theme }); @@ -59,6 +110,8 @@ export async function runConsultationWorkflow( headers: { "content-type": "application/json" }, body: JSON.stringify({ ...workflowInput, + // Correlates this call with the API access log when a run fails. + ...(options?.requestId ? { request_id: options.requestId } : {}), entry_mode: entryMode, question: workflowRequest.question, question_text: workflowRequest.question, @@ -77,9 +130,19 @@ export async function runConsultationWorkflow( signal, }); const data = await response.json().catch(() => null); - if (!response.ok || !data) throw new Error(data?.error || data?.message || `Jyotish API returned ${response.status}`); + if (!response.ok || !data) { + throw new ConsultationWorkflowError( + response.ok ? "workflow_empty_response" : httpFailureCode(response.status), + data?.error || data?.message || `Jyotish API returned ${response.status}`, + ); + } const parsed = consultationWorkflowResponseSchema.safeParse(data); - if (!parsed.success) throw new Error("Jyotish API returned an incomplete consultation contract"); + if (!parsed.success) { + throw new ConsultationWorkflowError( + "workflow_contract_invalid", + "Jyotish API returned an incomplete consultation contract", + ); + } return parsed.data; } diff --git a/frontend/tests/consultation-agentic-runtime.test.ts b/frontend/tests/consultation-agentic-runtime.test.ts index cc8eb3ae..e2cffbb4 100644 --- a/frontend/tests/consultation-agentic-runtime.test.ts +++ b/frontend/tests/consultation-agentic-runtime.test.ts @@ -5,7 +5,13 @@ import { 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"; @@ -212,6 +218,64 @@ test("a rejected workflow promise is cleared before a later tool call", async () 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!({ question: "队列满时的表现", theme: "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("personal Agent exposes the Jyotish Skill and named server tool", async () => { const state = createConsultationRuntimeState(); const agent = getJyotishAgent({