feat: add safe consultation cancellation

2026-07-17 11:57:28 +08:00
parent 95efa12b11
commit 2e193f594e
17 changed files with 1225 additions and 229 deletions
@@ -0,0 +1,50 @@
+export type ReplyTheme = "career" | "marriage" | "timing" | "general";
+
+const fallbackSuggestions: Record<ReplyTheme, readonly [string, string, string]> = {
+  career: ["我更适合怎样的职业路径？", "未来一年事业上要避开什么？", "我该如何发挥自己的优势？"],
+  marriage: ["我在关系里容易重复什么模式？", "怎样的伴侣更适合我？", "未来一年关系上要注意什么？"],
+  timing: ["接下来最值得把握的阶段是什么？", "哪些时期更适合主动行动？", "我现在应该优先准备什么？"],
+  general: ["未来一年，事业和收入该关注什么？", "我的关系模式是什么？", "未来哪些阶段值得把握？"],
+};
+
+function readSuggestions(value: unknown): string[] {
+  if (!Array.isArray(value)) return [];
+  return [...new Set(value
+    .filter((item): item is string => typeof item === "string")
+    .map((item) => item.replace(/\s+/g, " ").trim().slice(0, 80))
+    .filter(Boolean))].slice(0, 3);
+}
+
+function readTitle(value: string): string | undefined {
+  const title = value.replace(/\s+/g, " ").trim();
+  if (!title || /[\d\p{P}\p{S}]/u.test(title)) return undefined;
+  if (/\p{Script=Han}/u.test(title)) {
+    const length = Array.from(title.replace(/\s/g, "")).length;
+    return length >= 6 && length <= 14 ? title : undefined;
+  }
+  const words = title.split(" ").filter(Boolean);
+  return words.length >= 3 && words.length <= 7 && title.length <= 64 ? title : undefined;
+}
+
+export function parseAgentReply(value: string, theme: ReplyTheme) {
+  let suggestions: string[] = [];
+  let title: string | undefined;
+  const withoutSuggestions = value.replace(/<!--AYANAM_SUGGESTIONS:(\[[\s\S]*?\])-->/g, (_, json: string) => {
+    try {
+      suggestions = readSuggestions(JSON.parse(json));
+    } catch {
+      suggestions = [];
+    }
+    return "";
+  });
+  const text = withoutSuggestions.replace(/<!--AYANAM_TITLE:([\s\S]*?)-->/g, (_, rawTitle: string) => {
+    title = readTitle(rawTitle);
+    return "";
+  }).replace(/<!--AYANAM_[\s\S]*$/, "").trim();
+
+  return {
+    text,
+    suggestions: suggestions.length === 3 ? suggestions : [...fallbackSuggestions[theme]],
+    title,
+  };
+}
@@ -0,0 +1,57 @@
+import { z } from "zod";
+
+const creditResultSchema = z.object({
+  success: z.boolean(),
+  credits: z.number().int().nullable(),
+  error_code: z.string().nullable().optional(),
+});
+
+type CreditRpcName = "begin_consultation_credit" | "complete_consultation_credit" | "cancel_consultation_credit";
+type AccountingClient = {
+  rpc(
+    rpcName: CreditRpcName,
+    args: { p_user_id: string; p_request_id: string },
+  ): PromiseLike<{ data: unknown; error: { message: string } | null }>;
+};
+
+export type CreditResult = z.infer<typeof creditResultSchema>;
+
+export class CreditRpcError extends Error {
+  readonly code: string;
+
+  constructor(code: string) {
+    super(`Credit operation failed: ${code}`);
+    this.name = "CreditRpcError";
+    this.code = code;
+  }
+}
+
+export async function runCreditRpc(
+  accounting: AccountingClient,
+  rpcName: CreditRpcName,
+  userId: string,
+  requestId: string,
+): Promise<CreditResult> {
+  let lastError = "unknown_credit_error";
+
+  for (let attempt = 1; attempt <= 3; attempt += 1) {
+    try {
+      const { data, error } = await accounting.rpc(rpcName, {
+        p_user_id: userId,
+        p_request_id: requestId,
+      });
+      const candidate = Array.isArray(data) ? data[0] : data;
+      const parsed = creditResultSchema.safeParse(candidate);
+      if (!error && parsed.success) return parsed.data;
+      lastError = error?.message || "invalid_credit_response";
+    } catch (error) {
+      lastError = error instanceof Error ? error.message : "credit_request_failed";
+    }
+
+    if (attempt < 3) {
+      await new Promise((resolve) => setTimeout(resolve, attempt * 150));
+    }
+  }
+
+  throw new CreditRpcError(lastError);
+}
@@ -0,0 +1,67 @@
+type StreamHooks = {
+  readonly onComplete?: () => Promise<void>;
+  readonly onError?: (error: unknown, emitted: boolean) => Promise<void>;
+  readonly onCancel?: (emitted: boolean) => Promise<void>;
+};
+
+type StreamTextResponseOptions = StreamHooks & {
+  readonly mode: "engine" | "mastra";
+  readonly requestId: string;
+};
+
+export function streamTextResponse(
+  stream: AsyncIterable<string>,
+  options: StreamTextResponseOptions,
+) {
+  const iterator = stream[Symbol.asyncIterator]();
+  const encoder = new TextEncoder();
+  let settled = false;
+  let emitted = false;
+
+  const body = new ReadableStream<Uint8Array>({
+    async pull(controller) {
+      try {
+        const { done, value } = await iterator.next();
+        if (done) {
+          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;
+        }
+        if (/\S/.test(value)) emitted = true;
+        controller.enqueue(encoder.encode(value));
+      } catch (error) {
+        if (!settled) {
+          settled = true;
+          await options.onError?.(error, emitted);
+        }
+        controller.error(error);
+      }
+    },
+    async cancel() {
+      if (settled) return;
+      settled = true;
+      try {
+        await iterator.return?.();
+      } finally {
+        await options.onCancel?.(emitted);
+      }
+    },
+  });
+
+  return new Response(body, {
+    headers: {
+      "cache-control": "no-cache, no-transform",
+      "content-type": "text/plain; charset=utf-8",
+      "x-accel-buffering": "no",
+      "x-ayanam-mode": options.mode,
+      "x-ayanam-request-id": options.requestId,
+    },
+  });
+}