jesse-uxandGrok (agent loop)
a0e223e20c
开流前的注销检查可以和读会话、解析模型同时进行；审核改到扣点前等待，不再挡住第一行；模型版本可以短缓存；两个领域可以一起算。本机没有数据库中位数，也没有带星盘网络库的镜像，所以默认仍是关闭、缓存 0、领域逐个算。 自测：tsc 0 错；lint 0 error；相关测试失败 0、取消 0。test:db 未跑。 Co-Authored-By: Grok (agent loop)
1911 lines
80 KiB
TypeScript
1911 lines
80 KiB
TypeScript
import { NextResponse } from "next/server";
|
|
import { refuseWhenAccountDeletionPending } from "@/lib/account-deletion-server";
|
|
import {
|
|
consultationInputSchema,
|
|
consultationWorkflowReceipt,
|
|
getGeneralJyotishAgent,
|
|
getJyotishAgent,
|
|
getLegacyJyotishAgent,
|
|
getWindowJyotishAgent,
|
|
runConsultationWorkflow,
|
|
} from "@/mastra";
|
|
import { blocksPromptExtraction } from "@/lib/consult-safety";
|
|
import { consultDomainConcurrency, consultPrepParallelEnabled, modelVersionCacheTtlMs } from "@/lib/consult/prep-flags";
|
|
import {
|
|
ConsultationInputBlockedError,
|
|
consultPrepObservabilityEvent,
|
|
type ConsultPrepPhase,
|
|
} from "@/lib/consult/prep-log";
|
|
import { loadConsultTurnHead } from "@/lib/consult/turn-prep";
|
|
import { inputBlockedCopy, moderate, MODERATION_OUTPUT_REPLACED_COPY } from "@/lib/moderation";
|
|
import type { ModerationResult } from "@/lib/moderation";
|
|
import { recordModerationEvent } from "@/lib/moderation/log";
|
|
import { consultationDomainSchema } from "@/lib/consultation-domain-registry";
|
|
import { parseAgentReply } from "@/lib/agent-reply";
|
|
import { revisitQuestionForOffer, sessionHasRectificationOffer } from "@/lib/rectification-in-chat";
|
|
import { decideRectificationOffer, persistOpenedDirectOffer } from "@/lib/rectification-in-chat-server";
|
|
import { getRectificationEntrySummary } from "@/lib/rectification-agentic/v9/case-service";
|
|
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 { classifyConsultationTurn, type SmalltalkUsage } from "@/lib/consultation-smalltalk";
|
|
import { consultationUsageMetadata } from "@/lib/consultation-step-timing";
|
|
import { streamSmalltalkResponse } from "@/lib/stream-smalltalk-response";
|
|
import { streamFirstResponse } from "@/lib/stream-first-response";
|
|
import { consultationPublicActivityEvent, streamAgentResponse } from "@/lib/stream-agent-response";
|
|
import type { AgentExecutionReceipt, WorkflowReceipt } from "@/lib/consultation-agent-events";
|
|
import {
|
|
consultationContinueMessages,
|
|
followUpAnswerShapeInstruction,
|
|
natalAnswerShapeBody,
|
|
hasPriorAssistantAnswer,
|
|
natalUserTurnShape,
|
|
natalConsultationThinkingPlan,
|
|
type PublicThinkingSection,
|
|
RECTIFICATION_OFFER_RULE,
|
|
} from "@/lib/consultation-thinking-plan";
|
|
import {
|
|
AGENT_MAX_STEPS,
|
|
AGENT_TIMEOUT_MS,
|
|
CONSULTATION_ANSWER_TIMEOUT_MS,
|
|
createConsultationRunClock,
|
|
consultationContinueGenerationSettings,
|
|
consultationGenerationSettings,
|
|
consultationNatalPrepareStep,
|
|
consultationWindowPrepareStep,
|
|
createConsultationAgentContext,
|
|
type RectificationOfferHook,
|
|
createWindowConsultationAgentContext,
|
|
precomputeWindowConsultation,
|
|
windowPrecomputedPacketMessage,
|
|
consultationEvidenceCardTelemetry,
|
|
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 {
|
|
ConsultationSubjectError,
|
|
consultationSubjectBindingFromSession,
|
|
consultationSubjectFailureResponse,
|
|
} from "@/lib/consultation-subject-resolver";
|
|
import { CHART_SUBJECT_SELECT } from "@/lib/subject-birth";
|
|
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 = 480;
|
|
|
|
// 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: the tool phase (AGENT_TIMEOUT_MS) plus the answer phase's own
|
|
// clock (CONSULTATION_ANSWER_TIMEOUT_MS), 410s, plus setup and settlement
|
|
// (BUG-1051, BUG-1053, TASK-consult-answer-clock-20261001). Self-hosted `node server.js` does not enforce it; it
|
|
// documents the ceiling.
|
|
|
|
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」。精确应期一律不可用。大运、子运只说「在哪一段附近」，不写起止年月日。这只是粗看。" + RECTIFICATION_OFFER_RULE;
|
|
}
|
|
|
|
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 = item.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) {
|
|
const requestStartedAt = Date.now();
|
|
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 prepParallel = consultPrepParallelEnabled();
|
|
const prepPhases: ConsultPrepPhase[] = [];
|
|
const head = await loadConsultTurnHead({
|
|
parallel: prepParallel,
|
|
getUser: () => supabase.auth.getUser(),
|
|
readPendingDeletion: (userId) => refuseWhenAccountDeletionPending(userId),
|
|
readBody: async () => chatRequestSchema.safeParse(await request.json().catch(() => null)),
|
|
readSession: async (userId, parsed) => await supabase
|
|
.from("chat_sessions")
|
|
.select("id,model_id,model_config_version,session_type,messages,title,theme,chart_profile_id,chart_profile_name,chart_profile_role,context_summary")
|
|
.eq("id", parsed.sessionId)
|
|
.eq("user_id", userId)
|
|
.maybeSingle(),
|
|
resolveModel: async (chatSession) => {
|
|
if (!chatSession.model_id) return null;
|
|
return await resolveSessionLanguageModel(
|
|
chatSession.model_id,
|
|
chatSession.model_config_version,
|
|
);
|
|
},
|
|
});
|
|
if (!head.ok) {
|
|
// A pending account deletion freezes the account: nothing new is charged.
|
|
const frozen = head.frozen ? head.response : null;
|
|
if (frozen) return frozen;
|
|
return head.response;
|
|
}
|
|
const { user, parsed, chatSession, sessionModel } = head;
|
|
prepPhases.push(
|
|
{ phase: "prep.auth", durationMs: head.timings.authMs, status: "completed" },
|
|
{ phase: "prep.deletion", durationMs: head.timings.deletionMs, status: "completed" },
|
|
{ phase: "prep.session", durationMs: head.timings.sessionMs, status: "completed" },
|
|
{ phase: "prep.model", durationMs: head.timings.modelMs, status: "completed" },
|
|
{ phase: prepParallel ? "prep.flag.parallel-on" : "prep.flag.parallel-off", durationMs: 0, status: "completed" },
|
|
{ phase: modelVersionCacheTtlMs() > 0 ? "prep.flag.cache-on" : "prep.flag.cache-off", durationMs: 0, status: "completed" },
|
|
{ phase: consultDomainConcurrency() === 2 ? "prep.flag.domains-2" : "prep.flag.domains-1", durationMs: 0, status: "completed" },
|
|
);
|
|
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 },
|
|
);
|
|
}
|
|
// Content moderation, input side (2026-09-30): before any reservation, so a
|
|
// blocked question is never sent to a model and never charged.
|
|
const moderationContext = { route: "consult", userId: user.id, requestId: parsed.data.requestId, modelId: chatSession.model_id };
|
|
let inputModerationPending: Promise<ModerationResult> | null = null;
|
|
let inputModerationRecorded = false;
|
|
let moderationTimed = false;
|
|
let prepLogged = false;
|
|
const moderationStartedAt = Date.now();
|
|
function logPrep() {
|
|
if (prepLogged) return;
|
|
prepLogged = true;
|
|
logAgentObservability(consultPrepObservabilityEvent({
|
|
requestId,
|
|
sessionId,
|
|
phases: prepPhases,
|
|
}));
|
|
}
|
|
async function settleInputModeration(): Promise<ModerationResult | null> {
|
|
if (!inputModerationPending) return null;
|
|
const result = await inputModerationPending;
|
|
if (!moderationTimed) {
|
|
moderationTimed = true;
|
|
prepPhases.push({
|
|
phase: "prep.moderation",
|
|
durationMs: Math.max(0, Date.now() - moderationStartedAt),
|
|
status: result.verdict === "block" ? "failed" : "completed",
|
|
});
|
|
}
|
|
if (result.verdict !== "pass" && !inputModerationRecorded) {
|
|
inputModerationRecorded = true;
|
|
await recordModerationEvent(accounting, result, "input", moderationContext);
|
|
}
|
|
return result;
|
|
}
|
|
function contentBlockedResponse(result: ModerationResult) {
|
|
return NextResponse.json(
|
|
{ error: "无法处理该请求", code: "content_blocked", message: inputBlockedCopy(result) },
|
|
{ status: 400 },
|
|
);
|
|
}
|
|
if (!prepParallel) {
|
|
const inputModeration = await moderate(visibleQuestion, "input", moderationContext);
|
|
prepPhases.push({
|
|
phase: "prep.moderation",
|
|
durationMs: Math.max(0, Date.now() - moderationStartedAt),
|
|
status: inputModeration.verdict === "block" ? "failed" : "completed",
|
|
});
|
|
if (inputModeration.verdict !== "pass") await recordModerationEvent(accounting, inputModeration, "input", moderationContext);
|
|
if (inputModeration.verdict === "block") {
|
|
logPrep();
|
|
return contentBlockedResponse(inputModeration);
|
|
}
|
|
} else {
|
|
inputModerationPending = moderate(visibleQuestion, "input", moderationContext);
|
|
}
|
|
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>>>;
|
|
|
|
// Everything above answered with a real HTTP status: authentication, the
|
|
// request body, the session and its model. Everything below — subject and
|
|
// profile reads, the credit reservation, appending the question, the model
|
|
// classification, the agent itself — runs inside the response stream in the
|
|
// agentic runtime (BUG-1074): `runTurn` still returns the same Responses it
|
|
// always did, and `streamFirstResponse` relays them after a first frame that
|
|
// goes out before any of this work starts. The legacy runtime keeps the
|
|
// pre-stream order because its text/plain replies cannot ride an NDJSON stream.
|
|
const runTurn = async (): Promise<Response> => {
|
|
let prepared: PreparedConsultationRoute<ModelSelection, ConsultationPlan>;
|
|
const prepareStarted = Date.now();
|
|
try {
|
|
prepared = await prepareConsultationRoute<ModelSelection, ConsultationPlan>({
|
|
userId,
|
|
mode: parsed.data.consultationMode,
|
|
subject: {
|
|
binding: consultationSubjectBindingFromSession(chatSession),
|
|
clientBirth: parsed.data,
|
|
async loadOwnedChartProfile(chartProfileId) {
|
|
const { data, error } = await supabase
|
|
.from("chart_profiles")
|
|
.select(CHART_SUBJECT_SELECT)
|
|
.eq("id", chartProfileId)
|
|
.eq("user_id", userId)
|
|
.maybeSingle();
|
|
if (error) throw new ConsultationSubjectError("subject_unavailable");
|
|
if (!data || typeof data.id !== "string" || typeof data.user_id !== "string" || typeof data.role !== "string") {
|
|
return null;
|
|
}
|
|
const { id, user_id: userIdFromRow, role, ...typed } = data;
|
|
return { id, userId: userIdFromRow, role, profile: typed };
|
|
},
|
|
},
|
|
async loadProfile(profileUserId) {
|
|
const { data, error } = await supabase
|
|
.from("profiles")
|
|
.select("name,birth_date,active_birth_date,active_birth_timezone_offset,active_birth_provenance,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,gender")
|
|
.eq("id", profileUserId)
|
|
.single();
|
|
if (error || !data) throw new ConsultationProfileTruthError("profile_unavailable");
|
|
return data;
|
|
},
|
|
loadCandidateRange: (profileUserId) => loadReportCandidateRange(accounting, { userId: profileUserId }),
|
|
beforeReserve: async ({ consultationMode }) => {
|
|
prepPhases.push({
|
|
phase: "prep.subject",
|
|
durationMs: Math.max(0, Date.now() - prepareStarted),
|
|
status: "completed",
|
|
});
|
|
const inputModeration = await settleInputModeration();
|
|
if (inputModeration?.verdict === "block") throw new ConsultationInputBlockedError(inputModeration);
|
|
return createConsultationPlan({
|
|
userIntent: parsed.data.question,
|
|
theme: consultationTheme,
|
|
consultationMode,
|
|
modelCreditCost: sessionModel.creditCost,
|
|
});
|
|
},
|
|
reserve: async () => {
|
|
const reserveStarted = Date.now();
|
|
try {
|
|
return await 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,
|
|
};
|
|
},
|
|
);
|
|
} finally {
|
|
prepPhases.push({
|
|
phase: "prep.reserve",
|
|
durationMs: Math.max(0, Date.now() - reserveStarted),
|
|
status: "completed",
|
|
});
|
|
}
|
|
},
|
|
});
|
|
} catch (error) {
|
|
const blocked = await settleInputModeration();
|
|
if (blocked?.verdict === "block") {
|
|
logPrep();
|
|
return contentBlockedResponse(blocked);
|
|
}
|
|
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 ConsultationSubjectError) {
|
|
const failure = consultationSubjectFailureResponse(error);
|
|
return NextResponse.json(failure.body, { status: failure.status });
|
|
}
|
|
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 subjectName = prepared.subject?.name
|
|
|| prepared.serverChart?.name
|
|
|| prepared.declaredWindow?.name
|
|
|| "";
|
|
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 };
|
|
const appendStarted = Date.now();
|
|
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;
|
|
});
|
|
prepPhases.push({
|
|
phase: "prep.append",
|
|
durationMs: Math.max(0, Date.now() - appendStarted),
|
|
status: "completed",
|
|
});
|
|
} catch {
|
|
prepPhases.push({
|
|
phase: "prep.append",
|
|
durationMs: Math.max(0, Date.now() - appendStarted),
|
|
status: "failed",
|
|
});
|
|
logPrep();
|
|
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 usageStartedAt = Date.now();
|
|
let classificationUsage: SmalltalkUsage = {};
|
|
let classificationOutcome: string | undefined;
|
|
let classificationDurationMs: number | null = null;
|
|
let latencyState: ReturnType<typeof createConsultationRuntimeState> | null = null;
|
|
const titleStarted = Date.now();
|
|
const titleRead = supabase
|
|
.from("chat_sessions")
|
|
.select("title")
|
|
.eq("id", sessionId)
|
|
.eq("user_id", userId)
|
|
.maybeSingle();
|
|
const classifyStarted = Date.now();
|
|
const turn = parsed.data.entrypoint === undefined
|
|
? await classifyConsultationTurn({
|
|
model: selectedModel,
|
|
question: visibleQuestion,
|
|
history: chatSession.messages,
|
|
name: subjectName,
|
|
signal: request.signal,
|
|
onObservation(observation) {
|
|
if (!observation.late) {
|
|
classificationUsage = observation.usage ?? {};
|
|
classificationOutcome = observation.outcome;
|
|
classificationDurationMs = observation.durationMs;
|
|
}
|
|
const inputTokens = observation.usage?.inputTokens ?? 0;
|
|
const outputTokens = observation.usage?.outputTokens ?? 0;
|
|
logAgentObservability({
|
|
requestId, sessionId,
|
|
agentVersion: "consultation-smalltalk-v1",
|
|
modelVersion: String(selectedModel.configVersion),
|
|
policyVersion: "consultation-smalltalk-v1",
|
|
toolCalls: [],
|
|
classification: { durationMs: observation.durationMs },
|
|
contractPhases: [{
|
|
phase: observation.late ? "classification.late_usage" : `classification.${observation.outcome}`,
|
|
durationMs: observation.durationMs,
|
|
status: observation.outcome === "smalltalk" || observation.outcome === "consult" ? "completed" : "failed",
|
|
}],
|
|
...(observation.usage ? { inputTokens, outputTokens,
|
|
costMicrousd: Math.round((inputTokens * (selectedModel.inputCostMicrousdPerMillion ?? 0)
|
|
+ outputTokens * (selectedModel.outputCostMicrousdPerMillion ?? 0)) / 1_000_000) } : {}),
|
|
billingSettlementResult: "not_applicable",
|
|
});
|
|
},
|
|
})
|
|
: { kind: "consult" as const };
|
|
prepPhases.push({
|
|
phase: "prep.classify",
|
|
durationMs: classificationDurationMs ?? Math.max(0, Date.now() - classifyStarted),
|
|
status: "completed",
|
|
});
|
|
|
|
if (turn.kind === "smalltalk") {
|
|
logPrep();
|
|
return streamSmalltalkResponse({
|
|
requestId,
|
|
reply: turn.reply,
|
|
async complete() {
|
|
const actualUsage = await usagePayload(Promise.resolve({}));
|
|
const completion = await retryDetachedSettlement(async () => {
|
|
const { data, error } = await accounting.rpc("complete_consultation_free", {
|
|
p_user_id: userId,
|
|
p_request_id: requestId,
|
|
p_session_id: sessionId,
|
|
p_response_message: { role: "assistant", text: turn.reply, responseKind: "smalltalk" },
|
|
p_actual_usage: actualUsage,
|
|
});
|
|
const result = consultationCompletionSchema.safeParse(first(data ?? []));
|
|
if (error || !result.success || !result.data.success) {
|
|
throw new CreditRpcError(error?.message || (result.success ? result.data.error_code : "invalid_free_completion") || "free_completion_rejected");
|
|
}
|
|
return result.data;
|
|
});
|
|
logAgentObservability({ requestId, sessionId, agentVersion: "consultation-smalltalk-v1",
|
|
billingSettlementResult: completion.success ? "completed" : "failed" });
|
|
},
|
|
async onError() {
|
|
await cancel();
|
|
logAgentObservability({ requestId, sessionId, agentVersion: "consultation-smalltalk-v1",
|
|
billingSettlementResult: "failed", errorCode: "settlement_failed" });
|
|
},
|
|
});
|
|
}
|
|
|
|
const expectedTitle = typeof chatSession.title === "string" ? chatSession.title : "";
|
|
const { data: titleRow } = await titleRead;
|
|
prepPhases.push({
|
|
phase: "prep.title",
|
|
durationMs: Math.max(0, Date.now() - titleStarted),
|
|
status: "completed",
|
|
});
|
|
logPrep();
|
|
|
|
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 mergeUsage([usage, Promise.resolve(classificationUsage)]);
|
|
const usageRecord = resolved as Record<string, unknown>;
|
|
const cache = resolved.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,
|
|
metadata: {
|
|
...(cache ? { cache: { ...cache, hit: cache.readTokens > 0 } } : {}),
|
|
...(classificationOutcome || classificationDurationMs !== null ? { classification: {
|
|
...(classificationOutcome ? { outcome: classificationOutcome } : {}),
|
|
usageKnown: classificationUsage.inputTokens !== undefined || classificationUsage.outputTokens !== undefined,
|
|
inputTokens: classificationUsage.inputTokens ?? null,
|
|
outputTokens: classificationUsage.outputTokens ?? null,
|
|
durationMs: classificationDurationMs,
|
|
} } : {}),
|
|
...(latencyState?.modelStepTimings === undefined ? {} : consultationUsageMetadata({
|
|
modelSteps: latencyState.modelStepTimings,
|
|
answerReasoningMs: latencyState.answerReasoningMs,
|
|
})),
|
|
},
|
|
};
|
|
}
|
|
|
|
// Content moderation, output side (2026-09-30). The finished answer is
|
|
// checked once; a hit replaces it with a notice and the turn completes free.
|
|
let outputModerated = false;
|
|
let outputBlocked = false;
|
|
async function moderateAnswer(text: string): Promise<string | null> {
|
|
outputModerated = true;
|
|
const verdict = await moderate(text, "output", moderationContext);
|
|
if (verdict.verdict !== "pass") await recordModerationEvent(accounting, verdict, "output", moderationContext);
|
|
if (verdict.verdict !== "block") return null;
|
|
outputBlocked = true;
|
|
return MODERATION_OUTPUT_REPLACED_COPY;
|
|
}
|
|
|
|
async function completeModeratedResponse(usage: Promise<{ inputTokens?: number; outputTokens?: number }>): Promise<AgentSettlementResult> {
|
|
const actualUsage = await usagePayload(usage);
|
|
await retryDetachedSettlement(async () => {
|
|
const { data, error } = await accounting.rpc("complete_consultation_moderated", {
|
|
p_user_id: userId,
|
|
p_request_id: requestId,
|
|
p_session_id: sessionId,
|
|
p_response_message: { role: "assistant", text: MODERATION_OUTPUT_REPLACED_COPY, responseKind: "moderated" },
|
|
p_actual_usage: actualUsage,
|
|
});
|
|
const result = consultationCompletionSchema.safeParse(first(data ?? []));
|
|
if (error || !result.success || !result.data.success) {
|
|
throw new CreditRpcError(error?.message || (result.success ? result.data.error_code : "invalid_moderated_completion") || "moderated_completion_rejected");
|
|
}
|
|
return result.data;
|
|
});
|
|
return "completed";
|
|
}
|
|
|
|
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");
|
|
if (!outputModerated) await moderateAnswer(reply.text);
|
|
if (outputBlocked) return await completeModeratedResponse(usage);
|
|
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 } : {}),
|
|
// BUG-1248: the offer card hangs under this answer, also after a reload.
|
|
...(latencyState?.rectificationOffer ? { rectificationOffer: latencyState.rectificationOffer } : {}),
|
|
};
|
|
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;
|
|
}
|
|
}
|
|
|
|
// Explicit rectification (BUG-1306): one short sentence, same free settlement
|
|
// as smalltalk, so a turn with no reading does not spend a consultation credit.
|
|
// The RPC message is only role/text/responseKind. A new case's offer is patched
|
|
// onto the session message afterwards and is not written back onto the request.
|
|
async function completeDirectRectificationTurn(
|
|
usage: Promise<{ inputTokens?: number; outputTokens?: number }>,
|
|
): Promise<AgentSettlementResult> {
|
|
const entry = latencyState?.rectificationDirectEntry;
|
|
const offer = latencyState?.rectificationOffer;
|
|
if (!entry) throw new Error("direct_entry_missing");
|
|
try {
|
|
const actualUsage = await usagePayload(usage);
|
|
const responseMessage = { role: "assistant" as const, text: entry.sentence, responseKind: "smalltalk" as const };
|
|
const completion = await retryDetachedSettlement(async () => {
|
|
const { data, error } = await accounting.rpc("complete_consultation_free", {
|
|
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_free_completion");
|
|
}
|
|
if (!parsedCompletion.data.success && parsedCompletion.data.error_code !== "request_cancelled") {
|
|
throw new CreditRpcError(parsedCompletion.data.error_code || "free_completion_rejected");
|
|
}
|
|
if (parsedCompletion.data.success && entry.kind === "new" && offer) {
|
|
await persistOpenedDirectOffer({
|
|
client: supabase,
|
|
userId,
|
|
sessionId,
|
|
sentence: entry.sentence,
|
|
offer,
|
|
});
|
|
}
|
|
return parsedCompletion.data;
|
|
});
|
|
if (!completion.success && completion.error_code !== "request_cancelled") {
|
|
throw new CreditRpcError(completion.error_code || "free_completion_rejected");
|
|
}
|
|
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: prepared.subject?.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 });
|
|
latencyState = state;
|
|
// C2/C3 (BUG-1248): the offer tool asks the server, never decides itself.
|
|
const rectificationOffer: RectificationOfferHook = {
|
|
decide: () => decideRectificationOffer({
|
|
accounting,
|
|
userId,
|
|
subjectRole: prepared.subject?.role === "other" ? "other" : "self",
|
|
modelId: selectedModel.id,
|
|
}),
|
|
alreadyOffered: sessionHasRectificationOffer(chatSession?.messages),
|
|
revisitQuestion: revisitQuestionForOffer(chatSession?.messages, visibleQuestion, requestId),
|
|
sourceSessionId: sessionId,
|
|
loadResumableSessionId: async () => {
|
|
const summary = await getRectificationEntrySummary(accounting, userId);
|
|
const caseId = summary.latestResumable?.caseId;
|
|
if (!caseId) return null;
|
|
const { data, error } = await accounting
|
|
.from("agentic_rectification_cases")
|
|
.select("session_id")
|
|
.eq("user_id", userId)
|
|
.eq("id", caseId)
|
|
.maybeSingle();
|
|
if (error || !data) return null;
|
|
const resumableSessionId = (data as { session_id?: unknown }).session_id;
|
|
return typeof resumableSessionId === "string" && resumableSessionId ? resumableSessionId : null;
|
|
},
|
|
};
|
|
const hooks = createConsultationRuntimeHooks(state);
|
|
const usages: Promise<Usage>[] = [];
|
|
const agentStartedAt = Date.now();
|
|
let firstActivityMs = -1;
|
|
let firstTextMs = -1;
|
|
let logged = false;
|
|
// The finished natal answer, read once by the evidence-card record to find
|
|
// which card fields it quoted; never logged. A failed or cut run leaves it
|
|
// empty, so only finished answers count citations.
|
|
let answerForTelemetry = "";
|
|
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),
|
|
...(classificationDurationMs === null ? {} : { classification: { durationMs: classificationDurationMs } }),
|
|
...(errorCode === undefined ? {} : { errorCode }),
|
|
inputTokens,
|
|
outputTokens,
|
|
themeCoverage: state.workflowReceipt?.domains ?? [consultationTheme],
|
|
billingSettlementResult: settlementResult,
|
|
});
|
|
// D7: a separate six-field record, no session or user id in it.
|
|
const evidenceCard = consultationEvidenceCardTelemetry(state, answerForTelemetry);
|
|
if (evidenceCard) {
|
|
logAgentObservability({
|
|
runId: requestId,
|
|
agentVersion: "consultation-evidence-card-v2",
|
|
evidenceCard,
|
|
});
|
|
}
|
|
};
|
|
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，沿用服务器已选定的主题。"
|
|
: "如需新的个人星盘结论，必须调用服务器绑定的排盘工具。";
|
|
// The loop's own step after the calculation writes the answer (BUG-1053),
|
|
// so the answer shape the removed compose prompt used to carry is here.
|
|
// A session that already holds an answer gets the follow-up shape
|
|
// (BUG-1071): a direct answer, no skeleton. That is read from the stored
|
|
// history, never from the wording of the question.
|
|
const followUpTurn = hasPriorAssistantAnswer(history, { summaryText: historyWindow.summaryText });
|
|
const natalInstruction = `${natalToolInstruction}${natalUserTurnShape({ entrypoint: consultEntrypoint, history, summaryText: historyWindow.summaryText })}`;
|
|
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);
|
|
let windowPacketMessage: string | null = null;
|
|
// One run, two clocks (BUG-1053, BUG-1051). Tools and the window precompute
|
|
// keep the tool phase's AGENT_TIMEOUT_MS deadline. The model loop runs on
|
|
// that deadline until the calculation result is in hand, then on the answer
|
|
// clock (CONSULTATION_ANSWER_TIMEOUT_MS), because its next step writes the
|
|
// answer. Continuation and answer retries share the same answer clock.
|
|
// The general / no-birth-minute agent has no tools, so its loop is the
|
|
// answer from the first step and runs on the answer clock from the start.
|
|
const runClock = createConsultationRunClock({
|
|
toolPhaseMs: AGENT_TIMEOUT_MS,
|
|
answerMs: CONSULTATION_ANSWER_TIMEOUT_MS,
|
|
answerReady: () => state.consultationToolCompleted,
|
|
answerFromStart: usesPublicDailyGeneralAgent(consultationMode, generalDailyContext),
|
|
});
|
|
const agentAbortSignal = runClock.toolSignal;
|
|
const streamOptions = {
|
|
runId: requestId,
|
|
maxSteps: AGENT_MAX_STEPS,
|
|
abortSignal: runClock.loopSignal,
|
|
hooks,
|
|
...consultationGenerationSettings(selectedModel.model),
|
|
};
|
|
const natalStreamOptions = {
|
|
...streamOptions,
|
|
prepareStep: (input: { stepNumber: number }) => consultationNatalPrepareStep({
|
|
stepNumber: input.stepNumber,
|
|
rectificationOffer: true,
|
|
}),
|
|
};
|
|
const windowStreamOptions = {
|
|
...streamOptions,
|
|
prepareStep: (input: { stepNumber: number }) => consultationWindowPrepareStep({
|
|
stepNumber: input.stepNumber,
|
|
rectificationOffer: true,
|
|
}),
|
|
};
|
|
const answerPhaseSignal = runClock.answerSignal;
|
|
const startAnswerPhase = () => {
|
|
answerPhaseSignal();
|
|
};
|
|
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);
|
|
if (windowPacketMessage) {
|
|
baseMessages = [...baseMessages, { role: "user" as const, content: windowPacketMessage }];
|
|
}
|
|
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: [] };
|
|
|
|
const settleAnswer = (
|
|
output: string,
|
|
agentExecutionReceipt: AgentExecutionReceipt,
|
|
techniqueTruth: string,
|
|
workflow: WorkflowReceipt,
|
|
thinkingText?: string,
|
|
thinkingSections?: PublicThinkingSection[],
|
|
) => {
|
|
if (state.rectificationDirectEntry) {
|
|
return settleRun(() => completeDirectRectificationTurn(mergeUsage(usages)), undefined);
|
|
}
|
|
return settleRun(() => completeResponse(
|
|
output,
|
|
mergeUsage(usages),
|
|
techniqueTruth,
|
|
workflow,
|
|
agentExecutionReceipt,
|
|
thinkingText,
|
|
thinkingSections,
|
|
), undefined);
|
|
};
|
|
|
|
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 (retryHint?: string) => {
|
|
const retried = await agent.stream([
|
|
...baseMessages,
|
|
{ role: "user" as const, content: `上一轮没有输出任何回答文本。请直接给出这个问题的回答，不要只说明过程。${retryHint ? `\n${retryHint}` : ""}` },
|
|
], { ...streamOptions, abortSignal: answerPhaseSignal() });
|
|
usages.push(retried.totalUsage);
|
|
return retried.fullStream;
|
|
};
|
|
const continueAfterLength = async (output: string) => {
|
|
const continued = await agent.stream(consultationContinueMessages(baseMessages, output), {
|
|
...streamOptions,
|
|
abortSignal: answerPhaseSignal(),
|
|
...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({
|
|
moderateOutput: moderateAnswer,
|
|
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) => settleAnswer(
|
|
output,
|
|
agentExecutionReceipt,
|
|
generalDailyContext ? "public-panchanga-only" : "not-applicable",
|
|
workflowReceipt,
|
|
thinkingText,
|
|
thinkingSections,
|
|
),
|
|
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,
|
|
rectificationOffer,
|
|
});
|
|
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 (retryHint?: string) => {
|
|
const retried = await agent.stream([
|
|
...baseMessages,
|
|
{
|
|
role: "user" as const,
|
|
content: `服务器窗口计算已经完成，但上一轮没有输出任何回答文本。请重新取回本次计算结果，然后直接给出回答；不要只描述过程或工具调用。${retryHint ? `\n${retryHint}` : ""}`,
|
|
},
|
|
], { ...streamOptions, abortSignal: answerPhaseSignal() });
|
|
usages.push(retried.totalUsage);
|
|
return retried.fullStream;
|
|
};
|
|
// A continuation is a new stream and the agent keeps nothing between
|
|
// streams (BUG-1053). The precomputed packet is already in baseMessages;
|
|
// a packet the model fetched itself is not, so it travels along.
|
|
const continueAfterLength = async (output: string, evidence?: unknown) => {
|
|
const continued = await agent.stream(consultationContinueMessages(
|
|
baseMessages,
|
|
output,
|
|
windowPacketMessage ? undefined : evidence,
|
|
), {
|
|
...streamOptions,
|
|
abortSignal: answerPhaseSignal(),
|
|
...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({
|
|
moderateOutput: moderateAnswer,
|
|
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);
|
|
},
|
|
},
|
|
});
|
|
windowPacketMessage = windowPrecomputedPacketMessage(packet);
|
|
baseMessages = [
|
|
...baseMessages,
|
|
{ role: "user" as const, content: windowPacketMessage },
|
|
];
|
|
} 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,
|
|
stepScopedAnswer: true,
|
|
onAnswerPhase: startAnswerPhase,
|
|
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) => settleAnswer(
|
|
output,
|
|
agentExecutionReceipt,
|
|
state.techniqueTruth ?? "declared-window",
|
|
state.workflowReceipt ?? workflowReceipt,
|
|
thinkingText,
|
|
thinkingSections,
|
|
),
|
|
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],
|
|
followUp: followUpTurn,
|
|
});
|
|
const agentContext = createConsultationAgentContext({
|
|
userId,
|
|
sessionId,
|
|
requestId,
|
|
consultationMode: consultationMode === "verified_chart" ? "verified_chart" : "unverified_birth_time",
|
|
plan: prepared.preReserveResult,
|
|
theme: consultationTheme,
|
|
entrypoint: consultEntrypoint,
|
|
followUpTurn,
|
|
serverChart: prepared.serverChart,
|
|
abortSignal: agentAbortSignal,
|
|
state,
|
|
rectificationOffer,
|
|
});
|
|
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 (retryHint?: string) => {
|
|
const retried = await agent.stream([
|
|
...baseMessages,
|
|
{
|
|
role: "user" as const,
|
|
content: `服务器计算已经完成，但上一轮没有输出任何回答文本。请重新取回本次计算结果，然后直接给出回答；不要只描述过程或工具调用。${retryHint ? `\n${retryHint}` : ""}`,
|
|
},
|
|
], { ...natalStreamOptions, abortSignal: answerPhaseSignal() });
|
|
usages.push(retried.totalUsage);
|
|
return retried.fullStream;
|
|
};
|
|
// A continuation is a new stream and the agent keeps nothing between
|
|
// streams: without the calculation result the loop saw, it would continue
|
|
// the answer blind (BUG-1053).
|
|
const continueAfterLength = async (output: string, evidence?: unknown) => {
|
|
const continued = await agent.stream(consultationContinueMessages(baseMessages, output, evidence), {
|
|
...streamOptions,
|
|
abortSignal: answerPhaseSignal(),
|
|
...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 ?? "unknown",
|
|
...(state.techniqueAuditTable?.length
|
|
? { techniqueAuditTable: state.techniqueAuditTable }
|
|
: {}),
|
|
});
|
|
return streamAgentResponse({
|
|
moderateOutput: moderateAnswer,
|
|
runId: requestId,
|
|
requestId,
|
|
sideEvent: titleSideEvent,
|
|
state,
|
|
stream: result.fullStream,
|
|
requireTool: true,
|
|
retry,
|
|
retryForAnswer,
|
|
continueAfterLength,
|
|
stepScopedAnswer: true,
|
|
onAnswerPhase: startAnswerPhase,
|
|
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) => {
|
|
if (!state.rectificationDirectEntry) answerForTelemetry = output;
|
|
return settleAnswer(
|
|
output,
|
|
agentExecutionReceipt,
|
|
state.techniqueTruth ?? "unknown",
|
|
state.workflowReceipt ?? workflowReceipt,
|
|
thinkingText,
|
|
thinkingSections,
|
|
);
|
|
},
|
|
onError: (error) => settleRun(
|
|
cancel,
|
|
toAgentObservabilityErrorCode(error),
|
|
),
|
|
onCancel: () => settleRun(cancel, "cancelled"),
|
|
});
|
|
}
|
|
|
|
try {
|
|
const history = storedHistory;
|
|
const name = subjectName;
|
|
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);
|
|
// The grounded fallback writes from the injected workflow; same first-turn
|
|
// / follow-up split as the agentic path (BUG-1071), one statement of it.
|
|
const legacyNatalInstruction = hasPriorAssistantAnswer(history, { summaryText: historyWindow.summaryText })
|
|
? followUpAnswerShapeInstruction()
|
|
: `${natalAnswerShapeBody()}文末技法审计表由产品界面折叠展示，不写进正文。星盘事实只使用系统里已经注入的计算结果，不要复述内部字段、JSON 或再跑一遍咨询流程。`;
|
|
|
|
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: legacyNatalInstruction,
|
|
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: legacyNatalInstruction,
|
|
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 },
|
|
);
|
|
}
|
|
};
|
|
|
|
if (!shouldUseAgenticRuntime(user)) return runTurn();
|
|
return streamFirstResponse({
|
|
requestId,
|
|
run: runTurn,
|
|
startedAt: requestStartedAt,
|
|
onFirstFrame: (elapsedMs) => {
|
|
logAgentObservability({
|
|
requestId,
|
|
sessionId,
|
|
agentVersion: "consultation-stream-first-v1",
|
|
contractPhases: [{ phase: "first_byte", durationMs: elapsedMs, status: "completed" }],
|
|
billingSettlementResult: "not_applicable",
|
|
});
|
|
},
|
|
onUnhandled: (error) => {
|
|
const reason = error instanceof Error ? error.name : "UnknownError";
|
|
console.error(`[consult] turn failed before a response request=${requestId} reason=${reason}`);
|
|
},
|
|
});
|
|
}
|