1475 lines
60 KiB
TypeScript
1475 lines
60 KiB
TypeScript
import { NextResponse } from "next/server";
|
|
import {
|
|
consultationInputSchema,
|
|
consultationWorkflowReceipt,
|
|
getGeneralJyotishAgent,
|
|
getJyotishAgent,
|
|
getLegacyJyotishAgent,
|
|
getWindowJyotishAgent,
|
|
runConsultationWorkflow,
|
|
} from "@/mastra";
|
|
import { blocksPromptExtraction } from "@/lib/consult-safety";
|
|
import { consultationDomainSchema } from "@/lib/consultation-domain-registry";
|
|
import { parseAgentReply } from "@/lib/agent-reply";
|
|
import { createConsultationReplyMetadata } from "@/lib/consultation-reply-metadata";
|
|
import {
|
|
ConsultationPlanValidationError,
|
|
createConsultationPlan,
|
|
type ConsultationPlan,
|
|
} from "@/lib/consultation-plan";
|
|
import {
|
|
logAgentObservability,
|
|
settlementTelemetryOutcome,
|
|
toAgentObservabilityErrorCode,
|
|
type AgentSettlementResult,
|
|
} from "@/lib/agent-observability";
|
|
import {
|
|
consultationEntrypointSchema,
|
|
pinsConsultationDomains,
|
|
resolveConsultationQuestion,
|
|
shouldLoadGeneralDailyPanchanga,
|
|
} from "@/lib/consultation-entrypoint";
|
|
import { CreditRpcError } from "@/lib/consultation-billing";
|
|
import { cachedHistoryMessage, cachedSystemMessage, mergePromptCacheUsage, promptCacheUsage } from "@/lib/agent-generation-settings";
|
|
import { FeaturePricingError, resolveFeaturePricing } from "@/lib/feature-pricing";
|
|
import { reserveConsultationModel } from "@/lib/consultation-model-selection";
|
|
import { resolveSessionLanguageModel, loadLanguageModelCatalog } from "@/lib/model-catalog";
|
|
import { jsonForSupabaseSetupFailure } from "@/lib/api/service-unavailable";
|
|
import { createAdminSupabaseClient } from "@/lib/supabase/admin";
|
|
import { createServerSupabaseClient } from "@/lib/supabase/server";
|
|
import { streamTextResponse } from "@/lib/stream-text-response";
|
|
import { consultationPublicActivityEvent, streamAgentResponse } from "@/lib/stream-agent-response";
|
|
import type { AgentExecutionReceipt, WorkflowReceipt } from "@/lib/consultation-agent-events";
|
|
import { consultationComposePrompt, consultationContinuePrompt, natalConsultationThinkingPlan, type PublicThinkingSection } from "@/lib/consultation-thinking-plan";
|
|
import {
|
|
AGENT_MAX_STEPS,
|
|
AGENT_TIMEOUT_MS,
|
|
AGENT_SLICE_MAX_STEPS,
|
|
consultationContinueGenerationSettings,
|
|
consultationGenerationSettings,
|
|
consultationNatalPrepareStep,
|
|
consultationWindowPrepareStep,
|
|
createConsultationAgentContext,
|
|
createWindowConsultationAgentContext,
|
|
precomputeWindowConsultation,
|
|
windowPrecomputedPacketMessage,
|
|
consultationModelStepTelemetry,
|
|
consultationStepBudgetReceipt,
|
|
createConsultationRuntimeHooks,
|
|
createConsultationRuntimeState,
|
|
publicConsultationRuntimeSteps,
|
|
} from "@/mastra/consultation-tools";
|
|
import {
|
|
applyBirthTimeModeToWorkflowContext,
|
|
ACCEPTED_RANGE_READING_INSTRUCTION,
|
|
consultationBirthTimeModeSchema,
|
|
shouldRunBirthChartWorkflow,
|
|
shouldRunDeclaredWindowWorkflow,
|
|
type ConsultationBirthTimeMode,
|
|
} from "@/lib/consultation-birth-time-mode";
|
|
import {
|
|
ConsultationProfileTruthError,
|
|
prepareConsultationRoute,
|
|
type PreparedConsultationRoute,
|
|
} from "@/lib/consultation-route-service";
|
|
import { loadReportCandidateRange } from "@/lib/report-candidate-range";
|
|
import {
|
|
loadGeneralDailyPanchangaContext,
|
|
type GeneralDailyPanchangaContext,
|
|
} from "@/lib/general-daily-panchanga";
|
|
import {
|
|
consultationHistoryWindow,
|
|
consultationUserTurnContent,
|
|
isContextOverflowError,
|
|
lastConsultationPair,
|
|
parseSessionContextSummary,
|
|
} from "@/lib/consultation-session-history";
|
|
import {
|
|
checkpointSessionContextSummary,
|
|
generateSessionContextSummaryText,
|
|
} from "@/lib/session-context-summary";
|
|
import { persistGuardedSessionTitle } from "@/lib/session-title-guard";
|
|
import { generateSessionTitle, shouldGenerateSessionTitle } from "@/lib/session-title-agent";
|
|
import { z } from "zod";
|
|
|
|
export const runtime = "nodejs";
|
|
export const maxDuration = 120;
|
|
|
|
// The step budget, the wall-clock budget and the domain cap all bound this same
|
|
// run, so they are declared as one group in @/mastra/consultation-tools with the
|
|
// reasoning that ties them together. maxDuration above is the ceiling they must
|
|
// stay under; raising it here without raising that is meaningless.
|
|
|
|
const chatRequestMetadataSchema = z.object({
|
|
requestId: z.string().uuid(),
|
|
sessionId: z.string().uuid(),
|
|
modelId: z.string().trim().min(1).max(64),
|
|
name: z.string().trim().max(80).optional().default(""),
|
|
history: z
|
|
.array(
|
|
z.object({
|
|
role: z.enum(["user", "assistant"]),
|
|
text: z.string().max(4000),
|
|
}),
|
|
)
|
|
.max(20)
|
|
.optional()
|
|
.default([]),
|
|
});
|
|
|
|
const chartChatRequestSchema = consultationInputSchema.extend({
|
|
...chatRequestMetadataSchema.shape,
|
|
consultationMode: consultationBirthTimeModeSchema.exclude(["general_no_birth_time", "declared_birth_window"])
|
|
.optional()
|
|
.default("verified_chart"),
|
|
entrypoint: consultationEntrypointSchema.optional(),
|
|
}).strict();
|
|
|
|
const generalChatRequestSchema = z.object({
|
|
...chatRequestMetadataSchema.shape,
|
|
consultationMode: z.literal("general_no_birth_time"),
|
|
question: z.string().trim().min(1).max(500),
|
|
theme: consultationDomainSchema,
|
|
entrypoint: consultationEntrypointSchema.exclude(["birth_time_rectification"]).optional(),
|
|
}).strict();
|
|
|
|
const windowChatRequestSchema = z.object({
|
|
...chatRequestMetadataSchema.shape,
|
|
consultationMode: z.literal("declared_birth_window"),
|
|
question: z.string().trim().min(1).max(500),
|
|
theme: consultationDomainSchema,
|
|
entrypoint: consultationEntrypointSchema.exclude(["birth_time_rectification"]).optional(),
|
|
}).strict();
|
|
|
|
const chatRequestSchema = z.union([
|
|
generalChatRequestSchema,
|
|
windowChatRequestSchema,
|
|
chartChatRequestSchema,
|
|
]);
|
|
|
|
const consultationReservationSchema = z.object({
|
|
success: z.boolean(),
|
|
reservation_id: z.string().uuid().nullable(),
|
|
source: z.enum(["subscription", "credits"]).nullable(),
|
|
credits: z.number().int().nullable(),
|
|
subscription_id: z.string().uuid().nullable(),
|
|
reason: z.string().nullable(),
|
|
retry_after_seconds: z.number().int().nullable(),
|
|
});
|
|
|
|
const consultationCompletionSchema = z.object({
|
|
success: z.boolean(),
|
|
credits: z.number().int().nullable(),
|
|
error_code: z.string().nullable().optional(),
|
|
});
|
|
|
|
const consultationQuestionAppendSchema = z.object({
|
|
success: z.boolean(),
|
|
error_code: z.string().nullable().optional(),
|
|
});
|
|
|
|
function first<T>(value: T | T[]): T {
|
|
return Array.isArray(value) ? value[0] : value;
|
|
}
|
|
|
|
const detachedSettlementAttempts = 3;
|
|
// ponytail: Staging MVP ceiling—without a queue/worker, detached settlement gets only three short retries.
|
|
async function retryDetachedSettlement<T>(action: () => Promise<T>): Promise<T> {
|
|
let lastError: unknown;
|
|
for (let attempt = 1; attempt <= detachedSettlementAttempts; attempt += 1) {
|
|
try {
|
|
return await action();
|
|
} catch (error) {
|
|
lastError = error;
|
|
if (attempt < detachedSettlementAttempts) {
|
|
await new Promise((resolve) => setTimeout(resolve, attempt * 150));
|
|
}
|
|
}
|
|
}
|
|
throw lastError;
|
|
}
|
|
|
|
function currentTimeContext(now = new Date()) {
|
|
const chinaTime = new Date(now.getTime() + 8 * 60 * 60 * 1000)
|
|
.toISOString()
|
|
.replace("T", " ")
|
|
.slice(0, 19);
|
|
return `服务端当前时间（权威）：${now.toISOString()}；中国标准时间（UTC+8）：${chinaTime}。涉及“现在、今天、今年、未来几个月”等相对时间时，以此为准。`;
|
|
}
|
|
|
|
function generalDailyContextPrompt(context: GeneralDailyPanchangaContext | null) {
|
|
if (!context) return "";
|
|
return [
|
|
"以下是服务器计算并校验结构后的公共日历证据。只能在其边界内解释，不得补充个人命盘结论。",
|
|
"<public-daily-panchanga>",
|
|
JSON.stringify(context),
|
|
"</public-daily-panchanga>",
|
|
].join("\n");
|
|
}
|
|
|
|
function generalNoMinuteInstruction(hasPublicDaily: boolean) {
|
|
return hasPublicDaily
|
|
? "当前是无出生分钟的公共今日趋势咨询。可依据服务器提供的公共 Panchanga 摘要回答，但不得计算或推断个人星盘；不得补 00:00、时段中点或任何候选分钟。"
|
|
: "当前没有具体出生分钟。不得计算或推断个人星盘，也不得补 00:00、时段中点或任何候选分钟。请直接回答用户的问题：能答的部分照实说，需要出生分钟的部分明确标出限制。不要把整轮对话改成「一般知识或生时校正」二选一，也不要把服务端当前时间当成用户提供的出生时间。";
|
|
}
|
|
|
|
function declaredWindowInstruction() {
|
|
return "当前是声明出生窗口咨询，没有单一出生分钟。如需个人结构结论，必须调用 run-jyotish-window-consultation。探针时刻不是出生时间，不得把 00:00、时段中点、中午或任一探针写成出生分钟。只能根据稳定层作方向性回答；上升或宫位若在窗口内变化，列出可能星座，不得写成「你的上升是 X」。精确应期一律不可用。这只是粗看。";
|
|
}
|
|
|
|
function usesPublicDailyGeneralAgent(
|
|
consultationMode: ConsultationBirthTimeMode,
|
|
generalDailyContext: GeneralDailyPanchangaContext | null,
|
|
) {
|
|
return consultationMode === "general_no_birth_time"
|
|
|| (consultationMode === "declared_birth_window" && Boolean(generalDailyContext));
|
|
}
|
|
|
|
function chinaCalendarDate(now: Date) {
|
|
return new Date(now.getTime() + 8 * 60 * 60 * 1000).toISOString().slice(0, 10);
|
|
}
|
|
|
|
type Usage = {
|
|
inputTokens?: number;
|
|
outputTokens?: number;
|
|
cache?: ReturnType<typeof promptCacheUsage>;
|
|
};
|
|
|
|
function mergeUsage(usages: Promise<Usage>[]): Promise<Usage> {
|
|
return Promise.all(usages).then((items) => items.reduce<Usage>((total, item) => {
|
|
const usage = item && typeof item === "object" ? item as Record<string, unknown> : {};
|
|
const cache = promptCacheUsage(usage);
|
|
return {
|
|
inputTokens: (total.inputTokens ?? 0) + (typeof usage.inputTokens === "number" ? usage.inputTokens : 0),
|
|
outputTokens: (total.outputTokens ?? 0) + (typeof usage.outputTokens === "number" ? usage.outputTokens : 0),
|
|
cache: mergePromptCacheUsage([total.cache, cache]),
|
|
};
|
|
}, {} as Usage));
|
|
}
|
|
|
|
function shouldUseAgenticRuntime(user: { id: string; app_metadata?: Record<string, unknown> }) {
|
|
const mode = process.env.CONSULTATION_AGENTIC_RUNTIME?.trim().toLowerCase() ?? "enabled";
|
|
if (mode === "legacy") return false;
|
|
if (mode === "enabled") return true;
|
|
if (mode !== "canary") return true;
|
|
const ids = new Set((process.env.CONSULTATION_AGENTIC_CANARY_USER_IDS ?? "")
|
|
.split(",").map((value) => value.trim()).filter(Boolean));
|
|
const roles = Array.isArray(user.app_metadata?.roles) ? user.app_metadata.roles : [];
|
|
return ids.has(user.id) || user.app_metadata?.role === "admin" || roles.includes("admin");
|
|
}
|
|
|
|
function workflowStatus(status: string | undefined): "ready" | "degraded" | "blocked" {
|
|
return status === "ready" || status === "degraded" ? status : "blocked";
|
|
}
|
|
|
|
export async function POST(request: Request) {
|
|
let supabase: Awaited<ReturnType<typeof createServerSupabaseClient>>;
|
|
let accounting: ReturnType<typeof createAdminSupabaseClient>;
|
|
try {
|
|
supabase = await createServerSupabaseClient();
|
|
accounting = createAdminSupabaseClient();
|
|
} catch (error) {
|
|
return jsonForSupabaseSetupFailure(error, "POST /api/consult");
|
|
}
|
|
|
|
const {
|
|
data: { user },
|
|
error: authError,
|
|
} = await supabase.auth.getUser();
|
|
if (authError || !user) {
|
|
return NextResponse.json(
|
|
{ error: "请先登录", message: "登录后才能开始咨询。" },
|
|
{ status: 401 },
|
|
);
|
|
}
|
|
|
|
const parsed = chatRequestSchema.safeParse(
|
|
await request.json().catch(() => null),
|
|
);
|
|
if (!parsed.success) {
|
|
return NextResponse.json(
|
|
{ error: "出生资料或问题格式不正确", details: parsed.error.flatten() },
|
|
{ status: 400 },
|
|
);
|
|
}
|
|
|
|
const { data: chatSession, error: chatSessionError } = await supabase
|
|
.from("chat_sessions")
|
|
.select("id,model_id,model_config_version,session_type,messages,title,theme,chart_profile_role,context_summary")
|
|
.eq("id", parsed.data.sessionId)
|
|
.eq("user_id", user.id)
|
|
.maybeSingle();
|
|
if (chatSessionError) {
|
|
return NextResponse.json(
|
|
{ error: "暂时无法读取咨询会话", message: "请稍后重试。" },
|
|
{ status: 503 },
|
|
);
|
|
}
|
|
if (!chatSession) {
|
|
return NextResponse.json(
|
|
{ error: "咨询会话不存在", message: "请重新进入咨询。" },
|
|
{ status: 404 },
|
|
);
|
|
}
|
|
if (chatSession.session_type !== "consultation") {
|
|
return NextResponse.json(
|
|
{
|
|
error: "这是生时校正会话",
|
|
code: "session_not_consultation",
|
|
message: "请在生时校正里继续。",
|
|
},
|
|
{ status: 409 },
|
|
);
|
|
}
|
|
if (!chatSession.model_id || chatSession.model_id !== parsed.data.modelId) {
|
|
return NextResponse.json(
|
|
{ error: "会话模型已经变化", message: "请刷新会话后重新发送，本次不会扣点。" },
|
|
{ status: 409 },
|
|
);
|
|
}
|
|
const sessionModel = await resolveSessionLanguageModel(
|
|
chatSession.model_id,
|
|
chatSession.model_config_version,
|
|
);
|
|
|
|
if (!sessionModel) {
|
|
return NextResponse.json(
|
|
{
|
|
error: "模型暂不可用",
|
|
message: "当前会话绑定的数据库模型配置不可用，请联系管理员，本次不会扣点。",
|
|
},
|
|
{ status: 503 },
|
|
);
|
|
}
|
|
const sessionContextWindow = sessionModel.contextWindow;
|
|
|
|
if (parsed.data.entrypoint === "birth_time_rectification") {
|
|
return NextResponse.json(
|
|
{
|
|
error: "旧版生时校正入口已停用",
|
|
message: "请从首页生时校正卡片开始或继续对话式校正，本次不会扣点。",
|
|
},
|
|
{ status: 409 },
|
|
);
|
|
}
|
|
|
|
const requestTime = new Date();
|
|
const currentDate = chinaCalendarDate(requestTime);
|
|
|
|
const userId = user.id;
|
|
const requestId = parsed.data.requestId;
|
|
const sessionId = parsed.data.sessionId;
|
|
const consultationTheme = parsed.data.theme;
|
|
const visibleQuestion = parsed.data.question;
|
|
|
|
// Client `history` stays in the request schema for old bundles and is not read.
|
|
const contextSummary = parseSessionContextSummary(chatSession.context_summary);
|
|
const historyWindow = consultationHistoryWindow(chatSession.messages, contextSummary, {
|
|
contextWindow: sessionContextWindow,
|
|
});
|
|
const storedHistory = historyWindow.tail;
|
|
const userControlledPrompt = [
|
|
parsed.data.question,
|
|
historyWindow.summaryText,
|
|
...storedHistory.filter((message) => message.role === "user").map((message) => message.text),
|
|
].filter(Boolean).join("\n");
|
|
if (blocksPromptExtraction(userControlledPrompt)) {
|
|
return NextResponse.json(
|
|
{
|
|
error: "无法处理该请求",
|
|
message:
|
|
"我不能提供系统提示词、技能原文或任何密钥。你可以继续询问占星相关问题。",
|
|
},
|
|
{ status: 400 },
|
|
);
|
|
}
|
|
type SelectedModel = NonNullable<Awaited<ReturnType<typeof resolveSessionLanguageModel>>>;
|
|
type ReservationResult = { success: boolean; credits: number | null; error_code: string | null };
|
|
type ModelSelection = Awaited<ReturnType<typeof reserveConsultationModel<SelectedModel, ReservationResult>>>;
|
|
let prepared: PreparedConsultationRoute<ModelSelection, ConsultationPlan>;
|
|
try {
|
|
prepared = await prepareConsultationRoute<ModelSelection, ConsultationPlan>({
|
|
userId,
|
|
mode: parsed.data.consultationMode,
|
|
async loadProfile(profileUserId) {
|
|
const { data, error } = await supabase
|
|
.from("profiles")
|
|
.select("name,birth_date,reported_birth_time,active_birth_time,birth_time_source,birth_time_period,declared_window_start,declared_window_end,birth_time_status,country_code,province_code,city_code,district_code,latitude,longitude,timezone_offset,birth_place_label,birth_place_type,birth_place_provider,birth_place_provider_id,timezone_id,timezone_source,ayanamsa")
|
|
.eq("id", profileUserId)
|
|
.single();
|
|
if (error || !data) throw new ConsultationProfileTruthError("profile_unavailable");
|
|
return data;
|
|
},
|
|
loadCandidateRange: (profileUserId) => loadReportCandidateRange(accounting, { userId: profileUserId }),
|
|
beforeReserve: ({ consultationMode }) => createConsultationPlan({
|
|
userIntent: parsed.data.question,
|
|
theme: consultationTheme,
|
|
consultationMode,
|
|
modelCreditCost: sessionModel.creditCost,
|
|
}),
|
|
reserve: () => reserveConsultationModel(
|
|
chatSession.model_id,
|
|
(modelId) => sessionModel?.id === modelId ? sessionModel : null,
|
|
async (model) => {
|
|
const pricing = await resolveFeaturePricing(accounting, "chat.standard", model.id);
|
|
const { data, error } = await accounting.rpc("reserve_consultation_usage", {
|
|
p_user_id: userId,
|
|
p_request_id: requestId,
|
|
p_session_id: sessionId,
|
|
p_requested_model_id: model.id,
|
|
p_credit_cost: pricing.credit_cost,
|
|
});
|
|
const reservation = consultationReservationSchema.safeParse(first(data ?? []));
|
|
if (error || !reservation.success) {
|
|
throw new CreditRpcError(error?.message || "invalid_reservation_response");
|
|
}
|
|
return {
|
|
success: reservation.data.success,
|
|
credits: reservation.data.credits,
|
|
error_code: reservation.data.reason,
|
|
};
|
|
},
|
|
),
|
|
});
|
|
} catch (error) {
|
|
if (error instanceof FeaturePricingError) {
|
|
console.error(
|
|
`[billing] reservation failed request=${requestId} reason=${error.code}`,
|
|
);
|
|
return NextResponse.json(
|
|
{
|
|
error: "计费配置不可用",
|
|
message: "当前服务的计费配置尚未完成，请联系支持人员，本次不会扣点。",
|
|
code: "pricing_configuration_unavailable",
|
|
},
|
|
{ status: 503 },
|
|
);
|
|
}
|
|
if (error instanceof ConsultationPlanValidationError) {
|
|
return NextResponse.json(
|
|
{
|
|
error: "当前咨询计划不可用",
|
|
message: error.code === "plan_cost_ceiling_exceeded"
|
|
? "当前会话模型超出本次咨询的点数上限，请选择标准模型后重试，本次不会扣点。"
|
|
: "当前咨询模式与回答精度边界不一致，请刷新后重试，本次不会扣点。",
|
|
},
|
|
{ status: 409 },
|
|
);
|
|
}
|
|
if (error instanceof ConsultationProfileTruthError) {
|
|
const modeChanged = error.code === "mode_changed";
|
|
return NextResponse.json(
|
|
modeChanged
|
|
? {
|
|
error: "出生时间状态已经变化",
|
|
message: "请刷新后重新选择使用填报时间、一般咨询或先完成校正，本次不会扣点。",
|
|
}
|
|
: {
|
|
error: "暂时无法核对完整出生资料",
|
|
message: "出生日期、时间来源或出生地点资料不完整或不一致，请重新保存后再试，本次不会扣点。",
|
|
},
|
|
{ status: modeChanged ? 409 : 503 },
|
|
);
|
|
}
|
|
const reason = error instanceof Error ? error.name : "UnknownError";
|
|
console.error(
|
|
`[billing] reservation failed request=${requestId} reason=${reason}`,
|
|
);
|
|
return NextResponse.json(
|
|
{ error: "暂时无法确认咨询点数", message: "请稍后重试。" },
|
|
{ status: 503 },
|
|
);
|
|
}
|
|
|
|
const resolvedQuestion = resolveConsultationQuestion({
|
|
visibleQuestion: parsed.data.question,
|
|
entrypoint: parsed.data.entrypoint,
|
|
currentDate,
|
|
consultationMode: prepared.consultationMode,
|
|
});
|
|
|
|
const modelSelection = prepared.reservation;
|
|
|
|
if (modelSelection.status === "unavailable") {
|
|
return NextResponse.json(
|
|
{
|
|
error: "模型暂不可用",
|
|
message: "请选择其他模型后重新发送，本次不会扣除点数。",
|
|
},
|
|
{ status: 409 },
|
|
);
|
|
}
|
|
|
|
const selectedModel = modelSelection.model;
|
|
const reserveResult = modelSelection.reservation;
|
|
|
|
if (!reserveResult.success) {
|
|
const insufficient = reserveResult.error_code === "insufficient_credits";
|
|
return NextResponse.json(
|
|
{
|
|
error: insufficient ? "咨询点数不足" : "暂时无法扣除咨询点数",
|
|
message: insufficient
|
|
? "请先兑换咨询点数后再继续。"
|
|
: reserveResult.error_code || "请稍后重试。",
|
|
},
|
|
{ status: insufficient ? 402 : 503 },
|
|
);
|
|
}
|
|
|
|
|
|
async function cancel(): Promise<AgentSettlementResult> {
|
|
try {
|
|
const result = await retryDetachedSettlement(async () => {
|
|
const { data, error } = await accounting.rpc("cancel_consultation_credit", {
|
|
p_user_id: userId,
|
|
p_request_id: requestId,
|
|
});
|
|
const parsedCancellation = consultationCompletionSchema.safeParse(first(data ?? []));
|
|
if (error || !parsedCancellation.success) {
|
|
throw new CreditRpcError(error?.message || "invalid_cancellation_response");
|
|
}
|
|
return parsedCancellation.data;
|
|
});
|
|
if (!result.success && result.error_code !== "request_completed") {
|
|
throw new CreditRpcError(result.error_code || "cancellation_rejected");
|
|
}
|
|
return "cancelled";
|
|
} catch (error) {
|
|
const reason = error instanceof Error ? error.name : "UnknownError";
|
|
console.error(
|
|
`[billing] cancellation failed request=${requestId} reason=${reason}`,
|
|
);
|
|
return "failed";
|
|
}
|
|
}
|
|
|
|
let appendedQuestion: { success: boolean; error_code?: string | null };
|
|
try {
|
|
appendedQuestion = await retryDetachedSettlement(async () => {
|
|
const { data, error } = await accounting.rpc("append_consultation_question", {
|
|
p_user_id: userId,
|
|
p_request_id: requestId,
|
|
p_session_id: sessionId,
|
|
p_question_message: { role: "user", text: visibleQuestion },
|
|
});
|
|
const parsedAppend = consultationQuestionAppendSchema.safeParse(first(data ?? []));
|
|
if (error || !parsedAppend.success) {
|
|
throw new CreditRpcError(error?.message || "invalid_question_append_response");
|
|
}
|
|
return parsedAppend.data;
|
|
});
|
|
} catch {
|
|
await cancel();
|
|
return NextResponse.json(
|
|
{ error: "暂时无法保存问题", message: "请稍后重试。" },
|
|
{ status: 503 },
|
|
);
|
|
}
|
|
if (!appendedQuestion.success) {
|
|
await cancel();
|
|
if (appendedQuestion.error_code === "session_full") {
|
|
return NextResponse.json(
|
|
{
|
|
error: "这段对话已写满",
|
|
message: "这段对话已写满，开个新对话继续吧",
|
|
code: "session_full",
|
|
},
|
|
{ status: 409 },
|
|
);
|
|
}
|
|
return NextResponse.json(
|
|
{ error: "暂时无法保存问题", message: "请稍后重试。" },
|
|
{ status: 503 },
|
|
);
|
|
}
|
|
|
|
const expectedTitle = typeof chatSession.title === "string" ? chatSession.title : "";
|
|
const { data: titleRow } = await supabase
|
|
.from("chat_sessions")
|
|
.select("title")
|
|
.eq("id", sessionId)
|
|
.eq("user_id", userId)
|
|
.maybeSingle();
|
|
|
|
const usageStartedAt = Date.now();
|
|
async function checkpointConsultationContext() {
|
|
try {
|
|
const { data: sessionRow, error } = await supabase
|
|
.from("chat_sessions")
|
|
.select("messages, context_summary")
|
|
.eq("id", sessionId)
|
|
.eq("user_id", userId)
|
|
.maybeSingle();
|
|
if (error || !sessionRow) return;
|
|
const catalog = await loadLanguageModelCatalog();
|
|
const summaryModel = catalog.defaultModelId
|
|
? catalog.models.find((model) => model.id === catalog.defaultModelId) ?? selectedModel
|
|
: selectedModel;
|
|
await checkpointSessionContextSummary({
|
|
messages: sessionRow.messages,
|
|
summary: sessionRow.context_summary,
|
|
contextWindow: sessionContextWindow,
|
|
generateText: (prompt, signal) => generateSessionContextSummaryText(summaryModel, prompt, signal),
|
|
update: async (summary, seenUpdatedAt) => {
|
|
let query = supabase.from("chat_sessions")
|
|
.update({ context_summary: summary })
|
|
.eq("id", sessionId)
|
|
.eq("user_id", userId);
|
|
query = seenUpdatedAt
|
|
? query.eq("context_summary->>updatedAt", seenUpdatedAt)
|
|
: query.is("context_summary", null);
|
|
const { data, error: writeError } = await query.select("id");
|
|
if (writeError) throw writeError;
|
|
return Boolean(data?.length);
|
|
},
|
|
});
|
|
} catch (error) {
|
|
console.warn("session context summary failed", error);
|
|
}
|
|
}
|
|
async function usagePayload(usage: Promise<{ inputTokens?: number; outputTokens?: number }>) {
|
|
const resolved = await usage;
|
|
const usageRecord = resolved as Record<string, unknown>;
|
|
const cache = promptCacheUsage(usageRecord);
|
|
const inputTokens = Math.max(0, Math.trunc(typeof usageRecord.inputTokens === "number" ? usageRecord.inputTokens : 0));
|
|
const outputTokens = Math.max(0, Math.trunc(typeof usageRecord.outputTokens === "number" ? usageRecord.outputTokens : 0));
|
|
return {
|
|
eventKey: requestId,
|
|
actualModelId: selectedModel.id,
|
|
modelConfigVersion: selectedModel.configVersion,
|
|
inputTokens,
|
|
outputTokens,
|
|
costMicrousd: Math.round((
|
|
inputTokens * (selectedModel.inputCostMicrousdPerMillion ?? 0)
|
|
+ outputTokens * (selectedModel.outputCostMicrousdPerMillion ?? 0)
|
|
) / 1_000_000),
|
|
durationMs: Date.now() - usageStartedAt,
|
|
...(cache ? { metadata: { cache: { ...cache, hit: cache.readTokens > 0 } } } : {}),
|
|
};
|
|
}
|
|
|
|
async function completeResponse(
|
|
rawTransformedText: string,
|
|
usage: Promise<{ inputTokens?: number; outputTokens?: number }>,
|
|
techniqueTruth: string,
|
|
workflowReceipt: WorkflowReceipt,
|
|
agentExecutionReceipt?: AgentExecutionReceipt,
|
|
thinkingText?: string,
|
|
thinkingSections?: PublicThinkingSection[],
|
|
): Promise<AgentSettlementResult> {
|
|
try {
|
|
const reply = parseAgentReply(
|
|
rawTransformedText,
|
|
createConsultationReplyMetadata({ question: visibleQuestion }),
|
|
);
|
|
if (!reply.text) throw new Error("empty_agent_reply");
|
|
const persistedThinking = thinkingText?.trim().slice(0, 4_000);
|
|
const responseMessage = {
|
|
role: "assistant" as const,
|
|
text: reply.text,
|
|
...(persistedThinking ? { thinkingText: persistedThinking } : {}),
|
|
...(thinkingSections?.length ? { thinkingSections } : {}),
|
|
techniqueTruth,
|
|
workflowReceipt,
|
|
...(agentExecutionReceipt ? { agentExecutionReceipt } : {}),
|
|
};
|
|
const actualUsage = await usagePayload(usage);
|
|
const completion = await retryDetachedSettlement(async () => {
|
|
const { data, error } = await accounting.rpc("complete_consultation_response", {
|
|
p_user_id: userId,
|
|
p_request_id: requestId,
|
|
p_session_id: sessionId,
|
|
p_response_message: responseMessage,
|
|
p_actual_usage: actualUsage,
|
|
});
|
|
const parsedCompletion = consultationCompletionSchema.safeParse(first(data ?? []));
|
|
if (error || !parsedCompletion.success) {
|
|
throw new CreditRpcError(error?.message || "invalid_completion_response");
|
|
}
|
|
return parsedCompletion.data;
|
|
});
|
|
if (!completion.success && completion.error_code !== "request_cancelled") {
|
|
throw new CreditRpcError(completion.error_code || "completion_rejected");
|
|
}
|
|
if (completion.success) {
|
|
void checkpointConsultationContext();
|
|
}
|
|
return "completed";
|
|
} catch (error) {
|
|
await cancel();
|
|
throw error;
|
|
}
|
|
}
|
|
|
|
let settlement: Promise<AgentSettlementResult> | null = null;
|
|
function settleResult(action: () => Promise<AgentSettlementResult>) {
|
|
settlement ??= action();
|
|
return settlement;
|
|
}
|
|
async function settle(action: () => Promise<AgentSettlementResult>): Promise<void> {
|
|
await settleResult(action);
|
|
}
|
|
|
|
// A run that fails before streamAgentResponse exists never reaches its error
|
|
// path, so the closed observability event—the only place the tool failure
|
|
// code, per-step durations and the model finish reason are recorded—would be
|
|
// lost for exactly the runs that failed hardest. The agentic setup publishes
|
|
// its settle-and-log entry point here so the request-level catch below can
|
|
// still emit it.
|
|
const agenticFailure: { report?: (error: unknown) => Promise<void> } = {};
|
|
const titleAfterRpc = typeof titleRow?.title === "string" ? titleRow.title : expectedTitle;
|
|
const titleSideEvent = shouldGenerateSessionTitle({
|
|
title: expectedTitle,
|
|
sessionType: chatSession.session_type,
|
|
}, storedHistory)
|
|
? generateSessionTitle({
|
|
model: selectedModel,
|
|
question: parsed.data.question,
|
|
theme: consultationTheme,
|
|
chartRole: chatSession.chart_profile_role === "other" ? "other" : "self",
|
|
signal: request.signal,
|
|
}).then(async (title) => {
|
|
if (!title) return null;
|
|
await persistGuardedSessionTitle({
|
|
client: supabase,
|
|
sessionId,
|
|
userId,
|
|
expectedTitle: titleAfterRpc,
|
|
title,
|
|
});
|
|
return { type: "session.title" as const, title };
|
|
}).catch((error) => {
|
|
console.warn("session title failed", error);
|
|
return null;
|
|
})
|
|
: Promise.resolve(null);
|
|
|
|
async function runAgenticConsultation(
|
|
consultationMode: ConsultationBirthTimeMode,
|
|
history: Array<{ role: "user" | "assistant"; text: string }>,
|
|
name: string,
|
|
generalDailyContext: GeneralDailyPanchangaContext | null,
|
|
) {
|
|
// The recorded step list has to be able to hold everything the model loop
|
|
// can produce, otherwise a run that exhausts its steps also truncates the
|
|
// evidence of having done so.
|
|
const state = createConsultationRuntimeState({ plannedSteps: AGENT_MAX_STEPS });
|
|
const hooks = createConsultationRuntimeHooks(state);
|
|
const usages: Promise<Usage>[] = [];
|
|
const agentStartedAt = Date.now();
|
|
let firstActivityMs = -1;
|
|
let firstTextMs = -1;
|
|
let logged = false;
|
|
const markFirstActivity = () => { if (firstActivityMs < 0) firstActivityMs = Date.now() - agentStartedAt; };
|
|
const markFirstText = () => { if (firstTextMs < 0) firstTextMs = Date.now() - agentStartedAt; };
|
|
const consultEntrypoint = parsed.success ? parsed.data.entrypoint : undefined;
|
|
const logRun = async (
|
|
errorCode: string | undefined,
|
|
settlementResult: AgentSettlementResult,
|
|
) => {
|
|
if (logged) return;
|
|
logged = true;
|
|
const resolvedUsage: Usage = await mergeUsage(usages).catch(() => ({}));
|
|
const inputTokens = Math.max(0, Math.trunc(resolvedUsage.inputTokens ?? 0));
|
|
const outputTokens = Math.max(0, Math.trunc(resolvedUsage.outputTokens ?? 0));
|
|
const skillStep = state.steps.find((step) => step.kind === "skill");
|
|
const runStatus = errorCode === undefined
|
|
? "completed"
|
|
: errorCode === "cancelled"
|
|
? "cancelled"
|
|
: "failed";
|
|
logAgentObservability({
|
|
runId: requestId,
|
|
requestId,
|
|
sessionId,
|
|
agentVersion: "consultation-agentic-v1",
|
|
modelVersion: String(selectedModel.configVersion),
|
|
policyVersion: "consultation-runtime-contract-v1",
|
|
toolCalls: state.steps
|
|
.filter((step) => step.kind === "tool")
|
|
.map((step) => ({
|
|
name: step.name,
|
|
durationMs: Math.max(0, Math.trunc(step.durationMs ?? 0)),
|
|
status: step.status,
|
|
...(step.failureCode ? { failureCode: step.failureCode } : {}),
|
|
})),
|
|
contractPhases: [
|
|
{
|
|
phase: "skill.load",
|
|
...(skillStep?.durationMs === undefined
|
|
? {}
|
|
: { durationMs: Math.max(0, Math.trunc(skillStep.durationMs)) }),
|
|
status: state.jyotishSkillBound ? "completed" : "failed",
|
|
},
|
|
{
|
|
phase: "answer.first_activity",
|
|
...(firstActivityMs < 0 ? {} : { durationMs: firstActivityMs }),
|
|
status: firstActivityMs < 0 ? "skipped" : "completed",
|
|
},
|
|
{
|
|
phase: "answer.first_output",
|
|
...(firstTextMs < 0 ? {} : { durationMs: firstTextMs }),
|
|
status: firstTextMs < 0 ? "skipped" : "completed",
|
|
},
|
|
{
|
|
phase: "run.total",
|
|
durationMs: Date.now() - agentStartedAt,
|
|
status: runStatus,
|
|
},
|
|
{
|
|
phase: "billing.settled",
|
|
status: settlementResult,
|
|
},
|
|
],
|
|
retryCount: Math.max(0, usages.length - 1),
|
|
...consultationModelStepTelemetry(state),
|
|
...(errorCode === undefined ? {} : { errorCode }),
|
|
inputTokens,
|
|
outputTokens,
|
|
themeCoverage: state.workflowReceipt?.domains ?? [consultationTheme],
|
|
billingSettlementResult: settlementResult,
|
|
});
|
|
};
|
|
const settleRun = async (
|
|
action: () => Promise<AgentSettlementResult>,
|
|
errorCode: string | undefined,
|
|
) => {
|
|
try {
|
|
const actualSettlementResult = await settleResult(action);
|
|
const outcome = settlementTelemetryOutcome(actualSettlementResult, errorCode);
|
|
await logRun(outcome.errorCode, outcome.billingSettlementResult);
|
|
} catch (error) {
|
|
const outcome = settlementTelemetryOutcome("failed", errorCode);
|
|
await logRun(outcome.errorCode, outcome.billingSettlementResult);
|
|
throw error;
|
|
}
|
|
};
|
|
agenticFailure.report = async (error) => {
|
|
try {
|
|
await settleRun(cancel, toAgentObservabilityErrorCode(error));
|
|
} catch {
|
|
// Settlement already reported itself through logRun; the request-level
|
|
// failure response must not depend on it succeeding.
|
|
}
|
|
};
|
|
const cacheBoundary = cachedSystemMessage("【上下文缓存边界】后续内容为本轮请求输入。", selectedModel.model);
|
|
const natalToolInstruction = pinsConsultationDomains(consultEntrypoint)
|
|
? "如需新的个人星盘结论，必须调用服务器绑定的排盘工具。调用时不要填写 domains，沿用服务器已选定的主题。"
|
|
: "如需新的个人星盘结论，必须调用服务器绑定的排盘工具。";
|
|
const natalInstruction = consultEntrypoint === "daily_starlanguage"
|
|
? `${natalToolInstruction}按三节写：今日趋势、适合推进 / 需要避开、一个行动（把技法审计表和「探索性日提示，不是确定预测」放进最后一节）。不要复述内部 JSON 字段。`
|
|
: `${natalToolInstruction}事业/财富/婚恋/家庭先给口语开场，再按四个标题写结论：先回答你的问题、盘里支持这个判断的地方、时间怎么看、这周可以做的一件事。不要把统一参数或技法审计表写进正文。不要复述内部 JSON 字段。`;
|
|
const adoptedRangeNote = consultationMode === "verified_chart"
|
|
&& prepared.serverChart?.truth.birthTimeStatus === "accepted"
|
|
&& prepared.serverChart.toolInput.candidate_range
|
|
? ` ${ACCEPTED_RANGE_READING_INSTRUCTION}`
|
|
: "";
|
|
const modeInstruction = consultationMode === "general_no_birth_time" || (consultationMode === "declared_birth_window" && generalDailyContext)
|
|
? generalNoMinuteInstruction(Boolean(generalDailyContext))
|
|
: consultationMode === "declared_birth_window"
|
|
? declaredWindowInstruction()
|
|
: `${natalInstruction}${adoptedRangeNote}`;
|
|
const consultationBaseMessages = (overflow: boolean) => {
|
|
const tail = overflow ? lastConsultationPair(history) : history;
|
|
const mapped = tail.map((message) => message.role === "user"
|
|
? { role: "user" as const, content: message.text }
|
|
: { role: "assistant" as const, content: message.text });
|
|
const cachedTail = mapped.length > 0
|
|
? [...mapped.slice(0, -1), cachedHistoryMessage(mapped[mapped.length - 1]!, selectedModel.model)]
|
|
: mapped;
|
|
return [
|
|
...(cacheBoundary ? [cacheBoundary] : []),
|
|
...cachedTail,
|
|
{
|
|
role: "user" as const,
|
|
content: consultationUserTurnContent({
|
|
currentTime: currentTimeContext(requestTime),
|
|
name,
|
|
instruction: modeInstruction,
|
|
extra: generalDailyContextPrompt(generalDailyContext),
|
|
summaryText: historyWindow.summaryText,
|
|
droppedCount: historyWindow.droppedCount,
|
|
question: resolvedQuestion.modelQuestion,
|
|
}),
|
|
},
|
|
];
|
|
};
|
|
let baseMessages = consultationBaseMessages(false);
|
|
const agentAbortSignal = AbortSignal.timeout(AGENT_TIMEOUT_MS);
|
|
const streamOptions = {
|
|
runId: requestId,
|
|
maxSteps: AGENT_MAX_STEPS,
|
|
abortSignal: agentAbortSignal,
|
|
hooks,
|
|
...consultationGenerationSettings(selectedModel.model),
|
|
};
|
|
const natalStreamOptions = {
|
|
...streamOptions,
|
|
prepareStep: consultationNatalPrepareStep,
|
|
};
|
|
const windowStreamOptions = {
|
|
...streamOptions,
|
|
prepareStep: consultationWindowPrepareStep,
|
|
};
|
|
async function streamWithOverflowRetry(
|
|
agent: {
|
|
stream: (
|
|
messages: typeof baseMessages,
|
|
options: typeof streamOptions | typeof natalStreamOptions | typeof windowStreamOptions,
|
|
) => Promise<{ fullStream: AsyncIterable<unknown> | ReadableStream<unknown>; totalUsage: Promise<Usage> }>;
|
|
},
|
|
options: typeof streamOptions | typeof natalStreamOptions | typeof windowStreamOptions = streamOptions,
|
|
) {
|
|
try {
|
|
const result = await agent.stream(baseMessages, options);
|
|
usages.push(result.totalUsage);
|
|
return result;
|
|
} catch (error) {
|
|
if (!isContextOverflowError(error)) throw error;
|
|
// Same request, same wait: shrink to summary + last pair. Consultation
|
|
// clients cannot parse rectification `attempt.reset`, so the retry stays
|
|
// server-side and never opens a second user-visible wait.
|
|
baseMessages = consultationBaseMessages(true);
|
|
const overflow = await agent.stream(baseMessages, options);
|
|
usages.push(overflow.totalUsage);
|
|
return overflow;
|
|
}
|
|
}
|
|
const workflowReceipt: WorkflowReceipt = usesPublicDailyGeneralAgent(consultationMode, generalDailyContext)
|
|
? {
|
|
route: generalDailyContext ? "general-daily-panchanga" : "general-no-birth-time",
|
|
status: "ready",
|
|
preciseTiming: "blocked",
|
|
missingLayers: ["birth-minute"],
|
|
}
|
|
: consultationMode === "declared_birth_window"
|
|
? {
|
|
route: "declared-birth-window",
|
|
status: "blocked",
|
|
preciseTiming: "blocked",
|
|
missingLayers: ["birth-minute"],
|
|
}
|
|
: { route: "pending", status: "blocked", preciseTiming: "blocked", missingLayers: [] };
|
|
|
|
if (usesPublicDailyGeneralAgent(consultationMode, generalDailyContext)) {
|
|
state.workflowReceipt = workflowReceipt;
|
|
const agent = getGeneralJyotishAgent(selectedModel);
|
|
const result = await streamWithOverflowRetry(agent);
|
|
// This mode has no calculation to require and no chart method to bind,
|
|
// so there is no contract for a retry to repair.
|
|
const retryForAnswer = async () => {
|
|
const retried = await agent.stream([
|
|
...baseMessages,
|
|
{ role: "user" as const, content: "上一轮没有输出任何回答文本。请直接给出这个问题的回答，不要只说明过程。" },
|
|
], streamOptions);
|
|
usages.push(retried.totalUsage);
|
|
return retried.fullStream;
|
|
};
|
|
const continueAfterLength = async (output: string) => {
|
|
const continued = await agent.stream([
|
|
...baseMessages,
|
|
{ role: "assistant" as const, content: output },
|
|
{ role: "user" as const, content: consultationContinuePrompt(output) },
|
|
], {
|
|
...streamOptions,
|
|
...consultationContinueGenerationSettings(selectedModel.model),
|
|
});
|
|
usages.push(continued.totalUsage);
|
|
return continued.fullStream;
|
|
};
|
|
const executionReceipt = (): AgentExecutionReceipt => ({
|
|
runId: requestId,
|
|
runtime: "mastra-agentic",
|
|
skill: {
|
|
name: "jyotish-vedic-astrology",
|
|
loaded: state.jyotishSkillBound,
|
|
referenceReads: state.skillReferenceReadCount,
|
|
methodologySections: state.methodologySectionCount,
|
|
},
|
|
steps: publicConsultationRuntimeSteps(state),
|
|
stepBudget: consultationStepBudgetReceipt(state),
|
|
workflow: workflowReceipt,
|
|
techniqueTruth: generalDailyContext ? "public-panchanga-only" : "not-applicable",
|
|
...(state.techniqueAuditTable?.length
|
|
? { techniqueAuditTable: state.techniqueAuditTable }
|
|
: {}),
|
|
});
|
|
return streamAgentResponse({
|
|
runId: requestId,
|
|
requestId,
|
|
sideEvent: titleSideEvent,
|
|
state,
|
|
stream: result.fullStream,
|
|
requireTool: false,
|
|
retryForAnswer,
|
|
continueAfterLength,
|
|
continueAfterDisconnect: true,
|
|
pass4Mode: generalDailyContext ? "general_no_birth_time" : consultationMode,
|
|
toolStatus: () => "ready",
|
|
receipt: executionReceipt,
|
|
headers: { "x-jyotish-birth-time-mode": consultationMode },
|
|
onFirstActivity: markFirstActivity,
|
|
onFirstOutput: markFirstText,
|
|
onComplete: (output, agentExecutionReceipt, thinkingText, thinkingSections) => settleRun(() => completeResponse(
|
|
output,
|
|
mergeUsage(usages),
|
|
generalDailyContext ? "public-panchanga-only" : "not-applicable",
|
|
workflowReceipt,
|
|
agentExecutionReceipt,
|
|
thinkingText,
|
|
thinkingSections,
|
|
), undefined),
|
|
onError: (error) => settleRun(
|
|
cancel,
|
|
toAgentObservabilityErrorCode(error),
|
|
),
|
|
onCancel: () => settleRun(cancel, "cancelled"),
|
|
});
|
|
}
|
|
|
|
if (shouldRunDeclaredWindowWorkflow(consultationMode)) {
|
|
if (!prepared.declaredWindow) throw new Error("declared_window_truth_missing");
|
|
const agentContext = createWindowConsultationAgentContext({
|
|
userId,
|
|
sessionId,
|
|
requestId,
|
|
consultationMode: "declared_birth_window",
|
|
plan: prepared.preReserveResult,
|
|
theme: consultationTheme,
|
|
declaredWindow: prepared.declaredWindow,
|
|
abortSignal: agentAbortSignal,
|
|
state,
|
|
});
|
|
const agent = getWindowJyotishAgent(selectedModel, agentContext);
|
|
const retry = async () => {
|
|
const retried = await agent.stream([
|
|
...baseMessages,
|
|
{
|
|
role: "user" as const,
|
|
content: "运行合同不完整：本次尚未取得声明窗口计算结果。请调用 run-jyotish-window-consultation 完成计算，再据此回答；不要在工具参数中添加出生分钟。",
|
|
},
|
|
], windowStreamOptions);
|
|
usages.push(retried.totalUsage);
|
|
return retried.fullStream;
|
|
};
|
|
const retryForAnswer = async () => {
|
|
const retried = await agent.stream([
|
|
...baseMessages,
|
|
{
|
|
role: "user" as const,
|
|
content: "服务器窗口计算已经完成，但上一轮没有输出任何回答文本。请重新取回本次计算结果，然后直接给出回答；不要只描述过程或工具调用。",
|
|
},
|
|
], streamOptions);
|
|
usages.push(retried.totalUsage);
|
|
return retried.fullStream;
|
|
};
|
|
const continueAfterLength = async (output: string) => {
|
|
const continued = await agent.stream([
|
|
...baseMessages,
|
|
{ role: "assistant" as const, content: output },
|
|
{ role: "user" as const, content: consultationContinuePrompt(output) },
|
|
], {
|
|
...streamOptions,
|
|
...consultationContinueGenerationSettings(selectedModel.model),
|
|
});
|
|
usages.push(continued.totalUsage);
|
|
return continued.fullStream;
|
|
};
|
|
const executionReceipt = (): AgentExecutionReceipt => ({
|
|
runId: requestId,
|
|
runtime: "mastra-agentic",
|
|
skill: {
|
|
name: "jyotish-vedic-astrology",
|
|
loaded: state.jyotishSkillBound,
|
|
referenceReads: state.skillReferenceReadCount,
|
|
methodologySections: state.methodologySectionCount,
|
|
},
|
|
steps: publicConsultationRuntimeSteps(state),
|
|
stepBudget: consultationStepBudgetReceipt(state),
|
|
workflow: state.workflowReceipt ?? workflowReceipt,
|
|
techniqueTruth: state.techniqueTruth ?? "declared-window",
|
|
...(state.techniqueAuditTable?.length
|
|
? { techniqueAuditTable: state.techniqueAuditTable }
|
|
: {}),
|
|
});
|
|
return streamAgentResponse({
|
|
runId: requestId,
|
|
requestId,
|
|
sideEvent: titleSideEvent,
|
|
state,
|
|
warmup: async (send) => {
|
|
try {
|
|
const packet = await precomputeWindowConsultation(agentContext, {
|
|
question: resolvedQuestion.modelQuestion,
|
|
writer: {
|
|
custom: async (chunk) => {
|
|
if (!chunk || typeof chunk !== "object") return;
|
|
const value = chunk as { type?: unknown; data?: unknown };
|
|
if (value.type !== "data-jyotish-activity") return;
|
|
const event = consultationPublicActivityEvent(value.data);
|
|
if (event) send(event);
|
|
},
|
|
},
|
|
});
|
|
baseMessages = [
|
|
...baseMessages,
|
|
{ role: "user" as const, content: windowPrecomputedPacketMessage(packet) },
|
|
];
|
|
} catch {
|
|
// Leave the contract red; the model may still call the tool.
|
|
}
|
|
},
|
|
stream: async () => {
|
|
const streamed = await streamWithOverflowRetry(agent, windowStreamOptions);
|
|
return streamed.fullStream;
|
|
},
|
|
requireTool: true,
|
|
retry,
|
|
retryForAnswer,
|
|
continueAfterLength,
|
|
continueAfterDisconnect: true,
|
|
pass4Mode: consultationMode,
|
|
toolStatus: () => workflowStatus(state.workflowReceipt?.status),
|
|
receipt: executionReceipt,
|
|
headers: { "x-jyotish-birth-time-mode": consultationMode },
|
|
onFirstActivity: markFirstActivity,
|
|
onFirstOutput: markFirstText,
|
|
onComplete: (output, agentExecutionReceipt, thinkingText, thinkingSections) => settleRun(() => completeResponse(
|
|
output,
|
|
mergeUsage(usages),
|
|
state.techniqueTruth ?? "declared-window",
|
|
state.workflowReceipt ?? workflowReceipt,
|
|
agentExecutionReceipt,
|
|
thinkingText,
|
|
thinkingSections,
|
|
), undefined),
|
|
onError: (error) => settleRun(
|
|
cancel,
|
|
toAgentObservabilityErrorCode(error),
|
|
),
|
|
onCancel: () => settleRun(cancel, "cancelled"),
|
|
});
|
|
}
|
|
|
|
if (!prepared.serverChart) throw new Error("server_chart_truth_missing");
|
|
state.thinkingPlan = natalConsultationThinkingPlan({
|
|
domains: [consultationTheme],
|
|
});
|
|
const agentContext = createConsultationAgentContext({
|
|
userId,
|
|
sessionId,
|
|
requestId,
|
|
consultationMode: consultationMode === "verified_chart" ? "verified_chart" : "unverified_birth_time",
|
|
plan: prepared.preReserveResult,
|
|
theme: consultationTheme,
|
|
entrypoint: consultEntrypoint,
|
|
serverChart: prepared.serverChart,
|
|
abortSignal: agentAbortSignal,
|
|
state,
|
|
});
|
|
const agent = getJyotishAgent(selectedModel, agentContext);
|
|
const result = await streamWithOverflowRetry(agent, natalStreamOptions);
|
|
const retry = async () => {
|
|
const retried = await agent.stream([
|
|
...baseMessages,
|
|
{
|
|
role: "user" as const,
|
|
content: "运行合同不完整：本次尚未取得服务器计算结果。请调用 run-jyotish-consultation 完成计算，再据此回答；不要在工具参数中添加出生资料。",
|
|
},
|
|
], natalStreamOptions);
|
|
usages.push(retried.totalUsage);
|
|
return retried.fullStream;
|
|
};
|
|
// The tool caches this request's calculation, so this attempt gets the same
|
|
// evidence back without paying for it twice; keeping the tools available is
|
|
// what puts that evidence in front of the model at all.
|
|
const retryForAnswer = async () => {
|
|
const retried = await agent.stream([
|
|
...baseMessages,
|
|
{
|
|
role: "user" as const,
|
|
content: "服务器计算已经完成，但上一轮没有输出任何回答文本。请重新取回本次计算结果，然后直接给出回答；不要只描述过程或工具调用。",
|
|
},
|
|
], natalStreamOptions);
|
|
usages.push(retried.totalUsage);
|
|
return retried.fullStream;
|
|
};
|
|
const continueAfterLength = async (output: string) => {
|
|
const continued = await agent.stream([
|
|
...baseMessages,
|
|
{ role: "assistant" as const, content: output },
|
|
{ role: "user" as const, content: consultationContinuePrompt(output) },
|
|
], {
|
|
...streamOptions,
|
|
...consultationContinueGenerationSettings(selectedModel.model),
|
|
});
|
|
usages.push(continued.totalUsage);
|
|
return continued.fullStream;
|
|
};
|
|
const interpretFindings = async () => (
|
|
(state.thinkingPlan ?? []).map((section) => ({ id: section.id }))
|
|
);
|
|
const composeAnswer = async (_: readonly { id: string; text?: string }[] = [], retryHint?: string) => {
|
|
const composed = await agent.stream([
|
|
...baseMessages,
|
|
{ role: "user" as const, content: `${consultationComposePrompt()}${retryHint ? `\n${retryHint}` : ""}` },
|
|
], {
|
|
...streamOptions,
|
|
maxSteps: AGENT_SLICE_MAX_STEPS,
|
|
toolChoice: "none",
|
|
...consultationContinueGenerationSettings(selectedModel.model),
|
|
});
|
|
usages.push(composed.totalUsage);
|
|
return composed.fullStream;
|
|
};
|
|
const executionReceipt = (): AgentExecutionReceipt => ({
|
|
runId: requestId,
|
|
runtime: "mastra-agentic",
|
|
skill: {
|
|
name: "jyotish-vedic-astrology",
|
|
loaded: state.jyotishSkillBound,
|
|
referenceReads: state.skillReferenceReadCount,
|
|
methodologySections: state.methodologySectionCount,
|
|
},
|
|
steps: publicConsultationRuntimeSteps(state),
|
|
stepBudget: consultationStepBudgetReceipt(state),
|
|
workflow: state.workflowReceipt ?? workflowReceipt,
|
|
techniqueTruth: state.techniqueTruth ?? "unknown",
|
|
...(state.techniqueAuditTable?.length
|
|
? { techniqueAuditTable: state.techniqueAuditTable }
|
|
: {}),
|
|
});
|
|
return streamAgentResponse({
|
|
runId: requestId,
|
|
requestId,
|
|
sideEvent: titleSideEvent,
|
|
state,
|
|
stream: result.fullStream,
|
|
requireTool: true,
|
|
retry,
|
|
retryForAnswer,
|
|
continueAfterLength,
|
|
interpretFindings,
|
|
composeAnswer,
|
|
pass4Mode: consultationMode,
|
|
continueAfterDisconnect: true,
|
|
toolStatus: () => workflowStatus(state.workflowReceipt?.status),
|
|
receipt: executionReceipt,
|
|
headers: { "x-jyotish-birth-time-mode": consultationMode },
|
|
onFirstActivity: markFirstActivity,
|
|
onFirstOutput: markFirstText,
|
|
onComplete: (output, agentExecutionReceipt, thinkingText, thinkingSections) => settleRun(() => completeResponse(
|
|
output,
|
|
mergeUsage(usages),
|
|
state.techniqueTruth ?? "unknown",
|
|
state.workflowReceipt ?? workflowReceipt,
|
|
agentExecutionReceipt,
|
|
thinkingText,
|
|
thinkingSections,
|
|
), undefined),
|
|
onError: (error) => settleRun(
|
|
cancel,
|
|
toAgentObservabilityErrorCode(error),
|
|
),
|
|
onCancel: () => settleRun(cancel, "cancelled"),
|
|
});
|
|
}
|
|
|
|
try {
|
|
const history = storedHistory;
|
|
const name = prepared.serverChart?.name ?? prepared.declaredWindow?.name ?? parsed.data.name;
|
|
const consultationMode: ConsultationBirthTimeMode = prepared.consultationMode;
|
|
const generalDailyContext = shouldLoadGeneralDailyPanchanga({
|
|
consultationMode,
|
|
entrypoint: parsed.data.entrypoint,
|
|
visibleQuestion: parsed.data.question,
|
|
})
|
|
? await loadGeneralDailyPanchangaContext({
|
|
date: currentDate,
|
|
reference: prepared.generalDailyReference,
|
|
})
|
|
: null;
|
|
if (shouldUseAgenticRuntime(user) || shouldRunDeclaredWindowWorkflow(consultationMode)) {
|
|
return await runAgenticConsultation(consultationMode, history, name, generalDailyContext);
|
|
}
|
|
if (!shouldRunBirthChartWorkflow(consultationMode)) {
|
|
const cacheBoundary = cachedSystemMessage("【上下文缓存边界】后续内容为本轮请求输入。", selectedModel.model);
|
|
const result = await getGeneralJyotishAgent(selectedModel).stream([
|
|
...(cacheBoundary ? [cacheBoundary] : []),
|
|
{
|
|
role: "user",
|
|
content: consultationUserTurnContent({
|
|
currentTime: currentTimeContext(requestTime),
|
|
name,
|
|
instruction: generalNoMinuteInstruction(Boolean(generalDailyContext)),
|
|
extra: generalDailyContextPrompt(generalDailyContext),
|
|
summaryText: historyWindow.summaryText,
|
|
droppedCount: historyWindow.droppedCount,
|
|
question: resolvedQuestion.modelQuestion,
|
|
}),
|
|
},
|
|
]);
|
|
const workflowReceipt: WorkflowReceipt = {
|
|
route: generalDailyContext ? "general-daily-panchanga" : "general-no-birth-time",
|
|
status: "ready",
|
|
preciseTiming: "blocked",
|
|
missingLayers: ["birth-minute"],
|
|
};
|
|
const settleErrored = (emitted: boolean, output: string) => settle(
|
|
emitted
|
|
? () => completeResponse(
|
|
output,
|
|
result.totalUsage,
|
|
generalDailyContext ? "public-panchanga-only" : "not-applicable",
|
|
workflowReceipt,
|
|
)
|
|
: cancel,
|
|
);
|
|
return streamTextResponse(result.textStream, {
|
|
mode: "mastra",
|
|
requestId,
|
|
continueAfterDisconnect: true,
|
|
headers: {
|
|
"x-jyotish-workflow-route": workflowReceipt.route,
|
|
"x-jyotish-workflow-status": workflowReceipt.status,
|
|
"x-jyotish-technique-truth": generalDailyContext ? "public-panchanga-only" : "not-applicable",
|
|
"x-jyotish-precise-timing": workflowReceipt.preciseTiming,
|
|
"x-jyotish-missing-layers": workflowReceipt.missingLayers.join(","),
|
|
"x-jyotish-birth-time-mode": consultationMode,
|
|
},
|
|
onComplete: (rawTransformedText) => settle(() => completeResponse(
|
|
rawTransformedText,
|
|
result.totalUsage,
|
|
generalDailyContext ? "public-panchanga-only" : "not-applicable",
|
|
workflowReceipt,
|
|
)),
|
|
onError: (_error, emitted, output: string) => settleErrored(emitted, output),
|
|
onCancel: () => settle(cancel),
|
|
});
|
|
}
|
|
|
|
if (!prepared.serverChart) throw new Error("server_chart_truth_missing");
|
|
const toolInput = consultationInputSchema.parse({
|
|
...prepared.serverChart.toolInput,
|
|
// A user-reported concrete minute is a normal chart calculation. Keep its
|
|
// provenance, but let server evidence—not rectification purchase state—own
|
|
// precise-timing permission. Never reactivate the retired questionnaire.
|
|
entryMode: "direct_chart",
|
|
question: resolvedQuestion.modelQuestion,
|
|
theme: parsed.data.theme,
|
|
});
|
|
const workflowContext = applyBirthTimeModeToWorkflowContext(
|
|
await runConsultationWorkflow(toolInput, {
|
|
foreground: true,
|
|
plan: prepared.preReserveResult,
|
|
}),
|
|
consultationMode,
|
|
{ birthTimeSource: prepared.serverChart.truth.birthTimeSource },
|
|
);
|
|
const workflowReceipt = consultationWorkflowReceipt(workflowContext);
|
|
|
|
const cacheBoundary = cachedSystemMessage("【上下文缓存边界】后续内容为本轮请求输入。", selectedModel.model);
|
|
const legacyHistory = history.map((message) => message.role === "user"
|
|
? { role: "user" as const, content: message.text }
|
|
: { role: "assistant" as const, content: message.text });
|
|
const cachedLegacyHistory = legacyHistory.length > 0
|
|
? [...legacyHistory.slice(0, -1), cachedHistoryMessage(legacyHistory[legacyHistory.length - 1]!, selectedModel.model)]
|
|
: legacyHistory;
|
|
const legacyMessages = [
|
|
...(cacheBoundary ? [cacheBoundary] : []),
|
|
...cachedLegacyHistory,
|
|
{
|
|
role: "user" as const,
|
|
content: consultationUserTurnContent({
|
|
currentTime: currentTimeContext(requestTime),
|
|
name,
|
|
instruction: "先用不超过 400 字口语直接回答下面的问题，不要加标题；形状为：先给反差（表面 A，底下 B，命名成一个格局）、再说清谁在推谁在修、然后「别去应 X 的象，去扮演 Y 的象」、收尾最多三条短行动（破折号短句，每条不超过 20 字）；然后再按 skill Level 2 骨架写：原始结构、六步宫位、Yoga 表、时机、综合、文末技法审计表，最后才是现代生活。骨架不可省略。星盘事实只使用系统里已经注入的计算结果，不要复述内部字段、JSON 或再跑一遍咨询流程。",
|
|
summaryText: historyWindow.summaryText,
|
|
droppedCount: historyWindow.droppedCount,
|
|
question: resolvedQuestion.modelQuestion,
|
|
}),
|
|
},
|
|
];
|
|
const legacyAgent = getLegacyJyotishAgent(selectedModel, workflowContext);
|
|
let result;
|
|
try {
|
|
result = await legacyAgent.stream(legacyMessages);
|
|
} catch (error) {
|
|
if (!isContextOverflowError(error)) throw error;
|
|
const overflowHistory = lastConsultationPair(legacyHistory);
|
|
const overflowCached = overflowHistory.length > 0
|
|
? [...overflowHistory.slice(0, -1), cachedHistoryMessage(overflowHistory[overflowHistory.length - 1]!, selectedModel.model)]
|
|
: overflowHistory;
|
|
result = await legacyAgent.stream([
|
|
...(cacheBoundary ? [cacheBoundary] : []),
|
|
...overflowCached,
|
|
{
|
|
role: "user" as const,
|
|
content: consultationUserTurnContent({
|
|
currentTime: currentTimeContext(requestTime),
|
|
name,
|
|
instruction: "先用不超过 400 字口语直接回答下面的问题，不要加标题；形状为：先给反差（表面 A，底下 B，命名成一个格局）、再说清谁在推谁在修、然后「别去应 X 的象，去扮演 Y 的象」、收尾最多三条短行动（破折号短句，每条不超过 20 字）；然后再按 skill Level 2 骨架写：原始结构、六步宫位、Yoga 表、时机、综合、文末技法审计表，最后才是现代生活。骨架不可省略。星盘事实只使用系统里已经注入的计算结果，不要复述内部字段、JSON 或再跑一遍咨询流程。",
|
|
summaryText: historyWindow.summaryText,
|
|
droppedCount: historyWindow.droppedCount,
|
|
question: resolvedQuestion.modelQuestion,
|
|
}),
|
|
},
|
|
]);
|
|
}
|
|
const responseWorkflowReceipt = {
|
|
route: workflowReceipt.route,
|
|
status: workflowReceipt.status,
|
|
preciseTiming: workflowReceipt.preciseTiming,
|
|
missingLayers: workflowReceipt.missingLayers === "none"
|
|
? []
|
|
: workflowReceipt.missingLayers.split(",").map((item) => item.trim()).filter(Boolean),
|
|
};
|
|
const settleErrored = (emitted: boolean, output: string) => settle(
|
|
emitted
|
|
? () => completeResponse(
|
|
output,
|
|
result.totalUsage,
|
|
workflowReceipt.techniqueTruth,
|
|
responseWorkflowReceipt,
|
|
)
|
|
: cancel,
|
|
);
|
|
return streamTextResponse(result.textStream, {
|
|
mode: "mastra",
|
|
requestId,
|
|
continueAfterDisconnect: true,
|
|
headers: {
|
|
"x-jyotish-workflow-route": workflowReceipt.route,
|
|
"x-jyotish-workflow-status": workflowReceipt.status,
|
|
"x-jyotish-technique-truth": workflowReceipt.techniqueTruth,
|
|
"x-jyotish-precise-timing": workflowReceipt.preciseTiming,
|
|
"x-jyotish-missing-layers": workflowReceipt.missingLayers,
|
|
"x-jyotish-birth-time-mode": consultationMode,
|
|
},
|
|
onComplete: (rawTransformedText) => settle(() => completeResponse(
|
|
rawTransformedText,
|
|
result.totalUsage,
|
|
workflowReceipt.techniqueTruth,
|
|
responseWorkflowReceipt,
|
|
)),
|
|
onError: (_error, emitted, output: string) => settleErrored(emitted, output),
|
|
onCancel: () => settle(cancel),
|
|
});
|
|
} catch (error) {
|
|
if (agenticFailure.report) await agenticFailure.report(error);
|
|
else await cancel();
|
|
const reason = error instanceof Error ? error.name : "UnknownError";
|
|
console.error(
|
|
`[consult] generation failed request=${requestId} model=${selectedModel.id} reason=${reason}`,
|
|
);
|
|
return NextResponse.json(
|
|
{
|
|
error: "暂时无法生成解读",
|
|
message: "咨询服务暂时不可用，请稍后再试。",
|
|
recovery: "稍后重试，或换一个模型继续。",
|
|
},
|
|
{ status: 503 },
|
|
);
|
|
}
|
|
}
|