merge: sync latest staging into rectification v9

# Conflicts:
#	docs/BUG_HISTORY.md

frontend/src/app/api/admin/customers/reset/route.ts

@@ -0,0 +1,73 @@
+import { NextResponse } from "next/server";
+import { z } from "zod";
+
+import { isPostgresError, queryAdminRows } from "@/lib/admin/database";
+import {
+  adminErrorResponse,
+  invalidQueryResponse,
+  requestId,
+  requireHighRiskAdminMutation,
+} from "@/lib/admin/http";
+
+export const runtime = "nodejs";
+
+const resetSchema = z.object({
+  userId: z.string().uuid(),
+  confirmation: z.literal("RESET"),
+  reason: z.string().trim().min(1).max(500),
+});
+
+type ResetRow = {
+  user_id: string;
+  email: string;
+  credits: number;
+  chat_sessions_deleted: number;
+  chart_profiles_deleted: number;
+  synastry_reports_deleted: number;
+};
+
+export async function POST(request: Request) {
+  try {
+    const session = await requireHighRiskAdminMutation(
+      request,
+      "admin.users.manage_roles",
+    );
+    const parsed = resetSchema.safeParse(await request.json().catch(() => null));
+    if (!parsed.success) return invalidQueryResponse(parsed.error.flatten());
+
+    const rows = await queryAdminRows<ResetRow>(
+      "select * from public.admin_reset_customer_account($1, $2, $3, $4)",
+      [
+        session.user.id,
+        parsed.data.userId,
+        parsed.data.reason,
+        requestId(request),
+      ],
+    );
+    const row = rows[0];
+    if (!row) return NextResponse.json({ error: "用户不存在" }, { status: 404 });
+
+    return NextResponse.json({
+      data: {
+        userId: row.user_id,
+        email: row.email,
+        credits: row.credits,
+        deleted: {
+          chatSessions: row.chat_sessions_deleted,
+          chartProfiles: row.chart_profiles_deleted,
+          synastryReports: row.synastry_reports_deleted,
+        },
+      },
+    });
+  } catch (error) {
+    if (
+      isPostgresError(error)
+      && error.code === "P0002"
+      && error instanceof Error
+      && error.message.includes("admin_customer_not_found_or_identity_bridge_mismatch")
+    ) {
+      return NextResponse.json({ error: "用户不存在或账号数据不完整" }, { status: 404 });
+    }
+    return adminErrorResponse(error);
+  }
+}

frontend/src/app/api/consult/route.ts

@@ -4,6 +4,7 @@ import {
   consultationWorkflowReceipt,
   getGeneralJyotishAgent,
   getJyotishAgent,
+  getLegacyJyotishAgent,
   runConsultationWorkflow,
 } from "@/mastra";
 import { blocksPromptExtraction } from "@/lib/consult-safety";
@@ -18,6 +19,13 @@ import { resolveSessionLanguageModel } from "@/lib/model-catalog";
 import { createAdminSupabaseClient } from "@/lib/supabase/admin";
 import { createServerSupabaseClient } from "@/lib/supabase/server";
 import { streamTextResponse } from "@/lib/stream-text-response";
