diff --git a/frontend/src/app/api/consult/route.ts b/frontend/src/app/api/consult/route.ts
index b81688da..bc4a6707 100644
--- a/frontend/src/app/api/consult/route.ts
+++ b/frontend/src/app/api/consult/route.ts
@@ -7,16 +7,12 @@ import {
   runConsultationWorkflow,
 } from "@/mastra";
 import { blocksPromptExtraction } from "@/lib/consult-safety";
+import { parseAgentReply } from "@/lib/agent-reply";
 import {
   consultationEntrypointSchema,
   resolveConsultationQuestion,
 } from "@/lib/consultation-entrypoint";
-import {
-  authorizeUsage,
-  completeUsage,
-  CreditRpcError,
-  releaseUsage,
-} from "@/lib/consultation-billing";
+import { CreditRpcError } from "@/lib/consultation-billing";
 import { reserveConsultationModel } from "@/lib/consultation-model-selection";
 import { resolveSessionLanguageModel } from "@/lib/model-catalog";
 import { createAdminSupabaseClient } from "@/lib/supabase/admin";
@@ -72,6 +68,43 @@ const generalChatRequestSchema = z.object({
 
 const chatRequestSchema = z.union([generalChatRequestSchema, 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(),
+});
+
+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()
@@ -176,6 +209,8 @@ export async function POST(request: Request) {
 
   const userId = user.id;
   const requestId = parsed.data.requestId;
+  const sessionId = parsed.data.sessionId;
+  const consultationTheme = parsed.data.theme;
 
   const userControlledPrompt = [
     parsed.data.question,
@@ -210,17 +245,24 @@ export async function POST(request: Request) {
       reserve: () => reserveConsultationModel(
         chatSession.model_id,
         (modelId) => sessionModel?.id === modelId ? sessionModel : null,
-        (model) => authorizeUsage(accounting, {
-          userId,
-          requestId,
-          featureKey: "chat.standard",
-          requestedModelId: model.id,
-          creditCost: model.creditCost,
-        }).then((result) => ({
-          success: result.success,
-          credits: result.credits,
-          error_code: result.reason,
-        })),
+        async (model) => {
+          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: model.creditCost,
+          });
+          const reservation = consultationReservationSchema.safeParse(first(data ?? []));
+          if (error || !reservation.success) {
+            throw new CreditRpcError(error?.message || "invalid_reservation_response");
+          }
+          return {
+            success: reservation.data.success,
+            credits: reservation.data.credits,
+            error_code: reservation.data.reason,
+          };
+        },
       ),
     });
   } catch (error) {
@@ -277,9 +319,23 @@ export async function POST(request: Request) {
     );
   }
 
+
   async function cancel() {
     try {
-      await releaseUsage(accounting, userId, requestId, "consultation_cancelled");
+      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");
+      }
     } catch (error) {
       const reason = error instanceof Error ? error.name : "UnknownError";
       console.error(
@@ -289,25 +345,67 @@ export async function POST(request: Request) {
   }
 
   const usageStartedAt = Date.now();
-  async function complete(usage: Promise<{ inputTokens?: number; outputTokens?: number }>) {
+  async function usagePayload(usage: Promise<{ inputTokens?: number; outputTokens?: number }>) {
     const resolved = await usage;
     const inputTokens = Math.max(0, Math.trunc(resolved.inputTokens ?? 0));
     const outputTokens = Math.max(0, Math.trunc(resolved.outputTokens ?? 0));
-    const costMicrousd = Math.round((
-      inputTokens * (selectedModel.inputCostMicrousdPerMillion ?? 0)
-      + outputTokens * (selectedModel.outputCostMicrousdPerMillion ?? 0)
-    ) / 1_000_000);
-    const result = await completeUsage(accounting, userId, requestId, {
+    return {
       eventKey: requestId,
       actualModelId: selectedModel.id,
       modelConfigVersion: selectedModel.configVersion,
       inputTokens,
       outputTokens,
-      costMicrousd,
+      costMicrousd: Math.round((
+        inputTokens * (selectedModel.inputCostMicrousdPerMillion ?? 0)
+        + outputTokens * (selectedModel.outputCostMicrousdPerMillion ?? 0)
+      ) / 1_000_000),
       durationMs: Date.now() - usageStartedAt,
-    });
-    if (!result.success)
-      throw new CreditRpcError(result.error_code || "completion_rejected");
+    };
+  }
+
+  async function completeResponse(
+    rawTransformedText: string,
+    usage: Promise<{ inputTokens?: number; outputTokens?: number }>,
+    techniqueTruth: string,
+    workflowReceipt: {
+      route: string;
+      status: string;
+      preciseTiming: string;
+      missingLayers: readonly string[];
+    },
+  ) {
+    try {
+      const reply = parseAgentReply(rawTransformedText, consultationTheme);
+      if (!reply.text) throw new Error("empty_agent_reply");
+      const responseMessage = {
+        role: "assistant" as const,
+        text: reply.text,
+        suggestions: reply.suggestions,
+        techniqueTruth,
+        workflowReceipt,
+      };
+      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");
+      }
+    } catch (error) {
+      await cancel();
+      throw error;
+    }
   }
 
   let settlement: Promise<void> | null = null;
@@ -332,24 +430,43 @@ export async function POST(request: Request) {
           ].filter(Boolean).join("\n"),
         },
       ]);
-      const completeWithUsage = () => complete(result.totalUsage);
-      const settleInterrupted = (emitted: boolean) =>
-        settle(emitted ? completeWithUsage : cancel);
+      const workflowReceipt = {
+        route: "general-no-birth-time",
+        status: "ready",
+        preciseTiming: "blocked",
+        missingLayers: ["birth-minute"],
+      } as const;
+      const settleErrored = (emitted: boolean, output: string) => settle(
+        emitted
+          ? () => completeResponse(
+            output,
+            result.totalUsage,
+            "not-applicable",
+            workflowReceipt,
+          )
+          : cancel,
+      );
       return streamTextResponse(result.textStream, {
         transformText: createBirthTimeModeOutputGuard(consultationMode, false),
         mode: "mastra",
         requestId,
+        continueAfterDisconnect: true,
         headers: {
-          "x-jyotish-workflow-route": "general-no-birth-time",
-          "x-jyotish-workflow-status": "ready",
+          "x-jyotish-workflow-route": workflowReceipt.route,
+          "x-jyotish-workflow-status": workflowReceipt.status,
           "x-jyotish-technique-truth": "not-applicable",
-          "x-jyotish-precise-timing": "blocked",
-          "x-jyotish-missing-layers": "birth-minute",
+          "x-jyotish-precise-timing": workflowReceipt.preciseTiming,
+          "x-jyotish-missing-layers": workflowReceipt.missingLayers.join(","),
           "x-jyotish-birth-time-mode": consultationMode,
         },
-        onComplete: () => settle(completeWithUsage),
-        onError: (_error, emitted) => settleInterrupted(emitted),
-        onCancel: settleInterrupted,
+        onComplete: (rawTransformedText) => settle(() => completeResponse(
+          rawTransformedText,
+          result.totalUsage,
+          "not-applicable",
+          workflowReceipt,
+        )),
+        onError: (_error, emitted, output: string) => settleErrored(emitted, output),
+        onCancel: () => settle(cancel),
       });
     }
 
@@ -383,9 +500,24 @@ export async function POST(request: Request) {
         ].filter(Boolean).join("\n"),
       },
     ]);