+import { streamAgentResponse } from "@/lib/stream-agent-response";
+import type { AgentExecutionReceipt, WorkflowReceipt } from "@/lib/consultation-agent-events";
+import {
+  createConsultationAgentContext,
+  createConsultationRuntimeHooks,
+  createConsultationRuntimeState,
+} from "@/mastra/consultation-tools";
 import {
   applyBirthTimeModeToWorkflowContext,
   consultationBirthTimeModeSchema,
@@ -32,7 +40,7 @@ import {
 import { z } from "zod";
 
 export const runtime = "nodejs";
-export const maxDuration = 60;
+export const maxDuration = 120;
 
 const chatRequestMetadataSchema = z.object({
   requestId: z.string().uuid(),
@@ -117,6 +125,29 @@ function chinaCalendarDate(now: Date) {
   return new Date(now.getTime() + 8 * 60 * 60 * 1000).toISOString().slice(0, 10);
 }
 
+type Usage = { inputTokens?: number; outputTokens?: number };
+
+function mergeUsage(usages: Promise<Usage>[]): Promise<Usage> {
+  return Promise.all(usages).then((items) => items.reduce((total, item) => ({
+    inputTokens: (total.inputTokens ?? 0) + (item.inputTokens ?? 0),
+    outputTokens: (total.outputTokens ?? 0) + (item.outputTokens ?? 0),
+  }), {} as Usage));
+}
+
+function shouldUseAgenticRuntime(user: { id: string; app_metadata?: Record<string, unknown> }) {
+  const mode = process.env.CONSULTATION_AGENTIC_RUNTIME?.trim().toLowerCase() ?? "legacy";
+  if (mode === "enabled") return true;
+  if (mode !== "canary") return false;
+  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>;
@@ -228,7 +259,10 @@ export async function POST(request: Request) {
       { status: 400 },
     );
   }
-  let prepared;
+  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: Awaited<ReturnType<typeof prepareConsultationRoute<ModelSelection>>>;
   try {
     prepared = await prepareConsultationRoute({
       userId,
@@ -367,12 +401,8 @@ export async function POST(request: Request) {
     rawTransformedText: string,
     usage: Promise<{ inputTokens?: number; outputTokens?: number }>,
     techniqueTruth: string,
-    workflowReceipt: {
-      route: string;
-      status: string;
-      preciseTiming: string;
-      missingLayers: readonly string[];
-    },
+    workflowReceipt: WorkflowReceipt,
+    agentExecutionReceipt?: AgentExecutionReceipt,
   ) {
     try {
       const reply = parseAgentReply(rawTransformedText, consultationTheme);
@@ -383,6 +413,7 @@ export async function POST(request: Request) {
         suggestions: reply.suggestions,
         techniqueTruth,
         workflowReceipt,
+        ...(agentExecutionReceipt ? { agentExecutionReceipt } : {}),
       };
       const actualUsage = await usagePayload(usage);
       const completion = await retryDetachedSettlement(async () => {
@@ -414,10 +445,199 @@ export async function POST(request: Request) {
     return settlement;
   }
 
+  async function runAgenticConsultation(
+    consultationMode: ConsultationBirthTimeMode,
+    history: Array<{ role: "user" | "assistant"; text: string }>,
+    name: string,
+  ) {
+    const state = createConsultationRuntimeState();
+    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 logRun = (finishReason: string, settlementResult: string) => {
+      if (logged) return;
+      logged = true;
+      console.info([
+        "[consult-agentic]",
+        `request_id=${requestId}`,
+        `run_id=${requestId}`,
+        `session_id=${sessionId}`,
+        `model_id=${selectedModel.id}`,
+        `skill_loaded=${state.jyotishSkillLoaded}`,
+        `skill_reference_read_count=${state.skillReferenceReadCount}`,
+        `consultation_tool_call_count=${state.consultationToolCallCount}`,
+        `consultation_tool_duration_ms=${state.consultationToolDurationMs ?? -1}`,
+        `time_to_first_activity_ms=${firstActivityMs}`,
+        `time_to_first_text_ms=${firstTextMs}`,
+        `total_duration_ms=${Date.now() - agentStartedAt}`,
+        `finish_reason=${finishReason}`,
+        `settlement_result=${settlementResult}`,
+      ].join(" "));
+    };
+    const settleRun = async (action: () => Promise<void>, finishReason: string, settlementResult: string) => {
+      try {
+        await settle(action);
+        logRun(finishReason, settlementResult);
+      } catch (error) {
+        logRun("settlement_failed", "failed");
+        throw error;
+      }
+    };
+    const baseMessages = [
+      ...history.map((message) => message.role === "user"
+        ? { role: "user" as const, content: message.text }
+        : { role: "assistant" as const, content: message.text }),
+      {
+        role: "user" as const,
+        content: [
+          currentTimeContext(requestTime),
+          name ? `用户称呼：${name}` : "",
+          consultationMode === "general_no_birth_time"
+            ? "当前是用户明确选择的无出生分钟一般咨询。不得计算或推断个人星盘；不得补 00:00、时段中点或任何候选分钟。"
+            : "先加载 Jyotish Skill；如需新的个人星盘结论，必须调用服务器绑定的排盘工具。",
+          resolvedQuestion.modelQuestion,
+        ].filter(Boolean).join("\n"),
+      },
+    ];
+    const agentAbortSignal = AbortSignal.timeout(110_000);
+    const streamOptions = {
+      runId: requestId,
+      maxSteps: 6,
+      abortSignal: agentAbortSignal,
+      hooks,
+    };
+    const workflowReceipt: WorkflowReceipt = consultationMode === "general_no_birth_time"
+      ? { route: "general-no-birth-time", status: "ready", preciseTiming: "blocked", missingLayers: ["birth-minute"] }
+      : { route: "pending", status: "blocked", preciseTiming: "blocked", missingLayers: [] };
+
+    if (consultationMode === "general_no_birth_time") {
+      state.workflowReceipt = workflowReceipt;
+      const agent = getGeneralJyotishAgent(selectedModel);
+      const result = await agent.stream(baseMessages, streamOptions);
+      usages.push(result.totalUsage);
+      const retry = async () => {
+        const retried = await agent.stream([
+          ...baseMessages,
+          { role: "user" as const, content: "运行合同不完整：请先调用 skill({ name: \"jyotish-vedic-astrology\" }) 加载方法，再回答问题。" },
+        ], streamOptions);
+        usages.push(retried.totalUsage);
+        return retried.fullStream;
+      };
+      const executionReceipt = (): AgentExecutionReceipt => ({
+        runId: requestId,
+        runtime: "mastra-agentic",
+        skill: { name: "jyotish-vedic-astrology", loaded: state.jyotishSkillLoaded },
+        steps: state.steps,
+        workflow: workflowReceipt,
+        techniqueTruth: "not-applicable",
+      });
+      return streamAgentResponse({
+        runId: requestId,
+        requestId,
+        state,
+        stream: result.fullStream,
+        requireTool: false,
+        retry,
+        continueAfterDisconnect: true,
+        transformText: createBirthTimeModeOutputGuard(consultationMode, false),
+        toolStatus: () => "ready",
+        receipt: executionReceipt,
+        headers: { "x-jyotish-birth-time-mode": consultationMode },
+        onFirstActivity: markFirstActivity,
+        onFirstOutput: markFirstText,
+        onComplete: (output, agentExecutionReceipt) => settleRun(() => completeResponse(
+          output,
+          mergeUsage(usages),
+          "not-applicable",
+          workflowReceipt,
+          agentExecutionReceipt,
+        ), "completed", "completed"),
+        onError: (error) => settleRun(
+          cancel,
+          error instanceof Error ? error.message : "failed",
+          "cancelled",
+        ),
+        onCancel: () => settleRun(cancel, "cancelled", "cancelled"),
+      });
+    }
+
+    if (!prepared.serverChart) throw new Error("server_chart_truth_missing");
+    const agentContext = createConsultationAgentContext({
+      userId,
+      sessionId,
+      requestId,
+      consultationMode,
+      serverChart: prepared.serverChart,
+      abortSignal: agentAbortSignal,
+      state,
+    });
+    const agent = getJyotishAgent(selectedModel, agentContext);
+    const result = await agent.stream(baseMessages, streamOptions);
+    usages.push(result.totalUsage);
+    const retry = async () => {
+      const retried = await agent.stream([
+        ...baseMessages,
+        {
+          role: "user" as const,
+          content: "运行合同不完整：请先加载 jyotish-vedic-astrology Skill，再调用 run-jyotish-consultation 完成服务器计算；不要在工具参数中添加出生资料。",
+        },
+      ], streamOptions);
+      usages.push(retried.totalUsage);
+      return retried.fullStream;
+    };
+    const executionReceipt = (): AgentExecutionReceipt => ({
+      runId: requestId,
+      runtime: "mastra-agentic",
+      skill: { name: "jyotish-vedic-astrology", loaded: state.jyotishSkillLoaded },
+      steps: state.steps,
+      workflow: state.workflowReceipt ?? workflowReceipt,
+      techniqueTruth: state.techniqueTruth ?? "unknown",
+    });
+    return streamAgentResponse({
+      runId: requestId,
+      requestId,
+      state,
+      stream: result.fullStream,
+      requireTool: true,
+      retry,
+      continueAfterDisconnect: true,
+      transformText: (text) => createBirthTimeModeOutputGuard(
+        consultationMode,
+        state.workflowReceipt?.preciseTiming === "allowed",
+      )(text),
+      toolStatus: () => workflowStatus(state.workflowReceipt?.status),
+      receipt: executionReceipt,
+      headers: { "x-jyotish-birth-time-mode": consultationMode },
+      onFirstActivity: markFirstActivity,
+      onFirstOutput: markFirstText,
+      onComplete: (output, agentExecutionReceipt) => settleRun(() => completeResponse(
+        output,
+        mergeUsage(usages),
+        state.techniqueTruth ?? "unknown",
+        state.workflowReceipt ?? workflowReceipt,
+        agentExecutionReceipt,
+      ), "completed", "completed"),
+      onError: (error) => settleRun(
+        cancel,
+        error instanceof Error ? error.message : "failed",
+        "cancelled",
+      ),
+      onCancel: () => settleRun(cancel, "cancelled", "cancelled"),
+    });
+  }
+
   try {
     const { history } = parsed.data;
     const name = prepared.serverChart?.name ?? parsed.data.name;
     const consultationMode: ConsultationBirthTimeMode = prepared.consultationMode;
+    if (shouldUseAgenticRuntime(user)) {
+      return await runAgenticConsultation(consultationMode, history, name);
+    }
     if (!shouldRunBirthChartWorkflow(consultationMode)) {
       const result = await getGeneralJyotishAgent(selectedModel).stream([
         {
@@ -430,12 +650,12 @@ export async function POST(request: Request) {
           ].filter(Boolean).join("\n"),
         },
       ]);
-      const workflowReceipt = {
+      const workflowReceipt: WorkflowReceipt = {
         route: "general-no-birth-time",
         status: "ready",
         preciseTiming: "blocked",
         missingLayers: ["birth-minute"],
-      } as const;
+      };
       const settleErrored = (emitted: boolean, output: string) => settle(
         emitted
           ? () => completeResponse(
@@ -485,7 +705,7 @@ export async function POST(request: Request) {
     );
     const workflowReceipt = consultationWorkflowReceipt(workflowContext);
 
-    const result = await getJyotishAgent(selectedModel, workflowContext).stream([
+    const result = await getLegacyJyotishAgent(selectedModel, workflowContext).stream([
       ...history.map((message) => message.role === "user"
         ? { role: "user" as const, content: message.text }
         : { role: "assistant" as const, content: message.text }),

frontend/src/app/page.tsx

@@ -85,7 +85,12 @@ import {
   BALANCE_CHANGED_EVENT,
   membershipHref,
 } from "@/lib/membership";
-import { chatMessageViews, type ChatMessage } from "@/lib/chat-message-view";
+import { chatMessageViews, type AgentActivityView, type ChatMessage } from "@/lib/chat-message-view";
+import {
+  createNdjsonParser,
+  type AgentExecutionReceipt,
+  type ConsultationAgentPublicEvent,
+} from "@/lib/consultation-agent-events";
 import { writeChatSession } from "@/lib/chat-session-write-contract";
 import { consultationReportMarkdown } from "@/lib/consultation-report-export";
 import {
@@ -183,7 +188,7 @@ type ChatSession = {
 };
 
 type RequestError = { sessionId: string; message: string };
-type StreamingReply = { sessionId: string; text: string };
+type StreamingReply = { sessionId: string; text: string; activity?: AgentActivityView };
 type BirthPlace = {
   label: string;
   lat: number;
@@ -1103,6 +1108,9 @@ export default function Home() {
     || !account
     || !modelCatalog;
   const activeStreamingText = streamingReply && streamingReply.sessionId === activeSession?.id ? streamingReply.text : "";
+  const activeStreamingActivity = streamingReply && streamingReply.sessionId === activeSession?.id
+    ? streamingReply.activity
+    : undefined;
   const accountId = account?.user.id;
   const rectificationCardAction = resolveRectificationEntryAction(
     rectificationEntrySummary ?? {
@@ -2984,8 +2992,8 @@ export default function Home() {
       if (!response.body) {
         throw new ConsultationResponseError(502, "浏览器未收到可读取的回答流");
       }
-      const techniqueTruth = response.headers.get("x-jyotish-technique-truth") ?? "unknown";
-      const workflowReceipt = {
+      let techniqueTruth = response.headers.get("x-jyotish-technique-truth") ?? "unknown";
+      let workflowReceipt: AgentExecutionReceipt["workflow"] = {
         route: response.headers.get("x-jyotish-workflow-route") ?? "unknown",
         status: response.headers.get("x-jyotish-workflow-status") ?? "unknown",
         preciseTiming: response.headers.get("x-jyotish-precise-timing") ?? "unknown",
@@ -2994,26 +3002,63 @@ export default function Home() {
           .map((item) => item.trim())
           .filter((item) => item && item !== "none"),
       };
-
+      let agentExecutionReceipt: AgentExecutionReceipt | undefined;
+      let runCompleted = false;
       const reader = response.body.getReader();
       const decoder = new TextDecoder();
       let answer = "";
-
-      while (true) {
-        const { done, value } = await reader.read();
-        if (done) break;
-        answer += decoder.decode(value, { stream: true });
+      const updateStreamingAnswer = (activity?: AgentActivityView) => {
         const partialReply = parseAgentReply(answer, theme).text;
         latestPartialReply = partialReply;
-        setStreamingReply({ sessionId, text: partialReply });
+        setStreamingReply({ sessionId, text: partialReply, activity });
         if (partialReply && pendingConsultation.current?.requestId === requestId) {
-          pendingConsultation.current = {
-            ...pendingConsultation.current,
-            partialReply,
-          };
+          pendingConsultation.current = { ...pendingConsultation.current, partialReply };
         }
+      };
+      const updateActivity = (event: ConsultationAgentPublicEvent) => {
+        let activity: AgentActivityView | undefined;
+        if (event.type === "skill.started") {
+          activity = { phase: "loading-method", label: "正在读取印度占星分析规则…" };
+        } else if (event.type === "tool.started") {
+          activity = { phase: "chart-calculation", label: "正在计算本命盘…" };
+        } else if (event.type === "activity") {
+          activity = { phase: event.phase, label: event.label };
+        } else if (event.type === "tool.completed") {
+          activity = { phase: "evidence-validation", label: "正在核对可用证据…" };
+        } else if (event.type === "answer.delta") {
+          activity = { phase: "answer-composition", label: "正在组织回答…" };
+        }
+        if (activity) updateStreamingAnswer(activity);
+      };
+
+      if ((response.headers.get("content-type") ?? "").includes("application/x-ndjson")) {
+        const parser = createNdjsonParser((event) => {
+          if (event.type === "answer.delta") answer += event.text;
+          if (event.type === "run.completed") {
+            runCompleted = true;
+            agentExecutionReceipt = event.receipt;
+            workflowReceipt = event.receipt.workflow;
+            techniqueTruth = event.receipt.techniqueTruth ?? "unknown";
+          }
+          if (event.type === "run.failed") throw new ConsultationResponseError(502, event.message);
+          updateActivity(event);
+        });
+        while (true) {
+          const { done, value } = await reader.read();
+          if (done) break;
+          parser.push(decoder.decode(value, { stream: true }));
+        }
+        parser.finish(decoder.decode());
+        if (!runCompleted) throw new ConsultationResponseError(502, "Agent 回答未完成，本次不会保存为成功咨询。");
+      } else {
+        while (true) {
+          const { done, value } = await reader.read();
+          if (done) break;
+          answer += decoder.decode(value, { stream: true });
+          updateStreamingAnswer();
+        }
+        answer += decoder.decode();
       }
-      answer += decoder.decode();
       if (controller.signal.aborted) return Boolean(latestPartialReply);
       if (!answer.trim()) throw new Error("Agent 没有返回内容，请重试。");
       const reply = parseAgentReply(answer, theme);
@@ -3022,7 +3067,14 @@ export default function Home() {
       const completedSession: ChatSession = {
         ...userSession,
         title: userSession.title,
-        messages: [...userSession.messages, { role: "assistant", text: reply.text, suggestions: reply.suggestions, techniqueTruth, workflowReceipt }],
+        messages: [...userSession.messages, {
+          role: "assistant",
+          text: reply.text,
+          suggestions: reply.suggestions,
+          techniqueTruth,
+          workflowReceipt,
+          agentExecutionReceipt,
+        }],
         updatedAt: timestamp(),
       };
       updateSession(sessionId, () => completedSession);
@@ -3417,7 +3469,7 @@ export default function Home() {
           ) : (
             <div className="message-list" aria-busy={isLoading}>
               <span className="sr-only" aria-live="polite">{isLoading ? "Jyotisha 正在回答" : ""}</span>
-              {chatMessageViews(activeSession.messages, isLoading, activeStreamingText).map((message) => (
+              {chatMessageViews(activeSession.messages, isLoading, activeStreamingText, activeStreamingActivity).map((message) => (
                 <ChatMessageRow key={message.renderKey} message={message} />
               ))}
               {activeError && <p className="error-message" role="alert">{activeError}</p>}