-    const completeWithUsage = () => complete(result.totalUsage);
-    const settleInterrupted = (emitted: boolean) =>
-      settle(emitted ? completeWithUsage : cancel);
+    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, {
       transformText: createBirthTimeModeOutputGuard(
         consultationMode,
@@ -393,6 +525,7 @@ export async function POST(request: Request) {
       ),
       mode: "mastra",
       requestId,
+      continueAfterDisconnect: true,
       headers: {
         "x-jyotish-workflow-route": workflowReceipt.route,
         "x-jyotish-workflow-status": workflowReceipt.status,
@@ -401,9 +534,14 @@ export async function POST(request: Request) {
         "x-jyotish-missing-layers": workflowReceipt.missingLayers,
         "x-jyotish-birth-time-mode": consultationMode,
       },
-      onComplete: () => settle(completeWithUsage),
-      onError: (_error, emitted) => settleInterrupted(emitted),
-      onCancel: settleInterrupted,
+      onComplete: (rawTransformedText) => settle(() => completeResponse(
+        rawTransformedText,
+        result.totalUsage,
+        workflowReceipt.techniqueTruth,
+        responseWorkflowReceipt,
+      )),
+      onError: (_error, emitted, output: string) => settleErrored(emitted, output),
+      onCancel: () => settle(cancel),
     });
   } catch (error) {
     await cancel();
diff --git a/frontend/src/app/api/consult/status/route.ts b/frontend/src/app/api/consult/status/route.ts
new file mode 100644
index 00000000..376e21a2
--- /dev/null
+++ b/frontend/src/app/api/consult/status/route.ts
@@ -0,0 +1,110 @@
+import { NextResponse } from "next/server";
+import { z } from "zod";
+import { runCreditRpc } from "@/lib/consultation-billing";
+import { createAdminSupabaseClient } from "@/lib/supabase/admin";
+import { createServerSupabaseClient } from "@/lib/supabase/server";
+
+export const runtime = "nodejs";
+
+const idSchema = z.string().uuid();
+const reservedLeaseMs = 15 * 60 * 1000;
+// ponytail: Staging MVP ceiling—without a queue/worker, status lazily refunds reservations after this lease.
+function reservationLeaseExpired(updatedAt: unknown, now = Date.now()) {
+  if (typeof updatedAt !== "string") return false;
+  const timestamp = Date.parse(updatedAt);
+  return Number.isFinite(timestamp) && now - timestamp >= reservedLeaseMs;
+}
+
+export async function GET(request: Request) {
+  let supabase: Awaited<ReturnType<typeof createServerSupabaseClient>>;
+  try {
+    supabase = await createServerSupabaseClient();
+  } catch {
+    return NextResponse.json({ error: "服务尚未配置" }, { status: 503 });
+  }
+
+  const { data: { user }, error: authError } = await supabase.auth.getUser();
+  if (authError || !user) {
+    return NextResponse.json({ error: "请先登录" }, { status: 401 });
+  }
+
+  const params = new URL(request.url).searchParams;
+  const sessionIdValue = params.get("sessionId");
+  const requestIdValue = params.get("requestId");
+  const activeLookup = sessionIdValue === null && requestIdValue === null;
+  if (!activeLookup && (sessionIdValue === null || requestIdValue === null)) {
+    return NextResponse.json({ error: "咨询请求格式不正确" }, { status: 400 });
+  }
+
+  const sessionId = sessionIdValue === null ? null : idSchema.safeParse(sessionIdValue);
+  const requestId = requestIdValue === null ? null : idSchema.safeParse(requestIdValue);
+  if ((sessionId && !sessionId.success) || (requestId && !requestId.success)) {
+    return NextResponse.json({ error: "咨询请求格式不正确" }, { status: 400 });
+  }
+
+  let accounting: ReturnType<typeof createAdminSupabaseClient>;
+  try {
+    accounting = createAdminSupabaseClient();
+  } catch {
+    return NextResponse.json({ error: "服务尚未配置" }, { status: 503 });
+  }
+
+  let query = accounting
+    .from("consultation_requests")
+    .select("request_id,session_id,status,response_message,updated_at")
+    .eq("user_id", user.id);
+
+  if (activeLookup) {
+    query = query.eq("status", "reserved").order("created_at", { ascending: false }).limit(1);
+  } else {
+    query = query
+      .eq("session_id", sessionId!.data)
+      .eq("request_id", requestId!.data);
+  }
+
+  const { data, error } = await query.maybeSingle();
+  if (error) {
+    return NextResponse.json({ error: "暂时无法读取咨询状态" }, { status: 503 });
+  }
+  if (!data || typeof data.session_id !== "string") {
+    return NextResponse.json({ error: "咨询请求不存在" }, { status: 404 });
+  }
+
+  let statusData = data;
+  if (statusData.status === "reserved" && reservationLeaseExpired(statusData.updated_at)) {
+    let cancellation;
+    try {
+      cancellation = await runCreditRpc(
+        accounting,
+        "cancel_consultation_credit",
+        user.id,
+        statusData.request_id,
+      );
+    } catch {
+      return NextResponse.json({ error: "暂时无法回收超时咨询" }, { status: 503 });
+    }
+    if (!cancellation.success && cancellation.error_code !== "request_completed") {
+      return NextResponse.json({ error: "暂时无法回收超时咨询" }, { status: 503 });
+    }
+
+    const { data: settledData, error: settledError } = await accounting
+      .from("consultation_requests")
+      .select("request_id,session_id,status,response_message,updated_at")
+      .eq("user_id", user.id)
+      .eq("session_id", statusData.session_id)
+      .eq("request_id", statusData.request_id)
+      .maybeSingle();
+    if (settledError || !settledData || typeof settledData.session_id !== "string") {
+      return NextResponse.json({ error: "暂时无法读取咨询状态" }, { status: 503 });
+    }
+    statusData = settledData;
+  }
+
+  return NextResponse.json({
+    requestId: statusData.request_id,
+    sessionId: statusData.session_id,
+    status: statusData.status,
+    responseMessage: statusData.response_message,
+    updatedAt: statusData.updated_at,
+  });
+}
diff --git a/frontend/src/app/page.tsx b/frontend/src/app/page.tsx
index 0da71693..a0a20d3a 100644
--- a/frontend/src/app/page.tsx
+++ b/frontend/src/app/page.tsx
@@ -213,6 +213,13 @@ type DailyStarlanguageApiResponse = {
   boundary?: "not_deterministic_prediction";
 };
 type SessionReadResult = { readonly sessions: ChatSession[]; readonly fallbackSessionIds: string[] };
+type ConsultationStatus = {
+  readonly requestId: string;
+  readonly sessionId: string;
+  readonly status: "reserved" | "completed" | "cancelled";
+  readonly responseMessage?: unknown;
+  readonly updatedAt?: string;
+};
 type PendingConsultation = {
   readonly requestId: string;
   readonly sessionId: string;
@@ -224,10 +231,12 @@ type PendingConsultation = {
   readonly previousOnboardingState: boolean;
   readonly controller: AbortController;
   readonly cancelled: boolean;
-  readonly phase: "undo" | "streaming";
+  readonly phase: "undo" | "streaming" | "recovering";
   readonly partialReply: string;
 };
 const undoWindowMs = 2_500;
+const pendingConsultationStorageKey = "jyotisha.pending-consultation";
+const uuidPattern = /^[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i;
 const china = chinaLocations.country;
 
 const themes = defaultGuidedJyotishTopics;
@@ -850,6 +859,26 @@ class CancellationResponseError extends Error {
   }
 }
 
+class ConsultationResponseError extends Error {
+  readonly status: number;
+
+  constructor(status: number, message: string) {
+    super(message);
+    this.name = "ConsultationResponseError";
+    this.status = status;
+  }
+}
+
+class ConsultationStatusError extends Error {
+  readonly status: number;
+
+  constructor(status: number, message: string) {
+    super(message);
+    this.name = "ConsultationStatusError";
+    this.status = status;
+  }
+}
+
 function waitForUndoWindow(signal: AbortSignal) {
   return new Promise<void>((resolve) => {
     const finish = () => {
@@ -891,6 +920,45 @@ async function fetchSessions(signal?: AbortSignal): Promise<unknown> {
   return payload && typeof payload === "object" ? (payload as { sessions?: unknown }).sessions : null;
 }
 
+function parseConsultationStatus(payload: unknown, requestId?: string): ConsultationStatus {
+  if (!payload || typeof payload !== "object") throw new Error("后台回答状态无效");
+  const status = payload as Partial<ConsultationStatus>;
+  if (typeof status.requestId !== "string"
+    || typeof status.sessionId !== "string"
+    || (requestId && status.requestId !== requestId)
+    || (status.status !== "reserved" && status.status !== "completed" && status.status !== "cancelled")) {
+    throw new Error("后台回答状态无效");
+  }
+  return status as ConsultationStatus;
+}
+
+async function fetchConsultationStatus(sessionId: string, requestId: string, signal?: AbortSignal): Promise<ConsultationStatus> {
+  const response = await fetch(`/api/consult/status?sessionId=${encodeURIComponent(sessionId)}&requestId=${encodeURIComponent(requestId)}`, {
+    signal,
+    cache: "no-store",
+  });
+  const payload: unknown = await response.json().catch(() => null);
+  if (!response.ok) {
+    throw new ConsultationStatusError(
+      response.status,
+      payloadMessage(payload, "暂时无法恢复后台回答"),
+    );
+  }
+  const status = parseConsultationStatus(payload, requestId);
+  if (status.sessionId !== sessionId) throw new Error("后台回答状态无效");
+  return status;
+}
+
+async function fetchActiveConsultationStatus(signal?: AbortSignal): Promise<ConsultationStatus | null> {
+  const response = await fetch("/api/consult/status", { signal, cache: "no-store" });
+  const payload: unknown = await response.json().catch(() => null);
+  if (response.status === 404) return null;
+  if (!response.ok) throw new Error(payloadMessage(payload, "暂时无法恢复后台回答"));
+  const status = parseConsultationStatus(payload);
+  if (status.status !== "reserved") throw new Error("后台回答状态无效");
+  return status;
+}
+
 async function patchSessionModel(sessionId: string, modelId: string, signal?: AbortSignal) {
   const response = await fetch(`/api/sessions/${encodeURIComponent(sessionId)}`, {
     method: "PATCH",
@@ -938,9 +1006,10 @@ export default function Home() {
   const [draftTheme, setDraftTheme] = useState<Theme | null>(null);
   const [draftEntrypoint, setDraftEntrypoint] = useState<ConsultationEntrypoint | null>(null);
   const [composerNotice, setComposerNotice] = useState("");
-  const [consultationPhase, setConsultationPhase] = useState<"undo" | "streaming" | null>(null);
+  const [consultationPhase, setConsultationPhase] = useState<"undo" | "streaming" | "recovering" | null>(null);
   const [cancellationPending, setCancellationPending] = useState(false);
   const [pendingSessionId, setPendingSessionId] = useState<string | null>(null);
+  const [pendingRequestId, setPendingRequestId] = useState<string | null>(null);
   const [streamingReply, setStreamingReply] = useState<StreamingReply | null>(null);
   const [requestError, setRequestError] = useState<RequestError | null>(null);
   const [birthTimeConsultationConsent, setBirthTimeConsultationConsent] = useState<BirthTimeConsultationConsentState>(
@@ -976,6 +1045,9 @@ export default function Home() {
   const cancellationInFlight = useRef(false);
   const stoppedRequestAwaitingSettlement = useRef<string | null>(null);
   const stoppedSessionPersistence = useRef(new Map<string, Promise<void>>());
+  const consultationRecoveryWakeup = useRef<() => void>(() => undefined);
+  const consultationRecoveryCheck = useRef<() => void>(() => undefined);
+  const consultationStatusMissingCount = useRef(0);
   const modelPersistence = useRef(new SessionModelPersistenceQueue());
   const rectificationPersistence = useRef(new SessionModelPersistenceQueue());
   const modelSyncFailures = useRef(new Set<string>());
@@ -1163,6 +1235,48 @@ export default function Home() {
     && !activeSession?.messages.length;
   const daypartGreeting = greetingForHour(new Date().getHours());
 
+  function restoreConsultationRecovery(session: ChatSession, requestId: string) {
+    if (pendingConsultation.current) return;
+    const lastMessage = session.messages.at(-1);
+    const question = lastMessage?.role === "user" ? lastMessage.text : "";
+    const previousSession = lastMessage?.role === "user"
+      ? { ...session, messages: session.messages.slice(0, -1) }
+      : session;
+    pendingConsultation.current = {
+      requestId,
+      sessionId: session.id,
+      question,
+      entrypoint: null,
+      theme: session.theme,
+      previousSession,
+      optimisticSession: session,
+      previousOnboardingState: false,
+      controller: new AbortController(),
+      cancelled: false,
+      phase: "recovering",
+      partialReply: "",
+    };
+    setPendingSessionId(session.id);
+    setPendingRequestId(requestId);
+    setConsultationPhase("recovering");
+    setStreamingReply({ sessionId: session.id, text: "" });
+    setComposerNotice(navigator.onLine
+      ? "回答仍在后台生成，正在自动恢复。"
+      : "网络已断开，回答仍在后台生成；联网后会自动恢复。");
+  }
+
+  consultationRecoveryCheck.current = () => {
+    consultationRecoveryWakeup.current();
+    if (pendingConsultation.current || uiPreview.current) return;
+    void fetchActiveConsultationStatus()
+      .then((status) => {
+        if (status?.status !== "reserved") return;
+        const session = sessions.find((item) => item.id === status.sessionId);
+        if (session) restoreConsultationRecovery(session, status.requestId);
+      })
+      .catch(() => undefined);
+  };
+
   useEffect(() => {
     const controller = new AbortController();
     const bootstrapTimeout = window.setTimeout(() => {
@@ -1298,6 +1412,60 @@ export default function Home() {
           nextSessions = [initialSession];
         }
 
+        let reservedConsultation: ConsultationStatus | null = null;
+        let storedPending: { sessionId: string; requestId: string } | null = null;
+        const storedPendingJson = sessionStorage.getItem(pendingConsultationStorageKey);
+        if (storedPendingJson) {
+          try {
+            const parsedPending = JSON.parse(storedPendingJson) as Record<string, unknown>;
+            if (typeof parsedPending.sessionId === "string"
+              && typeof parsedPending.requestId === "string"
+              && uuidPattern.test(parsedPending.sessionId)
+              && uuidPattern.test(parsedPending.requestId)
+              && nextSessions.some((session) => session.id === parsedPending.sessionId)) {
+              storedPending = {
+                sessionId: parsedPending.sessionId,
+                requestId: parsedPending.requestId,
+              };
+            } else {
+              sessionStorage.removeItem(pendingConsultationStorageKey);
+            }
+          } catch {
+            sessionStorage.removeItem(pendingConsultationStorageKey);
+          }
+        }
+
+        if (storedPending) {
+          try {
+            const status = await fetchConsultationStatus(
+              storedPending.sessionId,
+              storedPending.requestId,
+              controller.signal,
+            );
+            if (status.status === "reserved") {
+              consultationStatusMissingCount.current = 0;
+              reservedConsultation = status;
+            } else {
+              sessionStorage.removeItem(pendingConsultationStorageKey);
+            }
+          } catch (caught) {
+            if (caught instanceof Error && caught.name === "AbortError") throw caught;
+            consultationStatusMissingCount.current = caught instanceof ConsultationStatusError && caught.status === 404 ? 1 : 0;
+            reservedConsultation = {
+              sessionId: storedPending.sessionId,
+              requestId: storedPending.requestId,
+              status: "reserved",
+            };
+          }
+        } else {
+          try {
+            reservedConsultation = await fetchActiveConsultationStatus(controller.signal);
+            consultationStatusMissingCount.current = 0;
+          } catch (caught) {
+            if (caught instanceof Error && caught.name === "AbortError") throw caught;
+          }
+        }
+
         if (controller.signal.aborted) return;
         const nextProfile = readProfile(nextAccount.profile);
         setAccount(nextAccount);
@@ -1308,6 +1476,10 @@ export default function Home() {
         setOnboardingStep(missingProfileStep(nextProfile) ?? "name");
         setSessions(nextSessions);
         setActiveSessionId(nextSessions[0].id);
+        if (reservedConsultation?.status === "reserved") {
+          const recoverySession = nextSessions.find((session) => session.id === reservedConsultation.sessionId);
+          if (recoverySession) restoreConsultationRecovery(recoverySession, reservedConsultation.requestId);
+        }
         if (modelCatalogResult.unavailable) {
           setComposerNotice("模型服务暂时不可用，当前无法发送问题。");
         } else if (parsedSessions.fallbackSessionIds.length > 0) {
@@ -1343,6 +1515,111 @@ export default function Home() {
     };
   }, []);
 
+  useEffect(() => {
+    if (!hydrated || uiPreview.current) return;
+    if (pendingSessionId && pendingRequestId) {
+      sessionStorage.setItem(pendingConsultationStorageKey, JSON.stringify({
+        sessionId: pendingSessionId,
+        requestId: pendingRequestId,
+      }));
+    } else {
+      consultationStatusMissingCount.current = 0;
+      sessionStorage.removeItem(pendingConsultationStorageKey);
+    }
+  }, [hydrated, pendingRequestId, pendingSessionId]);
+
+  useEffect(() => {
+    if (consultationPhase !== "recovering" || !pendingSessionId || !pendingRequestId || uiPreview.current) return;
+    const controller = new AbortController();
+    let timer = 0;
+    let polling = false;
+
+    const poll = async () => {
+      if (polling || controller.signal.aborted) return;
+      polling = true;
+      try {
+        if (!navigator.onLine) {
+          setComposerNotice("网络已断开，回答仍在后台生成；联网后会自动恢复。");
+          return;
+        }
+        const status = await fetchConsultationStatus(pendingSessionId, pendingRequestId, controller.signal);
+        if (status.status === "reserved") {
+          consultationStatusMissingCount.current = 0;
+          setComposerNotice("回答仍在后台生成，正在自动恢复。");
+          return;
+        }
+        if (status.status === "completed") {
+          const payload = await fetchSessions(controller.signal);
+          const parsed = readSessions(payload, modelCatalog);
+          setSessions(parsed.sessions);
+          setActiveSessionId((current) => parsed.sessions.some((session) => session.id === current)
+            ? current
+            : parsed.sessions[0]?.id ?? "");
+          pendingConsultation.current = null;
+          setPendingSessionId(null);
+          setPendingRequestId(null);
+          setConsultationPhase(null);
+          setStreamingReply(null);
+          setRequestError(null);
+          setComposerNotice("回答已恢复。");
+          void refreshAccount();
+          return;
+        }
+        pendingConsultation.current = null;
+        setPendingSessionId(null);
+        setPendingRequestId(null);
+        setConsultationPhase(null);
+        setStreamingReply(null);
+        setRequestError(null);
+        setComposerNotice("回答已取消；问题仍保留在聊天记录中，可重新发送。");
+      } catch (caught) {
+        if (controller.signal.aborted) return;
+        if (caught instanceof ConsultationStatusError && caught.status === 404) {
+          consultationStatusMissingCount.current += 1;
+          if (consultationStatusMissingCount.current >= 3) {
+            pendingConsultation.current = null;
+            setPendingSessionId(null);
+            setPendingRequestId(null);
+            setConsultationPhase(null);
+            setStreamingReply(null);
+            setRequestError({
+              sessionId: pendingSessionId,
+              message: "后台未找到本次咨询请求，请重新发送。",
+            });
+            setComposerNotice("后台未找到本次咨询请求，已停止恢复，请重新发送。");
+            return;
+          }
+          setComposerNotice("正在确认本次咨询请求是否已开始…");
+          return;
+        }
+        consultationStatusMissingCount.current = 0;
+        setComposerNotice(navigator.onLine
+          ? "回答仍在后台生成，正在自动恢复。"
+          : "网络已断开，回答仍在后台生成；联网后会自动恢复。");
+      } finally {
+        polling = false;
+        if (!controller.signal.aborted && pendingConsultation.current?.phase === "recovering") {
+          timer = window.setTimeout(() => void poll(), 1_750);
+        }
+      }
+    };
+
+    consultationRecoveryWakeup.current = () => {
+      window.clearTimeout(timer);
+      void poll();
+    };
+    if (consultationStatusMissingCount.current > 0) {
+      timer = window.setTimeout(() => void poll(), 1_750);
+    } else {
+      void poll();
+    }
+    return () => {
+      consultationRecoveryWakeup.current = () => undefined;
+      window.clearTimeout(timer);
+      controller.abort();
+    };
+  }, [consultationPhase, modelCatalog, pendingRequestId, pendingSessionId]);
+
   useEffect(() => {
     if (!hydrated || !shouldStreamOnboarding) return;
 
@@ -1459,14 +1736,25 @@ export default function Home() {
     const onBalanceChanged = () => void refreshAccount();
     const onPageShow = (event: PageTransitionEvent) => {
       if (event.persisted) void refreshAccount();
+      consultationRecoveryCheck.current();
+    };
+    const onOnline = () => consultationRecoveryCheck.current();
+    const onOffline = () => {
+      if (pendingConsultation.current?.phase === "recovering") {
+        setComposerNotice("网络已断开，回答仍在后台生成；联网后会自动恢复。");
+      }
     };
     window.addEventListener("storage", onBalanceStorage);
     window.addEventListener(BALANCE_CHANGED_EVENT, onBalanceChanged);
     window.addEventListener("pageshow", onPageShow);
+    window.addEventListener("online", onOnline);
+    window.addEventListener("offline", onOffline);
     return () => {
       window.removeEventListener("storage", onBalanceStorage);
       window.removeEventListener(BALANCE_CHANGED_EVENT, onBalanceChanged);
       window.removeEventListener("pageshow", onPageShow);
+      window.removeEventListener("online", onOnline);
+      window.removeEventListener("offline", onOffline);
     };
   }, [accountId, hydrated]);
 
@@ -2225,7 +2513,7 @@ export default function Home() {
     if (!pending || pending.cancelled) return;
 
     const isPreview = process.env.NODE_ENV === "development" && uiPreview.current;
-    if (pending.phase === "streaming" && !isPreview) {
+    if (pending.phase !== "undo" && !isPreview) {
       stoppedRequestAwaitingSettlement.current = pending.requestId;
       cancellationInFlight.current = true;
       setCancellationPending(true);
@@ -2239,26 +2527,90 @@ export default function Home() {
         messages: [...pending.optimisticSession.messages, { role: "assistant", text: pending.partialReply }],
         updatedAt: timestamp(),
       };
+      if (isPreview) {
+        updateSession(pending.sessionId, () => stoppedSession);
+        setStreamingReply(null);
+        setPendingSessionId(null);
+        setConsultationPhase(null);
+        setRequestError(null);
+        setComposerNotice("已停止回答，现有内容已保留。");
+        if (pendingConsultation.current?.requestId === pending.requestId) {
+          pendingConsultation.current = null;
+        }
+        return;
+      }
+
+      setComposerNotice("正在停止回答并申请退回本次点数…");
+      try {
+        await requestCancellation(pending.requestId);
+      } catch (error) {
+        cancellationRequests.current.delete(pending.requestId);
+        stoppedRequestAwaitingSettlement.current = null;
+        cancellationInFlight.current = false;
+        setCancellationPending(false);
+        pendingConsultation.current = {
+          ...pending,
+          controller: new AbortController(),
+          cancelled: false,
+          phase: "recovering",
+        };
+        setPendingSessionId(pending.sessionId);
+        setConsultationPhase("recovering");
+        setStreamingReply({ sessionId: pending.sessionId, text: pending.partialReply });
+        setRequestError(null);
+
+        if (error instanceof CancellationResponseError && error.status === 409) {
+          setComposerNotice("回答已完成，正在恢复服务端完整内容。");
+          try {
+            const status = await fetchConsultationStatus(pending.sessionId, pending.requestId);
+            if (status.status === "completed") {
+              const payload = await fetchSessions();
+              const parsed = readSessions(payload, modelCatalog);
+              setSessions(parsed.sessions);
+              setActiveSessionId((current) => parsed.sessions.some((session) => session.id === current)
+                ? current
+                : parsed.sessions[0]?.id ?? "");
+              pendingConsultation.current = null;
+              setPendingSessionId(null);
+              setConsultationPhase(null);
+              setStreamingReply(null);
+              setComposerNotice("回答已恢复。");
+              void refreshAccount();
+              return;
+            }
+          } catch {
+            // The recovery poll retries status and session reload.
+          }
+        } else {
+          setComposerNotice("停止请求尚未确认，正在自动恢复后台回答。");
+        }
+        window.setTimeout(() => consultationRecoveryWakeup.current(), 0);
+        return;
+      }
+
       updateSession(pending.sessionId, () => stoppedSession);
       setStreamingReply(null);
       setPendingSessionId(null);
       setConsultationPhase(null);
       setRequestError(null);
-      setComposerNotice("已停止回答。模型已开始生成，本次将计费，现有内容已保留。");
-      if (!isPreview) {
-        const persistence = persistSession(stoppedSession).catch((error) => {
-          setRequestError({
-            sessionId: pending.sessionId,
-            message: error instanceof Error ? error.message : "已停止的回答暂时无法同步。",
-          });
+      try {
+        await persistSession(stoppedSession);
+        setComposerNotice("已停止回答，现有内容已保留，本次点数已退回。");
+      } catch (error) {
+        setComposerNotice("本次点数已退回；现有内容暂时无法同步。");
+        setRequestError({
+          sessionId: pending.sessionId,
+          message: error instanceof Error ? error.message : "已停止的回答暂时无法同步。",
         });
-        stoppedSessionPersistence.current.set(pending.requestId, persistence);
       }
-      if (!isPreview) {
-        void refreshAccount();
-      } else if (pendingConsultation.current?.requestId === pending.requestId) {
+      cancellationRequests.current.delete(pending.requestId);
+      stoppedRequestAwaitingSettlement.current = null;
+      cancellationInFlight.current = false;
+      setCancellationPending(false);
+      if (pendingConsultation.current?.requestId === pending.requestId) {
         pendingConsultation.current = null;
       }
+      void refreshAccount();
       return;
     }
 
@@ -2288,6 +2640,11 @@ export default function Home() {
       pending.sessionId,
       "已停止，问题已放回输入框，本次未扣点。",
     );
+    cancellationRequests.current.delete(pending.requestId);
+    stoppedRequestAwaitingSettlement.current = null;
+    cancellationInFlight.current = false;
+    setCancellationPending(false);
+    if (pendingConsultation.current?.requestId === pending.requestId) pendingConsultation.current = null;
   }
 
   function completeConsultationInterface(requestId: string) {
@@ -2295,6 +2652,7 @@ export default function Home() {
     pendingConsultation.current = null;
     setStreamingReply(null);
     setPendingSessionId(null);
+    setPendingRequestId(null);
     setConsultationPhase(null);
   }
 
@@ -2358,7 +2716,9 @@ export default function Home() {
       : currentSession.messages;
     const userSession: ChatSession = {
       ...currentSession,
-      title: currentSession.title,
+      title: currentSession.messages.length === 0 && currentSession.title === "新对话"
+        ? resolveSessionTitle(question)
+        : currentSession.title,
       theme,
       messages: [...preservedMessages, { role: "user", text: question }],
       updatedAt: timestamp(),
@@ -2369,7 +2729,9 @@ export default function Home() {
     cancellationFeedbackRequest.current = null;
     setRequestError(null);
     setComposerNotice("");
+    consultationStatusMissingCount.current = 0;
     setPendingSessionId(sessionId);
+    setPendingRequestId(requestId);
     setConsultationPhase("undo");
     pendingConsultation.current = {
       requestId,
@@ -2417,9 +2779,7 @@ export default function Home() {
       ].join("\n"), theme);
       const previewSession: ChatSession = {
         ...userSession,
-        title: currentSession.messages.length === 0
-          ? resolveSessionTitle(question, previewReply.title)
-          : userSession.title,
+        title: userSession.title,
         messages: [...userSession.messages, {
           role: "assistant",
           text: previewReply.text,
@@ -2434,6 +2794,26 @@ export default function Home() {
 
     await waitForUndoWindow(controller.signal);
     if (controller.signal.aborted) return false;
+    try {
+      await persistSession(userSession);
+    } catch (caught) {
+      if (controller.signal.aborted) return false;
+      updateSession(sessionId, () => currentSession);
+      setOnboardingJustCompleted(previousOnboardingState);
+      if (activeSessionIdRef.current === sessionId) {
+        setDraft(originalQuestion);
+        setDraftTheme(theme);
+        setDraftEntrypoint(entrypoint);
+      }
+      setRequestError({
+        sessionId,
+        message: `${caught instanceof Error ? caught.message : "问题保存失败，请稍后重试。"} 问题已放回输入框。`,
+      });
+      setComposerNotice("问题保存失败，未开始生成；问题已放回输入框。");
+      completeConsultationInterface(requestId);
+      window.requestAnimationFrame(() => composerInput.current?.focus());
+      return false;
+    }
     if (pendingConsultation.current?.requestId === requestId) {
       pendingConsultation.current = {
         ...pendingConsultation.current,
@@ -2480,9 +2860,14 @@ export default function Home() {
         const errorPayload = contentType.includes("application/json") ? await response.json() : { message: await response.text() };
         if (response.status === 401) window.location.assign("/login");
         if (response.status === 402) window.location.assign(membershipHref("insufficient-credits"));
-        throw new Error(payloadMessage(errorPayload, "服务暂时不可用"));
+        throw new ConsultationResponseError(
+          response.status,
+          payloadMessage(errorPayload, "服务暂时不可用"),
+        );
+      }
+      if (!response.body) {
+        throw new ConsultationResponseError(502, "浏览器未收到可读取的回答流");
       }
-      if (!response.body) throw new Error("浏览器未收到可读取的回答流");
       const techniqueTruth = response.headers.get("x-jyotish-technique-truth") ?? "unknown";
       const workflowReceipt = {
         route: response.headers.get("x-jyotish-workflow-route") ?? "unknown",
@@ -2520,76 +2905,43 @@ export default function Home() {
 
       const completedSession: ChatSession = {
         ...userSession,
-        title: currentSession.messages.length === 0
-          ? resolveSessionTitle(question, reply.title)
-          : userSession.title,
+        title: userSession.title,
         messages: [...userSession.messages, { role: "assistant", text: reply.text, suggestions: reply.suggestions, techniqueTruth, workflowReceipt }],
         updatedAt: timestamp(),
       };
       updateSession(sessionId, () => completedSession);
       completeConsultationInterface(requestId);
-      try {
-        await persistSession(completedSession);
-      } catch (caught) {
-        void caught;
-        setComposerNotice("回答已保留；网络恢复后，下一次对话会继续同步完整记录。");
-      }
       void refreshAccount();
       return true;
     } catch (caught) {
       const cancelled = controller.signal.aborted;
       const ownsInterface = pendingConsultation.current?.requestId === requestId;
       const partialReply = latestPartialReply;
-      if (ownsInterface && !partialReply) {
-        updateSession(sessionId, () => currentSession);
-        setOnboardingJustCompleted(previousOnboardingState);
-        if (activeSessionIdRef.current === sessionId) {
-          setDraft(originalQuestion);
-          setDraftTheme(theme);
-          setDraftEntrypoint(entrypoint);
-        }
-        if (!cancelled) {
-          setRequestError({
-            sessionId,
-            message: `${caught instanceof Error ? caught.message : "服务暂时不可用，请稍后重试。"} 问题已放回输入框。`,
-          });
-          cancellationFeedbackRequest.current = requestId;
-          if (activeSessionIdRef.current === sessionId) {
-            setComposerNotice("问题已放回输入框，正在确认点数…");
-          }
-        }
+      if (!cancelled && ownsInterface && caught instanceof ConsultationResponseError) {
+        setRequestError({ sessionId, message: caught.message });
+        setComposerNotice(caught.message);
+        completeConsultationInterface(requestId);
+        return false;
       }
-      if (ownsInterface && !partialReply) {
-        await confirmCancellation(
-          requestId,
-          sessionId,
-          "问题已放回输入框，本次未扣点。",
-        );
-      } else if (!cancelled && ownsInterface) {
-        const interruptedSession: ChatSession = {
-          ...userSession,
-          messages: [...userSession.messages, { role: "assistant", text: partialReply }],
-          updatedAt: timestamp(),
+      if (!cancelled && ownsInterface && pendingConsultation.current) {
+        pendingConsultation.current = {
+          ...pendingConsultation.current,
+          phase: "recovering",
+          partialReply,
         };
-        updateSession(sessionId, () => interruptedSession);
-        try {
-          await persistSession(interruptedSession);
-          setRequestError({
-            sessionId,
-            message: "回答中途断开，已保留生成内容；请复制现有内容或继续追问，系统正在以账户记录为准同步点数。",
-          });
-        } catch (persistError) {
-          void persistError;
-          setComposerNotice("已保留当前回答；网络恢复后，下一次对话会继续同步完整记录。");
-        }
-        if (activeSessionIdRef.current === sessionId) {
-          setComposerNotice("回答中途断开，已保留现有内容；请继续追问或复制保存。");
-        }
+        setConsultationPhase("recovering");
+        setRequestError(null);
+        setComposerNotice(navigator.onLine
+          ? "连接中断，回答仍在后台生成，正在自动恢复。"
+          : "网络已断开，回答仍在后台生成；联网后会自动恢复。");
       }
       return Boolean(partialReply);
     } finally {
       cancellationRequests.current.delete(requestId);
-      completeConsultationInterface(requestId);
+      const pending = pendingConsultation.current;
+      if (pending?.requestId !== requestId || pending.phase !== "recovering") {
+        completeConsultationInterface(requestId);
+      }
       if (stoppedRequestAwaitingSettlement.current === requestId) {
         const persistence = stoppedSessionPersistence.current.get(requestId);
         if (persistence) {
@@ -2760,7 +3112,7 @@ export default function Home() {
           <div>
             <strong>{activeSession?.title || "新对话"}</strong>
             <span><i className={`status ${isLoading ? "status-loading" : "status-idle"}`} />{isLoading
-              ? (consultationPhase === "undo" ? "即将发送，可撤回" : activeStreamingText ? "正在回答" : "正在核对星盘信息")
+              ? (consultationPhase === "undo" ? "即将发送，可撤回" : consultationPhase === "recovering" ? "正在恢复后台回答" : activeStreamingText ? "正在回答" : "正在核对星盘信息")
               : rectificationSurfaceOpen || (!profileComplete && onboardingStep === "rectification")
                 ? "正在校正出生时间"
                 : personalChartAvailable ? "基于星盘证据回答" : "回答一般占星知识"}</span>
@@ -3020,8 +3372,8 @@ export default function Home() {
             {isLoading ? (
               <Button
                 className="composer-stop"
-                aria-label={consultationPhase === "undo" ? "撤回发送，本次不扣点" : activeStreamingText ? "停止回答，保留已生成内容" : "停止回答并申请退回本次点数"}
-                title={consultationPhase === "undo" ? "撤回发送，本次不扣点" : activeStreamingText ? "停止回答，本次已开始计费" : "停止回答"}
+                aria-label={consultationPhase === "undo" ? "撤回发送，本次不扣点" : activeStreamingText ? "停止回答，保留已生成内容并退回本次点数" : "停止回答并申请退回本次点数"}
+                title={consultationPhase === "undo" ? "撤回发送，本次不扣点" : activeStreamingText ? "停止回答，保留现有内容并申请退回本次点数" : "停止回答"}
                 size="icon"
                 type="button"
                 onClick={() => void stopResponse()}
diff --git a/frontend/src/lib/stream-text-response.ts b/frontend/src/lib/stream-text-response.ts
index 1010f246..8d8be84d 100644
--- a/frontend/src/lib/stream-text-response.ts
+++ b/frontend/src/lib/stream-text-response.ts
@@ -1,7 +1,7 @@
 type StreamHooks = {
   readonly onFirstOutput?: () => Promise<void>;
-  readonly onComplete?: () => Promise<void>;
-  readonly onError?: (error: unknown, emitted: boolean) => Promise<void>;
+  readonly onComplete?: (output: string) => Promise<void>;
+  readonly onError?: (error: unknown, emitted: boolean, output: string) => Promise<void>;
   readonly onCancel?: (emitted: boolean) => Promise<void>;
 };
 
@@ -10,6 +10,7 @@ type StreamTextResponseOptions = StreamHooks & {
   readonly requestId: string;
   readonly headers?: Record<string, string>;
   readonly transformText?: (text: string) => string;
+  readonly continueAfterDisconnect?: boolean;
 };
 
 const hiddenBlockOpeners = [
@@ -52,8 +53,9 @@ function createVisibleTextTransformer(transform: (text: string) => string) {
             rawBuffer = "";
             hidden = false;
           } else {
-            hiddenBuffer += rawBuffer;
-            rawBuffer = "";
+            const retainedLength = Math.min(2, rawBuffer.length);
+            hiddenBuffer += rawBuffer.slice(0, rawBuffer.length - retainedLength);
+            rawBuffer = rawBuffer.slice(rawBuffer.length - retainedLength);
           }
           break;
         }
@@ -149,16 +151,14 @@ export function streamTextResponse(
 ) {
   const iterator = stream[Symbol.asyncIterator]();
   const encoder = new TextEncoder();
-  // Keep a full natural-language clause unflushed so a later stream chunk cannot
-  // turn an allowed prefix into a disallowed timing or guaranteed conclusion.
-  const guardTailLength = options.transformText ? 1024 : 0;
   const visibleTransformer = options.transformText
     ? createVisibleTextTransformer(options.transformText)
     : null;
-  let pending = "";
   let settled = false;
   let cancellationStarted = false;
+  let disconnected = false;
   let emitted = false;
+  let fullOutput = "";
   let firstOutputSettlementStarted = false;
 
   function startFirstOutputSettlement(value: string) {
@@ -167,67 +167,70 @@ export function streamTextResponse(
     return options.onFirstOutput?.();
   }
 
-  async function enqueueOutput(
-    controller: ReadableStreamDefaultController<Uint8Array>,
+  async function output(
+    controller: ReadableStreamDefaultController<Uint8Array> | undefined,
     value: string,
   ) {
+    if (!value) return;
+    if (disconnected || !controller) {
+      fullOutput += value;
+      return;
+    }
     const firstOutputSettlement = startFirstOutputSettlement(value);
     controller.enqueue(encoder.encode(value));
+    fullOutput += value;
     if (/\S/.test(value)) emitted = true;
     await firstOutputSettlement;
   }
 
-  const body = new ReadableStream<Uint8Array>({
-    async pull(controller) {
-      try {
-        while (true) {
-          const next = await iterator.next();
+  async function consume(
+    controller: ReadableStreamDefaultController<Uint8Array> | undefined,
+  ) {
+    try {
+      while (true) {
+        const { done, value } = await iterator.next();
+        if (settled) return;
+        if (done) {
+          await output(controller, visibleTransformer ? visibleTransformer.finish("") : "");
           if (settled) return;
-          const { done, value } = next;
-          if (done) {
-            const finalText = visibleTransformer
-              ? visibleTransformer.finish(pending)
-              : pending;
-            if (finalText) {
-              await enqueueOutput(controller, finalText);
-            }
-            if (settled) return;
-            settled = true;
-            if (!emitted) {
-              const error = new Error("empty_stream");
-              await options.onError?.(error, false);
-              controller.error(error);
-              return;
-            }
-            await options.onComplete?.();
-            controller.close();
-            return;
-          }
-          pending += value;
-          if (pending.length <= guardTailLength) continue;
-
-          const stableLength = pending.length - guardTailLength;
-          const stable = pending.slice(0, stableLength);
-          pending = pending.slice(stableLength);
-          const transformed = visibleTransformer
-            ? visibleTransformer.push(stable)
-            : stable;
-          if (transformed) {
-            await enqueueOutput(controller, transformed);
-            return;
-          }
-        }
-      } catch (error) {
-        if (cancellationStarted) return;
-        if (!settled) {
           settled = true;
-          await options.onError?.(error, emitted);
+          if (!/\S/.test(fullOutput)) {
+            const error = new Error("empty_stream");
+            await options.onError?.(error, false, fullOutput);
+            if (!disconnected) controller?.error(error);
+            return;
+          }
+          await options.onComplete?.(fullOutput);
+          if (!disconnected) controller?.close();
+          return;
         }
-        controller.error(error);
+
+        const transformed = visibleTransformer
+          ? visibleTransformer.push(value)
+          : value;
+        await output(controller, transformed);
+        if (settled) return;
       }
+    } catch (error) {
+      if (cancellationStarted) return;
+      if (!settled) {
+        settled = true;
+        await options.onError?.(error, emitted, fullOutput);
+      }
+      if (!disconnected) controller?.error(error);
+    }
+  }
+
+  const body = new ReadableStream<Uint8Array>({
+    start(controller) {
+      void consume(controller).catch(() => {});
     },
     async cancel() {
       if (settled) return;
+      if (options.continueAfterDisconnect) {
+        disconnected = true;
+        return;
+      }
       settled = true;
       cancellationStarted = true;
       try {
diff --git a/frontend/supabase/migrations/20260808030000_consultation_stream_recovery.sql b/frontend/supabase/migrations/20260808030000_consultation_stream_recovery.sql
new file mode 100644
index 00000000..8d655d14
--- /dev/null
+++ b/frontend/supabase/migrations/20260808030000_consultation_stream_recovery.sql
@@ -0,0 +1,280 @@
+begin;
+
+alter table public.consultation_requests
+  add column if not exists session_id uuid references public.chat_sessions(id) on delete set null,
+  add column if not exists response_message jsonb;
+
+alter table public.consultation_requests
+  drop constraint if exists consultation_requests_response_message_check;
+alter table public.consultation_requests
+  add constraint consultation_requests_response_message_check
+  check (response_message is null or jsonb_typeof(response_message) = 'object');
+
+create index if not exists consultation_requests_user_session_created_idx
+  on public.consultation_requests (user_id, session_id, created_at desc);
+
+create index if not exists consultation_requests_user_status_created_idx
+  on public.consultation_requests (user_id, status, created_at desc);
+
+drop function if exists public.reserve_consultation_usage(uuid, text, text, integer);
+
+create or replace function public.reserve_consultation_usage(
+  p_user_id uuid,
+  p_request_id text,
+  p_session_id uuid,
+  p_requested_model_id text,
+  p_credit_cost integer
+)
+returns table(
+  success boolean,
+  reservation_id uuid,
+  source text,
+  credits integer,
+  subscription_id uuid,
+  reason text,
+  retry_after_seconds integer
+)
+language plpgsql
+security definer
+set search_path = ''
+as $$
+declare
+  v_authorization record;
+begin
+  if p_user_id is null or p_session_id is null or btrim(coalesce(p_request_id, '')) = '' then
+    return query select false, null::uuid, null::text, null::integer,
+      null::uuid, 'invalid_request'::text, null::integer;
+    return;
+  end if;
+
+  perform pg_advisory_xact_lock(hashtextextended(p_user_id::text || ':' || btrim(p_request_id), 0));
+
+  if not exists (
+    select 1
+    from public.chat_sessions as session
+    where session.id = p_session_id
+      and session.user_id = p_user_id
+      and session.session_type = 'consultation'
+  ) then
+    return query select false, null::uuid, null::text,
+      (select profile.credits from public.profiles as profile where profile.id = p_user_id),
+      null::uuid, 'session_missing'::text, null::integer;
+    return;
+  end if;
+
+  if exists (
+    select 1
+    from public.consultation_requests as request
+    where request.user_id = p_user_id
+      and request.request_id = btrim(p_request_id)
+  ) then
+    return query select false, null::uuid, null::text,
+      (select profile.credits from public.profiles as profile where profile.id = p_user_id),
+      null::uuid, 'request_conflict'::text, null::integer;
+    return;
+  end if;
+
+  select * into v_authorization
+  from public.authorize_usage(
+    p_user_id,
+    'chat.standard',
+    p_requested_model_id,
+    btrim(p_request_id),
+    p_credit_cost
+  );
+
+  if coalesce(v_authorization.success, false) then
+    insert into public.consultation_requests(user_id, request_id, session_id, status)
+    values (p_user_id, btrim(p_request_id), p_session_id, 'reserved');
+  end if;
+
+  return query select
+    v_authorization.success,
+    v_authorization.reservation_id,
+    v_authorization.source,
+    v_authorization.credits,
+    v_authorization.subscription_id,
+    v_authorization.reason,
+    v_authorization.retry_after_seconds;
+end;
+$$;
+
+revoke all on function public.reserve_consultation_usage(uuid, text, uuid, text, integer)
+  from public, anon, authenticated;
+grant execute on function public.reserve_consultation_usage(uuid, text, uuid, text, integer)
+  to service_role;
+do $$ begin
+  if exists(select 1 from pg_roles where rolname = 'admin_runtime') then
+    grant execute on function public.reserve_consultation_usage(uuid, text, uuid, text, integer)
+      to admin_runtime;
+  end if;
+end $$;
+
+create or replace function public.cancel_consultation_credit(p_user_id uuid, p_request_id text)
+returns table(success boolean, credits integer, error_code text)
+language plpgsql
+security definer
+set search_path = ''
+as $$
+declare
+  v_request public.consultation_requests%rowtype;
+  v_release record;
+  v_balance integer;
+begin
+  if p_user_id is null or btrim(coalesce(p_request_id, '')) = '' then
+    return query select false, null::integer, 'invalid_request'::text;
+    return;
+  end if;
+
+  perform pg_advisory_xact_lock(hashtextextended(p_user_id::text || ':' || btrim(p_request_id), 0));
+
+  select request.* into v_request
+  from public.consultation_requests as request
+  where request.user_id = p_user_id
+    and request.request_id = btrim(p_request_id)
+  for update;
+
+  if not found then
+    return query select false, null::integer, 'request_missing'::text;
+    return;
+  end if;
+
+  select profile.credits into v_balance
+  from public.profiles as profile
+  where profile.id = p_user_id;
+
+  if v_request.status = 'completed' then
+    return query select false, v_balance, 'request_completed'::text;
+    return;
+  end if;
+  if v_request.status = 'cancelled' then
+    return query select true, v_balance, null::text;
+    return;
+  end if;
+  if v_request.status <> 'reserved' then
+    return query select false, v_balance, 'invalid_request_status'::text;
+    return;
+  end if;
+
+  select * into v_release
+  from public.release_usage(p_user_id, btrim(p_request_id), 'consultation_cancelled');
+  if not coalesce(v_release.success, false) then
+    return query select false, v_release.credits, v_release.error_code;
+    return;
+  end if;
+
+  update public.consultation_requests
+  set status = 'cancelled', updated_at = clock_timestamp()
+  where user_id = p_user_id
+    and request_id = btrim(p_request_id)
+    and status = 'reserved';
+
+  return query select true, v_release.credits, null::text;
+end;
+$$;
+
+revoke all on function public.cancel_consultation_credit(uuid, text)
+  from public, anon, authenticated;
+grant execute on function public.cancel_consultation_credit(uuid, text)
+  to service_role;
+do $$ begin
+  if exists(select 1 from pg_roles where rolname = 'admin_runtime') then
+    grant execute on function public.cancel_consultation_credit(uuid, text)
+      to admin_runtime;
+  end if;
+end $$;
+
+create or replace function public.complete_consultation_response(
+  p_user_id uuid,
+  p_request_id text,
+  p_session_id uuid,
+  p_response_message jsonb,
+  p_actual_usage jsonb
+)
+returns table(success boolean, credits integer, error_code text)
+language plpgsql
+security definer
+set search_path = ''
+as $$
+declare
+  v_request public.consultation_requests%rowtype;
+  v_settlement record;
+  v_balance integer;
+begin
+  if p_user_id is null or p_session_id is null or btrim(coalesce(p_request_id, '')) = '' then
+    return query select false, null::integer, 'invalid_request'::text;
+    return;
+  end if;
+  if jsonb_typeof(p_response_message) <> 'object'
+    or p_response_message->>'role' <> 'assistant'
+    or btrim(coalesce(p_response_message->>'text', '')) = '' then
+    return query select false, null::integer, 'invalid_response_message'::text;
+    return;
+  end if;
+
+  perform pg_advisory_xact_lock(hashtextextended(p_user_id::text || ':' || btrim(p_request_id), 0));
+
+  select request.* into v_request
+  from public.consultation_requests as request
+  where request.user_id = p_user_id
+    and request.request_id = btrim(p_request_id)
+    and request.session_id = p_session_id
+  for update;
+
+  if not found then
+    return query select false, null::integer, 'request_missing'::text;
+    return;
+  end if;
+  if v_request.status = 'cancelled' then
+    return query select false, null::integer, 'request_cancelled'::text;
+    return;
+  end if;
+  if v_request.status = 'completed' then
+    select profile.credits into v_balance from public.profiles as profile where profile.id = p_user_id;
+    return query select v_request.response_message = p_response_message, v_balance,
+      case when v_request.response_message = p_response_message then null::text else 'response_conflict'::text end;
+    return;
+  end if;
+
+  update public.chat_sessions as session
+  set messages = session.messages || jsonb_build_array(p_response_message),
+      updated_at = clock_timestamp()
+  where session.id = p_session_id
+    and session.user_id = p_user_id
+    and session.session_type = 'consultation';
+  if not found then
+    return query select false, null::integer, 'session_missing'::text;
+    return;
+  end if;
+
+  update public.consultation_requests
+  set response_message = p_response_message,
+      updated_at = clock_timestamp()
+  where user_id = p_user_id and request_id = btrim(p_request_id);
+
+  select * into v_settlement
+  from public.complete_usage(p_user_id, btrim(p_request_id), coalesce(p_actual_usage, '{}'::jsonb));
+  if not coalesce(v_settlement.success, false) then
+    raise exception 'consultation_usage_settlement_failed:%', coalesce(v_settlement.error_code, 'unknown');
+  end if;
+
+  update public.consultation_requests
+  set status = 'completed', updated_at = clock_timestamp()
+  where user_id = p_user_id and request_id = btrim(p_request_id);
+
+  return query select true, v_settlement.credits, null::text;
+end;
+$$;
+
+revoke all on function public.complete_consultation_response(uuid, text, uuid, jsonb, jsonb)
+  from public, anon, authenticated;
+grant execute on function public.complete_consultation_response(uuid, text, uuid, jsonb, jsonb)
+  to service_role;
+do $$ begin
+  if exists(select 1 from pg_roles where rolname = 'admin_runtime') then
+    grant execute on function public.complete_consultation_response(uuid, text, uuid, jsonb, jsonb)
+      to admin_runtime;
+  end if;
+end $$;
+
+commit;
diff --git a/frontend/tests/application-billing-contract.test.ts b/frontend/tests/application-billing-contract.test.ts
index 56bbba02..014dc436 100644
--- a/frontend/tests/application-billing-contract.test.ts
+++ b/frontend/tests/application-billing-contract.test.ts
@@ -30,13 +30,13 @@ test("standard consultation resolves and settles the session-pinned model versio
   assert.match(consultRoute, /modelConfigVersion: selectedModel\.configVersion/);
 });
 
-test("standard consultation awaits real usage before its only permanent settlement", () => {
+test("standard consultation awaits real usage before durable response settlement", () => {
   assert.doesNotMatch(consultRoute, /recordActualUsage|void usage\.then/);
   assert.doesNotMatch(consultRoute, /inputTokens: 0,[\s\S]*outputTokens: 0,[\s\S]*costMicrousd: 0/);
-  assert.match(consultRoute, /async function complete\(usage: Promise<\{ inputTokens\?: number; outputTokens\?: number \}>\)/);
+  assert.match(consultRoute, /async function usagePayload\(usage: Promise<\{ inputTokens\?: number; outputTokens\?: number \}>\)/);
   assert.match(consultRoute, /const resolved = await usage;/);
-  assert.match(consultRoute, /await completeUsage\(accounting, userId, requestId, \{[\s\S]*inputTokens,[\s\S]*outputTokens,[\s\S]*costMicrousd/);
-  assert.match(consultRoute, /const completeWithUsage = \(\) => complete\(result\.totalUsage\)/g);
+  assert.match(consultRoute, /const actualUsage = await usagePayload\(usage\);[\s\S]*p_actual_usage: actualUsage/);
+  assert.equal(consultRoute.match(/result\.totalUsage/g)?.length, 4);
 });
 
 test("standard consultation forwards its stable reservation request as the usage event key", async () => {
@@ -57,7 +57,8 @@ test("standard consultation forwards its stable reservation request as the usage
   await completeUsage(accounting, "00000000-0000-4000-8000-000000000001", eventKey, usage);
 
   assert.deepEqual(calls.map((call) => (call.p_actual_usage as { eventKey: string }).eventKey), [eventKey, eventKey]);
-  assert.match(consultRoute, /completeUsage\(accounting, userId, requestId, \{[\s\S]*eventKey: requestId,/);
+  assert.match(consultRoute, /eventKey: requestId,/);
+  assert.match(consultRoute, /const actualUsage = await usagePayload\(usage\);[\s\S]*p_actual_usage: actualUsage/);
   assert.doesNotMatch(consultRoute, /eventKey:\s*(?:globalThis\.)?crypto\.randomUUID\(\)/);
 });
 
diff --git a/frontend/tests/consultation-recovery.test.ts b/frontend/tests/consultation-recovery.test.ts
new file mode 100644
index 00000000..0e5b1c02
--- /dev/null
+++ b/frontend/tests/consultation-recovery.test.ts
@@ -0,0 +1,120 @@
+import assert from "node:assert/strict";
+import { readFileSync } from "node:fs";
+import test from "node:test";
+
+const source = readFileSync(new URL("../src/app/page.tsx", import.meta.url), "utf8");
+const sendSource = source.slice(source.indexOf("  async function send("), source.indexOf("\n\n  useGSAP", source.indexOf("  async function send(")));
+const stopSource = source.slice(source.indexOf("  async function stopResponse("), source.indexOf("\n\n  function completeConsultationInterface"));
+
+test("consultation persists the optimistic user message before generation starts", () => {
+  const undo = sendSource.indexOf("await waitForUndoWindow(controller.signal)");
+  const persist = sendSource.indexOf("await persistSession(userSession)");
+  const consult = sendSource.indexOf('fetch("/api/consult"');
+
+  assert.ok(undo >= 0 && undo < persist && persist < consult);
+  assert.match(sendSource, /问题保存失败，未开始生成；问题已放回输入框。[\s\S]*?return false;[\s\S]*?fetch\("\/api\/consult"/);
+});
+
+test("only explicit stop can request consultation cancellation", () => {
+  const streamCatch = sendSource.indexOf("    } catch (caught) {", sendSource.indexOf('fetch("/api/consult"'));
+
+  assert.equal(source.match(/confirmCancellation\(/g)?.length, 2);
+  assert.match(stopSource, /await confirmCancellation\(/);
+  assert.ok(streamCatch >= 0);
+  assert.doesNotMatch(sendSource.slice(streamCatch), /confirmCancellation\(|\/api\/consult\/cancel/);
+});
+
+test("durable partial stop cancels before preserving content and recovers completed conflicts", () => {
+  const partialStop = stopSource.slice(
+    stopSource.indexOf("    if (pending.partialReply)"),
+    stopSource.indexOf("\n    updateSession(pending.sessionId, () => pending.previousSession)"),
+  );
+  const cancel = partialStop.indexOf("await requestCancellation(pending.requestId)");
+  const persist = partialStop.indexOf("await persistSession(stoppedSession)");
+  const conflict = partialStop.indexOf("error.status === 409");
+  const recoveryPersist = partialStop.slice(conflict, persist);
+
+  assert.ok(cancel >= 0 && cancel < persist);
+  assert.ok(conflict >= 0 && conflict < persist);
+  assert.doesNotMatch(recoveryPersist, /persistSession\(stoppedSession\)/);
+  assert.match(recoveryPersist, /fetchConsultationStatus\(pending.sessionId, pending.requestId\)/);
+  assert.match(recoveryPersist, /status.status === "completed"[\s\S]*?fetchSessions\(\)[\s\S]*?readSessions\(payload, modelCatalog\)/);
+  assert.match(partialStop, /已停止回答，现有内容已保留，本次点数已退回。/);
+  assert.match(source, /停止回答，保留已生成内容并退回本次点数/);
+  assert.match(source, /停止回答，保留现有内容并申请退回本次点数/);
+});
+
+test("explicit consultation HTTP failures unlock instead of entering recovery", () => {
+  assert.match(sendSource, /!response\.ok[\s\S]*throw new ConsultationResponseError\([\s\S]*response\.status/);
+  const explicitFailure = sendSource.slice(
+    sendSource.indexOf("caught instanceof ConsultationResponseError"),
+    sendSource.indexOf("if (!cancelled && ownsInterface && pendingConsultation.current)"),
+  );
+  assert.match(explicitFailure, /setRequestError\(\{ sessionId, message: caught\.message \}\)/);
+  assert.match(explicitFailure, /completeConsultationInterface\(requestId\)/);
+  assert.doesNotMatch(explicitFailure, /phase: "recovering"|setConsultationPhase\("recovering"\)/);
+});
+
+test("reserved consultations recover through the status endpoint", () => {
+  assert.match(source, /\/api\/consult\/status\?sessionId=\$\{encodeURIComponent\(sessionId\)\}&requestId=\$\{encodeURIComponent\(requestId\)\}/);
+  assert.match(source, /fetch\("\/api\/consult\/status", \{ signal, cache: "no-store" \}\)/);
+  assert.match(source, /reservedConsultation\?\.status === "reserved"[\s\S]*nextSessions\.find\(\(session\) => session\.id === reservedConsultation\.sessionId\)/);
+  assert.match(source, /status\?\.status !== "reserved"[\s\S]*sessions\.find\(\(item\) => item\.id === status\.sessionId\)/);
+  assert.match(source, /status: "reserved" \| "completed" \| "cancelled"/);
+  assert.match(source, /readonly responseMessage\?: unknown/);
+  assert.match(source, /phase: "recovering"/);
+  assert.match(source, /window\.setTimeout\(\(\) => void poll\(\), 1_750\)/);
+  assert.match(source, /status\.status === "completed"[\s\S]*?fetchSessions\(controller\.signal\)[\s\S]*?readSessions\(payload, modelCatalog\)/);
+  assert.match(source, /window\.addEventListener\("online", onOnline\)/);
+  assert.match(source, /window\.addEventListener\("pageshow", onPageShow\)/);
+  assert.match(source, /网络已断开，回答仍在后台生成；联网后会自动恢复。/);
+});
+
+test("three consecutive strict status misses unlock recovery while transient failures keep polling", () => {
+  assert.match(source, /class ConsultationStatusError extends Error[\s\S]*readonly status: number/);
+  assert.match(source, /throw new ConsultationStatusError\([\s\S]*response\.status/);
+  const recoveryEffect = source.slice(
+    source.indexOf('if (consultationPhase !== "recovering"'),
+    source.indexOf("}, [consultationPhase, modelCatalog, pendingRequestId, pendingSessionId])"),
+  );
+  assert.match(source, /const consultationStatusMissingCount = useRef\(0\)/);
+  assert.match(recoveryEffect, /status\.status === "reserved"[\s\S]*consultationStatusMissingCount\.current = 0/);
+  assert.match(recoveryEffect, /caught instanceof ConsultationStatusError && caught\.status === 404[\s\S]*consultationStatusMissingCount\.current \+= 1[\s\S]*consultationStatusMissingCount\.current >= 3/);
+  assert.match(recoveryEffect, /setPendingSessionId\(null\)[\s\S]*setPendingRequestId\(null\)[\s\S]*setConsultationPhase\(null\)/);
+  assert.match(recoveryEffect, /后台未找到本次咨询请求，已停止恢复，请重新发送。/);
+  assert.match(recoveryEffect, /consultationStatusMissingCount\.current = 0;[\s\S]*回答仍在后台生成，正在自动恢复。/);
+  assert.match(recoveryEffect, /consultationStatusMissingCount\.current > 0[\s\S]*window\.setTimeout\(\(\) => void poll\(\), 1_750\)[\s\S]*else \{[\s\S]*void poll\(\)/);
+});
+
+test("tab-local pending ids drive strict bootstrap recovery before the global fallback", () => {
+  const bootstrap = source.slice(
+    source.indexOf("let reservedConsultation: ConsultationStatus | null = null"),
+    source.indexOf("if (controller.signal.aborted) return;", source.indexOf("let reservedConsultation: ConsultationStatus | null = null")),
+  );
+  const storageSync = source.slice(
+    source.indexOf("if (!hydrated || uiPreview.current) return;", source.indexOf("}, []);")),
+    source.indexOf('if (consultationPhase !== "recovering"'),
+  );
+
+  assert.match(bootstrap, /sessionStorage\.getItem\(pendingConsultationStorageKey\)/);
+  assert.match(bootstrap, /uuidPattern\.test\(parsedPending\.sessionId\)[\s\S]*uuidPattern\.test\(parsedPending\.requestId\)[\s\S]*nextSessions\.some\(\(session\) => session\.id === parsedPending\.sessionId\)/);
+  assert.match(bootstrap, /else \{[\s\S]*sessionStorage\.removeItem\(pendingConsultationStorageKey\)[\s\S]*\} catch \{[\s\S]*sessionStorage\.removeItem\(pendingConsultationStorageKey\)/);
+  assert.ok(bootstrap.indexOf("if (storedPending)") < bootstrap.indexOf("fetchActiveConsultationStatus(controller.signal)"));
+  assert.match(bootstrap, /if \(storedPending\) \{[\s\S]*fetchConsultationStatus\([\s\S]*storedPending\.sessionId,[\s\S]*storedPending\.requestId,[\s\S]*\} else \{[\s\S]*fetchActiveConsultationStatus/);
+  assert.match(bootstrap, /status\.status === "reserved"[\s\S]*reservedConsultation = status;[\s\S]*else \{[\s\S]*sessionStorage\.removeItem\(pendingConsultationStorageKey\)/);
+  assert.match(bootstrap, /caught instanceof ConsultationStatusError && caught\.status === 404 \? 1 : 0/);
+  assert.match(storageSync, /pendingSessionId && pendingRequestId[\s\S]*sessionStorage\.setItem\(pendingConsultationStorageKey,[\s\S]*sessionId: pendingSessionId,[\s\S]*requestId: pendingRequestId/);
+  assert.match(storageSync, /else \{[\s\S]*sessionStorage\.removeItem\(pendingConsultationStorageKey\)/);
+});
+
+test("the first default consultation title is persisted with the user question", () => {
+  const userSessionBlock = sendSource.slice(
+    sendSource.indexOf("const userSession: ChatSession"),
+    sendSource.indexOf("const requestId = globalThis.crypto.randomUUID()"),
+  );
+  assert.match(userSessionBlock, /currentSession\.messages\.length === 0 && currentSession\.title === "新对话"[\s\S]*resolveSessionTitle\(question\)/);
+  assert.ok(sendSource.indexOf("await persistSession(userSession)") < sendSource.indexOf('fetch("/api/consult"'));
+  assert.doesNotMatch(sendSource, /persistSession\(completedSession\)/);
+  assert.match(sendSource, /const completedSession: ChatSession = \{[\s\S]*title: userSession\.title/);
+  assert.doesNotMatch(sendSource, /resolveSessionTitle\(question, reply\.title\)/);
+});
diff --git a/frontend/tests/consultation-stream-recovery.test.ts b/frontend/tests/consultation-stream-recovery.test.ts
new file mode 100644
index 00000000..f160f0d1
--- /dev/null
+++ b/frontend/tests/consultation-stream-recovery.test.ts
@@ -0,0 +1,98 @@
+import assert from "node:assert/strict";
+import { readFileSync } from "node:fs";
+import test from "node:test";
+
+const read = (path: string) => readFileSync(new URL(`../${path}`, import.meta.url), "utf8");
+const consultRoute = read("src/app/api/consult/route.ts");
+const statusRoute = read("src/app/api/consult/status/route.ts");
+const migration = read("supabase/migrations/20260808030000_consultation_stream_recovery.sql");
+
+test("reserves usage and binds the owned consultation session atomically", () => {
+  assert.match(migration, /add column if not exists session_id uuid references public\.chat_sessions\(id\) on delete set null/i);
+  assert.match(migration, /reserve_consultation_usage\([\s\S]*p_session_id uuid[\s\S]*from public\.chat_sessions as session[\s\S]*session\.id = p_session_id[\s\S]*session\.user_id = p_user_id[\s\S]*session\.session_type = 'consultation'/i);
+  assert.match(migration, /from public\.authorize_usage[\s\S]*insert into public\.consultation_requests\(user_id, request_id, session_id, status\)[\s\S]*values \(p_user_id, btrim\(p_request_id\), p_session_id, 'reserved'\)/i);
+  assert.match(migration, /reserve_consultation_usage\(uuid, text, uuid, text, integer\)/i);
+  assert.match(consultRoute, /rpc\("reserve_consultation_usage"[\s\S]*p_session_id: sessionId/);
+  assert.doesNotMatch(consultRoute, /from\("consultation_requests"\)[\s\S]*\.update\(\{ session_id: sessionId/);
+});
+
+test("persists transformed assistant metadata before atomically settling usage", () => {
+  assert.equal(consultRoute.match(/continueAfterDisconnect: true/g)?.length, 2);
+  assert.equal(consultRoute.match(/onComplete: \(rawTransformedText\) => settle\(\(\) => completeResponse\(/g)?.length, 2);
+  assert.match(consultRoute, /parseAgentReply\(rawTransformedText, consultationTheme\)/);
+  assert.match(consultRoute, /role: "assistant" as const,[\s\S]*suggestions: reply\.suggestions,[\s\S]*techniqueTruth,[\s\S]*workflowReceipt/);
+
+  const append = migration.indexOf("set messages = session.messages || jsonb_build_array(p_response_message)");
+  const store = migration.indexOf("set response_message = p_response_message");
+  const settle = migration.indexOf("from public.complete_usage");
+  const complete = migration.indexOf("set status = 'completed'");
+  assert.ok(append >= 0 && append < store && store < settle && settle < complete);
+});
+
+test("persists partial transformed output when the upstream stream errors", () => {
+  assert.doesNotMatch(consultRoute, /completeInterrupted|completeUsage\(accounting, userId, requestId/);
+  assert.equal(
+    consultRoute.match(/onError: \(_error, emitted, output: string\) => settleErrored\(emitted, output\)/g)?.length,
+    2,
+  );
+  assert.equal(consultRoute.match(/const settleErrored = \(emitted: boolean, output: string\) => settle\(/g)?.length, 2);
+  assert.equal(consultRoute.match(/emitted[\s\S]*?\? \(\) => completeResponse\([\s\S]*?output,[\s\S]*?result\.totalUsage,[\s\S]*?: cancel,/g)?.length, 2);
+  assert.equal(consultRoute.match(/onCancel: \(\) => settle\(cancel\)/g)?.length, 2);
+});
+
+test("best-effort cancels a failed or uncertain durable completion before rethrowing", () => {
+  const completion = consultRoute.slice(
+    consultRoute.indexOf("async function completeResponse("),
+    consultRoute.indexOf("let settlement:"),
+  );
+  assert.match(completion, /try \{[\s\S]*accounting\.rpc\("complete_consultation_response"/);
+  assert.match(completion, /catch \(error\) \{[\s\S]*await cancel\(\);[\s\S]*throw error;/);
+  assert.ok(completion.indexOf("await cancel();") > completion.indexOf('accounting.rpc("complete_consultation_response"'));
+});
+
+test("serializes explicit cancellation against response persistence and charging", () => {
+  const cancelFunction = migration.slice(
+    migration.indexOf("create or replace function public.cancel_consultation_credit"),
+    migration.indexOf("create or replace function public.complete_consultation_response"),
+  );
+  assert.match(cancelFunction, /pg_advisory_xact_lock\(hashtextextended\(p_user_id::text \|\| ':' \|\| btrim\(p_request_id\), 0\)\)/);
+  assert.match(cancelFunction, /from public\.consultation_requests[\s\S]*for update/);
+  assert.match(cancelFunction, /if v_request\.status = 'completed' then[\s\S]*'request_completed'/i);
+  assert.match(cancelFunction, /if v_request\.status = 'cancelled' then[\s\S]*select true/i);
+  assert.match(cancelFunction, /if v_request\.status <> 'reserved' then[\s\S]*from public\.release_usage[\s\S]*set status = 'cancelled'/i);
+  assert.match(cancelFunction, /grant execute on function public\.cancel_consultation_credit\(uuid, text\)[\s\S]*to service_role/);
+  assert.match(migration, /if v_request\.status = 'cancelled' then[\s\S]*'request_cancelled'/i);
+  assert.match(consultRoute, /cancel_consultation_credit/);
+  assert.match(consultRoute, /error_code !== "request_cancelled"/);
+});
+
+test("status endpoint supports one global reserved lookup and strict bound polling", () => {
+  assert.ok(statusRoute.indexOf("supabase.auth.getUser()") < statusRoute.indexOf("createAdminSupabaseClient()"));
+  assert.match(statusRoute, /activeLookup = sessionIdValue === null && requestIdValue === null/);
+  assert.match(statusRoute, /activeLookup[\s\S]*\.eq\("status", "reserved"\)\.order\("created_at", \{ ascending: false \}\)\.limit\(1\)/);
+  assert.match(statusRoute, /\.eq\("user_id", user\.id\)[\s\S]*\.eq\("session_id", sessionId!\.data\)[\s\S]*\.eq\("request_id", requestId!\.data\)/);
+  assert.match(statusRoute, /requestId: statusData\.request_id,[\s\S]*sessionId: statusData\.session_id,[\s\S]*status: statusData\.status/);
+  assert.doesNotMatch(statusRoute, /String\(data\.response_message\)|responseMessage:\s*data\.response_message as string/);
+});
+
+test("detached completion and cancellation use a bounded retry ceiling", () => {
+  assert.match(consultRoute, /const detachedSettlementAttempts = 3/);
+  assert.match(consultRoute, /ponytail: Staging MVP ceiling—without a queue\/worker/);
+  assert.match(consultRoute, /for \(let attempt = 1; attempt <= detachedSettlementAttempts; attempt \+= 1\)/);
+  assert.match(consultRoute, /setTimeout\(resolve, attempt \* 150\)/);
+  assert.match(consultRoute, /async function cancel\(\)[\s\S]*retryDetachedSettlement\(async \(\) =>[\s\S]*rpc\("cancel_consultation_credit"/);
+  assert.match(consultRoute, /async function completeResponse\([\s\S]*retryDetachedSettlement\(async \(\) =>[\s\S]*rpc\("complete_consultation_response"/);
+});
+
+test("status expires stale reservations after fifteen minutes but leaves fresh reservations active", () => {
+  assert.match(statusRoute, /const reservedLeaseMs = 15 \* 60 \* 1000/);
+  assert.match(statusRoute, /ponytail: Staging MVP ceiling—without a queue\/worker/);
+  assert.match(statusRoute, /Number\.isFinite\(timestamp\) && now - timestamp >= reservedLeaseMs/);
+  const leaseBlock = statusRoute.slice(
+    statusRoute.indexOf('if (statusData.status === "reserved" && reservationLeaseExpired(statusData.updated_at))'),
+    statusRoute.indexOf("return NextResponse.json({\n    requestId: statusData.request_id"),
+  );
+  assert.match(leaseBlock, /runCreditRpc\([\s\S]*"cancel_consultation_credit"/);
+  assert.match(leaseBlock, /\.eq\("session_id", statusData\.session_id\)[\s\S]*\.eq\("request_id", statusData\.request_id\)/);
+  assert.doesNotMatch(statusRoute.slice(0, statusRoute.indexOf('if (statusData.status === "reserved"')), /cancel_consultation_credit/);
+});
diff --git a/frontend/tests/stream-text-response.test.ts b/frontend/tests/stream-text-response.test.ts
index a9fcf28f..7e110125 100644
--- a/frontend/tests/stream-text-response.test.ts
+++ b/frontend/tests/stream-text-response.test.ts
@@ -26,6 +26,47 @@ test("durable settlement starts before the first response bytes are exposed", as
   assert.deepEqual(order, ["settled", "第一段", "completed"]);
 });
 
+test("transformed output emits at the first complete clause without reading ahead", async () => {
+  let markSecondReadStarted = () => {};
+  const secondReadStarted = new Promise<"read-ahead">((resolve) => {
+    markSecondReadStarted = () => resolve("read-ahead");
+  });
+  const never = new Promise<IteratorResult<string>>(() => {});
+  let reads = 0;
+  const reply: AsyncIterable<string> = {
+    [Symbol.asyncIterator]() {
+      return {
+        next() {
+          reads += 1;
+          if (reads === 1) {
+            return Promise.resolve({ done: false, value: "第一句。" });
+          }
+          markSecondReadStarted();
+          return never;
+        },
+        return() {
+          return Promise.resolve({ done: true, value: undefined });
+        },
+      };
+    },
+  };
+  const response = streamTextResponse(reply, {
+    mode: "mastra",
+    requestId: "00000000-0000-4000-8000-000000000104",
+    transformText: (text) => text,
+  });
+  const reader = response.body?.getReader();
+  assert.ok(reader);
+
+  const first = await Promise.race([reader.read(), secondReadStarted]);
+
+  assert.notEqual(first, "read-ahead");
+  if (first === "read-ahead") assert.fail("read-ahead");
+  assert.equal(first.done, false);
+  assert.equal(new TextDecoder().decode(first.value), "第一句。");
+  await reader.cancel();
+});
+
 test("cancelling while first-output settlement is pending observes committed output", async () => {
   let markSettlementStarted = () => {};
   const settlementStarted = new Promise<void>((resolve) => {
@@ -215,6 +256,192 @@ test("charges a consultation when cancellation happens after partial output", as
   assert.equal(completed, 1);
 });
 
+test("opt-in disconnect keeps consuming and completes with the full transformed output", async () => {
+  let releaseSecondChunk = () => {};
+  const secondChunk = new Promise<void>((resolve) => {
+    releaseSecondChunk = resolve;
+  });
+  let reads = 0;
+  let returnCalls = 0;
+  const reply: AsyncIterable<string> = {
+    [Symbol.asyncIterator]() {
+      return {
+        async next() {
+          reads += 1;
+          if (reads === 1) return { done: false, value: "第一段回答。".repeat(200) };
+          if (reads === 2) {
+            await secondChunk;
+            return { done: false, value: "第二段回答。" };
+          }
+          return { done: true, value: undefined };
+        },
+        return() {
+          returnCalls += 1;
+          return Promise.resolve({ done: true, value: undefined });
+        },
+      };
+    },
+  };
+  let cancelCalls = 0;
+  let completedOutput = "";
+  let markCompleted = () => {};
+  const completed = new Promise<"completed">((resolve) => {
+    markCompleted = () => resolve("completed");
+  });
+  let markCancelled = () => {};
+  const cancelled = new Promise<"cancelled">((resolve) => {
+    markCancelled = () => resolve("cancelled");
+  });
+  const response = streamTextResponse(reply, {
+    mode: "mastra",
+    requestId: "00000000-0000-4000-8000-000000000105",
+    transformText: (text) => text.replaceAll("回答", "安全回答"),
+    continueAfterDisconnect: true,
+    onCancel: async () => {
+      cancelCalls += 1;
+      markCancelled();
+    },
+    onComplete: async (output) => {
+      completedOutput = output;
+      markCompleted();
+    },
+  });
+  const reader = response.body?.getReader();
+  assert.ok(reader);
+  const first = await reader.read();
+  assert.equal(first.done, false);
+
+  await reader.cancel();
+  releaseSecondChunk();
+  assert.equal(await Promise.race([completed, cancelled]), "completed");
+
+  assert.equal(returnCalls, 0);
+  assert.equal(cancelCalls, 0);
+  assert.equal(
+    completedOutput,
+    `${"第一段安全回答。".repeat(200)}第二段安全回答。`,
+  );
+});
+
+test("starts draining without a reader and completes with the full transformed output", { timeout: 1_000 }, async () => {
+  let markFirstReadStarted = () => {};
+  const firstReadStarted = new Promise<void>((resolve) => {
+    markFirstReadStarted = resolve;
+  });
+  let releaseSecondChunk = () => {};
+  const secondChunk = new Promise<void>((resolve) => {
+    releaseSecondChunk = resolve;
+  });
+  let reads = 0;
+  const reply: AsyncIterable<string> = {
+    [Symbol.asyncIterator]() {
+      return {
+        async next() {
+          reads += 1;
+          if (reads === 1) {
+            markFirstReadStarted();
+            return { done: false, value: "第一段回答。" };
+          }
+          if (reads === 2) {
+            await secondChunk;
+            return { done: false, value: "第二段回答。" };
+          }
+          return { done: true, value: undefined };
+        },
+      };
+    },
+  };
+  let completedOutput = "";
+  let markCompleted = () => {};
+  const completed = new Promise<void>((resolve) => {
+    markCompleted = resolve;
+  });
+
+  const response = streamTextResponse(reply, {
+    mode: "mastra",
+    requestId: "00000000-0000-4000-8000-000000000107",
+    transformText: (text) => text.replaceAll("回答", "安全回答"),
+    continueAfterDisconnect: true,
+    onComplete: async (output) => {
+      completedOutput = output;
+      markCompleted();
+    },
+  });
+  assert.ok(response.body);
+  await firstReadStarted;
+
+  releaseSecondChunk();
+  await completed;
+
+  assert.equal(reads, 3);
+  assert.equal(completedOutput, "第一段安全回答。第二段安全回答。");
+});
+
+test("an early opt-in cancel keeps the started producer draining to completion", { timeout: 1_000 }, async () => {
+  let markFirstReadStarted = () => {};
+  const firstReadStarted = new Promise<void>((resolve) => {
+    markFirstReadStarted = resolve;
+  });
+  let releaseFirstChunk = () => {};
+  const firstChunk = new Promise<void>((resolve) => {
+    releaseFirstChunk = resolve;
+  });
+  let reads = 0;
+  let returnCalls = 0;
+  const reply: AsyncIterable<string> = {
+    [Symbol.asyncIterator]() {
+      return {
+        async next() {
+          reads += 1;
+          if (reads === 1) {
+            markFirstReadStarted();
+            await firstChunk;
+            return { done: false, value: "第一段回答。" };
+          }
+          if (reads === 2) return { done: false, value: "第二段回答。" };
+          return { done: true, value: undefined };
+        },
+        return() {
+          returnCalls += 1;
+          return Promise.resolve({ done: true, value: undefined });
+        },
+      };
+    },
+  };
+  let cancelCalls = 0;
+  let completeCalls = 0;
+  let completedOutput = "";
+  let markCompleted = () => {};
+  const completed = new Promise<void>((resolve) => {
+    markCompleted = resolve;
+  });
+  const response = streamTextResponse(reply, {
+    mode: "mastra",
+    requestId: "00000000-0000-4000-8000-000000000108",
+    transformText: (text) => text.replaceAll("回答", "安全回答"),
+    continueAfterDisconnect: true,
+    onCancel: async () => { cancelCalls += 1; },
+    onComplete: async (output) => {
+      completeCalls += 1;
+      completedOutput = output;
+      markCompleted();
+    },
+  });
+  const reader = response.body?.getReader();
+  assert.ok(reader);
+  await firstReadStarted;
+
+  await reader.cancel();
+  releaseFirstChunk();
+  await completed;
+
+  assert.equal(reads, 3);
+  assert.equal(returnCalls, 0);
+  assert.equal(cancelCalls, 0);
+  assert.equal(completeCalls, 1);
+  assert.equal(completedOutput, "第一段安全回答。第二段安全回答。");
+});
+
 test("refunds when cancellation happens before any output", async () => {
   // Given
   let completed = 0;
@@ -342,6 +569,27 @@ test("a transformed short reply that fails while buffered reports no emitted out
   assert.equal(observedEmitted, false);
 });
 
+test("an iterator error after partial output reports the full transformed output", async () => {
+  let observedOutput = "";
+  async function* reply() {
+    yield "第一段回答。";
+    yield "第二段回答。";
+    throw new Error("upstream_failed_after_output");
+  }
+  const response = streamTextResponse(reply(), {
+    mode: "mastra",
+    requestId: "00000000-0000-4000-8000-000000000106",
+    transformText: (text) => text.replaceAll("回答", "安全回答"),
+    onError: async (_error, emitted, output) => {
+      assert.equal(emitted, true);
+      observedOutput = output;
+    },
+  });
+
+  await assert.rejects(response.text(), /upstream_failed_after_output/);
+  assert.equal(observedOutput, "第一段安全回答。第二段安全回答。");
+});
+
 test("cancelling while transformed short output is buffered reports no emitted output", async () => {
   let markSecondReadStarted = () => {};
   const secondReadStarted = new Promise<void>((resolve) => {